api/veo_error_handler.py

"""
Veo 3.1 Official Error Handler
Based on official Veo 3.1 API documentation
"""

from enum import Enum
from typing import Dict, Tuple, Optional
import re


class VeoErrorType(Enum):
    """Official Veo 3.1 error types"""
    RESOURCE_EXHAUSTED = "429"
    FORBIDDEN = "403"
    SAFETY_FILTER = "safety"
    PUBLIC_ERROR_MINOR = "public_error_minor"
    CELEBRITY = "celebrity"  # Prominent public figure
    CHILD = "child"
    SEXUAL = "sexual"
    VIOLENCE = "violence"
    DANGEROUS = "dangerous"
    UNKNOWN = "unknown"


# Official Veo 3.1 Safety Filter Support Codes
# Source: Official API Documentation
SAFETY_CODES: Dict[str, str] = {
    # Child Safety
    "58061214": "Child",
    "17301594": "Child",
    
    # Celebrity / Public Figures
    "29310472": "Celebrity",
    "15236754": "Celebrity",
    
    # Sexual Content
    "90789179": "Sexual",
    "43188360": "Sexual",
    
    # Violence
    "61493863": "Violence",
    "56562880": "Violence",
    
    # Dangerous Content
    "62263041": "Dangerous",
}


def extract_support_code(error_message: str) -> Optional[str]:
    """
    Extract 8-digit support code from error message
    
    Args:
        error_message: The error message from Veo API
        
    Returns:
        Support code if found, None otherwise
    """
    # Look for 8-digit numbers (support codes)
    match = re.search(r'\b\d{8}\b', error_message)
    return match.group(0) if match else None


def get_safety_category(support_code: str) -> str:
    """
    Get safety category from support code
    
    Args:
        support_code: 8-digit support code
        
    Returns:
        Safety category name
    """
    return SAFETY_CODES.get(support_code, f"Unknown ({support_code})")


def is_safety_violation(error_message: str) -> bool:
    """
    Check if error is a safety filter violation
    
    Args:
        error_message: Error message from API
        
    Returns:
        True if safety violation detected
    """
    # Official keywords from documentation
    safety_keywords = [
        "safety filter",
        "safety violation",
        "content policy",
        "inappropriate content",
        "flagged for containing",
        "prominent public figure",
        "celebrity",
        "public figure",
        "politician",
        "moderation",
    ]
    
    error_lower = error_message.lower()
    
    # Check for keywords
    if any(keyword in error_lower for keyword in safety_keywords):
        return True
    
    # Check for support code
    if extract_support_code(error_message):
        return True
    
    return False


def is_public_error_minor(error_message: str) -> bool:
    """
    Check if error is PUBLIC_ERROR_MINOR (generic internal error)
    
    Args:
        error_message: Error message from API
        
    Returns:
        True if PUBLIC_ERROR_MINOR detected
    """
    return "PUBLIC_ERROR_MINOR" in error_message.upper()


def handle_veo_error(error_response: Dict) -> Tuple[VeoErrorType, str, Optional[str]]:
    """
    Handle Veo 3.1 errors according to official documentation
    
    Args:
        error_response: Error response from Veo API
        
    Returns:
        Tuple of (error_type, error_message, support_code)
    """
    error_code = str(error_response.get("code", ""))
    error_message = error_response.get("message", "")
    
    # Extract support code if present
    support_code = extract_support_code(error_message)
    
    # 429: Too many requests
    if "429" in error_code:
        return (
            VeoErrorType.RESOURCE_EXHAUSTED,
            "Too many requests. Wait 5-10 minutes and retry.",
            None
        )
    
    # 403: Insufficient permissions
    if "403" in error_code:
        return (
            VeoErrorType.FORBIDDEN,
            "Insufficient permissions. Check API configuration.",
            None
        )
    
    # PUBLIC_ERROR_MINOR: Generic internal error
    if is_public_error_minor(error_message):
        return (
            VeoErrorType.PUBLIC_ERROR_MINOR,
            "Internal processing error. Wait 5-10 minutes or simplify request.",
            None
        )
    
    # Safety filter with support code
    if support_code:
        category = get_safety_category(support_code)
        return (
            VeoErrorType.SAFETY_FILTER,
            f"Safety filter triggered: {category}",
            support_code
        )
    
    # Celebrity / Public Figure (no support code)
    if any(keyword in error_message.lower() for keyword in [
        "prominent public figure", "celebrity", "politician", "public figure"
    ]):
        return (
            VeoErrorType.CELEBRITY,
            "Content contains prominent public figure or celebrity",
            None
        )
    
    # Generic safety violation
    if is_safety_violation(error_message):
        return (
            VeoErrorType.SAFETY_FILTER,
            "Content policy violation detected",
            None
        )
    
    # Unknown error
    return (
        VeoErrorType.UNKNOWN,
        error_message or "Unknown error occurred",
        None
    )


def get_retry_strategy(error_type: VeoErrorType) -> Dict[str, any]:
    """
    Get recommended retry strategy for error type
    
    Args:
        error_type: The type of error
        
    Returns:
        Dict with retry strategy
    """
    strategies = {
        VeoErrorType.RESOURCE_EXHAUSTED: {
            "should_retry": True,
            "wait_seconds": 600,  # 10 minutes
            "max_retries": 2,
            "requires_fix": False,
        },
        VeoErrorType.PUBLIC_ERROR_MINOR: {
            "should_retry": True,
            "wait_seconds": 300,  # 5 minutes
            "max_retries": 2,
            "requires_fix": False,
        },
        VeoErrorType.CELEBRITY: {
            "should_retry": True,
            "wait_seconds": 2,
            "max_retries": 2,
            "requires_fix": True,  # Needs AI fix
        },
        VeoErrorType.SAFETY_FILTER: {
            "should_retry": True,
            "wait_seconds": 2,
            "max_retries": 2,
            "requires_fix": True,  # Needs AI fix
        },
        VeoErrorType.FORBIDDEN: {
            "should_retry": False,
            "wait_seconds": 0,
            "max_retries": 0,
            "requires_fix": False,
        },
        VeoErrorType.UNKNOWN: {
            "should_retry": True,
            "wait_seconds": 5,
            "max_retries": 1,
            "requires_fix": False,
        },
    }
    
    return strategies.get(error_type, strategies[VeoErrorType.UNKNOWN])


def should_auto_fix(error_type: VeoErrorType) -> bool:
    """
    Determine if error should trigger AI auto-fix
    
    Args:
        error_type: The type of error
        
    Returns:
        True if auto-fix should be attempted
    """
    fixable_types = [
        VeoErrorType.CELEBRITY,
        VeoErrorType.SAFETY_FILTER,
        VeoErrorType.CHILD,
        VeoErrorType.SEXUAL,
        VeoErrorType.VIOLENCE,
        VeoErrorType.DANGEROUS,
    ]
    return error_type in fixable_types