| from datetime import datetime, timezone |
| from enum import Enum |
| from typing import Any, Literal |
|
|
| from pydantic import BaseModel, Field, HttpUrl, field_validator |
|
|
|
|
| def utc_now() -> datetime: |
| return datetime.now(timezone.utc) |
|
|
|
|
| class TargetPlatform(str, Enum): |
| tiktok = "tiktok" |
| youtube_shorts = "youtube_shorts" |
| instagram_reels = "instagram_reels" |
|
|
|
|
| class ChannelProfile(BaseModel): |
| niche: str = Field(default="education", min_length=2, max_length=80) |
| niche_custom: str = Field(default="", max_length=80) |
| channel_description: str = Field(default="", max_length=700) |
| clip_style: str = Field(default="informative", min_length=2, max_length=80) |
| clip_length_seconds: int = Field(default=60, ge=15, le=180) |
| clip_count: int = Field(default=5, ge=1, le=20) |
| primary_language: str = Field(default="Thai", min_length=2, max_length=40) |
| target_platform: TargetPlatform = TargetPlatform.tiktok |
|
|
| @field_validator("niche", "niche_custom", "channel_description", "clip_style", "primary_language") |
| @classmethod |
| def clean_text(cls, value: str) -> str: |
| return value.strip() |
|
|
|
|
| class YoutubeJobRequest(BaseModel): |
| youtube_url: HttpUrl |
| profile: ChannelProfile |
|
|
|
|
| class TranscriptSegment(BaseModel): |
| id: str |
| start_seconds: float = Field(ge=0) |
| end_seconds: float = Field(ge=0) |
| text: str |
| language: str | None = None |
|
|
|
|
| class SubtitleCue(BaseModel): |
| """A single subtitle line with explicit timing relative to clip start.""" |
|
|
| start_seconds: float = Field(ge=0) |
| end_seconds: float = Field(ge=0) |
| text: str = "" |
|
|
|
|
| class SkipRange(BaseModel): |
| """A range to splice out of the middle of a clip (relative to clip start).""" |
|
|
| start_seconds: float = Field(ge=0) |
| end_seconds: float = Field(ge=0) |
|
|
|
|
| class ClipCandidate(BaseModel): |
| id: str |
| start_seconds: float = Field(ge=0) |
| end_seconds: float = Field(ge=0) |
| title: str |
| reason: str |
| score: float = Field(ge=0, le=100) |
| subtitle_text: str = "" |
| subtitle_cues: list[SubtitleCue] | None = None |
| skip_ranges: list[SkipRange] | None = None |
| video_url: str | None = None |
| download_url: str | None = None |
| approved: bool = False |
| deleted: bool = False |
| metadata: dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class ClipPatch(BaseModel): |
| start_seconds: float | None = Field(default=None, ge=0) |
| end_seconds: float | None = Field(default=None, ge=0) |
| subtitle_text: str | None = None |
| subtitle_cues: list[SubtitleCue] | None = None |
| skip_ranges: list[SkipRange] | None = None |
| approved: bool | None = None |
| deleted: bool | None = None |
|
|
|
|
| class RegenerateClipRequest(BaseModel): |
| clip_style: str | None = None |
| clip_length_seconds: int | None = Field(default=None, ge=15, le=180) |
| subtitle_text: str | None = None |
|
|
|
|
| class TranslateSubtitlesRequest(BaseModel): |
| target_language: str = Field(min_length=2, max_length=40) |
|
|
|
|
| class PolishSubtitlesRequest(BaseModel): |
| style: str | None = None |
|
|
|
|
| class JobSnapshot(BaseModel): |
| id: str |
| status: Literal["queued", "running", "completed", "failed"] |
| progress: float = Field(ge=0, le=1) |
| message: str |
| current_step: str = "" |
| step_index: int = Field(default=0, ge=0) |
| step_total: int = Field(default=6, ge=1) |
| active_clip_index: int = Field(default=0, ge=0) |
| active_clip_total: int = Field(default=0, ge=0) |
| source: dict[str, Any] |
| profile: ChannelProfile |
| transcript: list[TranscriptSegment] = Field(default_factory=list) |
| clips: list[ClipCandidate] = Field(default_factory=list) |
| timings: dict[str, float] = Field(default_factory=dict) |
| error: str | None = None |
| created_at: datetime = Field(default_factory=utc_now) |
| updated_at: datetime = Field(default_factory=utc_now) |
|
|
|
|
| class HealthResponse(BaseModel): |
| ok: bool |
| app: str |
| demo_mode: bool |
| accelerator: dict[str, Any] |
|
|