190 lines
6.6 KiB
Python
190 lines
6.6 KiB
Python
"""
|
|
YouTube Summarizer SDK Exception Classes
|
|
Custom exceptions for the YouTube Summarizer Python SDK
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any
|
|
|
|
|
|
class YouTubeSummarizerError(Exception):
|
|
"""Base exception class for YouTube Summarizer SDK"""
|
|
|
|
def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
|
|
self.message = message
|
|
self.details = details or {}
|
|
super().__init__(self.message)
|
|
|
|
def __str__(self) -> str:
|
|
if self.details:
|
|
return f"{self.message} (Details: {self.details})"
|
|
return self.message
|
|
|
|
|
|
class AuthenticationError(YouTubeSummarizerError):
|
|
"""Raised when API key authentication fails"""
|
|
|
|
def __init__(self, message: str = "Invalid API key or authentication failed"):
|
|
super().__init__(message)
|
|
|
|
|
|
class RateLimitError(YouTubeSummarizerError):
|
|
"""Raised when API rate limits are exceeded"""
|
|
|
|
def __init__(self, message: str, rate_limit_info: Optional[Dict[str, Any]] = None):
|
|
self.rate_limit_info = rate_limit_info or {}
|
|
super().__init__(message, rate_limit_info)
|
|
|
|
@property
|
|
def remaining(self) -> int:
|
|
"""Get remaining requests in current window"""
|
|
return int(self.rate_limit_info.get('remaining', 0))
|
|
|
|
@property
|
|
def reset_time(self) -> Optional[str]:
|
|
"""Get rate limit reset time"""
|
|
return self.rate_limit_info.get('reset')
|
|
|
|
@property
|
|
def limit(self) -> Optional[int]:
|
|
"""Get current rate limit"""
|
|
return self.rate_limit_info.get('limit')
|
|
|
|
|
|
class ValidationError(YouTubeSummarizerError):
|
|
"""Raised when request validation fails"""
|
|
|
|
def __init__(self, message: str, validation_errors: Optional[Dict[str, Any]] = None):
|
|
self.validation_errors = validation_errors or {}
|
|
super().__init__(message, validation_errors)
|
|
|
|
|
|
class APIError(YouTubeSummarizerError):
|
|
"""Raised for general API errors"""
|
|
|
|
def __init__(self, message: str, status_code: Optional[int] = None, response_data: Optional[Dict[str, Any]] = None):
|
|
self.status_code = status_code
|
|
self.response_data = response_data or {}
|
|
details = {"status_code": status_code, "response": response_data}
|
|
super().__init__(message, details)
|
|
|
|
|
|
class JobNotFoundError(YouTubeSummarizerError):
|
|
"""Raised when a job ID cannot be found"""
|
|
|
|
def __init__(self, job_id: str):
|
|
self.job_id = job_id
|
|
message = f"Job '{job_id}' not found or expired"
|
|
super().__init__(message, {"job_id": job_id})
|
|
|
|
|
|
class JobTimeoutError(YouTubeSummarizerError):
|
|
"""Raised when a job times out"""
|
|
|
|
def __init__(self, job_id: str, timeout_seconds: float):
|
|
self.job_id = job_id
|
|
self.timeout_seconds = timeout_seconds
|
|
message = f"Job '{job_id}' timed out after {timeout_seconds} seconds"
|
|
super().__init__(message, {"job_id": job_id, "timeout_seconds": timeout_seconds})
|
|
|
|
|
|
class TranscriptNotAvailableError(YouTubeSummarizerError):
|
|
"""Raised when transcript cannot be extracted from video"""
|
|
|
|
def __init__(self, video_id: str, reason: str = "No transcript available"):
|
|
self.video_id = video_id
|
|
self.reason = reason
|
|
message = f"Transcript not available for video '{video_id}': {reason}"
|
|
super().__init__(message, {"video_id": video_id, "reason": reason})
|
|
|
|
|
|
class VideoNotFoundError(YouTubeSummarizerError):
|
|
"""Raised when video cannot be found or accessed"""
|
|
|
|
def __init__(self, video_url: str, reason: str = "Video not found or private"):
|
|
self.video_url = video_url
|
|
self.reason = reason
|
|
message = f"Video not accessible '{video_url}': {reason}"
|
|
super().__init__(message, {"video_url": video_url, "reason": reason})
|
|
|
|
|
|
class ProcessingError(YouTubeSummarizerError):
|
|
"""Raised when video processing fails"""
|
|
|
|
def __init__(self, job_id: str, stage: str, reason: str):
|
|
self.job_id = job_id
|
|
self.stage = stage
|
|
self.reason = reason
|
|
message = f"Processing failed for job '{job_id}' at stage '{stage}': {reason}"
|
|
super().__init__(message, {"job_id": job_id, "stage": stage, "reason": reason})
|
|
|
|
|
|
class QuotaExceededError(YouTubeSummarizerError):
|
|
"""Raised when API quota is exceeded"""
|
|
|
|
def __init__(self, quota_type: str, reset_time: Optional[str] = None):
|
|
self.quota_type = quota_type
|
|
self.reset_time = reset_time
|
|
message = f"Quota exceeded for {quota_type}"
|
|
if reset_time:
|
|
message += f". Resets at {reset_time}"
|
|
details = {"quota_type": quota_type, "reset_time": reset_time}
|
|
super().__init__(message, details)
|
|
|
|
|
|
class WebSocketError(YouTubeSummarizerError):
|
|
"""Raised for WebSocket connection errors"""
|
|
|
|
def __init__(self, message: str, connection_state: Optional[str] = None):
|
|
self.connection_state = connection_state
|
|
super().__init__(message, {"connection_state": connection_state})
|
|
|
|
|
|
class ConfigurationError(YouTubeSummarizerError):
|
|
"""Raised for SDK configuration errors"""
|
|
|
|
def __init__(self, parameter: str, value: Any, expected: str):
|
|
self.parameter = parameter
|
|
self.value = value
|
|
self.expected = expected
|
|
message = f"Invalid configuration for '{parameter}': got {value}, expected {expected}"
|
|
super().__init__(message, {"parameter": parameter, "value": value, "expected": expected})
|
|
|
|
|
|
class RetryableError(YouTubeSummarizerError):
|
|
"""Base class for errors that can be retried"""
|
|
|
|
def __init__(self, message: str, retry_after: Optional[float] = None, details: Optional[Dict[str, Any]] = None):
|
|
self.retry_after = retry_after
|
|
super().__init__(message, details)
|
|
|
|
|
|
class TemporaryServiceError(RetryableError):
|
|
"""Raised for temporary service unavailability"""
|
|
|
|
def __init__(self, service_name: str, retry_after: Optional[float] = None):
|
|
self.service_name = service_name
|
|
message = f"Temporary error in {service_name} service"
|
|
if retry_after:
|
|
message += f". Retry after {retry_after} seconds"
|
|
super().__init__(message, retry_after, {"service": service_name})
|
|
|
|
|
|
class NetworkError(RetryableError):
|
|
"""Raised for network connectivity issues"""
|
|
|
|
def __init__(self, message: str = "Network connectivity error", retry_after: Optional[float] = None):
|
|
super().__init__(message, retry_after)
|
|
|
|
|
|
# Exception mapping for HTTP status codes
|
|
HTTP_STATUS_EXCEPTIONS = {
|
|
400: ValidationError,
|
|
401: AuthenticationError,
|
|
403: AuthenticationError,
|
|
404: JobNotFoundError,
|
|
429: RateLimitError,
|
|
500: APIError,
|
|
502: TemporaryServiceError,
|
|
503: TemporaryServiceError,
|
|
504: TemporaryServiceError,
|
|
} |