initialize

backend/app/services/media.py

@@ -0,0 +1,104 @@
+import subprocess
+from pathlib import Path
+from typing import List, Optional
+
+from app.config import COVER_DIR, HLS_DIR, PROCESSED_DIR
+
+
+class FFmpegError(RuntimeError):
+    pass
+
+
+def run_ffmpeg(command: List[str]) -> None:
+    completed = subprocess.run(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    if completed.returncode != 0:
+        raise FFmpegError(completed.stderr.strip() or "ffmpeg command failed")
+
+
+def build_filter(watermark_text: Optional[str]) -> Optional[str]:
+    if not watermark_text:
+        return None
+    escaped = watermark_text.replace(":", r"\:").replace("'", r"\\'")
+    return (
+        "drawtext=text='"
+        f"{escaped}"
+        "':fontcolor=white:fontsize=28:box=1:boxcolor=black@0.45:boxborderw=6:x=w-tw-24:y=h-th-24"
+    )
+
+
+def transcode_to_mp4(
+    source: Path, task_id: str, clip_seconds: Optional[int], watermark_text: Optional[str]
+) -> Path:
+    output = PROCESSED_DIR / f"{task_id}.mp4"
+    command = ["ffmpeg", "-y", "-i", str(source)]
+    if clip_seconds:
+        command.extend(["-t", str(clip_seconds)])
+
+    video_filter = build_filter(watermark_text)
+    if video_filter:
+        command.extend(["-vf", video_filter])
+
+    command.extend(
+        [
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-c:a",
+            "aac",
+            "-movflags",
+            "+faststart",
+            str(output),
+        ]
+    )
+    run_ffmpeg(command)
+    return output
+
+
+def generate_cover(source: Path, task_id: str) -> Path:
+    output = COVER_DIR / f"{task_id}.jpg"
+    command = [
+        "ffmpeg",
+        "-y",
+        "-ss",
+        "00:00:01",
+        "-i",
+        str(source),
+        "-frames:v",
+        "1",
+        str(output),
+    ]
+    run_ffmpeg(command)
+    return output
+
+
+def generate_hls(source_mp4: Path, task_id: str) -> Path:
+    task_dir = HLS_DIR / task_id
+    task_dir.mkdir(parents=True, exist_ok=True)
+    playlist = task_dir / "index.m3u8"
+    segment_pattern = task_dir / "segment_%03d.ts"
+    command = [
+        "ffmpeg",
+        "-y",
+        "-i",
+        str(source_mp4),
+        "-codec:v",
+        "libx264",
+        "-codec:a",
+        "aac",
+        "-hls_time",
+        "4",
+        "-hls_playlist_type",
+        "vod",
+        "-hls_segment_filename",
+        str(segment_pattern),
+        str(playlist),
+    ]
+    run_ffmpeg(command)
+    return playlist