jsondiff.py

import json
from pprint import pprint
import re
from typing import Dict, List, Any, Union
from deepdiff import DeepDiff


class JSONDiff:
    """A class for comparing JSON objects and generating structured diffs."""
    
    def __init__(self):
        pass
    
    def compare(self, json1: Union[str, Dict], json2: Union[str, Dict]) -> List[Dict[str, Any]]:
        """
        Compare two JSON objects and return a list of differences.
        
        Args:
            json1: First JSON object (string or dict)
            json2: Second JSON object (string or dict)
            
        Returns:
            List of diff objects with structure similar to Go version
        """
        # Parse JSON strings if needed
        if isinstance(json1, str):
            json1 = json.loads(json1)
        if isinstance(json2, str):
            json2 = json.loads(json2)
        
        # Use DeepDiff to find differences - treat arrays as atomic units
        diff_result = DeepDiff(json1, json2, ignore_order=False, view='tree')
        pprint(f"diff_result {diff_result}")
        
        diffs = []
        
        # Process different types of changes
        self._process_changes(diff_result, diffs)
        
        return diffs
    
    def _process_changes(self, diff_result: DeepDiff, diffs: List[Dict]):
        """Process all types of changes uniformly."""
        
        # Handle dictionary additions and removals
        change_mappings = [
            ('dictionary_item_added', 'added', lambda c: (None, c.t2, c.t2)),
            ('dictionary_item_removed', 'removed', lambda c: (c.t1, None, c.t1))
        ]
        
        for diff_key, change_type, value_func in change_mappings:
            if diff_key in diff_result:
                for change in diff_result[diff_key]:
                    path = self._convert_path_to_slash(change.path())
                    field_name = self._get_field_name_from_path(path)
                    from_val, to_val, type_val = value_func(change)
                    
                    diffs.append({
                        'field_name': field_name,
                        'change_type': change_type,
                        'from_value': self._serialize_value(from_val) if from_val is not None else None,
                        'to_value': self._serialize_value(to_val) if to_val is not None else None,
                        'full_path': path,
                        'value_type': self._get_value_type(type_val)
                    })
        
        # Handle type changes (currently disabled, can be enabled if needed)
        # if 'type_changes' in diff_result:
        #     for change in diff_result['type_changes']:
        #         path = self._convert_path_to_slash(change.path())
        #         field_name = self._get_field_name_from_path(path)
        #         
        #         diffs.append({
        #             'field_name': field_name,
        #             'change_type': 'modified',
        #             'from_value': self._serialize_value(change.t1),
        #             'to_value': self._serialize_value(change.t2),
        #             'full_path': path,
        #             'value_type': self._get_value_type(change.t2)
        #         })
        
        # Handle value changes - but group array element changes
        if 'values_changed' in diff_result:
            array_paths = {}
            non_array_changes = []
            
            for change in diff_result['values_changed']:
                path = change.path()
                
                # Check if this is an array element change (path ends with [number])
                if re.search(r'\[\d+\]$', path):
                    # Group by parent array
                    parent_path = self._get_array_parent_path(path)
                    if parent_path not in array_paths:
                        array_paths[parent_path] = True
                else:
                    non_array_changes.append(change)
            
            # Process non-array changes normally
            for change in non_array_changes:
                path = self._convert_path_to_slash(change.path())
                field_name = self._get_field_name_from_path(path)
                
                diffs.append({
                    'field_name': field_name,
                    'change_type': 'modified',
                    'from_value': self._serialize_value(change.t1),
                    'to_value': self._serialize_value(change.t2),
                    'full_path': path,
                    'value_type': self._get_value_type(change.t2)
                })
            
            # Process array changes as complete units
            for array_path in array_paths:
                path = self._convert_path_to_slash(array_path)
                field_name = self._get_field_name_from_path(path)
                
                # Get the full arrays
                old_array = self._get_value_at_path(diff_result.t1, array_path)
                new_array = self._get_value_at_path(diff_result.t2, array_path)
                
                diffs.append({
                    'field_name': field_name,
                    'change_type': 'modified',
                    'from_value': self._serialize_value(old_array),
                    'to_value': self._serialize_value(new_array),
                    'full_path': path,
                    'value_type': 'array'
                })
        
        # Handle array changes as complete unit changes
        if 'iterable_item_added' in diff_result or 'iterable_item_removed' in diff_result:
            # Group all array changes by their parent array
            array_paths = set()
            
            if 'iterable_item_added' in diff_result:
                for change in diff_result['iterable_item_added']:
                    array_path = self._get_array_parent_path(change.path())
                    array_paths.add(array_path)
            
            if 'iterable_item_removed' in diff_result:
                for change in diff_result['iterable_item_removed']:
                    array_path = self._get_array_parent_path(change.path())
                    array_paths.add(array_path)
            
            # Create diffs for each changed array
            for array_path in array_paths:
                path = self._convert_path_to_slash(array_path)
                field_name = self._get_field_name_from_path(path)
                
                # Get the full arrays
                old_array = self._get_value_at_path(diff_result.t1, array_path)
                new_array = self._get_value_at_path(diff_result.t2, array_path)
                
                diffs.append({
                    'field_name': field_name,
                    'change_type': 'modified',
                    'from_value': self._serialize_value(old_array),
                    'to_value': self._serialize_value(new_array),
                    'full_path': path,
                    'value_type': 'array'
                })
    
    def _convert_path_to_slash(self, path: str) -> str:
        """Convert root['key'][0] to /key/0 format."""
        if not path or path == "root":
            return ""
        
        # Extract all parts from brackets
        matches = re.findall(r'\[([^\]]+)\]', path)
        
        # Clean and join with slashes
        parts = []
        for part in matches:
            # Remove quotes
            if part.startswith("'") and part.endswith("'"):
                part = part[1:-1]
            elif part.startswith('"') and part.endswith('"'):
                part = part[1:-1]
            parts.append(part)
        
        return "/" + "/".join(parts) if parts else ""
    
    def _get_field_name_from_path(self, path: str) -> str:
        """Extract field name from slash path."""
        if not path or path == "/":
            return "root"
        
        parts = path.strip('/').split('/')
        return parts[-1] if parts else "root"
    
    def _get_array_parent_path(self, path: str) -> str:
        """Get the parent array path from an array element path."""
        # For root['arr'][0] -> root['arr']
        # For root['fruits']['arr'][1] -> root['fruits']['arr']
        # For root['matrix'][0][1][2] -> root['matrix'] (only the first array level)
        
        # Find the topmost array in the hierarchy
        # This groups all nested array changes under the root array field
        matches = re.findall(r'root(?:\[[^\]]+\])*', path)
        if matches:
            base = matches[0]
            # Find first numeric index and remove everything from there
            first_numeric = re.search(r'\[\d+\]', base)
            if first_numeric:
                return base[:first_numeric.start()]
        return path
    
    def _get_value_at_path(self, obj: Any, path: str) -> Any:
        """Get value at the specified path."""
        if not path or path == "root":
            return obj
        
        try:
            current = obj
            matches = re.findall(r'\[([^\]]+)\]', path)
            
            for part in matches:
                # Remove quotes
                if part.startswith("'") and part.endswith("'"):
                    part = part[1:-1]
                elif part.startswith('"') and part.endswith('"'):
                    part = part[1:-1]
                
                if part.isdigit():
                    current = current[int(part)]
                else:
                    current = current[part]
            
            return current
        except (KeyError, IndexError, TypeError):
            return None
    
    def _serialize_value(self, value: Any) -> str | None:
        """Serialize a value to string representation."""
        if value is None:
            return 'null'
        if isinstance(value, str):
            return value
        if isinstance(value, (int, float, bool)):
            return json.dumps(value)  # This preserves the original type when parsing
        return json.dumps(value)
    
    def _get_value_type(self, value: Any) -> str:
        """Get the type name for a value."""
        if isinstance(value, dict):
            return "object"
        elif isinstance(value, list):
            return "array"
        elif isinstance(value, bool):  # Check bool before int since bool is subclass of int
            return "boolean"
        elif isinstance(value, str):
            return "string"
        elif isinstance(value, int):
            return "number"
        elif isinstance(value, float):
            return "number"
        else:
            return "unknown"


class JSONReconstructor:
    """A class for reconstructing original JSON from current JSON and diffs."""
    
    def __init__(self):
        pass
    
    def reverse_diff(self, current_json: Union[str, Dict], diffs: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        Reconstruct the original JSON from current JSON and diffs.
        
        Args:
            current_json: The current JSON object
            diffs: List of differences to reverse
            
        Returns:
            The reconstructed original JSON object
        """
        if isinstance(current_json, str):
            current_json = json.loads(current_json)
        
        # Create a deep copy to avoid modifying the original
        original = json.loads(json.dumps(current_json))
        
        # Process diffs in reverse order to reconstruct
        for diff in reversed(diffs):
            self._apply_reverse_diff(original, diff)
        
        return original
    
    def _apply_reverse_diff(self, obj: Dict[str, Any] | List[Any], diff: Dict[str, Any]):
        """Apply a single reverse diff to reconstruct original state."""
        path = diff['full_path']
        change_type = diff['change_type']
        
        if change_type == 'added':
            # If something was added, remove it to get original
            self._remove_at_path(obj, path)
        elif change_type == 'removed':
            # If something was removed, add it back
            from_value = diff['from_value']
            if from_value is not None:
                try:
                    value = json.loads(from_value)
                except (json.JSONDecodeError, TypeError):
                    value = from_value
                self._set_at_path(obj, path, value)
        elif change_type == 'modified':
            # If something was modified, set it back to original value
            from_value = diff['from_value']
            if from_value is not None:
                try:
                    value = json.loads(from_value)
                except (json.JSONDecodeError, TypeError):
                    value = from_value
                self._set_at_path(obj, path, value)
    
    def _set_at_path(self, obj: Dict[str, Any] | List[Any], path: str, value: Any):
        """Set a value at a specific slash path."""
        if not path:
            return
        
        parts = path.strip('/').split('/')
        if not parts:
            return
        
        current = obj
        
        # Navigate to parent
        for part in parts[:-1]:
            if part.isdigit():
                if isinstance(current, list):
                    current = current[int(part)]
                else:
                    current = current[part]  # type: ignore
            else:
                if isinstance(current, dict):
                    if part not in current:
                        current[part] = {}
                    current = current[part]
                else:
                    return  # Can't navigate further
        
        # Set the final value
        final_part = parts[-1]
        if final_part.isdigit():
            if isinstance(current, list):
                idx = int(final_part)
                if idx < len(current):
                    current[idx] = value
            else:
                current[final_part] = value  # type: ignore
        else:
            if isinstance(current, dict):
                current[final_part] = value
    
    def _remove_at_path(self, obj: Dict[str, Any] | List[Any], path: str):
        """Remove a value at a specific slash path."""
        if not path:
            return
        
        parts = path.strip('/').split('/')
        if not parts:
            return
        
        current = obj
        
        # Navigate to parent
        for part in parts[:-1]:
            if part.isdigit():
                if isinstance(current, list):
                    current = current[int(part)]
                else:
                    current = current[part]  # type: ignore
            else:
                if isinstance(current, dict):
                    current = current[part]
                else:
                    return  # Can't navigate further
        
        # Remove the final value
        final_part = parts[-1]
        if final_part.isdigit():
            if isinstance(current, list) and int(final_part) < len(current):
                current.pop(int(final_part))
        else:
            if isinstance(current, dict) and final_part in current:
                del current[final_part]