thrillwiki_django_no_react/history_tracking/utils.py

from typing import Dict, Any, List, Optional, TypeVar, Type, Union, cast
from django.core.exceptions import ValidationError
from .models import VersionBranch, ChangeSet
from django.utils import timezone
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AbstractUser
from django.db.models import Model

UserModel = TypeVar('UserModel', bound=AbstractUser)
User = cast(Type[UserModel], get_user_model())

def _handle_source_target_resolution(change: ChangeSet) -> Dict[str, Any]:
    resolved = {}
    for record in change.historical_records.all():
        resolved[f"{record.instance_type}_{record.instance_pk}"] = record
    return resolved
def _handle_manual_resolution(
    conflict_id: str,
    source_change: ChangeSet,
    manual_resolutions: Dict[str, str],
    user: Optional[UserModel]
) -> Dict[str, Any]:
    manual_content = manual_resolutions.get(conflict_id)
    if not manual_content:
        raise ValidationError(f"Manual resolution missing for conflict {conflict_id}")
    
    resolved = {}
    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': '~'
            }
        )
        for field, value in manual_content.items():
            setattr(new_record, field, value)
        resolved[f"{new_record.instance_type}_{new_record.instance_pk}"] = new_record
    return resolved

def resolve_conflicts(
    source_branch: VersionBranch,
    target_branch: VersionBranch,
    resolutions: Dict[str, str],
    manual_resolutions: Dict[str, str],
    user: Optional[UserModel] = 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':
            resolved_content.update(_handle_source_target_resolution(source_change))
        elif resolution_type == 'target':
            resolved_content.update(_handle_source_target_resolution(target_change))
        elif resolution_type == 'manual':
            resolved_content.update(_handle_manual_resolution(
                conflict_id, source_change, manual_resolutions, user
            ))
    
    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'
    )
    
    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