Pac-cogs/videoarchiver/utils/exceptions.py

"""Custom exceptions for VideoArchiver"""

from typing import Optional, Dict, Any
from enum import Enum, auto

__all__ = [
    'ErrorSeverity',
    'ErrorContext',
    'VideoArchiverError',
    'VideoDownloadError',
    'VideoProcessingError',
    'VideoVerificationError',
    'VideoUploadError',
    'VideoCleanupError',
    'FileCleanupError',
    'ConfigurationError',
    'PermissionError',
    'NetworkError',
    'ResourceError',
    'QueueError',
    'ComponentError',
    'DiscordAPIError',
    'ResourceExhaustedError',
    'ProcessingError',
    'CleanupError',
    'FileOperationError',
    'ProcessorError',
    'ValidationError',
    'DisplayError',
    'URLExtractionError',
    'MessageHandlerError',
    'QueueHandlerError',
    'QueueProcessorError',
    'FFmpegError',
    'DatabaseError',
    'HealthCheckError',
    'TrackingError',
    'CommandError',
    'EventError',
    'CogError'
]


class ErrorSeverity(Enum):
    """Severity levels for errors"""

    LOW = auto()
    MEDIUM = auto()
    HIGH = auto()
    CRITICAL = auto()


class ErrorContext:
    """Context information for errors"""

    def __init__(
        self,
        component: str,
        operation: str,
        details: Optional[Dict[str, Any]] = None,
        severity: ErrorSeverity = ErrorSeverity.MEDIUM,
    ) -> None:
        self.component = component
        self.operation = operation
        self.details = details or {}
        self.severity = severity

    def __str__(self) -> str:
        return (
            f"[{self.severity.name}] {self.component}.{self.operation}: "
            f"{', '.join(f'{k}={v}' for k, v in self.details.items())}"
        )


class VideoArchiverError(Exception):
    """Base exception for VideoArchiver errors"""

    def __init__(self, message: str, context: Optional[ErrorContext] = None) -> None:
        self.context = context
        super().__init__(f"{context}: {message}" if context else message)


class VideoDownloadError(VideoArchiverError):
    """Error downloading video"""

    pass


class VideoProcessingError(VideoArchiverError):
    """Error processing video"""

    pass


class VideoVerificationError(VideoArchiverError):
    """Error verifying video"""

    pass


class VideoUploadError(VideoArchiverError):
    """Error uploading video"""

    pass


class VideoCleanupError(VideoArchiverError):
    """Error cleaning up video files"""

    pass


class FileCleanupError(VideoArchiverError):
    """Error cleaning up files"""

    pass


class ConfigurationError(VideoArchiverError):
    """Error in configuration"""

    pass


class PermissionError(VideoArchiverError):
    """Error with file permissions"""

    pass


class NetworkError(VideoArchiverError):
    """Error with network operations"""

    def __init__(
        self,
        message: str,
        url: Optional[str] = None,
        status_code: Optional[int] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.url = url
        self.status_code = status_code
        details = f" (URL: {url}" + (
            f", Status: {status_code})" if status_code else ")"
        )
        super().__init__(message + details, context)


class ResourceError(VideoArchiverError):
    """Error with system resources"""

    pass


class QueueError(VideoArchiverError):
    """Error with queue operations"""

    pass


class ComponentError(VideoArchiverError):
    """Error with component initialization or cleanup"""

    pass


class DiscordAPIError(VideoArchiverError):
    """Error with Discord API operations"""

    def __init__(
        self,
        message: str,
        status_code: Optional[int] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.status_code = status_code
        details = f" (Status: {status_code})" if status_code else ""
        super().__init__(f"Discord API Error: {message}{details}", context)


class ResourceExhaustedError(VideoArchiverError):
    """Error when system resources are exhausted"""

    def __init__(
        self,
        message: str,
        resource_type: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.resource_type = resource_type
        details = f" (Type: {resource_type})" if resource_type else ""
        super().__init__(f"Resource exhausted: {message}{details}", context)


class ProcessingError(VideoArchiverError):
    """Error during video processing"""

    pass


class CleanupError(VideoArchiverError):
    """Error during cleanup operations"""

    pass


class FileOperationError(VideoArchiverError):
    """Error during file operations"""

    def __init__(
        self,
        message: str,
        path: Optional[str] = None,
        operation: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.path = path
        self.operation = operation
        details = []
        if path:
            details.append(f"Path: {path}")
        if operation:
            details.append(f"Operation: {operation}")
        details_str = f" ({', '.join(details)})" if details else ""
        super().__init__(f"File operation error: {message}{details_str}", context)


class ProcessorError(VideoArchiverError):
    """Error in video processor operations"""

    pass


class ValidationError(VideoArchiverError):
    """Error in message or content validation"""

    pass


class DisplayError(VideoArchiverError):
    """Error in status display operations"""

    pass


class URLExtractionError(VideoArchiverError):
    """Error extracting URLs from content"""

    def __init__(
        self,
        message: str,
        url: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.url = url
        details = f" (URL: {url})" if url else ""
        super().__init__(f"URL extraction error: {message}{details}", context)


class MessageHandlerError(VideoArchiverError):
    """Error in message handling operations"""

    def __init__(
        self,
        message: str,
        message_id: Optional[int] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.message_id = message_id
        details = f" (Message ID: {message_id})" if message_id else ""
        super().__init__(f"Message handler error: {message}{details}", context)


class QueueHandlerError(VideoArchiverError):
    """Error in queue handling operations"""

    pass


class QueueProcessorError(VideoArchiverError):
    """Error in queue processing operations"""

    pass


class FFmpegError(VideoArchiverError):
    """Error in FFmpeg operations"""

    def __init__(
        self,
        message: str,
        command: Optional[str] = None,
        exit_code: Optional[int] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.command = command
        self.exit_code = exit_code
        details = []
        if command:
            details.append(f"Command: {command}")
        if exit_code is not None:
            details.append(f"Exit Code: {exit_code}")
        details_str = f" ({', '.join(details)})" if details else ""
        super().__init__(f"FFmpeg error: {message}{details_str}", context)


class DatabaseError(VideoArchiverError):
    """Error in database operations"""

    def __init__(
        self,
        message: str,
        query: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.query = query
        details = f" (Query: {query})" if query else ""
        super().__init__(f"Database error: {message}{details}", context)


class HealthCheckError(VideoArchiverError):
    """Error in health check operations"""

    def __init__(
        self,
        message: str,
        component: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.component = component
        details = f" (Component: {component})" if component else ""
        super().__init__(f"Health check error: {message}{details}", context)


class TrackingError(VideoArchiverError):
    """Error in progress tracking operations"""

    def __init__(
        self,
        message: str,
        operation: Optional[str] = None,
        item_id: Optional[str] = None,
        context: Optional[ErrorContext] = None,
    ) -> None:
        self.operation = operation
        self.item_id = item_id
        details = []
        if operation:
            details.append(f"Operation: {operation}")
        if item_id:
            details.append(f"Item ID: {item_id}")
        details_str = f" ({', '.join(details)})" if details else ""
        super().__init__(f"Progress tracking error: {message}{details_str}", context)


class CommandError(VideoArchiverError):
    """Error in command execution"""

    pass


class EventError(VideoArchiverError):
    """Error in event handling"""

    pass

class CogError(VideoArchiverError):
    """Error in cog operations"""
    def __init__(
        self,
        message: str,
        cog_name: Optional[str] = None,
        context: Optional[ErrorContext] = None
    ) -> None:
        self.cog_name = cog_name
        details = f" (Cog: {cog_name})" if cog_name else ""
        super().__init__(f"Cog error: {message}{details}", context)