branderson/songtrace
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: rebuild a song, redoing only the stages the edit invalidated

Browse Source
This commit is contained in:
branderson
2026-07-27 01:59:36 -07:00
parent 99c62822b4
commit c016e3c09d
2 changed files with 233 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/karaoke_maker/web.py
+82 -12
View File
@@ -13,8 +13,10 @@ from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from . import rebuild as rebuild_policy
from .config import Config
from .pipeline import STAGE_ORDER, PipelineError, SongInput, run_pipeline
from .pipeline import (STAGE_ORDER, PipelineError, SongInput, run_pipeline,
song_slug)
STATIC = Path(__file__).parent / "static"
STAGE_LINE = re.compile(r"^\[(\w+)\] (?:running|cached, skipping)")
@@ -48,6 +50,21 @@ async def _save_upload(upload: UploadFile, dest: Path) -> None:
await run_in_threadpool(shutil.copyfileobj, upload.file, f)
async def _store_uploads(cfg: Config, job_id: str, video: UploadFile | None,
cover: UploadFile | None) -> tuple[Path | None, Path | None]:
updir = cfg.work_dir / "uploads" / job_id
paths = []
for upload, default in ((video, "video"), (cover, "cover")):
if not (upload and upload.filename):
paths.append(None)
continue
updir.mkdir(parents=True, exist_ok=True)
dest = updir / _safe_filename(upload.filename, default)
await _save_upload(upload, dest)
paths.append(dest)
return paths[0], paths[1]
def _cleanup_uploads(cfg: Config, job_id: str, video_path: Path | None) -> None:
"""Drop the video upload a finished job no longer needs.
@@ -82,6 +99,21 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
lock = threading.Lock()
todo: queue.Queue[str] = queue.Queue()
def _prepare_rebuild(job: Job, log) -> str | None:
"""Relocate cached work and report the stage this rebuild restarts from."""
if not job.rebuilt_from:
return None
with lock:
parent = jobs.get(job.rebuilt_from)
if parent is None:
return None
force_from = rebuild_policy.plan(parent.inp, job.inp)
changed = rebuild_policy.changed_fields(parent.inp, job.inp)
rebuild_policy.relocate_work(parent.inp, job.inp, cfg)
log(f"[rebuild] restarting from {force_from or 'the first incomplete stage'}"
f" ({', '.join(changed) if changed else 'nothing'} changed)")
return force_from
def worker():
while True:
job_id = todo.get()
@@ -97,7 +129,8 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
job.stage = m[1]
try:
result = runner(job.inp, cfg, log=log)
force_from = _prepare_rebuild(job, log)
result = runner(job.inp, cfg, log=log, force_from=force_from)
with lock:
job.status = "done"
job.output_dir = str(result.output_dir)
@@ -152,16 +185,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
raise HTTPException(400, f"bpm must be a finite value > 0, got {bpm!r}")
job_id = uuid.uuid4().hex
updir = cfg.work_dir / "uploads" / job_id
video_path = cover_path = None
if has_video:
updir.mkdir(parents=True, exist_ok=True)
video_path = updir / _safe_filename(video.filename, "video")
await _save_upload(video, video_path)
if cover and cover.filename:
updir.mkdir(parents=True, exist_ok=True)
cover_path = updir / _safe_filename(cover.filename, "cover")
await _save_upload(cover, cover_path)
video_path, cover_path = await _store_uploads(cfg, job_id, video, cover)
inp = SongInput(title=title, artist=artist, lyrics_text=lyrics,
youtube_url=youtube_url, video_path=video_path,
@@ -174,6 +198,52 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
parent = _finished_job(job_id)
return _enqueue(uuid.uuid4().hex, parent.inp, rebuilt_from=job_id)
def _require_source(parent_inp: SongInput, inp: SongInput,
force_from: str | None) -> None:
"""Refuse a rebuild that will rerun acquire against an upload we no longer have."""
if inp.video_path is None or inp.video_path.exists():
return
if force_from != "acquire":
# acquire is cached under the old slug until the worker renames it
for slug in {song_slug(parent_inp), song_slug(inp)}:
work = cfg.work_dir / slug
if (work / "meta.json").exists() and (work / "audio.wav").exists():
return
raise HTTPException(400, "the original video upload is no longer available; "
"attach the video file again")
@app.post("/api/jobs/{job_id}/rebuild", status_code=201)
async def rebuild(job_id: str, title: str = Form(...), artist: str = Form(...),
lyrics: str = Form(...), youtube_url: str | None = Form(None),
genre: str | None = Form(None), year: int | None = Form(None),
language: str | None = Form(None), bpm: float | None = Form(None),
video: UploadFile | None = File(None),
cover: UploadFile | None = File(None)):
parent = _finished_job(job_id)
# a cleared field arrives as "" from some clients; "" and None must not
# look like different inputs to plan()
youtube_url, genre, language = (youtube_url or None, genre or None,
language or None)
has_video = bool(video and video.filename)
if youtube_url and has_video:
raise HTTPException(400, "provide exactly one of youtube_url / video")
if bpm is not None and (not math.isfinite(bpm) or bpm <= 0):
raise HTTPException(400, f"bpm must be a finite value > 0, got {bpm!r}")
new_id = uuid.uuid4().hex
video_path, cover_path = await _store_uploads(cfg, new_id, video, cover)
if not youtube_url and not has_video:
youtube_url, video_path = parent.inp.youtube_url, parent.inp.video_path
if cover_path is None:
cover_path = parent.inp.cover_path
inp = SongInput(title=title, artist=artist, lyrics_text=lyrics,
youtube_url=youtube_url, video_path=video_path,
genre=genre, year=year, language=language,
cover_path=cover_path, bpm_override=bpm)
_require_source(parent.inp, inp, rebuild_policy.plan(parent.inp, inp))
return _enqueue(new_id, inp, rebuilt_from=job_id)
@app.get("/api/stages")
def list_stages():
return {"stages": list(STAGE_ORDER)}
tests/test_web.py
+151 -1
View File
@@ -7,7 +7,7 @@ import pytest
from fastapi.testclient import TestClient
from karaoke_maker.config import Config
from karaoke_maker.pipeline import STAGE_ORDER, PipelineError, PipelineResult
from karaoke_maker.pipeline import STAGE_ORDER, PipelineError, PipelineResult, song_slug
from karaoke_maker.web import create_app
@@ -669,3 +669,153 @@ def test_input_reports_bare_filenames_only(tmp_path):
def test_input_of_an_unknown_job_is_404(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
assert client.get("/api/jobs/nope/input").status_code == 404
def capture_runner_factory(seen):
def capture(inp, cfg, log=print, force_from=None, stages=None):
seen.append({"inp": inp, "force_from": force_from})
return PipelineResult(output_dir=cfg.output_dir / song_slug(inp))
return capture
def rebuild_form(**overrides):
"""The form as the page posts it — empty fields are stripped, not sent blank."""
data = {"title": "T", "artist": "A", "lyrics": "x", "youtube_url": "http://y"}
data.update(overrides)
return {k: v for k, v in data.items() if v != ""}
def test_rebuild_forces_the_stage_the_edit_invalidated(tmp_path):
seen = []
client, _ = make_client(tmp_path, capture_runner_factory(seen))
parent = make_done_job(client)
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(lyrics="new words")).json()["id"]
wait_status(client, child, "done")
assert seen[1]["force_from"] == "align"
assert seen[1]["inp"].lyrics_text == "new words"
def test_rebuild_with_no_edits_forces_nothing(tmp_path):
seen = []
client, _ = make_client(tmp_path, capture_runner_factory(seen))
parent = make_done_job(client)
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form()).json()["id"]
wait_status(client, child, "done")
assert seen[1]["force_from"] is None
def test_rebuild_renames_the_work_dir_on_a_retitle(tmp_path):
seen = []
client, cfg = make_client(tmp_path, capture_runner_factory(seen))
parent = make_done_job(client)
(cfg.work_dir / "A - T").mkdir(parents=True)
(cfg.work_dir / "A - T" / "audio.wav").write_bytes(b"x")
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(title="T2")).json()["id"]
wait_status(client, child, "done")
assert not (cfg.work_dir / "A - T").exists()
assert (cfg.work_dir / "A - T2" / "audio.wav").exists()
assert seen[1]["force_from"] == "grid"
def test_rebuild_fails_the_job_on_a_work_dir_collision(tmp_path):
client, cfg = make_client(tmp_path, ok_runner)
parent = make_done_job(client)
(cfg.work_dir / "A - T").mkdir(parents=True)
(cfg.work_dir / "A - T2").mkdir(parents=True)
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(title="T2")).json()["id"]
job = wait_status(client, child, "failed")
assert "already exists" in job["error"]
def test_rebuild_logs_what_changed_without_faking_a_stage(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
parent = make_done_job(client)
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(year="1998")).json()["id"]
job = wait_status(client, child, "done")
line = next(l for l in job["log"] if l.startswith("[rebuild]"))
assert "grid" in line and "year" in line
assert job["stage"] == "acquire" # from ok_runner, not from the rebuild line
def test_rebuild_inherits_the_parents_source(tmp_path):
seen = []
client, cfg = make_client(tmp_path, capture_runner_factory(seen))
resp = client.post("/api/jobs", data={"title": "T", "artist": "A", "lyrics": "x"},
files={"video": ("v.mp4", io.BytesIO(b"data"), "video/mp4")})
parent = resp.json()["id"]
wait_status(client, parent, "done")
(cfg.work_dir / "A - T").mkdir(parents=True)
for name in ("meta.json", "audio.wav"):
(cfg.work_dir / "A - T" / name).write_bytes(b"x")
child = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(youtube_url="", year="1998")).json()["id"]
wait_status(client, child, "done")
assert seen[1]["inp"].video_path == seen[0]["inp"].video_path
def test_rebuild_rejects_a_missing_upload_that_would_be_needed(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
resp = client.post("/api/jobs", data={"title": "T", "artist": "A", "lyrics": "x"},
files={"video": ("v.mp4", io.BytesIO(b"data"), "video/mp4")})
parent = resp.json()["id"]
wait_status(client, parent, "done")
resp = client.post(f"/api/jobs/{parent}/rebuild",
data=rebuild_form(youtube_url="", language="de"))
assert resp.status_code == 400
assert "attach the video file again" in resp.json()["detail"]
def test_rebuild_accepts_a_replacement_source(tmp_path):
seen = []
client, _ = make_client(tmp_path, capture_runner_factory(seen))
parent = make_done_job(client)
child = client.post(f"/api/jobs/{parent}/rebuild", data=rebuild_form(youtube_url=""),
files={"video": ("new.mp4", io.BytesIO(b"d"), "video/mp4")}
).json()["id"]
wait_status(client, child, "done")
assert seen[1]["inp"].youtube_url is None
assert seen[1]["inp"].video_path.name == "new.mp4"
assert seen[1]["force_from"] == "acquire"
def test_rebuild_rejects_two_sources(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
parent = make_done_job(client)
resp = client.post(f"/api/jobs/{parent}/rebuild", data=rebuild_form(),
files={"video": ("v.mp4", io.BytesIO(b"d"), "video/mp4")})
assert resp.status_code == 400
def test_rebuild_rejects_a_bad_bpm(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
parent = make_done_job(client)
resp = client.post(f"/api/jobs/{parent}/rebuild", data=rebuild_form(bpm="0"))
assert resp.status_code == 400
def test_rebuild_of_a_running_job_is_409(tmp_path):
release = threading.Event()
def blocking(inp, cfg, log=print, force_from=None, stages=None):
release.wait(5)
return PipelineResult(output_dir=cfg.output_dir / "X")
client, _ = make_client(tmp_path, blocking)
parent = client.post("/api/jobs", data=rebuild_form()).json()["id"]
try:
wait_status(client, parent, "running")
resp = client.post(f"/api/jobs/{parent}/rebuild", data=rebuild_form(year="1998"))
assert resp.status_code == 409
finally:
release.set()
wait_status(client, parent, "done")
def test_rebuild_of_an_unknown_job_is_404(tmp_path):
client, _ = make_client(tmp_path, ok_runner)
assert client.post("/api/jobs/nope/rebuild", data=rebuild_form()).status_code == 404