thrillwiki_django_no_react/history_tracking/utils.py

from typing import Dict, Any, List, Optional
from django.core.exceptions import ValidationError
from .models import VersionBranch, ChangeSet
from django.utils import timezone
from django.contrib.auth import get_user_model

User = get_user_model()

def resolve_conflicts(
    source_branch: VersionBranch,
    target_branch: VersionBranch,
    resolutions: Dict[str, str],
    manual_resolutions: Dict[str, str],
    user: Optional[User] = None
) -> ChangeSet:
    """
    Resolve merge conflicts between branches
    
    Args:
        source_branch: Source branch of the merge
        target_branch: Target branch of the merge
        resolutions: Dict mapping conflict IDs to resolution type ('source', 'target', 'manual')
        manual_resolutions: Dict mapping conflict IDs to manual resolution content
        user: User performing the resolution
    
    Returns:
        ChangeSet: The changeset recording the conflict resolution
    """
    if not resolutions:
        raise ValidationError("No resolutions provided")
    
    resolved_content = {}
    
    for conflict_id, resolution_type in resolutions.items():
        source_id, target_id = conflict_id.split('_')
        source_change = ChangeSet.objects.get(pk=source_id)
        target_change = ChangeSet.objects.get(pk=target_id)
        
        if resolution_type == 'source':
            # Use source branch version
            for record in source_change.historical_records.all():
                resolved_content[f"{record.instance_type}_{record.instance_pk}"] = record
        
        elif resolution_type == 'target':
            # Use target branch version
            for record in target_change.historical_records.all():
                resolved_content[f"{record.instance_type}_{record.instance_pk}"] = record
        
        elif resolution_type == 'manual':
            # Use manual resolution
            manual_content = manual_resolutions.get(conflict_id)
            if not manual_content:
                raise ValidationError(f"Manual resolution missing for conflict {conflict_id}")
            
            # Create new historical record with manual content
            base_record = source_change.historical_records.first()
            if base_record:
                new_record = base_record.__class__(
                    **{
                        **base_record.__dict__,
                        'id': base_record.id,
                        'history_date': timezone.now(),
                        'history_user': user,
                        'history_change_reason': 'Manual conflict resolution',
                        'history_type': '~'
                    }
                )
                # Apply manual changes
                for field, value in manual_content.items():
                    setattr(new_record, field, value)
                resolved_content[f"{new_record.instance_type}_{new_record.instance_pk}"] = new_record
    
    # Create resolution changeset
    resolution_changeset = ChangeSet.objects.create(
        branch=target_branch,
        created_by=user,
        description=f"Resolved conflicts from '{source_branch.name}'",
        metadata={
            'resolution_type': 'conflict_resolution',
            'source_branch': source_branch.name,
            'resolved_conflicts': list(resolutions.keys())
        },
        status='applied'
    )
    
    # Add resolved records to changeset
    for record in resolved_content.values():
        resolution_changeset.historical_records.add(record)
    
    return resolution_changeset

def get_change_diff(change: ChangeSet) -> List[Dict[str, Any]]:
    """
    Get a structured diff of changes in a changeset
    
    Args:
        change: The changeset to analyze
    
    Returns:
        List of diffs for each changed record
    """
    diffs = []
    
    for record in change.historical_records.all():
        diff = {
            'model': record.instance_type.__name__,
            'id': record.instance_pk,
            'type': record.history_type,
            'date': record.history_date,
            'user': record.history_user_display,
            'changes': {}
        }
        
        if record.history_type == '~':  # Modified
            previous = record.prev_record
            if previous:
                diff['changes'] = record.diff_against_previous
        elif record.history_type == '+':  # Added
            diff['changes'] = {
                field: {'old': None, 'new': str(getattr(record, field))}
                for field in record.__dict__
                if not field.startswith('_') and field not in [
                    'history_date', 'history_id', 'history_type',
                    'history_user_id', 'history_change_reason'
                ]
            }
        elif record.history_type == '-':  # Deleted
            diff['changes'] = {
                field: {'old': str(getattr(record, field)), 'new': None}
                for field in record.__dict__
                if not field.startswith('_') and field not in [
                    'history_date', 'history_id', 'history_type',
                    'history_user_id', 'history_change_reason'
                ]
            }
        
        diffs.append(diff)
    
    return diffs