diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 709dc9e..6680cbd 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -3,7 +3,7 @@ import shutil import threading import time from dataclasses import replace -from typing import Sequence +from typing import Callable, Sequence import psycopg @@ -16,6 +16,41 @@ from lyra_worker.types import Candidate, MBTarget _AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"} +# Reject a download whose total playtime falls below this fraction of MusicBrainz's expected +# total — catches truncated files / preview clips substituted for real tracks that keep the +# track COUNT right. Generous, so edition/encoding differences don't false-positive; bonus +# tracks (over-long) are always fine. +_DURATION_MIN_RATIO = 0.85 + + +def _measure_staged_duration_s(staging: str) -> float | None: + """Sum the playtime (seconds) of the non-empty staged audio via mutagen. Returns None if + nothing could be measured (no audio, or an unreadable container) so the caller skips the + check rather than failing a good download on a probe hiccup. Lazy mutagen import; not + exercised by the offline fakes (which stage no real audio).""" + try: + import mutagen + except Exception: + return None + total = 0.0 + measured = False + for root, _dirs, files in os.walk(staging): + for f in files: + if os.path.splitext(f)[1].lower() not in _AUDIO_EXT: + continue + path = os.path.join(root, f) + try: + if os.path.getsize(path) == 0: + continue + audio = mutagen.File(path) + length = getattr(getattr(audio, "info", None), "length", None) + except Exception: + continue + if length: + total += float(length) + measured = True + return total if measured else None + def _count_staged_audio(staging: str) -> int: """Count NON-EMPTY audio files anywhere under the staging dir (streamrip nests them in a @@ -236,6 +271,7 @@ def run_pipeline( dest_root: str = "/music", staging_root: str | None = None, upgrade_cutoff: int | None = None, + measure_duration: Callable[[str], float | None] = _measure_staged_duration_s, ) -> None: """Real staged acquisition using source-agnostic adapters. Fakes in this plan.""" names = [a.name for a in adapters] @@ -334,6 +370,16 @@ def run_pipeline( _fail(conn, job_id, "incomplete download") return + # Duration check: even with the right track count, a truncated file or a short preview + # substituted for a real track shows up as a total playtime well under MB's expected + # total. Only applies when both are known (measured is None for the offline fakes). + if target.total_duration_s: + measured = measure_duration(staging) + if measured is not None and measured < target.total_duration_s * _DURATION_MIN_RATIO: + _fail(conn, job_id, + f"download too short ({measured / 60:.0f} of ~{target.total_duration_s / 60:.0f} min)") + return + final = album_dir(dest_root, target) existing_q = _library_quality(conn, target) new_q = quality_class(winner.quality) diff --git a/worker/tests/test_pipeline.py b/worker/tests/test_pipeline.py index fd2f795..107d3a9 100644 --- a/worker/tests/test_pipeline.py +++ b/worker/tests/test_pipeline.py @@ -124,6 +124,39 @@ def test_dedupe_skips_already_in_library(conn): assert cur.fetchone()[0] == 1 # not duplicated +class _DurationResolver: + """Resolver that stamps a known MB total duration on the target (the real MB path).""" + def __init__(self, total_s, tracks=10): + self._total, self._tracks = total_s, tracks + def resolve(self, artist, album): + from lyra_worker.types import MBTarget + return MBTarget(artist=artist, album=album, track_count=self._tracks, total_duration_s=self._total) + + +def test_rejects_a_download_far_shorter_than_the_mb_total(conn): + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + # MB says ~40 min; only ~5 min of audio staged (truncated/preview tracks) → reject. + run_pipeline(conn, job_id, [FakeQobuz()], resolver=_DurationResolver(2400), + measure_duration=lambda _s: 300.0, dest_root="/tmp/lib") + state, _stage = _job_state(conn, job_id) + assert state == "needs_attention" + with conn.cursor() as cur: + cur.execute('SELECT error FROM "Job" WHERE id = %s', (job_id,)) + assert "too short" in cur.fetchone()[0] + cur.execute('SELECT count(*) FROM "LibraryItem" WHERE album = %s', ("In Rainbows",)) + assert cur.fetchone()[0] == 0 # nothing imported + + +def test_accepts_a_download_within_the_duration_tolerance(conn): + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + # ~39 of ~40 min → within the 85% tolerance → imports normally. + run_pipeline(conn, job_id, [FakeQobuz()], resolver=_DurationResolver(2400), + measure_duration=lambda _s: 2340.0, dest_root="/tmp/lib") + assert _job_state(conn, job_id) == ("imported", "import") + + def test_force_reacquires_an_album_already_in_library(conn): # First acquisition puts the album in the library. job1 = insert_request(conn, artist="Radiohead", album="In Rainbows")