feat: map input changes to the pipeline stage they invalidate

2026-07-27 01:35:51 -07:00
parent d65e55b137
commit 6536496a12
2 changed files with 199 additions and 0 deletions
@@ -0,0 +1,73 @@
+import shutil
+from pathlib import Path
+
+from .config import Config
+from .pipeline import STAGE_ORDER, PipelineError, SongInput, song_slug
+
+# The earliest stage that reads each SongInput field. Editing a field means every
+# stage from here onward is stale. Adding a field to SongInput without adding it
+# here would let an edit invalidate nothing at all, so tests pin this to the
+# dataclass.
+FIELD_STAGE = {
+    "youtube_url": "acquire",
+    "video_path": "acquire",
+    "language": "transcribe",
+    "lyrics_text": "align",
+    "title": "grid",
+    "artist": "grid",
+    "genre": "grid",
+    "year": "grid",
+    "bpm_override": "grid",
+    "cover_path": "package",
+}
+
+
+def changed_fields(old: SongInput, new: SongInput) -> list[str]:
+    return [f for f in FIELD_STAGE if getattr(old, f) != getattr(new, f)]
+
+
+def plan(old: SongInput, new: SongInput) -> str | None:
+    """The stage to restart from, or None when nothing changed.
+
+    None is not "do nothing": run_pipeline with no force_from resumes at the
+    first stage whose outputs are missing, which is exactly what a rerun wants.
+    """
+    changed = changed_fields(old, new)
+    if not changed:
+        return None
+    return min((FIELD_STAGE[f] for f in changed), key=STAGE_ORDER.index)
+
+
+def relocate_work(old: SongInput, new: SongInput, cfg: Config) -> bool:
+    """Move the cached work dir when title/artist changed its slug.
+
+    Nothing inside a work dir depends on the slug except song.txt, which `grid`
+    rewrites — and FIELD_STAGE sends title/artist edits to `grid`. So a retitle
+    costs a rename plus grid and package, not the whole GPU pipeline.
+    """
+    old_dir = cfg.work_dir / song_slug(old)
+    new_dir = cfg.work_dir / song_slug(new)
+    if old_dir == new_dir or not old_dir.exists():
+        return False
+    if new_dir.exists():
+        raise PipelineError(
+            f"a work folder already exists for {song_slug(new)!r}; "
+            "rename or remove it before rebuilding")
+    new_dir.parent.mkdir(parents=True, exist_ok=True)
+    shutil.move(str(old_dir), str(new_dir))
+    return True
+
+
+def superseded_output(old_dir: str | None, new_dir: str,
+                      output_root: Path) -> Path | None:
+    """The previous build's folder, when the rebuild replaced it under a new slug.
+
+    Returns None unless the path is strictly inside output_root, so a bad job
+    record cannot turn the caller's removal into an arbitrary delete.
+    """
+    if not old_dir:
+        return None
+    old, new = Path(old_dir).resolve(), Path(new_dir).resolve()
+    if old == new or output_root.resolve() not in old.parents:
+        return None
+    return old
@@ -0,0 +1,126 @@
+import dataclasses
+from pathlib import Path
+
+import pytest
+
+from karaoke_maker.config import Config
+from karaoke_maker.pipeline import STAGE_ORDER, PipelineError, SongInput
+from karaoke_maker.rebuild import (FIELD_STAGE, changed_fields, plan,
+                                   relocate_work, superseded_output)
+
+
+def song(**kw):
+    base = dict(title="T", artist="A", lyrics_text="la", youtube_url="http://y")
+    return SongInput(**{**base, **kw})
+
+
+def cfg_for(tmp_path):
+    return Config(whisper_url="u", work_dir=tmp_path / "work",
+                  output_dir=tmp_path / "out")
+
+
+def test_every_song_input_field_is_mapped():
+    """A field added to SongInput without a FIELD_STAGE entry would make edits to
+    it invalidate nothing, silently building from stale intermediates."""
+    actual = {f.name for f in dataclasses.fields(SongInput)}
+    assert actual - set(FIELD_STAGE) == set()
+    assert set(FIELD_STAGE) - actual == set()
+
+
+def test_every_mapped_stage_is_a_real_stage():
+    assert set(FIELD_STAGE.values()) <= set(STAGE_ORDER)
+
+
+def test_no_change_means_no_force_from():
+    assert plan(song(), song()) is None
+
+
+@pytest.mark.parametrize("field,value,stage", [
+    ("youtube_url", "http://other", "acquire"),
+    ("video_path", Path("/tmp/v.mp4"), "acquire"),
+    ("language", "de", "transcribe"),
+    ("lyrics_text", "different words", "align"),
+    ("title", "Other", "grid"),
+    ("artist", "Other", "grid"),
+    ("genre", "rock", "grid"),
+    ("year", 1999, "grid"),
+    ("bpm_override", 123.0, "grid"),
+    ("cover_path", Path("/tmp/c.jpg"), "package"),
+])
+def test_single_field_maps_to_its_stage(field, value, stage):
+    assert plan(song(), song(**{field: value})) == stage
+
+
+def test_earliest_stage_wins_when_several_change():
+    new = song(lyrics_text="different", year=1999, language="de")
+    assert plan(song(), new) == "transcribe"
+
+
+def test_changed_fields_lists_what_moved():
+    assert sorted(changed_fields(song(), song(year=1999, genre="rock"))) == \
+        ["genre", "year"]
+
+
+def test_slug_fields_invalidate_no_later_than_the_stage_that_writes_the_slug():
+    """song.txt embeds the slug in #MP3/#COVER/#VIDEO and is written by `grid`,
+    so a title/artist edit must rerun grid or earlier for relocate_work to be safe."""
+    for field in ("title", "artist"):
+        assert STAGE_ORDER.index(FIELD_STAGE[field]) <= STAGE_ORDER.index("grid")
+
+
+def test_relocate_moves_the_work_dir_when_the_slug_changes(tmp_path):
+    cfg = cfg_for(tmp_path)
+    src = cfg.work_dir / "A - T"
+    src.mkdir(parents=True)
+    (src / "audio.wav").write_bytes(b"x")
+    assert relocate_work(song(), song(title="Fixed"), cfg) is True
+    assert not src.exists()
+    assert (cfg.work_dir / "A - Fixed" / "audio.wav").read_bytes() == b"x"
+
+
+def test_relocate_is_a_noop_when_the_slug_is_unchanged(tmp_path):
+    cfg = cfg_for(tmp_path)
+    src = cfg.work_dir / "A - T"
+    src.mkdir(parents=True)
+    assert relocate_work(song(), song(year=1999), cfg) is False
+    assert src.exists()
+
+
+def test_relocate_is_a_noop_when_there_is_nothing_to_move(tmp_path):
+    cfg = cfg_for(tmp_path)
+    assert relocate_work(song(), song(title="Fixed"), cfg) is False
+
+
+def test_relocate_refuses_to_clobber_an_existing_work_dir(tmp_path):
+    cfg = cfg_for(tmp_path)
+    (cfg.work_dir / "A - T").mkdir(parents=True)
+    (cfg.work_dir / "A - Fixed").mkdir(parents=True)
+    with pytest.raises(PipelineError, match="already exists"):
+        relocate_work(song(), song(title="Fixed"), cfg)
+    assert (cfg.work_dir / "A - T").exists()
+
+
+def test_superseded_output_names_the_replaced_folder(tmp_path):
+    root = tmp_path / "out"
+    assert superseded_output(str(root / "A - Old"), str(root / "A - New"), root) == \
+        (root / "A - Old").resolve()
+
+
+def test_superseded_output_is_none_when_the_slug_did_not_change(tmp_path):
+    root = tmp_path / "out"
+    assert superseded_output(str(root / "A - T"), str(root / "A - T"), root) is None
+
+
+def test_superseded_output_is_none_without_a_previous_build(tmp_path):
+    root = tmp_path / "out"
+    assert superseded_output(None, str(root / "A - T"), root) is None
+
+
+def test_superseded_output_refuses_a_path_outside_the_output_root(tmp_path):
+    root = tmp_path / "out"
+    assert superseded_output(str(tmp_path / "elsewhere"), str(root / "A - T"), root) is None
+
+
+def test_superseded_output_refuses_the_output_root_itself(tmp_path):
+    root = tmp_path / "out"
+    assert superseded_output(str(root), str(root / "A - T"), root) is None