potato/annotation_history.py

"""
Annotation History Module

This module provides comprehensive tracking of all annotation actions with fine-grained
timestamp metadata. It enables performance analysis, quality assurance, and future
undo functionality.

Key Components:
- AnnotationAction: Dataclass representing a single annotation action
- AnnotationHistoryManager: Utility class for creating and analyzing annotation actions
- Performance metrics calculation and suspicious activity detection
"""

import uuid
import datetime
import logging
from dataclasses import dataclass, asdict
from typing import Optional, Dict, Any, List
import json

logger = logging.getLogger(__name__)


@dataclass
class AnnotationAction:
    """
    Represents a single annotation action with full metadata.

    This class captures all information about an annotation change, including
    timing data, user context, and action details for comprehensive tracking.
    """
    action_id: str  # UUID for unique identification
    timestamp: datetime.datetime  # Precise timestamp
    user_id: str
    instance_id: str
    action_type: str  # 'add_label', 'update_label', 'delete_label', 'add_span', 'update_span', 'delete_span'
    schema_name: str
    label_name: str
    old_value: Optional[Any]  # Previous value (for updates/deletes)
    new_value: Optional[Any]  # New value (for adds/updates)
    span_data: Optional[Dict]  # For span annotations (start, end, text)
    session_id: str  # Browser session identifier
    client_timestamp: Optional[datetime.datetime]  # Frontend timestamp
    server_processing_time_ms: int  # Server processing time
    metadata: Dict[str, Any]  # Additional metadata (browser info, etc.)

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for serialization"""
        data = asdict(self)
        data['timestamp'] = self.timestamp.isoformat()
        if self.client_timestamp:
            data['client_timestamp'] = self.client_timestamp.isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'AnnotationAction':
        """Create from dictionary"""
        data['timestamp'] = datetime.datetime.fromisoformat(data['timestamp'])
        if data.get('client_timestamp'):
            data['client_timestamp'] = datetime.datetime.fromisoformat(data['client_timestamp'])
        return cls(**data)

    def __str__(self) -> str:
        """String representation for logging"""
        return f"AnnotationAction({self.action_type}: {self.schema_name}:{self.label_name} = {self.new_value})"


class AnnotationHistoryManager:
    """
    Manages annotation history and provides analytics.

    This class provides utilities for creating annotation actions, calculating
    performance metrics, and detecting suspicious activity patterns.
    """

    @staticmethod
    def create_action(
        user_id: str,
        instance_id: str,
        action_type: str,
        schema_name: str,
        label_name: str,
        old_value: Optional[Any],
        new_value: Optional[Any],
        span_data: Optional[Dict] = None,
        session_id: str = None,
        client_timestamp: Optional[datetime.datetime] = None,
        server_processing_time_ms: int = 0,
        metadata: Optional[Dict] = None
    ) -> AnnotationAction:
        """
        Create a new annotation action with current timestamp.

        Args:
            user_id: The user performing the action
            instance_id: The instance being annotated
            action_type: Type of action (add_label, update_label, etc.)
            schema_name: Name of the annotation schema
            label_name: Name of the specific label
            old_value: Previous value (for updates/deletes)
            new_value: New value (for adds/updates)
            span_data: Span annotation data (start, end, text)
            session_id: Browser session identifier
            client_timestamp: Frontend timestamp
            server_processing_time_ms: Server processing time in milliseconds
            metadata: Additional metadata

        Returns:
            AnnotationAction object with current timestamp
        """
        return AnnotationAction(
            action_id=str(uuid.uuid4()),
            timestamp=datetime.datetime.now(),
            user_id=user_id,
            instance_id=instance_id,
            action_type=action_type,
            schema_name=schema_name,
            label_name=label_name,
            old_value=old_value,
            new_value=new_value,
            span_data=span_data,
            session_id=session_id or "unknown",
            client_timestamp=client_timestamp,
            server_processing_time_ms=server_processing_time_ms,
            metadata=metadata or {}
        )

    @staticmethod
    def calculate_performance_metrics(actions: List[AnnotationAction]) -> Dict[str, Any]:
        """
        Calculate performance metrics from action history.

        Args:
            actions: List of annotation actions to analyze

        Returns:
            Dictionary containing performance metrics
        """
        if not actions:
            return {
                'total_actions': 0,
                'average_action_time_ms': 0,
                'fastest_action_time_ms': 0,
                'slowest_action_time_ms': 0,
                'actions_per_minute': 0,
                'total_processing_time_ms': 0
            }

        processing_times = [a.server_processing_time_ms for a in actions]
        total_time = sum(processing_times)

        # Calculate actions per minute
        if len(actions) > 1:
            time_span = (actions[-1].timestamp - actions[0].timestamp).total_seconds() / 60
            actions_per_minute = len(actions) / time_span if time_span > 0 else 0
        else:
            actions_per_minute = 0

        return {
            'total_actions': len(actions),
            'average_action_time_ms': total_time / len(actions),
            'fastest_action_time_ms': min(processing_times),
            'slowest_action_time_ms': max(processing_times),
            'actions_per_minute': actions_per_minute,
            'total_processing_time_ms': total_time
        }

    @staticmethod
    def detect_suspicious_activity(actions: List[AnnotationAction],
                                 fast_threshold_ms: int = 500,
                                 burst_threshold_seconds: int = 2) -> Dict[str, Any]:
        """
        Detect potentially suspicious annotation activity.

        Args:
            actions: List of annotation actions to analyze
            fast_threshold_ms: Threshold for considering an action "too fast"
            burst_threshold_seconds: Threshold for burst activity detection

        Returns:
            Dictionary containing suspicious activity analysis
        """
        if not actions:
            return {
                'suspicious_actions': [],
                'fast_actions_count': 0,
                'burst_actions_count': 0,
                'suspicious_score': 0
            }

        suspicious_actions = []
        fast_actions = []
        burst_actions = []

        # Detect fast actions
        for action in actions:
            if action.server_processing_time_ms < fast_threshold_ms:
                fast_actions.append(action)
                suspicious_actions.append(action)

        # Detect burst activity (multiple actions in quick succession)
        for i in range(1, len(actions)):
            time_diff = (actions[i].timestamp - actions[i-1].timestamp).total_seconds()
            if time_diff < burst_threshold_seconds:
                burst_actions.append(actions[i])
                if actions[i] not in suspicious_actions:
                    suspicious_actions.append(actions[i])

        # Calculate suspicious score (0-100)
        total_actions = len(actions)
        fast_percentage = (len(fast_actions) / total_actions) * 100 if total_actions > 0 else 0
        burst_percentage = (len(burst_actions) / total_actions) * 100 if total_actions > 0 else 0

        suspicious_score = min(100, (fast_percentage * 0.6) + (burst_percentage * 0.4))

        return {
            'suspicious_actions': suspicious_actions,
            'fast_actions_count': len(fast_actions),
            'burst_actions_count': len(burst_actions),
            'fast_actions_percentage': fast_percentage,
            'burst_actions_percentage': burst_percentage,
            'suspicious_score': suspicious_score,
            'suspicious_level': _get_suspicious_level(suspicious_score)
        }

    @staticmethod
    def get_actions_by_time_range(actions: List[AnnotationAction],
                                 start_time: datetime.datetime,
                                 end_time: datetime.datetime) -> List[AnnotationAction]:
        """
        Filter actions by time range.

        Args:
            actions: List of annotation actions
            start_time: Start of time range
            end_time: End of time range

        Returns:
            Filtered list of actions within the time range
        """
        return [action for action in actions
                if start_time <= action.timestamp <= end_time]

    @staticmethod
    def get_actions_by_instance(actions: List[AnnotationAction],
                               instance_id: str) -> List[AnnotationAction]:
        """
        Filter actions by instance ID.

        Args:
            actions: List of annotation actions
            instance_id: Instance ID to filter by

        Returns:
            Filtered list of actions for the specified instance
        """
        return [action for action in actions if action.instance_id == instance_id]

    @staticmethod
    def get_actions_by_type(actions: List[AnnotationAction],
                           action_type: str) -> List[AnnotationAction]:
        """
        Filter actions by action type.

        Args:
            actions: List of annotation actions
            action_type: Action type to filter by

        Returns:
            Filtered list of actions of the specified type
        """
        return [action for action in actions if action.action_type == action_type]


def _get_suspicious_level(score: float) -> str:
    """Convert suspicious score to level description."""
    if score < 10:
        return "Normal"
    elif score < 30:
        return "Low"
    elif score < 60:
        return "Medium"
    elif score < 80:
        return "High"
    else:
        return "Very High"