fix artwork

maple-py/src/maple/artwork.py

@@ -59,9 +59,12 @@ def extract_artwork(mp3: Path, usb_root: Path, next_id: int, cache: dict[str, in
     abs_small = usb_root / rel_small.lstrip("/")
     abs_large = usb_root / rel_large.lstrip("/")
     abs_small.parent.mkdir(parents=True, exist_ok=True)
-    if not abs_small.exists():
-        abs_small.write_bytes(_resize_jpeg(data, SMALL))
-    if not abs_large.exists():
-        abs_large.write_bytes(_resize_jpeg(data, LARGE))
+    # Always overwrite — the artwork_id↔file mapping is recomputed every
+    # export, and a stale a<id>.jpg left over from a previous run would
+    # silently render the wrong cover on the CDJ. The intra-run dedup
+    # (cache hit at the top of this function) already prevents wasteful
+    # rewrites within one export.
+    abs_small.write_bytes(_resize_jpeg(data, SMALL))
+    abs_large.write_bytes(_resize_jpeg(data, LARGE))
     cache[sha] = aid
     return aid, rel_small

maple-py/src/maple/cli.py

@@ -316,7 +316,7 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple
         logger.warning(
             "{} validation issue(s) across {} track(s)",
             len(validation_issues),
-            len({(i.artist, i.title) for i in validation_issues}),
+            len({i.track.display_name for i in validation_issues}),
         )
         log_issues(validation_issues, report=False)
     else:

maple-py/src/maple/collection/cli.py

@@ -164,7 +164,7 @@ async def main_async(argv: list[str]) -> int:
             )
             for issue in new_issues:
                 logger.warning(
-                    "{} - {}: {}", issue.artist, issue.title, issue.reason,
+                    "{}: {}: {}", issue.track.display_name, issue.field, issue.reason,
                 )
             issues.extend(new_issues)
             for pl, _ in affected:

maple-py/src/maple/collection/core.py

@@ -261,7 +261,7 @@ class Collection:
             bound.info("ok")
             return
         for i in issues:
-            bound.warning(f"[{i.playlist}] {i.artist} - {i.title}: {i.reason}")
+            bound.warning(f"{i.track.display_name}: {i.field}: {i.reason}")
 
     # --- indexing ---------------------------------------------------------- #
 

maple-py/src/maple/track.py

@@ -5,21 +5,9 @@ A *collection* is a sequence of `Track`s persisted under
 rekordbox USB at `data/usb/`. This dataclass is the bridge: any code that
 talks about "a track" in maple should hold one of these.
 
-Field semantics:
-  - Identity-ish fields (`isrc`, `title`, `artists`, `album`) start populated
-    from whichever source first knows the track — Spotify, a file's existing
-    tags, or a manual entry.
-  - Spotify-only fields (`spotify_track_id`, `release_date`, …) stay empty
-    for tracks that didn't come through the Spotify path.
-  - On-disk fields (`path`, `bitrate`, …) are populated once the audio is
-    actually on disk somewhere.
-  - Analytical fields (`bpm`, `beats`, …) populate during the audio-analysis
-    pass and are cached under `<collection>/_analysis/<isrc>.json`.
-
-The existing `maple.model.Track` is a *denormalised pdb row* — it carries
-artist_id/album_id/genre_id FKs that only exist in the context of a
-specific export. `Track.to_pdb_row(...)` projects this canonical Track
-into that shape using caller-supplied FK indices.
+`Track.validate()` is the project's single source of truth for what makes
+a track "exportable". Every check that used to be scattered across
+`maple/validate.py` lives here now; both pipelines call `track.validate()`.
 """
 from __future__ import annotations
 
@@ -29,11 +17,25 @@ from pathlib import Path
 from maple import model
 
 
+@dataclass
+class TrackIssue:
+    """A single failed validation criterion, bound to the `Track` it
+    came from. Repr-friendly via `__str__`."""
+
+    track: "Track"
+    field: str
+    reason: str
+
+    def __str__(self) -> str:
+        return f"{self.track.display_name}: {self.field}: {self.reason}"
+
+
 @dataclass
 class Track:
-    """Canonical track. Defaults intentionally permissive so partially-filled
-    instances flow through the pipeline; validation (`maple.validate`)
-    decides what counts as "complete enough to export"."""
+    """Canonical track metadata. Every field has a concrete default so the
+    type is always populated — "missing" data shows up as `""`/`0`/`False`
+    and is caught by `validate()`, not by Optional-handling at every call
+    site."""
 
     # ---- identity --------------------------------------------------------- #
     isrc: str = ""
@@ -53,14 +55,18 @@ class Track:
     duration_ms: int = 0
 
     # ---- on-disk audio --------------------------------------------------- #
-    path: Path | None = None             # current best file location
+    # `path` is the only legitimately-Optional field — a Track may exist
+    # purely as a Spotify entry before the audio has been downloaded.
+    path: Path | None = None
     file_size: int = 0
     bitrate: int = 0                     # kbps
     sample_rate: int = 0                 # Hz
     sample_depth: int = 0                # bits per sample
+    has_artwork: bool = False            # embedded cover-art presence
+    has_cached_analysis: bool = False    # collection/_analysis/<isrc>.json exists
 
     # ---- analytical ------------------------------------------------------ #
-    bpm: float | None = None             # median-of-intervals tempo
+    bpm: float = 0.0                     # 0.0 = unknown
 
     # ---- provenance ------------------------------------------------------ #
     source: str = ""                     # "spotify", "local", "bandcamp", …
@@ -120,10 +126,13 @@ class Track:
         )
 
     @classmethod
-    def from_file(cls, path: Path) -> "Track":
-        """Construct from on-disk audio via mutagen. Fields not in the file's
-        tags stay at their defaults; the caller can `merge_spotify` later to
-        fill them in canonically."""
+    def from_file(cls, path: Path, *, analysis_cache_dir: Path | None = None) -> "Track":
+        """Construct from on-disk audio via mutagen.
+
+        When `analysis_cache_dir` is provided, also sets
+        `has_cached_analysis` from `<cache_dir>/<isrc>.json` so
+        `.validate()` can check it without a separate call site.
+        """
         from mutagen import File as MutagenFile
 
         out = cls(path=path, source="local")
@@ -149,9 +158,9 @@ class Track:
             from maple.cli import split_genre_tags
             out.genres = split_genre_tags(gtag)
         try:
-            out.bpm = float(first("bpm")) if first("bpm") else None
+            out.bpm = float(first("bpm")) if first("bpm") else 0.0
         except ValueError:
-            out.bpm = None
+            out.bpm = 0.0
         try:
             out.track_number = int(first("tracknumber").split("/")[0])
         except ValueError:
@@ -167,14 +176,20 @@ class Track:
             out.file_size = path.stat().st_size
         except OSError:
             pass
+        # Presence flags so validate() can decide without extra plumbing.
+        from maple.collection.validate import has_embedded_art
+        out.has_artwork = has_embedded_art(path)
+        if analysis_cache_dir is not None and out.isrc:
+            from maple.analyze import cache_path_for
+            cp = cache_path_for(analysis_cache_dir, out.isrc)
+            out.has_cached_analysis = cp is not None and cp.exists()
         return out
 
     # ---- enrichment ------------------------------------------------------ #
 
     def merge_spotify(self, d: dict) -> "Track":
-        """Overlay Spotify-canonical fields onto this Track *in place* — used
-        when we have a local file but Spotify carries better identity / genre
-        / release date. Empty Spotify fields are NOT applied (no clobbering)."""
+        """Overlay Spotify-canonical fields onto this Track in place. Empty
+        Spotify fields are NOT applied (no clobbering)."""
         if d.get("title"):
             self.title = d["title"]
         if d.get("artists"):
@@ -207,10 +222,81 @@ class Track:
 
     def merge_analysis(self, analysis: dict) -> "Track":
         """Merge fields from `maple.analyze.analyze_audio`'s output."""
-        if analysis.get("bpm") is not None:
+        if analysis.get("bpm"):
             self.bpm = float(analysis["bpm"])
         return self
 
+    # ---- validation ------------------------------------------------------ #
+
+    def validate(self) -> list[TrackIssue]:
+        """Return the list of failed criteria. Empty list = ready to export.
+
+        The single place every "is this track exportable?" check lives.
+        Both `just collection` and `just export` call this — there is no
+        separate validation entrypoint.
+        """
+        issues: list[TrackIssue] = []
+        def fail(field_name: str, reason: str) -> None:
+            issues.append(TrackIssue(self, field_name, reason))
+
+        # Required text tags — the spotify-canonical set.
+        if not self.title:
+            fail("title", "tag missing")
+        if not self.artists:
+            fail("artist", "tag missing")
+        if not self.album:
+            fail("album", "tag missing")
+        if not self.release_date:
+            fail("date", "tag missing")
+        if not self.genres:
+            fail("genre", "tag missing")
+        if not self.isrc:
+            fail("isrc", "tag missing")
+
+        # Analytical.
+        if self.bpm <= 0:
+            fail("bpm", "tag missing")
+
+        # On-disk.
+        if not self.has_artwork:
+            fail("artwork", "no embedded cover art")
+        if not self.has_cached_analysis:
+            fail("analysis", "no cached analysis")
+
+        # Sanity — only check when we have something to check.
+        if self.path is not None and self.duration_ms <= 0:
+            fail("duration", "≤ 0s — corrupt file or unreadable container")
+        return issues
+
+    def compare(self, canonical: "Track", *, fuzzy_threshold: float = 0.5) -> list[TrackIssue]:
+        """Diff this Track against a canonical one (typically a Spotify
+        `from_spotify` Track). Returns mismatch issues — empty when the
+        two agree on every populated field."""
+        from maple.collection.index import first_artist, fuzzy_ratio, norm, norm_loose
+        out: list[TrackIssue] = []
+        def diff(field_name: str, sv: object, fv: object) -> None:
+            out.append(TrackIssue(self, field_name, f"spotify={sv!r} file={fv!r}"))
+
+        sp_artist = ", ".join(canonical.artists)
+        f_artist = ", ".join(self.artists)
+        if sp_artist and fuzzy_ratio(norm(first_artist(sp_artist)), norm(first_artist(f_artist))) < fuzzy_threshold:
+            diff("artist", sp_artist, f_artist)
+        if canonical.title and fuzzy_ratio(norm_loose(canonical.title), norm_loose(self.title)) < fuzzy_threshold:
+            diff("title", canonical.title, self.title)
+        if canonical.album and self.album and fuzzy_ratio(norm_loose(canonical.album), norm_loose(self.album)) < fuzzy_threshold:
+            diff("album", canonical.album, self.album)
+        elif canonical.album and not self.album:
+            diff("album", canonical.album, self.album)
+        if canonical.isrc and self.isrc and canonical.isrc.upper() != self.isrc.upper():
+            diff("isrc", canonical.isrc, self.isrc)
+        if canonical.release_date and self.release_date and canonical.release_date[:4] != self.release_date[:4]:
+            diff("date", canonical.release_date, self.release_date)
+        if canonical.duration_ms and self.duration_ms:
+            delta = abs(canonical.duration_ms - self.duration_ms) / 1000.0
+            if delta > 5.0:
+                diff("duration", f"{canonical.duration_s}s", f"{self.duration_s}s")
+        return out
+
     # ---- export adapter -------------------------------------------------- #
 
     def to_pdb_row(
@@ -226,13 +312,7 @@ class Track:
         date_added: str = "",
         tempo_x100: int | None = None,
     ) -> model.Track:
-        """Project this canonical Track into the denormalised pdb row shape.
-
-        FK ids (`artist_id`, `album_id`, …) and the per-export `file_path` /
-        `analyze_path` are caller-supplied because they only exist in the
-        context of a specific export. `tempo_x100` defaults to `round(bpm *
-        100)` when we have a BPM; pass through otherwise.
-        """
+        """Project this canonical Track into the denormalised pdb row shape."""
         if tempo_x100 is None:
             tempo_x100 = round(self.bpm * 100) if self.bpm else 0
         return model.Track(

maple-py/src/maple/validate.py

@@ -1,72 +1,22 @@
-"""Cross-cutting per-track validator.
+"""Thin compatibility layer over `Track.validate()`.
 
-A track is "valid" iff every export criterion holds. Both `just collection`
-(`Collection.process_track`) and `just export` (`cli.build`) must invoke
-this — there is no separate "collection-only" or "export-only" validation
-in maple. Add new criteria in one place; both pipelines pick them up.
+Validation logic lives on `maple.track.Track` itself — this module just
+re-exports `TrackIssue` and offers helpers so callers don't all need to
+construct `Track` objects inline.
 
-Criteria (extend as they emerge):
-    - ffprobe-decodable
-    - duration > 0
-    - title / artist / album / date / genre / isrc / bpm tags present
-    - embedded cover art present
-    - analytical BPM cached or fresh (cache key = ISRC)
-    - if `spotify_track` supplied: tags within fuzzy match of Spotify truth
-
-`validate_track` returns a list of `TrackIssue`. Empty list = passing.
-Callers decide whether to warn-and-continue (the default) or fail.
+The `validate_track(path, ...)` wrapper is kept for the few sites that
+hand around `Path`s rather than `Track`s; new code should prefer
+`Track.from_file(path).validate()`.
 """
 from __future__ import annotations
 
-from dataclasses import dataclass
 from pathlib import Path
 
 from loguru import logger
-from mutagen import File as MutagenFile
 
-from maple.analyze import cache_path_for
-from maple.collection.index import (
-    first_artist,
-    fuzzy_ratio,
-    norm,
-    norm_loose,
-)
-from maple.collection.validate import (
-    DURATION_TOLERANCE_S,
-    ffprobe_ok,
-    file_duration,
-    has_embedded_art,
-)
+from maple.track import Track, TrackIssue
 
-
-@dataclass
-class TrackIssue:
-    """A single validation finding.
-
-    Shape is (playlist, artist, title, reason) — `reason` is a
-    `"<field>: <detail>"` string so structured fields and free-text
-    descriptions share one container.
-    """
-
-    playlist: str
-    artist: str
-    title: str
-    reason: str
-
-
-REQUIRED_TAGS = ("title", "artist", "album", "date", "genre", "bpm", "isrc")
-
-
-def _file_tags(path: Path) -> dict[str, str] | None:
-    try:
-        audio = MutagenFile(path, easy=True)
-    except Exception as e:
-        logger.warning("could not open {} for validation: {}", path.name, e)
-        return None
-    if audio is None or audio.tags is None:
-        return {k: "" for k in REQUIRED_TAGS}
-    tags = audio.tags
-    return {k: (tags.get(k) or [""])[0] if hasattr(tags, "get") else "" for k in REQUIRED_TAGS}
+__all__ = ["Track", "TrackIssue", "validate_track", "log_issues"]
 
 
 def validate_track(
@@ -76,87 +26,22 @@ def validate_track(
     analysis_cache_dir: Path | None = None,
     playlist: str = "",
 ) -> list[TrackIssue]:
-    """Run every criterion against `path`. Returns the list of failed checks.
+    """Construct a `Track` from `path` and run its validator.
 
-    Optional context broadens the checks:
-      - `spotify_track`: enables fuzzy tag comparison against Spotify truth.
-      - `analysis_cache_dir`: enables the "analysis cached" check via ISRC.
-      - `playlist`: tags every issue with playlist context (collection flow).
+    `playlist` is unused — kept in the signature for API compatibility
+    with callers that still pass it.
     """
-    tags = _file_tags(path)
-    if tags is None:
-        return [TrackIssue(playlist, "", path.name, "tags: could not read file")]
-
-    artist = tags.get("artist") or (
-        ", ".join(spotify_track.get("artists") or []) if spotify_track else ""
-    )
-    title = tags.get("title") or (spotify_track.get("title") if spotify_track else "") or path.stem
-    issues: list[TrackIssue] = []
-
-    def fail(field: str, detail: str) -> None:
-        issues.append(TrackIssue(playlist, artist, title, f"{field}: {detail}"))
-
-    # ---- audio integrity --------------------------------------------------- #
-    if not ffprobe_ok(path):
-        fail("decode", f"ffprobe rejected {path.name}")
-    duration = file_duration(path)
-    if duration <= 0:
-        fail("duration", "≤ 0s — corrupt file or unreadable container")
-
-    # ---- required tags ----------------------------------------------------- #
-    for key in REQUIRED_TAGS:
-        if not (tags.get(key) or "").strip():
-            fail(key, "tag missing")
-
-    # ---- artwork ----------------------------------------------------------- #
-    if not has_embedded_art(path):
-        fail("artwork", "no embedded cover art")
-
-    # ---- analysis cached --------------------------------------------------- #
-    isrc = (tags.get("isrc") or "").strip()
-    if analysis_cache_dir is not None and isrc:
-        cp = cache_path_for(analysis_cache_dir, isrc)
-        if cp is None or not cp.exists():
-            fail("analysis", f"no cached analysis at {cp}")
-
-    # ---- Spotify reconciliation -------------------------------------------- #
+    del playlist  # not used; TrackIssue carries the track now
+    track = Track.from_file(path, analysis_cache_dir=analysis_cache_dir)
+    issues = track.validate()
     if spotify_track is not None:
-        sp_artist = ", ".join(spotify_track.get("artists") or [])
-        sp_title = (spotify_track.get("title") or "").strip()
-        sp_album = (spotify_track.get("album") or "").strip()
-        sp_isrc = (spotify_track.get("isrc") or "").strip()
-        sp_duration = (spotify_track.get("duration_ms") or 0) / 1000.0
-        sp_date = (spotify_track.get("release_date") or "").strip()
-
-        f_artist = tags.get("artist") or ""
-        f_title = tags.get("title") or ""
-        f_album = tags.get("album") or ""
-        f_isrc = tags.get("isrc") or ""
-        f_date = tags.get("date") or ""
-
-        if sp_artist and fuzzy_ratio(norm(first_artist(sp_artist)), norm(first_artist(f_artist))) < 0.5:
-            fail("artist", f"spotify={sp_artist!r} file={f_artist!r}")
-        if sp_title and fuzzy_ratio(norm_loose(sp_title), norm_loose(f_title)) < 0.5:
-            fail("title", f"spotify={sp_title!r} file={f_title!r}")
-        if sp_album and f_album and fuzzy_ratio(norm_loose(sp_album), norm_loose(f_album)) < 0.5:
-            fail("album", f"spotify={sp_album!r} file={f_album!r}")
-        elif sp_album and not f_album:
-            fail("album", f"spotify={sp_album!r} file=''")
-        if sp_isrc and f_isrc and sp_isrc.upper() != f_isrc.upper():
-            fail("isrc", f"spotify={sp_isrc!r} file={f_isrc!r}")
-        if sp_date and f_date and sp_date[:4] != f_date[:4]:
-            fail("date", f"spotify={sp_date!r} file={f_date!r}")
-        if sp_duration > 0 and abs(duration - sp_duration) > DURATION_TOLERANCE_S:
-            fail("duration", f"file {duration:.1f}s vs spotify {sp_duration:.1f}s")
-
+        issues.extend(track.compare(Track.from_spotify(spotify_track)))
     return issues
 
 
 def log_issues(issues: list[TrackIssue], *, report: bool = True) -> None:
-    """Emit every issue via loguru.warning. When `report=True`, also bind
-    `extra["report"]` so any attached file-sink (Collection.attach_report_sink
-    or the export-side equivalent) captures the line."""
+    """Emit every issue via `loguru.warning`. With `report=True`, also
+    binds `extra["report"]=True` so a report-file sink can capture them."""
     bound = logger.bind(report=True) if report else logger
     for i in issues:
-        prefix = f"[{i.playlist}] " if i.playlist else ""
-        bound.warning("{}{} - {}: {}", prefix, i.artist, i.title, i.reason)
+        bound.warning("{}", i)