feat: package stage producing final song folder

src/karaoke_maker/stages/__init__.py

@@ -3,6 +3,7 @@ from .acquire import acquire_stage
 from .align import align_stage
 from .grid import grid_stage
 from .notes import notes_stage
+from .package import package_stage
 from .pitch import pitch_stage
 from .separate import separate_stage
 from .transcribe import transcribe_stage
@@ -10,4 +11,4 @@ from .transcribe import transcribe_stage
 
 def all_stages() -> list[Stage]:
     return [acquire_stage(), separate_stage(), transcribe_stage(), align_stage(),
-            pitch_stage(), notes_stage(), grid_stage()]
+            pitch_stage(), notes_stage(), grid_stage(), package_stage()]

src/karaoke_maker/stages/package.py

@@ -0,0 +1,75 @@
+import json
+import shutil
+from pathlib import Path
+
+from PIL import Image
+
+from ..media import ffmpeg, ffprobe_codec
+from ..pipeline import Ctx, Stage, song_slug
+from ..ultrastar import parse_ultrastar, serialize_ultrastar
+
+GOOD_VIDEO = {("webm", "vp9"), ("webm", "vp8"),
+              ("mov,mp4,m4a,3gp,3g2,mj2", "h264")}
+
+
+def prepare_cover(src: Path, dest_jpg: Path) -> None:
+    img = Image.open(src).convert("RGB")
+    side = min(img.size)
+    left = (img.width - side) // 2
+    top = (img.height - side) // 2
+    img = img.crop((left, top, left + side, top + side))
+    if side > 800:
+        img = img.resize((800, 800))
+    img.save(dest_jpg, "JPEG", quality=90)
+
+
+def _set_header(txt_path: Path, key: str, value: str | None) -> None:
+    song = parse_ultrastar(txt_path.read_text(encoding="utf-8"))
+    replaced = False
+    headers = []
+    for k, v in song.headers:
+        if k.upper() == key.upper():
+            replaced = True
+            if value is not None:
+                headers.append((key, value))
+        else:
+            headers.append((k, v))
+    if value is not None and not replaced:
+        headers.append((key, value))
+    song.headers = headers
+    txt_path.write_text(serialize_ultrastar(song), encoding="utf-8")
+
+
+def _run(ctx: Ctx) -> None:
+    slug = song_slug(ctx.inp)
+    out = ctx.cfg.output_dir / slug
+    out.mkdir(parents=True, exist_ok=True)
+    txt = out / f"{slug}.txt"
+    shutil.copyfile(ctx.work / "song.txt", txt)
+    ffmpeg("-i", str(ctx.work / "audio.wav"), "-b:a", "320k",
+           str(out / f"{slug}.mp3"))
+    meta = json.loads((ctx.work / "meta.json").read_text())
+    src_video = ctx.work / meta["video_file"]
+    fmt, codec = ffprobe_codec(src_video)
+    if (fmt, codec) in GOOD_VIDEO:
+        shutil.copyfile(src_video, out / f"{slug}{src_video.suffix}")
+    else:
+        ffmpeg("-i", str(src_video), "-c:v", "libx264", "-crf", "20",
+               "-preset", "medium", "-c:a", "aac", str(out / f"{slug}.mp4"))
+        _set_header(txt, "VIDEO", f"{slug}.mp4")
+    cover_src = ctx.inp.cover_path
+    if cover_src is None and meta.get("thumbnail"):
+        cover_src = ctx.work / meta["thumbnail"]
+    if cover_src and Path(cover_src).exists():
+        prepare_cover(Path(cover_src), out / f"{slug}.jpg")
+    else:
+        ctx.log("no cover available; dropping #COVER header")
+        _set_header(txt, "COVER", None)
+
+
+def package_stage() -> Stage:
+    def outputs(ctx: Ctx):
+        slug = song_slug(ctx.inp)
+        out = ctx.cfg.output_dir / slug
+        return [out / f"{slug}.txt", out / f"{slug}.mp3"]
+    return Stage(name="package", outputs=outputs, run=_run)

tests/test_grid.py

@@ -213,4 +213,4 @@ def test_grid_stage_uses_bpm_override(tmp_path, monkeypatch):
 
 def test_grid_stage_registered():
     names = [s.name for s in all_stages()]
-    assert names[-1] == "grid" and names[-2] == "notes"
+    assert names[-1] == "package" and names[-2] == "grid"

tests/test_package.py

@@ -0,0 +1,104 @@
+import json
+from pathlib import Path
+
+from PIL import Image
+
+from karaoke_maker.config import Config
+from karaoke_maker.pipeline import STAGE_ORDER, Ctx, SongInput
+from karaoke_maker.stages import all_stages, package
+from karaoke_maker.ultrastar import get_header, parse_ultrastar
+
+SONG_TXT = ("#TITLE:T\n#ARTIST:A\n#MP3:A - T.mp3\n#COVER:A - T.jpg\n"
+            "#VIDEO:A - T.webm\n#BPM:200\n#GAP:0\n: 0 1 0 la\nE\n")
+
+
+def make_ctx(tmp_path, cover=None):
+    work = tmp_path / "work"
+    work.mkdir()
+    (work / "song.txt").write_text(SONG_TXT)
+    (work / "audio.wav").write_bytes(b"w")
+    (work / "source_video.webm").write_bytes(b"v")
+    (work / "meta.json").write_text(json.dumps(
+        {"video_file": "source_video.webm", "thumbnail": None}))
+    inp = SongInput(title="T", artist="A", lyrics_text="x", cover_path=cover)
+    cfg = Config(whisper_url="u", output_dir=tmp_path / "out", work_dir=tmp_path)
+    return Ctx(inp=inp, cfg=cfg, work=work, log=lambda m: None)
+
+
+def make_cover(tmp_path):
+    p = tmp_path / "c.png"
+    Image.new("RGB", (640, 480), "red").save(p)
+    return p
+
+
+def test_package_copies_compatible_video(tmp_path, monkeypatch):
+    monkeypatch.setattr(package, "ffmpeg", lambda *a: Path(a[-1]).write_bytes(b"m"))
+    monkeypatch.setattr(package, "ffprobe_codec", lambda p: ("webm", "vp9"))
+    ctx = make_ctx(tmp_path, cover=make_cover(tmp_path))
+    package.package_stage().run(ctx)
+    out = tmp_path / "out" / "A - T"
+    assert (out / "A - T.txt").exists()
+    assert (out / "A - T.mp3").exists()
+    assert (out / "A - T.webm").read_bytes() == b"v"
+    img = Image.open(out / "A - T.jpg")
+    assert img.width == img.height
+
+
+def test_package_reencodes_and_rewrites_video_header(tmp_path, monkeypatch):
+    monkeypatch.setattr(package, "ffmpeg", lambda *a: Path(a[-1]).write_bytes(b"m"))
+    monkeypatch.setattr(package, "ffprobe_codec", lambda p: ("avi", "mpeg4"))
+    ctx = make_ctx(tmp_path, cover=make_cover(tmp_path))
+    package.package_stage().run(ctx)
+    out = tmp_path / "out" / "A - T"
+    song = parse_ultrastar((out / "A - T.txt").read_text())
+    assert get_header(song, "VIDEO") == "A - T.mp4"
+    # VIDEO keeps its original slot, ahead of BPM/GAP
+    keys = [k for k, _ in song.headers]
+    assert keys.index("VIDEO") < keys.index("BPM") < keys.index("GAP")
+
+
+def test_package_no_cover_drops_header(tmp_path, monkeypatch):
+    monkeypatch.setattr(package, "ffmpeg", lambda *a: Path(a[-1]).write_bytes(b"m"))
+    monkeypatch.setattr(package, "ffprobe_codec", lambda p: ("webm", "vp9"))
+    ctx = make_ctx(tmp_path, cover=None)
+    package.package_stage().run(ctx)
+    song = parse_ultrastar((tmp_path / "out" / "A - T" / "A - T.txt").read_text())
+    assert get_header(song, "COVER") is None
+
+
+def test_package_uses_work_thumbnail_when_no_cover_path(tmp_path, monkeypatch):
+    monkeypatch.setattr(package, "ffmpeg", lambda *a: Path(a[-1]).write_bytes(b"m"))
+    monkeypatch.setattr(package, "ffprobe_codec", lambda p: ("webm", "vp9"))
+    ctx = make_ctx(tmp_path, cover=None)
+    make_cover(tmp_path).rename(ctx.work / "thumb.png")
+    meta = json.loads((ctx.work / "meta.json").read_text())
+    meta["thumbnail"] = "thumb.png"
+    (ctx.work / "meta.json").write_text(json.dumps(meta))
+    package.package_stage().run(ctx)
+    out = tmp_path / "out" / "A - T"
+    assert (out / "A - T.jpg").exists()
+    song = parse_ultrastar((out / "A - T.txt").read_text())
+    assert get_header(song, "COVER") == "A - T.jpg"
+
+
+def test_package_outputs_txt_and_mp3():
+    ctx_outputs = package.package_stage().outputs
+    cfg = Config(whisper_url="u", output_dir=Path("out"), work_dir=Path("work"))
+    inp = SongInput(title="T", artist="A", lyrics_text="x")
+    ctx = Ctx(inp=inp, cfg=cfg, work=Path("work"), log=lambda m: None)
+    assert ctx_outputs(ctx) == [Path("out/A - T/A - T.txt"), Path("out/A - T/A - T.mp3")]
+
+
+def test_slash_in_title_is_stripped_from_folder_and_filenames(tmp_path, monkeypatch):
+    monkeypatch.setattr(package, "ffmpeg", lambda *a: Path(a[-1]).write_bytes(b"m"))
+    monkeypatch.setattr(package, "ffprobe_codec", lambda p: ("webm", "vp9"))
+    ctx = make_ctx(tmp_path, cover=make_cover(tmp_path))
+    ctx.inp.title = "T/2"
+    package.package_stage().run(ctx)
+    out = tmp_path / "out" / "A - T2"
+    assert (out / "A - T2.txt").exists()
+    assert (out / "A - T2.mp3").exists()
+
+
+def test_all_stages_matches_stage_order():
+    assert [s.name for s in all_stages()] == STAGE_ORDER