feat(backend): youtube karaoke pipeline orchestrator

tests/test_youtube_pipeline.py

@@ -0,0 +1,91 @@
+from unittest.mock import MagicMock
+from pathlib import Path
+from youtube_pipeline import run_pipeline, PipelineDeps
+from youtube import YoutubeFetchError, VideoTooLong
+
+
+def _deps(tmp_path, existing_song=None):
+    repo = MagicMock()
+    repo.get_song.return_value = existing_song
+    storage = MagicMock()
+    storage.public_url.side_effect = lambda k: f"https://pub.example.com/{k}"
+
+    def fake_fetch_meta(url, max_duration_sec):
+        return {
+            "youtube_id": "abc12345678",
+            "title": "T", "artist": "A",
+            "duration_sec": 100, "thumbnail_url": "https://thumb",
+        }
+
+    def fake_download_audio(url, yt_id, out_dir):
+        p = Path(out_dir) / f"{yt_id}.m4a"
+        p.write_bytes(b"audio")
+        return p
+
+    def fake_separate(audio_path):
+        out = audio_path.parent / "no_vocals.mp3"
+        out.write_bytes(b"instrumental")
+        return out
+
+    return PipelineDeps(
+        repo=repo,
+        storage=storage,
+        fetch_metadata=fake_fetch_meta,
+        download_audio=fake_download_audio,
+        separate_vocals=fake_separate,
+        work_dir=tmp_path,
+        max_duration_sec=600,
+    ), repo, storage
+
+
+def test_pipeline_cache_hit_skips_download(tmp_path):
+    cached = {
+        "youtube_id": "abc", "slug": "t-a", "status": "ready",
+        "r2_key": "karaoke/abc.mp3",
+    }
+    deps, repo, storage = _deps(tmp_path, existing_song=cached)
+    deps.repo.get_song.return_value = cached
+
+    result = run_pipeline("https://youtu.be/abc12345678", "job1", deps)
+
+    assert result["slug"] == "t-a"
+    deps.storage.upload_file.assert_not_called()
+    repo.update_job.assert_called_with("job1", status="ready", error=None)
+
+
+def test_pipeline_full_path_uploads_and_persists(tmp_path):
+    deps, repo, storage = _deps(tmp_path, existing_song=None)
+
+    result = run_pipeline("https://youtu.be/abc12345678", "job1", deps)
+
+    storage.upload_file.assert_called_once()
+    args, kwargs = storage.upload_file.call_args
+    assert kwargs.get("key", args[1]).startswith("karaoke/")
+    repo.upsert_song.assert_called_once()
+    upserted = repo.upsert_song.call_args[0][0]
+    assert upserted["youtube_id"] == "abc12345678"
+    assert upserted["status"] == "ready"
+    assert upserted["r2_key"].endswith(".mp3")
+    assert result["slug"] == upserted["slug"]
+
+
+def test_pipeline_fetch_error_marks_failed(tmp_path):
+    deps, repo, storage = _deps(tmp_path)
+    def boom(*a, **k): raise YoutubeFetchError("blocked")
+    deps = deps._replace(fetch_metadata=boom)
+
+    result = run_pipeline("https://youtu.be/abc12345678", "job1", deps)
+
+    assert result["status"] == "failed_fetch"
+    repo.update_job.assert_called_with("job1", status="failed_fetch", error="blocked")
+    storage.upload_file.assert_not_called()
+
+
+def test_pipeline_too_long_marks_failed(tmp_path):
+    deps, repo, storage = _deps(tmp_path)
+    def boom(*a, **k): raise VideoTooLong("9999s")
+    deps = deps._replace(fetch_metadata=boom)
+
+    result = run_pipeline("https://youtu.be/abc12345678", "job1", deps)
+
+    assert result["status"] == "failed_fetch"

youtube_pipeline.py

@@ -0,0 +1,79 @@
+from pathlib import Path
+from typing import Any, Callable, NamedTuple
+from db import SongRepo, slugify
+from youtube import (
+    extract_youtube_id, YoutubeFetchError, VideoTooLong, InvalidYoutubeUrl,
+)
+
+
+class PipelineDeps(NamedTuple):
+    repo: SongRepo
+    storage: Any  # R2Storage
+    fetch_metadata: Callable[..., dict]
+    download_audio: Callable[..., Path]
+    separate_vocals: Callable[[Path], Path]
+    work_dir: Path
+    max_duration_sec: int
+
+
+def run_pipeline(url: str, job_id: str, deps: PipelineDeps) -> dict:
+    try:
+        youtube_id = extract_youtube_id(url)
+    except InvalidYoutubeUrl:
+        deps.repo.update_job(job_id, status="failed_fetch", error="invalid url")
+        return {"status": "failed_fetch", "error": "invalid url"}
+
+    # Cache hit
+    cached = deps.repo.get_song(youtube_id)
+    if cached and cached.get("status") == "ready" and cached.get("r2_key"):
+        deps.repo.update_job(job_id, status="ready", error=None)
+        return {"status": "ready", "slug": cached["slug"], "youtube_id": youtube_id}
+
+    # Fetch metadata
+    deps.repo.update_job(job_id, status="fetching")
+    try:
+        meta = deps.fetch_metadata(url, max_duration_sec=deps.max_duration_sec)
+    except (YoutubeFetchError, VideoTooLong) as e:
+        deps.repo.update_job(job_id, status="failed_fetch", error=str(e))
+        return {"status": "failed_fetch", "error": str(e)}
+
+    # Download
+    work = deps.work_dir / youtube_id
+    work.mkdir(parents=True, exist_ok=True)
+    try:
+        audio_path = deps.download_audio(url, youtube_id, work)
+    except YoutubeFetchError as e:
+        deps.repo.update_job(job_id, status="failed_fetch", error=str(e))
+        return {"status": "failed_fetch", "error": str(e)}
+
+    # Separate
+    deps.repo.update_job(job_id, status="separating")
+    try:
+        instrumental_path = deps.separate_vocals(audio_path)
+    except Exception as e:
+        deps.repo.update_job(job_id, status="failed_separation", error=str(e)[:200])
+        return {"status": "failed_separation", "error": str(e)}
+
+    # Upload
+    deps.repo.update_job(job_id, status="uploading")
+    key = f"karaoke/{youtube_id}.mp3"
+    try:
+        deps.storage.upload_file(instrumental_path, key, content_type="audio/mpeg")
+    except Exception as e:
+        deps.repo.update_job(job_id, status="failed_other", error=str(e)[:200])
+        return {"status": "failed_other", "error": str(e)}
+
+    # Persist
+    slug = slugify(meta["title"], meta["artist"])
+    deps.repo.upsert_song({
+        "youtube_id": youtube_id,
+        "slug": slug,
+        "title": meta["title"],
+        "artist": meta["artist"],
+        "duration_sec": meta["duration_sec"],
+        "r2_key": key,
+        "thumbnail_url": meta["thumbnail_url"],
+        "status": "ready",
+    })
+    deps.repo.update_job(job_id, status="ready", error=None)
+    return {"status": "ready", "slug": slug, "youtube_id": youtube_id}