""" TranscodeJob Schema Definition Source of truth for job data model. """ from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any, Dict, Optional from uuid import UUID class JobStatus(str, Enum): """Status of a transcode/trim job.""" PENDING = "pending" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" @dataclass class TranscodeJob: """ A transcoding or trimming job in the queue. Jobs can either: - Transcode using a preset (full re-encode) - Trim only (stream copy with -c:v copy -c:a copy) A trim-only job has no preset and uses stream copy. """ id: UUID # Input source_asset_id: UUID # Configuration preset_id: Optional[UUID] = None preset_snapshot: Dict[str, Any] = field( default_factory=dict ) # Copy at creation time # Trimming (optional) trim_start: Optional[float] = None # seconds trim_end: Optional[float] = None # seconds # Output output_filename: str = "" output_path: Optional[str] = None output_asset_id: Optional[UUID] = None # Status & Progress status: JobStatus = JobStatus.PENDING progress: float = 0.0 # 0.0 to 100.0 current_frame: Optional[int] = None current_time: Optional[float] = None # seconds processed speed: Optional[str] = None # "2.5x" error_message: Optional[str] = None # Worker tracking celery_task_id: Optional[str] = None execution_arn: Optional[str] = None # AWS Step Functions execution ARN priority: int = 0 # Lower = higher priority # Timestamps created_at: Optional[datetime] = None started_at: Optional[datetime] = None completed_at: Optional[datetime] = None @property def is_trim_only(self) -> bool: """Check if this is a trim-only job (stream copy, no transcode).""" return self.preset_id is None and ( self.trim_start is not None or self.trim_end is not None )