from __future__ import annotations import json from datetime import datetime from typing import Annotated, Literal from pydantic import ( BaseModel, ConfigDict, Field, StrictBool, StrictFloat, StrictInt, model_validator, ) EDITOR_SCHEMA_VERSION = 1 EDITOR_STATE_MAX_BYTES = 1_048_576 EDITOR_ID_MAX_LENGTH = 128 EDITOR_LABEL_MAX_LENGTH = 500 MetadataValue = str | StrictInt | StrictFloat | StrictBool | None Metadata = dict[str, MetadataValue] class EditorModel(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) class ClipTransform(EditorModel): x: float = Field(allow_inf_nan=False, ge=-100_000, le=100_000) y: float = Field(allow_inf_nan=False, ge=-100_000, le=100_000) scale_x: float = Field(alias="scaleX", allow_inf_nan=False, gt=0, le=100) scale_y: float = Field(alias="scaleY", allow_inf_nan=False, gt=0, le=100) rotation: float = Field(allow_inf_nan=False, ge=-36_000, le=36_000) class BaseClip(EditorModel): id: str = Field(min_length=1, max_length=EDITOR_ID_MAX_LENGTH) track_id: str = Field(alias="trackId", min_length=1, max_length=EDITOR_ID_MAX_LENGTH) label: str = Field(min_length=1, max_length=EDITOR_LABEL_MAX_LENGTH) start_ms: int = Field(alias="startMs", ge=0) duration_ms: int = Field(alias="durationMs", gt=0) visible: StrictBool opacity: float = Field(allow_inf_nan=False, ge=0, le=1) metadata: Metadata = Field(default_factory=dict, max_length=100) class SourceClip(BaseClip): asset_id: str = Field(alias="assetId", min_length=1, max_length=36) source_start_ms: int = Field(alias="sourceStartMs", ge=0) source_duration_ms: int = Field(alias="sourceDurationMs", gt=0) @model_validator(mode="after") def validate_source_range(self) -> SourceClip: if self.source_duration_ms != self.duration_ms: raise ValueError("Source duration must match timeline duration") return self class MediaClip(SourceClip): kind: Literal["media"] media_type: Literal["video", "image"] = Field(alias="mediaType") transform: ClipTransform volume: float = Field(allow_inf_nan=False, ge=0, le=1) class AudioClip(SourceClip): kind: Literal["audio"] volume: float = Field(allow_inf_nan=False, ge=0, le=1) fade_in_ms: int = Field(alias="fadeInMs", ge=0) fade_out_ms: int = Field(alias="fadeOutMs", ge=0) @model_validator(mode="after") def validate_fades(self) -> AudioClip: if self.fade_in_ms > self.duration_ms or self.fade_out_ms > self.duration_ms: raise ValueError("Audio fades cannot exceed clip duration") return self class CaptionStyle(EditorModel): align: Literal["left", "center", "right"] position: Literal["top", "center", "bottom"] class CaptionClip(BaseClip): kind: Literal["caption"] text: str = Field(max_length=10_000) style: CaptionStyle class EffectClip(BaseClip): kind: Literal["effect"] effect_type: str = Field(alias="effectType", min_length=1, max_length=100) target_clip_id: str | None = Field( default=None, alias="targetClipId", min_length=1, max_length=EDITOR_ID_MAX_LENGTH ) parameters: Metadata = Field(default_factory=dict, max_length=100) TimelineClip = Annotated[ MediaClip | AudioClip | CaptionClip | EffectClip, Field(discriminator="kind") ] class Track(EditorModel): id: str = Field(min_length=1, max_length=EDITOR_ID_MAX_LENGTH) type: Literal["video", "audio", "caption", "overlay"] name: str = Field(min_length=1, max_length=200) order: int = Field(ge=0) muted: StrictBool locked: StrictBool visible: StrictBool clips: list[TimelineClip] @model_validator(mode="after") def validate_clip_ownership(self) -> Track: for clip in self.clips: if clip.track_id != self.id: raise ValueError("Clip trackId must reference its containing track") valid = { "video": {"media"}, "audio": {"audio"}, "caption": {"caption"}, "overlay": {"media", "caption", "effect"}, }[self.type] if clip.kind not in valid: raise ValueError(f"Clip kind {clip.kind} is invalid for a {self.type} track") return self class Transition(EditorModel): id: str = Field(min_length=1, max_length=EDITOR_ID_MAX_LENGTH) type: str = Field(min_length=1, max_length=100) from_clip_id: str = Field(alias="fromClipId", min_length=1, max_length=EDITOR_ID_MAX_LENGTH) to_clip_id: str = Field(alias="toClipId", min_length=1, max_length=EDITOR_ID_MAX_LENGTH) duration_ms: int = Field(alias="durationMs", gt=0) metadata: Metadata = Field(default_factory=dict, max_length=100) class Marker(EditorModel): id: str = Field(min_length=1, max_length=EDITOR_ID_MAX_LENGTH) time_ms: int = Field(alias="timeMs", ge=0) label: str = Field(min_length=1, max_length=500) color_token: str | None = Field(default=None, alias="colorToken", max_length=100) class Timeline(EditorModel): time_unit: Literal["milliseconds"] = Field(alias="timeUnit") tracks: list[Track] transitions: list[Transition] markers: list[Marker] @model_validator(mode="after") def validate_structure(self) -> Timeline: track_ids = [track.id for track in self.tracks] track_orders = [track.order for track in self.tracks] clip_ids = [clip.id for track in self.tracks for clip in track.clips] marker_ids = [marker.id for marker in self.markers] transition_ids = [item.id for item in self.transitions] for label, values in ( ("track", track_ids), ("track order", track_orders), ("clip", clip_ids), ("marker", marker_ids), ("transition", transition_ids), ): if len(values) != len(set(values)): raise ValueError(f"Duplicate {label} identifiers are not allowed") if track_orders and set(track_orders) != set(range(len(track_orders))): raise ValueError("Track order must be contiguous and zero-based") known_clips = set(clip_ids) for transition in self.transitions: if ( transition.from_clip_id not in known_clips or transition.to_clip_id not in known_clips ): raise ValueError("Transition clip references are invalid") if transition.from_clip_id == transition.to_clip_id: raise ValueError("A transition must reference two clips") for track in self.tracks: for clip in track.clips: if isinstance(clip, EffectClip) and ( clip.target_clip_id is not None and clip.target_clip_id not in known_clips ): raise ValueError("Effect target clip reference is invalid") return self class EditorRenderSettings(EditorModel): format: Literal["mp4", "webm"] width: int = Field(ge=2) height: int = Field(ge=2) frame_rate: float = Field(alias="frameRate", allow_inf_nan=False, gt=0, le=120) @model_validator(mode="after") def validate_even_resolution(self) -> EditorRenderSettings: if self.width % 2 or self.height % 2: raise ValueError("Render width and height must be even") return self class EditorDocument(EditorModel): schema_version: Literal[EDITOR_SCHEMA_VERSION] = Field(alias="schemaVersion") project_id: str = Field(alias="projectId", min_length=1, max_length=36) timeline: Timeline render_settings: EditorRenderSettings = Field(alias="renderSettings") def json_bytes(self) -> bytes: return json.dumps( self.model_dump(by_alias=True), ensure_ascii=False, allow_nan=False, separators=(",", ":"), ).encode("utf-8") def asset_ids(self) -> set[str]: return { clip.asset_id for track in self.timeline.tracks for clip in track.clips if isinstance(clip, (MediaClip, AudioClip)) } def duration_ms(self) -> int: return max( ( clip.start_ms + clip.duration_ms for track in self.timeline.tracks for clip in track.clips ), default=0, ) class EditorSaveRequest(EditorModel): expected_revision: int = Field(ge=0) schema_version: Literal[EDITOR_SCHEMA_VERSION] state: EditorDocument @model_validator(mode="after") def validate_versions(self) -> EditorSaveRequest: if self.state.schema_version != self.schema_version: raise ValueError("Editor schema versions do not match") if len(self.state.json_bytes()) > EDITOR_STATE_MAX_BYTES: raise ValueError("Editor state exceeds maximum allowed size") return self class EditorStateResponse(BaseModel): project_id: str revision: int = Field(ge=1) schema_version: int state: EditorDocument created_at: datetime updated_at: datetime updated_by: str class ProjectRenderCreate(EditorModel): editor_revision: int = Field(ge=1) output_format: Literal["mp4", "webm"] = "mp4" width: int = Field(default=1920, ge=2) height: int = Field(default=1080, ge=2) frame_rate: float = Field(default=30, allow_inf_nan=False, gt=0, le=120) quality: Literal["draft", "standard", "high"] = "standard" preset: Literal["fast", "balanced", "quality"] = "balanced" @model_validator(mode="after") def validate_resolution(self) -> ProjectRenderCreate: if self.width % 2 or self.height % 2: raise ValueError("Render width and height must be even") return self class ProjectRenderResponse(BaseModel): id: str project_id: str editor_revision: int status: Literal["queued", "processing", "completed", "failed", "cancelling", "cancelled"] render_settings: dict[str, object] progress: None = None output_asset_id: str | None error_code: str | None error_message: str | None attempt_count: int created_at: datetime started_at: datetime | None completed_at: datetime | None cancelled_at: datetime | None updated_at: datetime class ProjectRenderListResponse(BaseModel): items: list[ProjectRenderResponse]