feat: pipeline framework with stage caching and force-from

src/karaoke_maker/pipeline.py

@@ -0,0 +1,76 @@
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable
+
+from .config import Config
+
+class PipelineError(Exception):
+    pass
+
+@dataclass
+class SongInput:
+    title: str
+    artist: str
+    lyrics_text: str
+    youtube_url: str | None = None
+    video_path: Path | None = None
+    genre: str | None = None
+    year: int | None = None
+    language: str | None = None
+    cover_path: Path | None = None      # local file or URL already fetched by front-end
+    bpm_override: float | None = None
+
+@dataclass
+class Stage:
+    name: str
+    outputs: Callable[["Ctx"], list[Path]]
+    run: Callable[["Ctx"], None]
+
+@dataclass
+class Ctx:
+    inp: SongInput
+    cfg: Config
+    work: Path                          # per-song work dir (created)
+    log: Callable[[str], None]
+
+@dataclass
+class PipelineResult:
+    output_dir: Path
+    warnings: list[str] = field(default_factory=list)
+
+STAGE_ORDER = ["acquire", "separate", "transcribe", "align", "pitch", "notes", "grid", "package"]
+
+def song_slug(inp: SongInput) -> str:
+    name = f"{inp.artist} - {inp.title}"
+    return name.replace("/", "").replace("\0", "")
+
+def _real_stages() -> list[Stage]:
+    from . import stages
+    return stages.all_stages()
+
+def run_pipeline(inp: SongInput, cfg: Config, log=print,
+                 force_from: str | None = None,
+                 stages: list[Stage] | None = None) -> PipelineResult:
+    if stages is None:
+        stages = _real_stages()
+    names = [s.name for s in stages]
+    if force_from is not None and force_from not in names:
+        raise PipelineError(f"unknown stage {force_from!r}, expected one of {names}")
+    work = cfg.work_dir / song_slug(inp)
+    work.mkdir(parents=True, exist_ok=True)
+    ctx = Ctx(inp=inp, cfg=cfg, work=work, log=log)
+    forcing = False
+    for stage in stages:
+        if stage.name == force_from:
+            forcing = True
+        if not forcing and all(p.exists() for p in stage.outputs(ctx)):
+            log(f"[{stage.name}] cached, skipping")
+            continue
+        log(f"[{stage.name}] running")
+        stage.run(ctx)
+    warnings = []
+    notes_file = work / "notes_sec.json"
+    if notes_file.exists():
+        warnings = json.loads(notes_file.read_text()).get("warnings", [])
+    return PipelineResult(output_dir=cfg.output_dir / song_slug(inp), warnings=warnings)

tests/test_pipeline.py

@@ -0,0 +1,48 @@
+import json
+from pathlib import Path
+from karaoke_maker.config import Config
+from karaoke_maker.pipeline import SongInput, Stage, run_pipeline, song_slug
+
+def make_cfg(tmp_path):
+    return Config(whisper_url="http://x/v1", output_dir=tmp_path / "out",
+                  work_dir=tmp_path / "work")
+
+def make_inp():
+    return SongInput(title="T/est", artist="A", lyrics_text="la la")
+
+def fake_stage(name, calls):
+    def outputs(ctx):
+        return [ctx.work / f"{name}.done"]
+    def run(ctx):
+        calls.append(name)
+        (ctx.work / f"{name}.done").write_text("x")
+    return Stage(name=name, outputs=outputs, run=run)
+
+def test_slug_strips_slash():
+    assert song_slug(make_inp()) == "A - Test"
+
+def test_runs_all_stages_then_skips_cached(tmp_path):
+    cfg, inp, calls = make_cfg(tmp_path), make_inp(), []
+    stages = [fake_stage("s1", calls), fake_stage("s2", calls)]
+    run_pipeline(inp, cfg, log=lambda m: None, stages=stages)
+    assert calls == ["s1", "s2"]
+    run_pipeline(inp, cfg, log=lambda m: None, stages=stages)
+    assert calls == ["s1", "s2"]
+
+def test_force_from_reruns_tail(tmp_path):
+    cfg, inp, calls = make_cfg(tmp_path), make_inp(), []
+    stages = [fake_stage("s1", calls), fake_stage("s2", calls), fake_stage("s3", calls)]
+    run_pipeline(inp, cfg, log=lambda m: None, stages=stages)
+    calls.clear()
+    run_pipeline(inp, cfg, log=lambda m: None, stages=stages, force_from="s2")
+    assert calls == ["s2", "s3"]
+
+def test_warnings_surfaced(tmp_path):
+    cfg, inp = make_cfg(tmp_path), make_inp()
+    def run(ctx):
+        (ctx.work / "notes_sec.json").write_text(
+            json.dumps({"notes": [], "breaks": [], "warnings": ["w1"]}))
+        (ctx.work / "n.done").write_text("x")
+    st = Stage(name="n", outputs=lambda ctx: [ctx.work / "n.done"], run=run)
+    result = run_pipeline(inp, cfg, log=lambda m: None, stages=[st])
+    assert result.warnings == ["w1"]