thrillwiki_django_no_react/history_tracking/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from simple_history.signals import post_create_historical_record
from django.contrib.auth import get_user_model
from django.db import transaction
from .models import VersionBranch, ChangeSet, HistoricalModel
from .managers import ChangeTracker
import threading

User = get_user_model()

# Thread-local storage for tracking active changesets
_changeset_context = threading.local()

def get_current_branch():
    """Get the currently active branch for the thread"""
    return getattr(_changeset_context, 'current_branch', None)

def set_current_branch(branch):
    """Set the active branch for the current thread"""
    _changeset_context.current_branch = branch

def clear_current_branch():
    """Clear the active branch for the current thread"""
    if hasattr(_changeset_context, 'current_branch'):
        del _changeset_context.current_branch

class ChangesetContextManager:
    """Context manager for tracking changes in a specific branch"""
    
    def __init__(self, branch, user=None):
        self.branch = branch
        self.user = user
        self.previous_branch = None
    
    def __enter__(self):
        self.previous_branch = get_current_branch()
        set_current_branch(self.branch)
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        set_current_branch(self.previous_branch)

@receiver(post_create_historical_record)
def handle_history_record(sender, instance, history_instance, **kwargs):
    """Handle creation of historical records by adding them to changesets"""
    # Only handle records from HistoricalModel subclasses
    if not isinstance(instance, HistoricalModel):
        return
    
    branch = get_current_branch()
    if not branch:
        # If no branch is set, use the default branch
        branch, _ = VersionBranch.objects.get_or_create(
            name='main',
            defaults={
                'metadata': {
                    'type': 'default_branch',
                    'created_automatically': True
                }
            }
        )
    
    # Create or get active changeset for the current branch
    changeset = getattr(_changeset_context, 'active_changeset', None)
    if not changeset:
        changeset = ChangeSet.objects.create(
            branch=branch,
            created_by=history_instance.history_user,
            description=f"Automatic change tracking: {history_instance.history_type}",
            metadata={
                'auto_tracked': True,
                'model': instance._meta.model_name,
                'history_type': history_instance.history_type
            },
            status='applied'
        )
        _changeset_context.active_changeset = changeset
    
    # Add the historical record to the changeset
    changeset.historical_records.add(history_instance)

@receiver(post_save, sender=ChangeSet)
def handle_changeset_save(sender, instance, created, **kwargs):
    """Handle changeset creation by updating related objects"""
    if created and instance.status == 'applied':
        # Clear the active changeset if this is the one we were using
        active_changeset = getattr(_changeset_context, 'active_changeset', None)
        if active_changeset and active_changeset.id == instance.id:
            delattr(_changeset_context, 'active_changeset')
        
        # Update branch metadata
        branch = instance.branch
        if not branch.metadata.get('first_change'):
            branch.metadata['first_change'] = instance.created_at.isoformat()
        branch.metadata['last_change'] = instance.created_at.isoformat()
        branch.metadata['change_count'] = branch.changesets.count()
        branch.save()

def start_changeset(branch, user=None, description=None):
    """Start a new changeset in the given branch"""
    changeset = ChangeSet.objects.create(
        branch=branch,
        created_by=user,
        description=description or "Manual changeset",
        status='pending'
    )
    _changeset_context.active_changeset = changeset
    return changeset

def commit_changeset(success=True):
    """Commit the current changeset"""
    changeset = getattr(_changeset_context, 'active_changeset', None)
    if changeset:
        changeset.status = 'applied' if success else 'failed'
        changeset.save()
        delattr(_changeset_context, 'active_changeset')
    return changeset

class ChangesetManager:
    """Context manager for handling changesets"""
    
    def __init__(self, branch, user=None, description=None):
        self.branch = branch
        self.user = user
        self.description = description
        self.changeset = None
    
    def __enter__(self):
        self.changeset = start_changeset(
            self.branch,
            self.user,
            self.description
        )
        return self.changeset
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        commit_changeset(success=exc_type is None)