From da1ed5cabacf178c46b962de82c63a2a0d6c3674 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Fri, 10 Jul 2026 21:18:09 +0200 Subject: [PATCH] docs: add YouTube adapter plan (slice 1, plan 3b) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../plans/2026-07-10-lyra-youtube-adapter.md | 495 ++++++++++++++++++ 1 file changed, 495 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-10-lyra-youtube-adapter.md diff --git a/docs/superpowers/plans/2026-07-10-lyra-youtube-adapter.md b/docs/superpowers/plans/2026-07-10-lyra-youtube-adapter.md new file mode 100644 index 0000000..836367d --- /dev/null +++ b/docs/superpowers/plans/2026-07-10-lyra-youtube-adapter.md @@ -0,0 +1,495 @@ +# Lyra YouTube Adapter Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship Lyra's first REAL source adapter — YouTube via yt-dlp — implementing the existing `SourceAdapter` contract, and wire it into `registry.build_adapters()` so the running worker actually acquires music from YouTube. The adapter's logic is unit-tested offline against a fake yt-dlp client; the real network path is exercised only by an opt-in live test the user runs. + +**Architecture:** `YouTubeAdapter` depends on a small injectable `YtClient` seam (two methods: `search_album`, `download`). Tests inject a `FakeYtClient` returning recorded-shape dicts — no network, no credentials, deterministic. The real `YtDlpClient` wraps `yt_dlp.YoutubeDL` and is constructed only inside `registry.build_adapters()`. Adding YouTube changes nothing in the pipeline/ranker/confidence — they already operate purely on `Candidate`/`SourceAdapter`. + +**Tech Stack:** Python 3.12 worker; `yt-dlp` (new dependency); `ffmpeg` in the worker image (for `FFmpegExtractAudio`). Tests: pytest with a fake client (offline) + an opt-in live test gated by `LYRA_LIVE_TESTS`. + +## Global Constraints + +- Python 3.12 in the worker container; local test runs may use newer 3.x — keep deps version-flexible. +- **New dependency allowed:** `yt-dlp` (this is the first plan that adds real source tooling). No other new deps. +- The `SourceAdapter` protocol, `Candidate`/`Quality`/`DownloadResult`/`MBTarget` types, the ranker, confidence, and pipeline are UNCHANGED. The adapter only *implements* the protocol. +- `Candidate.source_tier` is the single source of truth for ranking (set by the adapter / stamped by the pipeline). YouTube: `name="youtube"`, `tier=2`, quality is lossy (`Quality(fmt="OPUS", lossless=False, bitrate_kbps=160)`). +- The real `YtDlpClient` is NOT unit-tested offline (it hits the network); it is covered by an OPT-IN live test skipped unless env `LYRA_LIVE_TESTS` is set. Never make normal CI/test runs depend on network. +- `registry.build_adapters()` returns only REAL, usable adapters. After this plan that is `[YouTubeAdapter(YtDlpClient())]`; Qobuz/Soulseek are added by later plans. The fake adapters remain in `adapters/fakes.py` for pipeline tests only. +- Every task ends with a commit. + +## Shared interfaces (defined by tasks below; listed for consistency) + +```python +# worker/lyra_worker/adapters/youtube.py (Task 1) +class YtClient(Protocol): + def search_album(self, artist: str, album: str) -> list[dict]: ... # each: {source_ref, title, artist, track_count} + def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: ... # {track_count, path}; raises on failure + +class YouTubeAdapter: # implements SourceAdapter + name = "youtube"; tier = 2 + def __init__(self, client: YtClient): ... + def health(self) -> bool: ... + def search(self, target: MBTarget) -> list[Candidate]: ... + def download(self, candidate: Candidate, dest: str, on_progress) -> DownloadResult: ... + +# worker/lyra_worker/adapters/_ytdlp.py (Task 2) +class YtDlpClient: # real YtClient wrapping yt_dlp.YoutubeDL + def search_album(self, artist, album) -> list[dict]: ... + def download(self, source_ref, dest, on_progress) -> dict: ... +``` + +--- + +### Task 1: YouTubeAdapter + client seam (offline, fake client) + +**Files:** +- Create: `worker/lyra_worker/adapters/youtube.py` +- Test: `worker/tests/test_youtube_adapter.py` + +**Interfaces:** +- Consumes: `MBTarget`, `Candidate`, `Quality`, `DownloadResult` (types.py); `SourceAdapter` (adapters/base.py). +- Produces: the `YtClient` Protocol and `YouTubeAdapter` (see Shared interfaces). `search` maps each client result dict to a `Candidate(source="youtube", source_ref, matched_artist=result["artist"], matched_album=result["title"], quality=lossy OPUS 160, track_count, source_tier=2)`. `download` calls `client.download`; on exception returns `DownloadResult(ok=False, error=str(e))`, else `DownloadResult(ok=True, path, track_count)`. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_youtube_adapter.py`: +```python +from lyra_worker.adapters.youtube import YouTubeAdapter +from lyra_worker.types import Candidate, MBTarget, Quality + +TARGET = MBTarget(artist="Radiohead", album="In Rainbows", track_count=10) + + +class FakeYtClient: + def __init__(self, results=None, fail=False): + self._results = results if results is not None else [ + {"source_ref": "yt:playlist:abc", "title": "In Rainbows", "artist": "Radiohead", "track_count": 10} + ] + self._fail = fail + self.downloaded = [] + + def search_album(self, artist, album): + return self._results + + def download(self, source_ref, dest, on_progress): + if self._fail: + raise RuntimeError("network boom") + on_progress(1.0) + self.downloaded.append((source_ref, dest)) + return {"track_count": 10, "path": dest} + + +def test_name_and_tier(): + a = YouTubeAdapter(FakeYtClient()) + assert a.name == "youtube" + assert a.tier == 2 + assert a.health() is True + + +def test_search_maps_results_to_candidates(): + cands = YouTubeAdapter(FakeYtClient()).search(TARGET) + assert len(cands) == 1 + c = cands[0] + assert c.source == "youtube" + assert c.source_ref == "yt:playlist:abc" + assert c.matched_artist == "Radiohead" + assert c.matched_album == "In Rainbows" + assert c.track_count == 10 + assert c.source_tier == 2 + assert c.quality.lossless is False + + +def test_search_empty_when_no_results(): + assert YouTubeAdapter(FakeYtClient(results=[])).search(TARGET) == [] + + +def test_download_success(): + client = FakeYtClient() + cand = YouTubeAdapter(client).search(TARGET)[0] + seen = [] + result = YouTubeAdapter(client).download(cand, "/tmp/x", seen.append) + assert result.ok is True + assert result.track_count == 10 + assert seen and seen[-1] == 1.0 + + +def test_download_failure_returns_not_ok(): + client = FakeYtClient(fail=True) + cand = YouTubeAdapter(FakeYtClient()).search(TARGET)[0] + result = YouTubeAdapter(client).download(cand, "/tmp/x", lambda _p: None) + assert result.ok is False + assert result.error +``` + +- [ ] **Step 2: Run the test to verify it fails** + +Run: +```bash +cd worker +worker/.venv/bin/python -m pytest tests/test_youtube_adapter.py -v +``` +Expected: FAIL — `ModuleNotFoundError: No module named 'lyra_worker.adapters.youtube'`. + +- [ ] **Step 3: Implement the adapter** + +`worker/lyra_worker/adapters/youtube.py`: +```python +from typing import Callable, Protocol + +from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality + +_YT_QUALITY = Quality(fmt="OPUS", lossless=False, bitrate_kbps=160) + + +class YtClient(Protocol): + def search_album(self, artist: str, album: str) -> list[dict]: + """Return album matches: dicts with keys source_ref, title, artist, track_count.""" + ... + + def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: + """Download into dest; return {track_count, path}. Raise on failure.""" + ... + + +class YouTubeAdapter: + name = "youtube" + tier = 2 + + def __init__(self, client: YtClient): + self._client = client + + def health(self) -> bool: + return True + + def search(self, target: MBTarget) -> list[Candidate]: + candidates: list[Candidate] = [] + for r in self._client.search_album(target.artist, target.album): + candidates.append( + Candidate( + source="youtube", + source_ref=r["source_ref"], + matched_artist=r.get("artist", ""), + matched_album=r.get("title", ""), + quality=_YT_QUALITY, + track_count=r.get("track_count", 1) or 1, + source_tier=self.tier, + ) + ) + return candidates + + def download( + self, candidate: Candidate, dest: str, on_progress: Callable[[float], None] + ) -> DownloadResult: + try: + info = self._client.download(candidate.source_ref, dest, on_progress) + except Exception as e: # network/extraction failures -> fall-through in the pipeline + return DownloadResult(ok=False, error=str(e)) + return DownloadResult( + ok=True, + path=info.get("path", dest), + track_count=info.get("track_count", candidate.track_count), + ) +``` + +- [ ] **Step 4: Run the test to verify it passes** + +Run: +```bash +cd worker +worker/.venv/bin/python -m pytest tests/test_youtube_adapter.py -v +``` +Expected: PASS — all five tests green. + +- [ ] **Step 5: Commit** + +```bash +git add worker/lyra_worker/adapters/youtube.py worker/tests/test_youtube_adapter.py +git commit -m "feat: add YouTubeAdapter with injectable client seam" +``` + +--- + +### Task 2: Real YtDlpClient + yt-dlp dependency + ffmpeg + opt-in live test + +**Files:** +- Create: `worker/lyra_worker/adapters/_ytdlp.py` +- Create: `worker/tests/test_youtube_live.py` +- Modify: `worker/requirements.txt` +- Modify: `worker/Dockerfile` + +**Interfaces:** +- Consumes: `yt_dlp`. +- Produces: `YtDlpClient` implementing the `YtClient` seam using `yt_dlp.YoutubeDL`. `search_album` runs a `ytsearch5:` query and maps entries; `download` fetches bestaudio into `dest` with a progress hook and `FFmpegExtractAudio`. + +- [ ] **Step 1: Add the dependency and ffmpeg** + +Append to `worker/requirements.txt`: +``` +yt-dlp>=2024.1 +``` +Install into the venv: `worker/.venv/bin/pip install -r worker/requirements.txt` + +In `worker/Dockerfile`, add `ffmpeg` to the image (yt-dlp needs it for audio extraction). Change the first lines from: +```dockerfile +FROM python:3.12-slim +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +``` +to: +```dockerfile +FROM python:3.12-slim +WORKDIR /app +RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg && rm -rf /var/lib/apt/lists/* +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +``` + +- [ ] **Step 2: Implement the real client** + +`worker/lyra_worker/adapters/_ytdlp.py`: +```python +import os +from typing import Callable + +import yt_dlp + + +class YtDlpClient: + """Real YtClient backed by yt_dlp. Not unit-tested offline; see test_youtube_live.py.""" + + def search_album(self, artist: str, album: str) -> list[dict]: + query = f"ytsearch5:{artist} {album} full album" + opts = { + "quiet": True, + "no_warnings": True, + "extract_flat": "in_playlist", + "skip_download": True, + } + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(query, download=False) + results: list[dict] = [] + for e in (info.get("entries") or []): + if not e: + continue + results.append( + { + "source_ref": e.get("url") or e.get("webpage_url") or e.get("id"), + "title": e.get("title") or "", + "artist": e.get("uploader") or e.get("channel") or "", + "track_count": e.get("playlist_count") or 1, + } + ) + return results + + def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: + os.makedirs(dest, exist_ok=True) + + def hook(d: dict) -> None: + if d.get("status") == "downloading": + total = d.get("total_bytes") or d.get("total_bytes_estimate") + if total: + on_progress(min(1.0, d.get("downloaded_bytes", 0) / total)) + elif d.get("status") == "finished": + on_progress(1.0) + + opts = { + "quiet": True, + "no_warnings": True, + "format": "bestaudio/best", + "outtmpl": os.path.join(dest, "%(playlist_index)s-%(title)s.%(ext)s"), + "progress_hooks": [hook], + "postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "opus"}], + } + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(source_ref, download=True) + entries = info.get("entries") if info.get("entries") is not None else [info] + return {"track_count": len([e for e in entries if e]), "path": dest} +``` + +- [ ] **Step 3: Write the opt-in live test** + +`worker/tests/test_youtube_live.py`: +```python +import os + +import pytest + +from lyra_worker.adapters._ytdlp import YtDlpClient + +pytestmark = pytest.mark.skipif( + not os.environ.get("LYRA_LIVE_TESTS"), + reason="live network test; set LYRA_LIVE_TESTS=1 to run", +) + + +def test_live_search_returns_results(): + results = YtDlpClient().search_album("Radiohead", "In Rainbows") + assert len(results) >= 1 + assert all(r["source_ref"] for r in results) +``` + +- [ ] **Step 4: Verify (offline) the module imports and the live test is skipped** + +Run: +```bash +cd worker +worker/.venv/bin/python -c "from lyra_worker.adapters._ytdlp import YtDlpClient; print('import ok')" +worker/.venv/bin/python -m pytest tests/test_youtube_live.py -v +``` +Expected: prints `import ok`; the live test is reported **skipped** (LYRA_LIVE_TESTS not set). Do NOT set LYRA_LIVE_TESTS here — the live test is for the user to run on their network. + +- [ ] **Step 5: Commit** + +```bash +git add worker/lyra_worker/adapters/_ytdlp.py worker/tests/test_youtube_live.py worker/requirements.txt worker/Dockerfile +git commit -m "feat: add real yt-dlp client and opt-in live test; ffmpeg in worker image" +``` + +--- + +### Task 3: Wire the real YouTube adapter into the registry + +**Files:** +- Modify: `worker/lyra_worker/registry.py` +- Modify: `worker/tests/test_registry.py` + +**Interfaces:** +- Consumes: `YouTubeAdapter` (Task 1), `YtDlpClient` (Task 2). +- Produces: `build_adapters()` returns `[YouTubeAdapter(YtDlpClient())]` — the real, usable adapters. (Qobuz/Soulseek added by later plans.) + +- [ ] **Step 1: Update the registry test (write the new expectation first)** + +Replace `worker/tests/test_registry.py` with: +```python +from lyra_worker.adapters.youtube import YouTubeAdapter +from lyra_worker.registry import build_adapters + + +def test_registry_returns_real_youtube_adapter(): + adapters = build_adapters() + names = [a.name for a in adapters] + assert "youtube" in names + yt = next(a for a in adapters if a.name == "youtube") + assert isinstance(yt, YouTubeAdapter) + assert yt.tier == 2 + assert yt.health() is True + + +def test_registry_adapter_names_are_unique(): + names = [a.name for a in build_adapters()] + assert len(names) == len(set(names)) +``` + +- [ ] **Step 2: Run the test to verify it fails** + +Run: +```bash +cd worker +worker/.venv/bin/python -m pytest tests/test_registry.py -v +``` +Expected: FAIL — the old `build_adapters` returns the three fakes, so `isinstance(yt, YouTubeAdapter)` fails (the youtube entry is a `FakeYouTube`). + +- [ ] **Step 3: Update the registry** + +Replace `worker/lyra_worker/registry.py` with: +```python +from lyra_worker.adapters._ytdlp import YtDlpClient +from lyra_worker.adapters.base import SourceAdapter +from lyra_worker.adapters.youtube import YouTubeAdapter + + +def build_adapters() -> list[SourceAdapter]: + """The enabled real source adapters, best-tier first. + + Only real, usable adapters are returned. Plan 3c/3d add Qobuz (streamrip) + and Soulseek (slskd), each gated on its config being present via health(). + """ + return [YouTubeAdapter(YtDlpClient())] +``` + +- [ ] **Step 4: Run the test to verify it passes** + +Run: +```bash +cd worker +worker/.venv/bin/python -m pytest tests/test_registry.py -v +``` +Expected: PASS — both tests green. + +- [ ] **Step 5: Commit** + +```bash +git add worker/lyra_worker/registry.py worker/tests/test_registry.py +git commit -m "feat: wire real YouTube adapter into the registry" +``` + +--- + +### Task 4: Pipeline integration with the real adapter class + +**Files:** +- Test: `worker/tests/test_youtube_pipeline.py` + +**Interfaces:** +- Consumes: `run_pipeline` (pipeline.py), `YouTubeAdapter` (Task 1), the `conn` fixture + `insert_request`. +- Produces: proof that the REAL `YouTubeAdapter` class (driven by a fake yt-dlp client, so offline) flows through the full pipeline to `imported`, writing a `LibraryItem` and a chosen `Candidate` with `source="youtube"`. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_youtube_pipeline.py`: +```python +from lyra_worker.adapters.youtube import YouTubeAdapter +from lyra_worker.claim import claim_next +from lyra_worker.pipeline import run_pipeline +from tests.conftest import insert_request +from tests.test_youtube_adapter import FakeYtClient + + +def test_youtube_adapter_flows_through_pipeline(conn): + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + run_pipeline(conn, job_id, [YouTubeAdapter(FakeYtClient())], dest_root="/tmp/lib") + + with conn.cursor() as cur: + cur.execute('SELECT state, "currentStage" FROM "Job" WHERE id = %s', (job_id,)) + assert cur.fetchone() == ("imported", "import") + cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) + assert cur.fetchone()[0] == "youtube" + cur.execute('SELECT source FROM "LibraryItem" WHERE artist = %s', ("Radiohead",)) + assert cur.fetchone()[0] == "youtube" +``` + +- [ ] **Step 2: Run the test to verify it fails, then passes** + +Run: +```bash +cd worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_youtube_pipeline.py -v +``` +Expected: This test should PASS immediately — it exercises already-implemented code (the adapter from Task 1 + the existing pipeline) with no new production code. If it FAILS, investigate a real integration mismatch (e.g. the FakeYtClient's default result must score above the confidence gate against "Radiohead"/"In Rainbows" — it does, since title/artist match exactly). This task is a characterization/integration test; no implementation step is expected. + +- [ ] **Step 3: Run the full worker suite** + +Run: +```bash +cd worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest -v +``` +Expected: PASS — every suite green (youtube adapter, youtube pipeline, registry, config, crypto, pipeline, ranker, etc.), the live test skipped, output pristine. + +- [ ] **Step 4: Commit** + +```bash +git add worker/tests/test_youtube_pipeline.py +git commit -m "test: verify real YouTubeAdapter flows through the pipeline" +``` + +--- + +## Notes for the next plans + +`registry.build_adapters()` now returns the real YouTube adapter. Plan 3c adds `QobuzAdapter` (streamrip) and Plan 3d adds `SoulseekAdapter` (slskd REST client + an slskd service in docker-compose), each reading credentials via `get_config(conn)` and included in `build_adapters()` only when configured (checked through `health()`). The `YtClient`/injected-client pattern established here (real client separate from the adapter, fake client for offline tests, opt-in live test) is the template for both. Plan 4 then adds real MusicBrainz resolution in `intake` (filling `MBTarget.track_count`/`total_duration_s`, which will make the confidence gate and YouTube's `track_count` reporting meaningfully stricter) and real tagging/file-move in `tag`/`import`.