From 4cc746810b9757f7d02d34d2f45941fbba47b9fb Mon Sep 17 00:00:00 2001 From: mecattaf Date: Thu, 13 Aug 2026 14:30:13 +0200 Subject: [PATCH 1/3] music-acquire: add verified acquisition cascade --- flake.nix | 3 + home/home.nix | 1 + overlays/default.nix | 3 + pkgs/music-acquire/README.md | 24 + pkgs/music-acquire/acquire.py | 8 + pkgs/music-acquire/default.nix | 74 + pkgs/music-acquire/music_acquire/__init__.py | 3 + pkgs/music-acquire/music_acquire/backend.py | 1645 +++++++++++++++++ pkgs/music-acquire/music_acquire/cli.py | 611 ++++++ pkgs/music-acquire/music_acquire/state.py | 318 ++++ .../music_acquire/verification.py | 222 +++ .../tests/fixtures/stage1-kirschberg.jsonl | 1 + .../tests/fixtures/stage2-corrida.jsonl | 1 + .../tests/fixtures/stage2-isrc-pommade.jsonl | 1 + .../music-acquire/tests/test_music_acquire.py | 589 ++++++ 15 files changed, 3504 insertions(+) create mode 100644 pkgs/music-acquire/README.md create mode 100644 pkgs/music-acquire/acquire.py create mode 100644 pkgs/music-acquire/default.nix create mode 100644 pkgs/music-acquire/music_acquire/__init__.py create mode 100644 pkgs/music-acquire/music_acquire/backend.py create mode 100644 pkgs/music-acquire/music_acquire/cli.py create mode 100644 pkgs/music-acquire/music_acquire/state.py create mode 100644 pkgs/music-acquire/music_acquire/verification.py create mode 100644 pkgs/music-acquire/tests/fixtures/stage1-kirschberg.jsonl create mode 100644 pkgs/music-acquire/tests/fixtures/stage2-corrida.jsonl create mode 100644 pkgs/music-acquire/tests/fixtures/stage2-isrc-pommade.jsonl create mode 100644 pkgs/music-acquire/tests/test_music_acquire.py diff --git a/flake.nix b/flake.nix index e9104183..b242dadc 100644 --- a/flake.nix +++ b/flake.nix @@ -481,6 +481,7 @@ local-ai-monthly mactahoe-gtk-theme mactahoe-icon-theme + music-acquire sfmono-liga ; @@ -525,6 +526,8 @@ # The RAW out-of-store dotfiles are never checked at switch, so check them here. checks.${system} = { + music-acquire = pkgs.music-acquire; + nas-topology = let nas = self.nixosConfigurations.nas.config; diff --git a/home/home.nix b/home/home.nix index 1e596663..38136291 100644 --- a/home/home.nix +++ b/home/home.nix @@ -472,6 +472,7 @@ in backlog-md # bespoke pkg via overlay — see pkgs/backlog-md.nix pkgs.crm # vendored personal CRM CLI; data stays at its built-in notes path pkgs.dcal # vendored calendar CLI; data lives under XDG, nothing in git + music-acquire # evidence-gated SoundCloud → YouTube → capture acquisition cliamp # terminal music player → navidrome. overlay pkg, see pkgs/cliamp.nix uv # Astral Python pkg/project manager. "hot" overlay pkg — rides nixpkgs-fresh HEAD (flake.nix), so it stays latest independent of the main pin. diff --git a/overlays/default.nix b/overlays/default.nix index ca1272c5..a194e4b8 100644 --- a/overlays/default.nix +++ b/overlays/default.nix @@ -80,6 +80,9 @@ final: prev: { # Personal git-backed CRM CLI, vendored with its package definition. crm = final.callPackage ../pkgs/crm/nix/package.nix { }; + # Evidence-gated, resumable front door over the music acquisition campaign. + music-acquire = final.callPackage ../pkgs/music-acquire { }; + # Headless calendar CLI, vendored with its package definition. dcal = final.callPackage ../pkgs/dcal/nix/package.nix { }; diff --git a/pkgs/music-acquire/README.md b/pkgs/music-acquire/README.md new file mode 100644 index 00000000..a6a1830f --- /dev/null +++ b/pkgs/music-acquire/README.md @@ -0,0 +1,24 @@ +# music-acquire + +`acquire` is the durable front door for the music-consolidation acquisition +cascade. It writes an immutable worklist, an append-only evidence ledger, and a +separate archive resume gate under `~/.local/state/music-acquire//`. + +```text +acquire tracklist worklist.jsonl [--source goldcast|synapson|vent-2024] +acquire soundcloud ARTIST [--likes] [--reposts] [--sets] +acquire bandcamp ARTIST [--alias "Project Pablo"] +acquire ytmusic ARTIST +acquire status [--batch NAME] [--json] +acquire resume --batch NAME +``` + +Global controls are `--batch`, `--dry-run`, `--limit`, `--no-capture`, and +`--out`. Final downloads always retain the source audio stream; no extraction, +format coercion, or recoding option is used. SoundCloud and YouTube cookies are +copied from agenix into the batch's mode-0600 cookie directory before use. + +The calibrated fingerprint and capture engines remain in +`~/mecattaf/music-consolidation/scripts/`; set `MUSIC_CONSOLIDATION_REPO` if that +checkout lives elsewhere. Capture is dispatched to `worker` by default and can +be changed with `MUSIC_ACQUIRE_CAPTURE_HOST`. diff --git a/pkgs/music-acquire/acquire.py b/pkgs/music-acquire/acquire.py new file mode 100644 index 00000000..64434886 --- /dev/null +++ b/pkgs/music-acquire/acquire.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +"""Executable entry point for the packaged music-acquire utility.""" + +from music_acquire.cli import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pkgs/music-acquire/default.nix b/pkgs/music-acquire/default.nix new file mode 100644 index 00000000..4b4bd290 --- /dev/null +++ b/pkgs/music-acquire/default.nix @@ -0,0 +1,74 @@ +{ + lib, + stdenvNoCC, + makeWrapper, + python3, + bash, + chromaprint, + coreutils, + curl, + ffmpeg, + jq, + nodejs, + openssh, + pipewire, + yt-dlp, +}: +let + python = python3.withPackages (ps: [ ps.websocket-client ]); +in +stdenvNoCC.mkDerivation { + pname = "music-acquire"; + version = "2026.08.13"; + + src = builtins.path { + path = ./.; + name = "music-acquire-src"; + }; + + nativeBuildInputs = [ + makeWrapper + python + ]; + + dontBuild = true; + doCheck = true; + + checkPhase = '' + runHook preCheck + PYTHONPATH=$PWD ${python}/bin/python -m unittest discover -s tests -v + runHook postCheck + ''; + + installPhase = '' + runHook preInstall + install -d $out/libexec/music-acquire $out/bin + cp acquire.py $out/libexec/music-acquire/ + cp -r music_acquire $out/libexec/music-acquire/ + + makeWrapper ${python}/bin/python $out/bin/acquire \ + --add-flags $out/libexec/music-acquire/acquire.py \ + --set FPCALC ${chromaprint}/bin/fpcalc \ + --prefix PATH : ${ + lib.makeBinPath [ + bash + chromaprint + coreutils + curl + ffmpeg + jq + nodejs + openssh + pipewire + yt-dlp + ] + } + runHook postInstall + ''; + + meta = { + description = "Resumable, evidence-gated SoundCloud/YouTube/Bandcamp music acquisition"; + mainProgram = "acquire"; + platforms = lib.platforms.linux; + }; +} diff --git a/pkgs/music-acquire/music_acquire/__init__.py b/pkgs/music-acquire/music_acquire/__init__.py new file mode 100644 index 00000000..221070eb --- /dev/null +++ b/pkgs/music-acquire/music_acquire/__init__.py @@ -0,0 +1,3 @@ +"""Reusable, evidence-gated music acquisition.""" + +__version__ = "2026.08.13" diff --git a/pkgs/music-acquire/music_acquire/backend.py b/pkgs/music-acquire/music_acquire/backend.py new file mode 100644 index 00000000..75ccfc07 --- /dev/null +++ b/pkgs/music-acquire/music_acquire/backend.py @@ -0,0 +1,1645 @@ +"""Production source enumeration and acquisition backends. + +Network-facing behavior is kept behind this object so the orchestration can be +regression-tested without touching SoundCloud, YouTube, Bandcamp, or PipeWire. +""" + +from __future__ import annotations + +import hashlib +import html +import http.cookiejar +import importlib.util +import json +import os +import re +import shlex +import shutil +import socket +import subprocess +import tempfile +import time +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Any, Iterable + +from .state import BatchState +from .verification import ( + BER_ACCEPT, + DUR_TOL_ACOUSTIC, + artist_agrees, + core_and_version, + metadata_verdict, + norm, + title_agrees, + version_agrees, +) + + +AUDIO_EXTENSIONS = { + ".aac", + ".aif", + ".aiff", + ".alac", + ".flac", + ".m4a", + ".m4b", + ".mka", + ".mp2", + ".mp3", + ".oga", + ".ogg", + ".opus", + ".wav", + ".webm", + ".wma", +} + +RATE_LIMIT_MARKERS = ( + "rate-limited", + "rate limited", + "http error 429", + "too many requests", + "sign in to confirm you're not a bot", +) +GONE_MARKERS = ( + "http error 404", + "not found", + "was removed", + "has been removed", + "no longer available", + "does not exist", +) +DRM_MARKERS = ("drm protected", "premium only", "not available for free accounts") +UA = ( + "music-acquire/2026.08.13 " + "(https://github.com/mecattaf/dotfiles; contact: thomas@leger.run)" +) + + +def stable_stem(item_id: str) -> str: + value = re.sub(r"[/\\\x00-\x1f]", "-", str(item_id)).strip(" .") + if not value: + value = "item" + if len(value.encode("utf-8")) > 180: + digest = hashlib.sha256(value.encode()).hexdigest()[:16] + value = value[:140].rstrip() + "-" + digest + return value + + +def is_rate_limited(text: str | None) -> bool: + value = (text or "").lower() + return any(marker in value for marker in RATE_LIMIT_MARKERS) + + +def classify_failure(text: str | None) -> str: + value = (text or "").lower() + if any(marker in value for marker in DRM_MARKERS): + return "drm" + if any(marker in value for marker in GONE_MARKERS): + return "gone" + return "retryable" + + +class LiveBackend: + """The real, deliberately conservative three-stage acquisition cascade.""" + + def __init__( + self, + state: BatchState, + out: Path, + campaign_repo: Path, + *, + no_capture: bool = False, + capture_host: str = "worker", + environ: dict[str, str] | None = None, + ): + self.state = state + self.out = out.expanduser().resolve() + self.campaign_repo = campaign_repo.expanduser().resolve() + self.no_capture = no_capture + self.capture_host = capture_host + self.environ = dict(os.environ if environ is None else environ) + self.out.mkdir(parents=True, exist_ok=True) + self.work = self.state.path / "work" + self.previews = self.state.path / "previews" + self.work.mkdir(parents=True, exist_ok=True) + self.previews.mkdir(parents=True, exist_ok=True) + self.sc_cookie = self.state.cookies_path / "soundcloud.txt" + self.yt_cookie = self.state.cookies_path / "youtube-music.txt" + self._client_id: str | None = None + self._fpmatch = None + self._duplicate_sc: dict[str, str] = {} + self._duplicate_item: dict[str, str] = {} + self._duplicate_yt: dict[str, str] = {} + self._mb_cache: dict[str, dict[str, Any] | None] = {} + self.capture_info: dict[str, Any] = {} + self._load_duplicate_index() + + # -- process and HTTP mechanics ----------------------------------------- + + def _run( + self, + command: list[str], + *, + timeout: float = 300, + input_text: str | None = None, + env: dict[str, str] | None = None, + ) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + capture_output=True, + text=True, + timeout=timeout, + input=input_text, + env=env, + check=False, + ) + + @staticmethod + def _copy_cookie(source: str, destination: Path) -> bool: + src = Path(source) + if not src.is_file() or not os.access(src, os.R_OK): + return False + destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + shutil.copyfile(src, destination) + os.chmod(destination, 0o600) + return True + + def prepare(self) -> dict[str, Any]: + sc_source = self.environ.get( + "MUSIC_ACQUIRE_SC_COOKIES", "/run/agenix/soundcloud-cookies" + ) + yt_source = self.environ.get( + "MUSIC_ACQUIRE_YT_COOKIES", "/run/agenix/youtube-music-cookies" + ) + have_sc = self._copy_cookie(sc_source, self.sc_cookie) + have_yt = self._copy_cookie(yt_source, self.yt_cookie) + + entitlement: str | None = None + entitlement_error: str | None = None + if have_sc: + try: + me = self._soundcloud_get("me", authenticated=True) + entitlement = ( + (((me or {}).get("consumer_subscription") or {}).get("product") or {}).get( + "id" + ) + ) + except Exception as error: # the header records uncertainty; work may continue + entitlement_error = f"{type(error).__name__}: {error}"[:240] + + host = socket.gethostname().split(".")[0] + capture_reachable = host == self.capture_host + capture_busy = False + if host != self.capture_host: + try: + probe_script = ( + "if systemctl --user is-active --quiet mc-tier3.service; " + "then echo busy; else echo ready; fi" + ) + probe = self._run( + [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=5", + self.capture_host, + "sh -lc " + shlex.quote(probe_script), + ], + timeout=10, + ) + capture_reachable = probe.returncode == 0 + capture_busy = probe.stdout.strip().endswith("busy") + except (OSError, subprocess.TimeoutExpired): + capture_reachable = False + else: + probe = self._run( + ["systemctl", "--user", "is-active", "--quiet", "mc-tier3.service"], + timeout=5, + ) + capture_busy = probe.returncode == 0 + + self.capture_info = { + "requested": not self.no_capture, + "host": self.capture_host, + "host_reachable": capture_reachable, + "host_busy": capture_busy, + "entitlement": entitlement, + "entitlement_active": entitlement == "consumer-high-tier", + "entitlement_error": entitlement_error, + "available": ( + not self.no_capture + and capture_reachable + and not capture_busy + and entitlement == "consumer-high-tier" + ), + "soundcloud_cookie": have_sc, + "youtube_cookie": have_yt, + } + return self.capture_info + + def _get_bytes(self, url: str, *, authenticated: bool = False) -> bytes: + headers = {"User-Agent": UA} + if authenticated and self.sc_cookie.exists(): + jar = http.cookiejar.MozillaCookieJar(str(self.sc_cookie)) + jar.load(ignore_discard=True, ignore_expires=True) + # SoundCloud's browser session keeps oauth_token as a host-only + # soundcloud.com cookie. It is intentionally not sent to the + # api-v2 subdomain; the web client promotes it to the OAuth header. + token = next( + (cookie.value for cookie in jar if cookie.name == "oauth_token"), + None, + ) + if token: + headers["Authorization"] = "OAuth " + token + request = urllib.request.Request(url, headers=headers) + opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) + with opener.open(request, timeout=45) as response: + return response.read() + request = urllib.request.Request(url, headers=headers) + with urllib.request.urlopen(request, timeout=45) as response: + return response.read() + + def _scrape_client_id(self) -> str: + if self._client_id: + return self._client_id + page = self._get_bytes("https://soundcloud.com/").decode("utf-8", "replace") + assets = re.findall( + r'src="(https://a-v2\.sndcdn\.com/assets/[^"]+\.js)"', page + ) + for asset in assets: + body = self._get_bytes(asset).decode("utf-8", "replace") + match = re.search(r'client_id\s*[=:]\s*"([A-Za-z0-9]{32})"', body) + if match: + self._client_id = match.group(1) + return self._client_id + raise RuntimeError("could not scrape SoundCloud's public client_id") + + def _soundcloud_get( + self, + endpoint: str, + params: dict[str, str | int] | None = None, + *, + authenticated: bool = False, + ) -> Any: + query = dict(params or {}) + query["client_id"] = self._scrape_client_id() + url = "https://api-v2.soundcloud.com/" + endpoint.lstrip("/") + url += "?" + urllib.parse.urlencode(query) + return json.loads(self._get_bytes(url, authenticated=authenticated)) + + def _yt_entries(self, url: str, *, cookies: Path | None = None) -> list[dict[str, Any]]: + command = [ + "yt-dlp", + "--ignore-config", + "--flat-playlist", + "--dump-json", + "--ignore-errors", + "--no-warnings", + ] + if cookies and cookies.exists(): + command += ["--cookies", str(cookies)] + command.append(url) + result = self._run(command, timeout=900) + rows = [] + for line in result.stdout.splitlines(): + try: + row = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(row, dict): + rows.append(row) + if not rows and result.returncode != 0: + raise RuntimeError((result.stderr or "enumeration failed")[-400:]) + return rows + + # -- source enumeration -------------------------------------------------- + + def _resolve_soundcloud_artist(self, value: str) -> str: + if value.startswith(("http://", "https://")): + return value.rstrip("/") + result = self._soundcloud_get("search/users", {"q": value, "limit": 20}) + users = result.get("collection") or [] + exact = [user for user in users if norm(user.get("username")) == norm(value)] + chosen = (exact or users or [None])[0] + if not chosen or not chosen.get("permalink_url"): + raise RuntimeError(f"SoundCloud artist not found: {value}") + return str(chosen["permalink_url"]).rstrip("/") + + @staticmethod + def _sc_flat_item(row: dict[str, Any], source: str) -> dict[str, Any] | None: + item_id = row.get("id") + url = row.get("webpage_url") or row.get("url") + if not item_id or not url or "/sets/" in str(url): + return None + return { + "id": str(item_id), + "sc_id": str(item_id), + "url": url, + "query": row.get("title") or url, + "title": row.get("title"), + "artist": row.get("uploader") or row.get("channel"), + "duration_s": row.get("duration"), + "source": source, + "appearances": [], + } + + def enumerate_soundcloud( + self, + artist: str, + *, + likes: bool = False, + reposts: bool = False, + sets: bool = False, + ) -> tuple[str, list[dict[str, Any]]]: + artist_url = self._resolve_soundcloud_artist(artist) + feeds = ["tracks"] + if likes: + feeds.append("likes") + if reposts: + feeds.append("reposts") + if sets: + feeds.append("sets") + items: dict[str, dict[str, Any]] = {} + source = f"soundcloud:{artist_url}" + for feed in feeds: + rows = self._yt_entries(f"{artist_url}/{feed}", cookies=self.sc_cookie) + if feed == "sets": + set_urls = [ + row.get("webpage_url") or row.get("url") + for row in rows + if row.get("webpage_url") or row.get("url") + ] + rows = [] + for set_url in set_urls: + rows.extend(self._yt_entries(str(set_url), cookies=self.sc_cookie)) + for row in rows: + item = self._sc_flat_item(row, source) + if item: + items.setdefault(item["id"], item) + return artist_url, list(items.values()) + + def enumerate_bandcamp( + self, artist_url: str, *, alias: str | None = None + ) -> list[dict[str, Any]]: + base = artist_url.rstrip("/") + releases = self._yt_entries(f"{base}/music") + items: dict[str, dict[str, Any]] = {} + for release in releases: + release_url = release.get("webpage_url") or release.get("url") + if not release_url: + continue + tracks = self._yt_entries(str(release_url)) + if not tracks: + tracks = [release] + for track in tracks: + track_id = track.get("id") + track_url = track.get("webpage_url") or track.get("url") + if not track_id or not track_url: + continue + item_id = f"bc-{track_id}" + items.setdefault( + item_id, + { + "id": item_id, + "bandcamp_id": str(track_id), + "url": track_url, + "query": track.get("title") or track_url, + "title": track.get("title"), + "artist": track.get("artist") or track.get("uploader"), + "release_url": release_url, + "release_title": release.get("title"), + "alias": alias, + "source": f"bandcamp:{base}", + "appearances": [], + }, + ) + return list(items.values()) + + def _resolve_youtube_artist(self, value: str) -> str: + if value.startswith(("http://", "https://")): + return value.rstrip("/") + rows = self._yt_entries(f"ytsearch10:{value}", cookies=self.yt_cookie) + ranked = [] + for row in rows: + channel_url = row.get("channel_url") or row.get("uploader_url") + channel = row.get("channel") or row.get("uploader") or "" + if not channel_url: + continue + score = 0 + if norm(channel).removesuffix(" topic") == norm(value): + score += 5 + if artist_agrees([value], f"{channel} {row.get('title') or ''}"): + score += 2 + if str(channel).endswith("- Topic"): + score += 1 + ranked.append((score, str(channel_url).rstrip("/"))) + if not ranked: + raise RuntimeError(f"YouTube Music artist not found: {value}") + ranked.sort(reverse=True) + return ranked[0][1] + + def enumerate_ytmusic(self, artist: str) -> tuple[str, list[dict[str, Any]]]: + artist_url = self._resolve_youtube_artist(artist) + rows: list[dict[str, Any]] = [] + errors = [] + for suffix in ("/releases", "/videos"): + try: + rows.extend(self._yt_entries(artist_url + suffix, cookies=self.yt_cookie)) + except RuntimeError as error: + errors.append(str(error)) + if not rows: + rows = self._yt_entries(artist_url, cookies=self.yt_cookie) + items: dict[str, dict[str, Any]] = {} + for row in rows: + video_id = row.get("id") + url = row.get("webpage_url") or row.get("url") + if video_id and url: + item_id = f"yt-{video_id}" + items.setdefault( + item_id, + { + "id": item_id, + "yt_id": str(video_id), + "url": url, + "query": row.get("title") or url, + "title": row.get("title"), + "artist": row.get("channel") or row.get("uploader"), + "source": f"ytmusic:{artist_url}", + "appearances": [], + }, + ) + if not items and errors: + raise RuntimeError(errors[-1]) + return artist_url, list(items.values()) + + # -- duplicate suppression ---------------------------------------------- + + def _load_duplicate_index(self) -> None: + staging = Path( + self.environ.get("MUSIC_ACQUIRE_STAGING", "/mnt/nas/music-staging") + ) + provenance = Path( + self.environ.get( + "MUSIC_ACQUIRE_PROVENANCE", + str(self.campaign_repo / "ledgers" / "staging-provenance.tsv"), + ) + ) + if provenance.exists(): + with provenance.open(encoding="utf-8") as handle: + header = handle.readline().rstrip("\n").split("\t") + index = {name: position for position, name in enumerate(header)} + for line in handle: + fields = line.rstrip("\n").split("\t") + if len(fields) < len(header): + continue + relpath = fields[index.get("relpath", 2)] + path = str(staging / relpath) + sc_id = fields[index.get("sc_id", 5)] if "sc_id" in index else "" + if sc_id: + self._duplicate_sc.setdefault(sc_id, path) + if "extra" in index: + for part in fields[index["extra"]].split(";"): + key, _, value = part.partition("=") + if key == "acquire_id" and value: + self._duplicate_item.setdefault(value, path) + if key == "yt_id" and value: + self._duplicate_yt.setdefault(value, path) + + # Before beets runs there is deliberately no provenance snapshot yet. + # Index the campaign's actual ingest files by their established naming + # contract so an already-held SoundCloud id still short-circuits without + # a network request (tier-1 uses `` - title``; tiers 2/3 use ````). + ingest = staging / "_ingest" + lane_roots = { + "soundcloud": ingest / "soundcloud", + "tier2": ingest / "tier2", + "tier3": ingest / "tier3", + } + for lane, root in lane_roots.items(): + if not root.is_dir(): + continue + for path in root.rglob("*"): + if not path.is_file() or path.suffix.lower() not in AUDIO_EXTENSIONS: + continue + if lane == "soundcloud": + match = re.match(r"([0-9]+) - ", path.name) + sc_id = match.group(1) if match else "" + else: + sc_id = path.stem if path.stem.isdigit() else "" + if sc_id: + self._duplicate_sc.setdefault(sc_id, str(path)) + + state_root = self.state.root + if state_root.exists(): + for ledger in state_root.glob("*/ledger.jsonl"): + final: dict[str, dict[str, Any]] = {} + with ledger.open(encoding="utf-8") as handle: + for line in handle: + try: + row = json.loads(line) + except json.JSONDecodeError: + continue + if row.get("record") == "item" and row.get("id"): + final[str(row["id"])] = row + for item_id, row in final.items(): + if not str(row.get("disposition") or "").startswith("ok_"): + continue + path = row.get("path") + if path: + self._duplicate_item.setdefault(item_id, str(path)) + if row.get("sc_id"): + self._duplicate_sc.setdefault(str(row["sc_id"]), str(path)) + if row.get("yt_id"): + self._duplicate_yt.setdefault(str(row["yt_id"]), str(path)) + + def duplicate(self, item: dict[str, Any], sc_track: dict[str, Any] | None = None): + item_id = str(item["id"]) + if item_id in self._duplicate_item: + return {"path": self._duplicate_item[item_id], "duplicate_id": item_id} + sc_id = "" if (sc_track or {}).get("segment") else str( + (sc_track or {}).get("id") or item.get("sc_id") or "" + ) + if sc_id and sc_id in self._duplicate_sc: + return {"path": self._duplicate_sc[sc_id], "duplicate_sc_id": sc_id} + yt_id = str(item.get("yt_id") or "") + if yt_id and yt_id in self._duplicate_yt: + return {"path": self._duplicate_yt[yt_id], "duplicate_yt_id": yt_id} + return None + + def remember(self, item: dict[str, Any], row: dict[str, Any]) -> None: + path = row.get("path") + if not path: + return + self._duplicate_item[str(item["id"])] = str(path) + if row.get("sc_id"): + self._duplicate_sc[str(row["sc_id"])] = str(path) + if row.get("yt_id"): + self._duplicate_yt[str(row["yt_id"])] = str(path) + + # -- SoundCloud resolution and direct acquisition ----------------------- + + @staticmethod + def _soundcloud_track(track: dict[str, Any]) -> dict[str, Any]: + publisher = track.get("publisher_metadata") or {} + transcodings = ((track.get("media") or {}).get("transcodings") or []) + user = track.get("user") or {} + return { + "id": str(track["id"]), + "resolved": True, + "title": track.get("title"), + "url": track.get("permalink_url"), + "uploader": user.get("permalink"), + "uploader_name": user.get("username"), + "policy": track.get("policy"), + "monetization_model": track.get("monetization_model"), + "duration_ms": track.get("duration"), + "full_duration_ms": track.get("full_duration") or track.get("duration"), + "artwork_url": track.get("artwork_url"), + "pm_artist": publisher.get("artist"), + "pm_album_title": publisher.get("album_title"), + "pm_release_title": publisher.get("release_title"), + "pm_isrc": publisher.get("isrc"), + "has_preview": any(value.get("snipped") for value in transcodings), + } + + @staticmethod + def _description_segments( + description: str | None, full_duration_ms: int | None + ) -> list[dict[str, Any]]: + """Parse timestamped long-form tracklists without guessing boundaries.""" + starts: list[tuple[float, str]] = [] + timestamp = re.compile( + r"(?[0-9]{1,2}):)?" + r"(?P[0-9]{1,2}):(?P[0-9]{2})(?![0-9])" + ) + for line in (description or "").splitlines(): + match = timestamp.search(line) + if not match: + continue + hours = int(match.group("h") or 0) + minutes = int(match.group("m")) + seconds = int(match.group("s")) + if minutes >= 60 or seconds >= 60: + continue + start = float(hours * 3600 + minutes * 60 + seconds) + title = timestamp.sub(" ", line) + title = re.sub( + r"^[\s\[\](){}#|]*[0-9]{1,3}\s*[.):-]\s*", "", title + ) + title = re.sub(r"[\[\](){}|]", " ", title) + title = re.sub(r"\s+", " ", title).strip(" -–—.:#") + if title: + starts.append((start, title)) + starts.sort(key=lambda value: value[0]) + duration_s = float(full_duration_ms or 0) / 1000.0 + segments = [] + for index, (start, title) in enumerate(starts): + end = starts[index + 1][0] if index + 1 < len(starts) else duration_s + if end > start: + segments.append({"start_s": start, "end_s": end, "title": title}) + return segments + + def resolve_soundcloud(self, item: dict[str, Any]) -> dict[str, Any]: + try: + sc_id = item.get("sc_id") + if sc_id: + tracks = self._soundcloud_get("tracks", {"ids": str(sc_id)}) + if not tracks: + return {"status": "gone", "reason": "track_gone_upstream"} + return {"status": "found", "track": self._soundcloud_track(tracks[0])} + + if not str(item.get("source") or "").startswith("tracklist:"): + return {"status": "absent", "reason": "not_a_text_source"} + candidates = [] + segment_candidates = [] + reference_title = item.get("title") or item.get("query") + artists = [item.get("artist") or ""] + search_queries = [] + seen_queries = set() + for query in (item.get("query"), item.get("title"), item.get("artist")): + key = norm(query) + if query and key not in seen_queries: + seen_queries.add(key) + search_queries.append(query) + seen_candidates = set() + for query_index, query in enumerate(search_queries): + result = self._soundcloud_get( + "search/tracks", {"q": query, "limit": 20} + ) + for candidate in result.get("collection") or []: + candidate_id = str(candidate.get("id") or "") + if not candidate_id or candidate_id in seen_candidates: + continue + seen_candidates.add(candidate_id) + user = candidate.get("user") or {} + candidate_text = ( + f"{candidate.get('title') or ''} " + f"{user.get('username') or ''}" + ) + uploader_text = ( + f"{user.get('username') or ''} " + f"{user.get('permalink') or ''}" + ) + direct_title = title_agrees( + reference_title, + candidate.get("title"), + [*artists, item.get("album") or ""], + ) + candidate_artist = not artists[0] or artist_agrees( + artists, candidate_text + ) + if direct_title and candidate_artist: + exact = norm(candidate.get("title")) == norm(reference_title) + candidates.append((not exact, candidate)) + + # A timestamped official album/mix can carry the requested + # constituent track even when its container title differs. + # Require its uploader to agree with the artist so a DJ-set + # description is never mistaken for a clean source recording. + if artists[0] and artist_agrees(artists, uploader_text): + for segment in self._description_segments( + candidate.get("description"), + candidate.get("full_duration") or candidate.get("duration"), + ): + if title_agrees( + reference_title, + segment["title"], + [*artists, item.get("album") or ""], + ): + segment_candidates.append((candidate, segment)) + if candidates or segment_candidates: + break + if query_index + 1 < len(search_queries): + time.sleep( + float(self.environ.get("MUSIC_ACQUIRE_SC_API_SLEEP", "0.35")) + ) + if not candidates: + if segment_candidates: + candidate, segment = segment_candidates[0] + resolved = self._soundcloud_track(candidate) + segment_duration_ms = round( + (segment["end_s"] - segment["start_s"]) * 1000 + ) + resolved.update( + { + "title": segment["title"], + "parent_title": candidate.get("title"), + "full_duration_ms": segment_duration_ms, + "duration_ms": segment_duration_ms, + "segment": segment, + "has_preview": False, + "pm_artist": item.get("artist"), + "pm_release_title": item.get("title"), + "pm_album_title": item.get("album") + or candidate.get("title"), + "pm_isrc": None, + } + ) + return {"status": "found", "track": resolved} + return {"status": "absent", "reason": "no_soundcloud_candidate"} + candidates.sort(key=lambda pair: pair[0]) + return {"status": "found", "track": self._soundcloud_track(candidates[0][1])} + except Exception as error: + return { + "status": "retryable", + "reason": "soundcloud_resolution_failed", + "detail": f"{type(error).__name__}: {error}"[:300], + } + + def _audio_for_stem(self, stem: str, directory: Path | None = None) -> Path | None: + root = directory or self.out + for path in sorted(root.glob(stem + ".*")): + if path.suffix.lower() in AUDIO_EXTENSIONS and not path.name.endswith(".part"): + if path.stat().st_size > 0: + return path + return None + + def _download_native( + self, + *, + url: str, + item_id: str, + cookies: Path | None, + format_selector: str, + timeout: float = 1800, + ) -> dict[str, Any]: + stem = stable_stem(item_id) + held = self._audio_for_stem(stem) + if held: + return {"status": "ok", "path": str(held), "bytes": held.stat().st_size} + command = [ + "yt-dlp", + "--ignore-config", + "--no-playlist", + "--no-progress", + "--no-warnings", + "--sleep-requests", + "1", + "--retries", + "5", + "--fragment-retries", + "5", + "--no-overwrites", + "--write-info-json", + "--write-thumbnail", + "--embed-thumbnail", + "--embed-metadata", + "-f", + format_selector, + "-o", + str(self.out / (stem + ".%(ext)s")), + ] + if cookies and cookies.exists(): + command += ["--cookies", str(cookies)] + command.append(url) + try: + result = self._run(command, timeout=timeout) + except subprocess.TimeoutExpired: + return {"status": "retryable", "reason": "download_timeout"} + audio = self._audio_for_stem(stem) + if audio: + return { + "status": "ok", + "path": str(audio), + "bytes": audio.stat().st_size, + "download_note": (result.stderr or "")[-240:] or None, + } + detail = (result.stderr or result.stdout or "download produced no audio")[-400:] + return { + "status": classify_failure(detail), + "reason": "download_failed", + "detail": detail, + } + + def download_soundcloud( + self, item: dict[str, Any], track: dict[str, Any] + ) -> dict[str, Any]: + if not self.sc_cookie.exists(): + return {"status": "retryable", "reason": "soundcloud_cookie_unavailable"} + if track.get("segment"): + return self._download_soundcloud_segment(item, track) + return self._download_native( + url=str(track.get("url") or item.get("url")), + item_id=str(item["id"]), + cookies=self.sc_cookie, + format_selector="download/bestaudio/best", + ) + + def _download_soundcloud_segment( + self, item: dict[str, Any], track: dict[str, Any] + ) -> dict[str, Any]: + """Stream-copy one timestamped track from an official long-form upload.""" + output_stem = stable_stem(str(item["id"])) + held = self._audio_for_stem(output_stem) + if held: + return {"status": "ok", "path": str(held), "bytes": held.stat().st_size} + + parent_stem = "sc-parent-" + stable_stem(str(track["id"])) + parent = self._audio_for_stem(parent_stem, self.work) + if not parent: + command = [ + "yt-dlp", + "--ignore-config", + "--no-playlist", + "--no-progress", + "--no-warnings", + "--sleep-requests", + "1", + "--retries", + "5", + "--fragment-retries", + "5", + "--no-overwrites", + "--write-info-json", + "--write-thumbnail", + "--embed-thumbnail", + "--embed-metadata", + "--cookies", + str(self.sc_cookie), + "-f", + "download/bestaudio/best", + "-o", + str(self.work / (parent_stem + ".%(ext)s")), + str(track["url"]), + ] + try: + result = self._run(command, timeout=3600) + except subprocess.TimeoutExpired: + return {"status": "retryable", "reason": "download_timeout"} + parent = self._audio_for_stem(parent_stem, self.work) + if not parent: + detail = ( + result.stderr or result.stdout or "download produced no audio" + )[-400:] + return { + "status": classify_failure(detail), + "reason": "download_failed", + "detail": detail, + } + + segment = track["segment"] + duration = float(segment["end_s"]) - float(segment["start_s"]) + final = self.out / (output_stem + parent.suffix.lower()) + # Keep the temporary beside the destination so the final rename is + # atomic even when state lives locally and the ingest lane is on NAS. + temporary = self.out / ( + "." + output_stem + ".segment.part" + parent.suffix.lower() + ) + try: + temporary.unlink(missing_ok=True) + except OSError: + return {"status": "retryable", "reason": "stale_segment_output"} + command = [ + "ffmpeg", + "-v", + "error", + "-nostdin", + "-ss", + f"{float(segment['start_s']):.3f}", + "-i", + str(parent), + "-t", + f"{duration:.3f}", + "-map", + "0", + "-c", + "copy", + "-avoid_negative_ts", + "make_zero", + "-metadata", + f"title={item.get('title') or segment['title']}", + str(temporary), + ] + result = self._run(command, timeout=max(300, duration + 120)) + if ( + result.returncode != 0 + or not temporary.exists() + or temporary.stat().st_size == 0 + ): + return { + "status": "retryable", + "reason": "segment_stream_copy_failed", + "detail": (result.stderr or "")[-300:], + } + os.replace(temporary, final) + + parent_info = self.work / (parent_stem + ".info.json") + if parent_info.exists(): + try: + info = json.loads(parent_info.read_text(encoding="utf-8")) + info["music_acquire_segment"] = { + **segment, + "parent_sc_id": str(track["id"]), + "parent_sc_url": track.get("url"), + } + (self.out / (output_stem + ".info.json")).write_text( + json.dumps(info, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + except (OSError, json.JSONDecodeError): + pass + for artwork in self.work.glob(parent_stem + ".*"): + if artwork.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}: + destination = self.out / (output_stem + artwork.suffix.lower()) + if not destination.exists(): + shutil.copyfile(artwork, destination) + + return { + "status": "ok", + "path": str(final), + "bytes": final.stat().st_size, + "segment": { + **segment, + "parent_sc_id": str(track["id"]), + "parent_sc_url": track.get("url"), + }, + } + + # -- MusicBrainz and YouTube verification ------------------------------- + + def _musicbrainz_get(self, endpoint: str, params: dict[str, str]) -> Any: + query = urllib.parse.urlencode({**params, "fmt": "json"}) + request = urllib.request.Request( + f"https://musicbrainz.org/ws/2/{endpoint}?{query}", + headers={"User-Agent": UA, "Accept": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=45) as response: + result = json.loads(response.read()) + time.sleep(float(self.environ.get("MUSIC_ACQUIRE_MB_SLEEP", "1"))) + return result + + @staticmethod + def _mb_recording(row: dict[str, Any], isrc: str | None = None) -> dict[str, Any]: + credits = row.get("artist-credit") or row.get("artist_credit") or [] + artists = [] + for credit in credits: + if isinstance(credit, dict): + artist = credit.get("artist") or {} + name = artist.get("name") or credit.get("name") + if name: + artists.append(name) + if not artists and row.get("artist"): + artists = [row["artist"]] + isrcs = row.get("isrcs") or ([isrc] if isrc else []) + return { + "mbid": row.get("id") or row.get("mbid"), + "title": row.get("title"), + "artist": artists[0] if artists else None, + "artists": artists, + "length_ms": row.get("length") or row.get("length_ms"), + "isrc": (isrcs or [None])[0], + } + + def _mb_for_sc(self, track: dict[str, Any]) -> dict[str, Any] | None: + isrc = track.get("pm_isrc") + if not isrc: + return None + cache_key = "isrc:" + str(isrc) + if cache_key in self._mb_cache: + return self._mb_cache[cache_key] + + local = self.campaign_repo / "ledgers" / "musicbrainz-isrc.jsonl" + state_local = Path.home() / ".local/state/music-tier2/musicbrainz-isrc.jsonl" + for path in (local, state_local): + if not path.exists(): + continue + with path.open(encoding="utf-8") as handle: + for line in handle: + try: + row = json.loads(line) + except json.JSONDecodeError: + continue + if row.get("isrc") != isrc or not row.get("found"): + continue + recordings = row.get("recordings") or [] + if recordings: + chosen = min( + recordings, + key=lambda value: abs( + (value.get("length_ms") or 0) + - (track.get("full_duration_ms") or 0) + ), + ) + result = self._mb_recording(chosen, str(isrc)) + self._mb_cache[cache_key] = result + return result + try: + response = self._musicbrainz_get( + f"isrc/{urllib.parse.quote(str(isrc))}", + {"inc": "recordings+artist-credits"}, + ) + recordings = response.get("recordings") or [] + if recordings: + chosen = min( + recordings, + key=lambda value: abs( + (value.get("length") or 0) - (track.get("full_duration_ms") or 0) + ), + ) + result = self._mb_recording(chosen, str(isrc)) + self._mb_cache[cache_key] = result + return result + except Exception: + pass + self._mb_cache[cache_key] = None + return None + + def _mb_for_text(self, item: dict[str, Any]) -> dict[str, Any] | None: + artist = item.get("artist") or "" + title = item.get("title") or item.get("query") or "" + cache_key = "text:" + norm(f"{artist} {title}") + if cache_key in self._mb_cache: + return self._mb_cache[cache_key] + query = f'recording:"{title}"' + if artist: + query += f' AND artist:"{artist}"' + try: + response = self._musicbrainz_get( + "recording/", {"query": query, "limit": "10", "inc": "isrcs"} + ) + except Exception: + self._mb_cache[cache_key] = None + return None + candidates = [] + for row in response.get("recordings") or []: + record = self._mb_recording(row) + if not record.get("length_ms"): + continue + if not title_agrees( + title, record.get("title"), [artist, item.get("album") or ""] + ): + continue + if artist and not artist_agrees([artist], " ".join(record.get("artists") or [])): + continue + candidates.append(record) + result = candidates[0] if candidates else None + self._mb_cache[cache_key] = result + return result + + def _yt_search(self, query: str, count: int = 8) -> tuple[list[dict[str, Any]], str]: + command = [ + "yt-dlp", + "--ignore-config", + "--no-warnings", + "--flat-playlist", + "--print", + "%(id)s\t%(duration)s\t%(channel)s\t%(title)s", + ] + if self.yt_cookie.exists(): + command += ["--cookies", str(self.yt_cookie)] + command.append(f"ytsearch{count}:{query}") + try: + result = self._run(command, timeout=150) + except subprocess.TimeoutExpired: + return [], "timeout" + candidates = [] + for line in result.stdout.splitlines(): + fields = line.split("\t") + if len(fields) != 4: + continue + video_id, duration, channel, title = fields + try: + duration_s = float(duration) + except ValueError: + continue + candidates.append( + { + "id": video_id, + "duration": duration_s, + "channel": channel, + "title": title, + } + ) + error = result.stderr or "" + if result.returncode != 0 and not error: + error = f"yt-dlp search exited {result.returncode}" + return candidates, error + + def _preview(self, track: dict[str, Any]) -> Path | None: + path = self.previews / (stable_stem(str(track["id"])) + ".mp3") + if path.exists() and path.stat().st_size > 20_000: + return path + command = [ + "yt-dlp", + "--ignore-config", + "--no-warnings", + "-f", + "bestaudio", + "-o", + str(path), + str(track["url"]), + ] + try: + self._run(command, timeout=180) + except subprocess.TimeoutExpired: + return None + return path if path.exists() and path.stat().st_size > 20_000 else None + + def _yt_section(self, video_id: str) -> tuple[Path | None, str | None]: + section_dir = Path(tempfile.mkdtemp(prefix="yt-section-", dir=self.work)) + stem = section_dir / "candidate" + command = [ + "yt-dlp", + "--ignore-config", + "--no-warnings", + "--sleep-requests", + "1", + "-f", + "bestaudio", + "--download-sections", + "*0-90", + "--force-keyframes-at-cuts", + "-o", + str(stem) + ".%(ext)s", + ] + if self.yt_cookie.exists(): + command += ["--cookies", str(self.yt_cookie)] + command.append(f"https://www.youtube.com/watch?v={video_id}") + try: + result = self._run(command, timeout=360) + except subprocess.TimeoutExpired: + return None, "timeout" + files = [path for path in section_dir.iterdir() if path.is_file()] + audio = next( + (path for path in files if path.suffix.lower() in AUDIO_EXTENSIONS), None + ) + return audio, None if audio else (result.stderr or "section download failed")[-400:] + + def _load_fpmatch(self): + if self._fpmatch is not None: + return self._fpmatch + path = self.campaign_repo / "scripts" / "fpmatch.py" + if not path.exists(): + raise RuntimeError(f"missing campaign fingerprint matcher: {path}") + spec = importlib.util.spec_from_file_location("music_acquire_fpmatch", path) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot import {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + self._fpmatch = module + return module + + @staticmethod + def _identity(item: dict[str, Any], track: dict[str, Any] | None, mb: dict | None): + titles: list[str] = [] + artists: list[str] = [] + if mb: + titles.append(mb.get("title") or "") + artists.extend(mb.get("artists") or [mb.get("artist") or ""]) + if track: + titles.extend( + value + for value in (track.get("pm_release_title"), track.get("title")) + if value + ) + artists.extend( + value + for value in (track.get("pm_artist"), track.get("uploader_name")) + if value + ) + else: + titles.extend(value for value in (item.get("title"), item.get("query")) if value) + artists.extend(value for value in (item.get("artist"),) if value) + return titles, artists + + def verify_youtube( + self, item: dict[str, Any], track: dict[str, Any] | None + ) -> dict[str, Any]: + if not self.yt_cookie.exists(): + return {"status": "retryable", "reason": "youtube_cookie_unavailable"} + mb = self._mb_for_sc(track) if track else self._mb_for_text(item) + titles, artists = self._identity(item, track, mb) + if track: + duration_ms = track.get("full_duration_ms") or track.get("duration_ms") + target_s = float(duration_ms) / 1000.0 if duration_ms else None + else: + target_s = ( + float(mb["length_ms"]) / 1000.0 if mb and mb.get("length_ms") else None + ) + if target_s is None: + return {"status": "unverified", "reason": "no_reference_duration"} + + queries = [] + if mb: + queries.append(f"{mb.get('artist') or ''} {mb.get('title') or ''}".strip()) + if item.get("artist") and item.get("title"): + queries.append(f"{item['artist']} {item['title']}") + if track and track.get("pm_artist") and track.get("pm_release_title"): + queries.append(f"{track['pm_artist']} {track['pm_release_title']}") + queries.append(item.get("query") or (titles[0] if titles else "")) + unique_queries = [] + seen_queries = set() + for query in queries: + key = norm(query) + if query and key not in seen_queries: + seen_queries.add(key) + unique_queries.append(query) + + candidates: dict[str, dict[str, Any]] = {} + search_errors = [] + for query in unique_queries[:4]: + rows, error = self._yt_search(query) + search_errors.append(error) + for candidate in rows: + if abs(candidate["duration"] - target_s) <= 8.0: + candidates.setdefault(candidate["id"], candidate) + if len(candidates) >= 3: + break + if not candidates: + if any(is_rate_limited(error) for error in search_errors): + return {"status": "retryable", "reason": "throttled_during_search"} + if any(error.strip() for error in search_errors): + return { + "status": "retryable", + "reason": "youtube_search_failed", + "detail": next(error[-300:] for error in search_errors if error.strip()), + } + return { + "status": "unverified", + "reason": "no_candidate", + "queries": unique_queries, + } + + def score(candidate: dict[str, Any]) -> tuple[float, str]: + value = 0.0 + context = [ + *artists, + item.get("album") or "", + (track or {}).get("pm_album_title") or "", + ] + if any( + title_agrees(title, candidate["title"], context) for title in titles + ): + value += 3 + if artist_agrees(artists, f"{candidate['title']} {candidate['channel']}"): + value += 2 + if str(candidate.get("channel") or "").endswith("- Topic"): + value += 1.5 + value -= abs(candidate["duration"] - target_s) / 4.0 + return (-value, candidate["id"]) + + ranked = sorted(candidates.values(), key=score)[:3] + preview = self._preview(track) if track and track.get("has_preview") else None + rejects = [] + accepted = None + section_failures = 0 + if preview: + matcher = self._load_fpmatch() + for candidate in ranked: + section, error = self._yt_section(candidate["id"]) + if error and is_rate_limited(error): + return { + "status": "retryable", + "reason": "throttled_during_verification", + "detail": error[-240:], + } + if not section: + section_failures += 1 + rejects.append({**candidate, "rejected": "section_download_failed"}) + continue + try: + result = matcher.compare(str(preview), str(section)) + finally: + shutil.rmtree(section.parent, ignore_errors=True) + duration_delta = abs(candidate["duration"] - target_s) + if result and result[0] <= BER_ACCEPT and duration_delta <= DUR_TOL_ACOUSTIC: + accepted = { + "candidate": candidate, + "verdict": "acoustic", + "evidence": { + "ber": round(result[0], 4), + "offset_s": result[1], + "overlap_s": round(result[2], 1), + "dur_ref_s": round(target_s, 1), + "dur_got_s": candidate["duration"], + }, + } + break + rejects.append( + { + **candidate, + "rejected": f"ber_{round(result[0], 4) if result else 'unreadable'}", + } + ) + if not accepted and section_failures == len(ranked): + return { + "status": "retryable", + "reason": "no_audio_fetched_for_any_candidate", + "candidates": rejects, + } + else: + for candidate in ranked: + verdict, evidence = metadata_verdict( + reference_titles=titles, + reference_artists=artists, + candidate_title=candidate["title"], + candidate_channel=candidate["channel"], + candidate_duration_s=candidate["duration"], + source_duration_s=target_s, + mb_recording=mb, + reference_context=[ + item.get("album") or "", + (track or {}).get("pm_album_title") or "", + ], + ) + if verdict: + accepted = { + "candidate": candidate, + "verdict": verdict, + "evidence": evidence, + } + break + rejects.append({**candidate, "rejected": evidence}) + + if not accepted: + return { + "status": "unverified", + "reason": "unverified", + "queries": unique_queries, + "candidates": rejects, + } + candidate = accepted["candidate"] + downloaded = self._download_native( + url=f"https://www.youtube.com/watch?v={candidate['id']}", + item_id=str(item["id"]), + cookies=self.yt_cookie, + format_selector="bestaudio/best", + ) + if downloaded.get("status") != "ok": + return { + "status": "retryable", + "reason": "verified_but_download_failed", + "detail": downloaded.get("detail") or downloaded.get("reason"), + "yt_id": candidate["id"], + "verdict": accepted["verdict"], + "evidence": accepted["evidence"], + } + return { + "status": "ok", + "path": downloaded["path"], + "bytes": downloaded.get("bytes"), + "yt_id": candidate["id"], + "yt_title": candidate["title"], + "yt_channel": candidate["channel"], + "verdict": accepted["verdict"], + "evidence": accepted["evidence"], + "also_considered": rejects, + } + + def download_ytmusic(self, item: dict[str, Any]) -> dict[str, Any]: + result = self._download_native( + url=str(item["url"]), + item_id=str(item["id"]), + cookies=self.yt_cookie, + format_selector="bestaudio/best", + ) + if result.get("status") == "ok": + result.update( + { + "yt_id": item.get("yt_id"), + "verdict": "source", + "evidence": {"artist_source": item.get("source")}, + } + ) + return result + + def retry_youtube_download( + self, item: dict[str, Any], previous: dict[str, Any] + ) -> dict[str, Any]: + """Retry a proven video id without paying for verification again.""" + if not self.yt_cookie.exists(): + return {"status": "retryable", "reason": "youtube_cookie_unavailable"} + video_id = str(previous["yt_id"]) + downloaded = self._download_native( + url=f"https://www.youtube.com/watch?v={video_id}", + item_id=str(item["id"]), + cookies=self.yt_cookie, + format_selector="bestaudio/best", + ) + if downloaded.get("status") != "ok": + return { + "status": "retryable", + "reason": "verified_but_download_failed", + "detail": downloaded.get("detail") or downloaded.get("reason"), + "yt_id": video_id, + "verdict": previous.get("verdict"), + "evidence": previous.get("evidence") or {}, + } + return { + "status": "ok", + "path": downloaded["path"], + "bytes": downloaded.get("bytes"), + "yt_id": video_id, + "yt_title": previous.get("yt_title"), + "yt_channel": previous.get("yt_channel"), + "verdict": previous.get("verdict"), + "evidence": previous.get("evidence") or {}, + "repaired_from": "verified_but_download_failed", + } + + def download_bandcamp(self, item: dict[str, Any]) -> dict[str, Any]: + result = self._download_native( + url=str(item["url"]), + item_id=str(item["id"]), + cookies=None, + format_selector="mp3-128/bestaudio/best", + ) + if result.get("status") == "ok": + result.update( + { + "bandcamp_id": item.get("bandcamp_id"), + "verdict": "source", + "evidence": { + "release_url": item.get("release_url"), + "alias": item.get("alias"), + "purchases_attempted": False, + }, + } + ) + return result + + # -- worker-resident realtime capture ----------------------------------- + + def capture(self, item: dict[str, Any], track: dict[str, Any]) -> dict[str, Any]: + if track.get("segment"): + return { + "status": "unavailable", + "reason": "capture_unsupported_segment", + } + info = self.capture_info + if self.no_capture: + return {"status": "unavailable", "reason": "capture_unavailable"} + if info.get("host_busy"): + return {"status": "retryable", "reason": "capture_host_busy"} + if not info.get("host_reachable") or not info.get("entitlement_active"): + return {"status": "unavailable", "reason": "capture_unavailable"} + if socket.gethostname().split(".")[0] == self.capture_host: + return self._capture_local(item, track) + return self._capture_remote(item, track) + + def _capture_work_item(self, item: dict[str, Any], track: dict[str, Any]) -> dict[str, Any]: + return { + "id": stable_stem(str(item["id"])), + "url": track.get("url"), + "title": track.get("title") or item.get("title"), + "full_duration_ms": track.get("full_duration_ms"), + "artwork_url": track.get("artwork_url"), + } + + @staticmethod + def _capture_result(ledger: Path, capture_id: str) -> dict[str, Any]: + final = None + if ledger.exists(): + with ledger.open(encoding="utf-8") as handle: + for line in handle: + try: + row = json.loads(line) + except json.JSONDecodeError: + continue + if str(row.get("id")) == capture_id: + final = row + if not final: + return {"status": "retryable", "reason": "capture_produced_no_ledger_row"} + if final.get("status") == "ok": + return { + "status": "ok", + "path": final.get("path"), + "evidence": final.get("evidence") or {}, + } + reason = str(final.get("reason") or "capture_failed") + if "fingerprint_mismatch" in reason or "duration_mismatch" in reason: + return { + "status": "rejected", + "reason": "capture_verification_failed", + "evidence": final.get("evidence") or {}, + "detail": reason, + } + return { + "status": "retryable", + "reason": reason, + "evidence": final.get("evidence") or {}, + } + + def _capture_local(self, item: dict[str, Any], track: dict[str, Any]) -> dict[str, Any]: + script = self.campaign_repo / "scripts" / "tier3-capture.py" + if not script.exists(): + return {"status": "unavailable", "reason": "capture_engine_missing"} + capture_state = self.state.path / "capture" + capture_state.mkdir(parents=True, exist_ok=True) + capture_item = self._capture_work_item(item, track) + worklist = capture_state / "worklist.jsonl" + worklist.write_text(json.dumps(capture_item, ensure_ascii=False) + "\n", encoding="utf-8") + if self.sc_cookie.exists(): + shutil.copyfile(self.sc_cookie, capture_state / "sc-cookies.txt") + os.chmod(capture_state / "sc-cookies.txt", 0o600) + preview = self.previews / (stable_stem(str(track["id"])) + ".mp3") + capture_previews = capture_state / "previews" + capture_previews.mkdir(exist_ok=True) + if preview.exists(): + shutil.copyfile(preview, capture_previews / (capture_item["id"] + ".mp3")) + env = dict(self.environ) + env.update( + { + "TIER3_STATE": str(capture_state), + "TIER3_OUT": str(self.out), + "TIER3_PREVIEWS": str(capture_previews), + "TIER3_LIMIT": "1", + } + ) + result = self._run( + ["python3", str(script), str(worklist)], timeout=7200, env=env + ) + outcome = self._capture_result( + capture_state / "tier3-captures.jsonl", capture_item["id"] + ) + if outcome.get("status") != "ok" and result.returncode != 0: + outcome.setdefault("detail", (result.stderr or "")[-300:]) + return outcome + + def _capture_remote(self, item: dict[str, Any], track: dict[str, Any]) -> dict[str, Any]: + capture_item = self._capture_work_item(item, track) + batch = stable_stem(self.state.batch) + remote_state = f"/home/tom/.local/state/music-acquire-capture/{batch}" + remote_out = f"/home/tom/music-staging/_ingest/acquire-{batch}" + remote_repo = self.environ.get( + "MUSIC_ACQUIRE_REMOTE_REPO", "/home/tom/mecattaf/music-consolidation" + ) + setup = ( + f"mkdir -p {shlex.quote(remote_state)}/previews {shlex.quote(remote_out)}; " + f"install -m 600 /run/agenix/soundcloud-cookies " + f"{shlex.quote(remote_state)}/sc-cookies.txt; " + f"cat > {shlex.quote(remote_state)}/worklist-one.jsonl" + ) + result = self._run( + ["ssh", self.capture_host, "bash", "-lc", shlex.quote(setup)], + input_text=json.dumps(capture_item, ensure_ascii=False) + "\n", + timeout=30, + ) + if result.returncode != 0: + return { + "status": "retryable", + "reason": "capture_remote_setup_failed", + "detail": (result.stderr or "")[-300:], + } + preview = self.previews / (stable_stem(str(track["id"])) + ".mp3") + if preview.exists(): + copied = self._run( + [ + "scp", + "-q", + str(preview), + f"{self.capture_host}:{remote_state}/previews/{capture_item['id']}.mp3", + ], + timeout=120, + ) + if copied.returncode != 0: + return {"status": "retryable", "reason": "capture_preview_copy_failed"} + run_command = ( + f"TIER3_STATE={shlex.quote(remote_state)} " + f"TIER3_OUT={shlex.quote(remote_out)} " + f"TIER3_PREVIEWS={shlex.quote(remote_state + '/previews')} " + "TIER3_LIMIT=1 " + "FPCALC=/home/tom/.local/state/music-campaign/deps/chromaprint/bin/fpcalc " + f"python3 {shlex.quote(remote_repo + '/scripts/tier3-capture.py')} " + f"{shlex.quote(remote_state + '/worklist-one.jsonl')}" + ) + try: + ran = self._run( + ["ssh", self.capture_host, "bash", "-lc", shlex.quote(run_command)], + timeout=7200, + ) + except subprocess.TimeoutExpired: + return {"status": "retryable", "reason": "capture_timeout"} + local_ledger = self.state.path / "capture-remote-ledger.jsonl" + copied_ledger = self._run( + [ + "scp", + "-q", + f"{self.capture_host}:{remote_state}/tier3-captures.jsonl", + str(local_ledger), + ], + timeout=120, + ) + if copied_ledger.returncode != 0: + return { + "status": "retryable", + "reason": "capture_ledger_copy_failed", + "detail": (ran.stderr or "")[-300:], + } + outcome = self._capture_result(local_ledger, capture_item["id"]) + if outcome.get("status") == "ok": + remote_path = str(outcome.get("path")) + suffix = Path(remote_path).suffix or ".flac" + local_path = self.out / (stable_stem(str(item["id"])) + suffix) + copied_audio = self._run( + ["scp", "-q", f"{self.capture_host}:{remote_path}", str(local_path)], + timeout=1800, + ) + if copied_audio.returncode != 0 or not local_path.exists(): + return {"status": "retryable", "reason": "capture_audio_copy_failed"} + outcome["path"] = str(local_path) + return outcome diff --git a/pkgs/music-acquire/music_acquire/cli.py b/pkgs/music-acquire/music_acquire/cli.py new file mode 100644 index 00000000..94b44000 --- /dev/null +++ b/pkgs/music-acquire/music_acquire/cli.py @@ -0,0 +1,611 @@ +"""Command-line front door for the acquisition cascade.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import socket +import sys +import time +import unicodedata +from pathlib import Path +from typing import Any, Iterable + +from . import __version__ +from .backend import LiveBackend +from .state import BatchState, all_batch_statuses +from .verification import norm + + +DEFAULT_STATE_ROOT = Path("~/.local/state/music-acquire") +DEFAULT_CAMPAIGN_REPO = Path("~/mecattaf/music-consolidation") +DEFAULT_STAGING = Path("/mnt/nas/music-staging/_ingest") + + +def slug(value: str) -> str: + text = unicodedata.normalize("NFKD", value) + text = "".join(char for char in text if not unicodedata.combining(char)) + text = re.sub(r"[^A-Za-z0-9]+", "-", text.lower()).strip("-") + return text[:80] or "batch" + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + rows = [] + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, 1): + if not line.strip(): + continue + try: + row = json.loads(line) + except json.JSONDecodeError as error: + raise ValueError(f"{path}:{line_number}: {error}") from error + if not isinstance(row, dict): + raise ValueError(f"{path}:{line_number}: expected a JSON object") + rows.append(row) + return rows + + +def tracklist_items(path: Path, source_filter: str | None = None) -> list[dict[str, Any]]: + merged: dict[str, dict[str, Any]] = {} + order: list[str] = [] + for row in load_jsonl(path): + appearances = list(row.get("appearances") or []) + if source_filter and not any( + appearance.get("source") == source_filter for appearance in appearances + ): + continue + item_id = str(row.get("key") or row.get("id") or norm(row.get("query") or "")) + if not item_id: + raise ValueError("tracklist row has no key, id, or usable query") + source = source_filter or ( + appearances[0].get("source") if len({a.get("source") for a in appearances}) == 1 else "mixed" + ) + item = { + **row, + "id": item_id, + "key": row.get("key") or item_id, + "source": f"tracklist:{source}", + "appearances": appearances, + } + if item_id not in merged: + merged[item_id] = item + order.append(item_id) + continue + current = merged[item_id] + seen = { + json.dumps(appearance, sort_keys=True, ensure_ascii=False) + for appearance in current.get("appearances") or [] + } + for appearance in appearances: + key = json.dumps(appearance, sort_keys=True, ensure_ascii=False) + if key not in seen: + current.setdefault("appearances", []).append(appearance) + seen.add(key) + return [merged[item_id] for item_id in order] + + +def _without_status(outcome: dict[str, Any]) -> dict[str, Any]: + return {key: value for key, value in outcome.items() if key != "status" and value is not None} + + +class Acquirer: + """Deterministic orchestration over an injectable acquisition backend.""" + + def __init__( + self, + state: BatchState, + backend: Any, + *, + limit: int = 0, + backoff_seconds: float | None = None, + dry_limit: int | None = None, + ): + self.state = state + self.backend = backend + self.limit = max(0, limit) + self.backoff_seconds = ( + float(os.environ.get("MUSIC_ACQUIRE_BACKOFF", "3600")) + if backoff_seconds is None + else backoff_seconds + ) + self.dry_limit = ( + int(os.environ.get("MUSIC_ACQUIRE_DRY_LIMIT", "4")) + if dry_limit is None + else dry_limit + ) + self.dry_streak = 0 + self.final = state.final_rows() + + def _record(self, item: dict[str, Any], disposition: str, **fields: Any): + row = self.state.record(item, disposition, **fields) + self.final[str(item["id"])] = row + if disposition.startswith("ok_"): + self.dry_streak = 0 + self.backend.remember(item, row) + elif disposition == "retryable" and fields.get("reason") in { + "throttled_during_search", + "throttled_during_verification", + "no_audio_fetched_for_any_candidate", + }: + self.dry_streak += 1 + if self.dry_streak >= self.dry_limit: + self.state.append( + { + "record": "event", + "ts": row["ts"], + "event": "backoff", + "reason": "consecutive_dry_attempts", + "seconds": self.backoff_seconds, + } + ) + self.dry_streak = 0 + if self.backoff_seconds > 0: + time.sleep(self.backoff_seconds) + return row + + def _duplicate(self, item: dict[str, Any], track: dict[str, Any] | None = None): + duplicate = self.backend.duplicate(item, track) + if duplicate: + return self._record(item, "skipped_duplicate", **duplicate) + return None + + @staticmethod + def _sc_fields(track: dict[str, Any]) -> dict[str, Any]: + if track.get("segment"): + return { + "sc_id": None, + "sc_parent_id": str(track["id"]), + "sc_url": track.get("url"), + } + return {"sc_id": str(track["id"]), "sc_url": track.get("url")} + + def _direct_source(self, item: dict[str, Any], verb: str): + if self._duplicate(item): + return + if verb == "bandcamp": + outcome = self.backend.download_bandcamp(item) + disposition = "ok_bandcamp" + else: + outcome = self.backend.download_ytmusic(item) + disposition = "ok_youtube" + status = outcome.get("status") + if status == "ok": + self._record(item, disposition, **_without_status(outcome)) + elif status == "gone": + self._record(item, "gone", **_without_status(outcome)) + else: + self._record(item, "retryable", **_without_status(outcome)) + + def _cascade(self, item: dict[str, Any]): + if self._duplicate(item): + return + previous = self.final.get(str(item["id"])) or {} + if ( + previous.get("disposition") == "retryable" + and previous.get("stage") == "youtube" + and previous.get("reason") == "verified_but_download_failed" + and previous.get("yt_id") + ): + youtube = self.backend.retry_youtube_download(item, previous) + if youtube.get("status") == "ok": + self._record( + item, + "ok_youtube", + sc_id=previous.get("sc_id"), + sc_url=previous.get("sc_url"), + **_without_status(youtube), + ) + else: + self._record( + item, + "retryable", + sc_id=previous.get("sc_id"), + sc_url=previous.get("sc_url"), + **_without_status(youtube), + stage="youtube", + ) + return + + resolved = self.backend.resolve_soundcloud(item) + status = resolved.get("status") + if status == "retryable": + self._record(item, "retryable", **_without_status(resolved)) + return + if status == "gone": + self._record(item, "gone", **_without_status(resolved)) + return + + track = resolved.get("track") if status == "found" else None + if track and self._duplicate(item, track): + return + + resume_capture = ( + previous.get("disposition") == "retryable" + and previous.get("stage") == "capture" + ) or ( + previous.get("disposition") == "fallthrough" + and previous.get("reason") == "capture_unavailable" + ) + if resume_capture and track: + self._finish_capture(item, track, previous.get("stage2_reason")) + return + + resume_youtube = ( + previous.get("disposition") == "retryable" + and previous.get("stage") == "youtube" + ) + if track and not resume_youtube: + direct = self.backend.download_soundcloud(item, track) + direct_status = direct.get("status") + if direct_status == "ok": + self._record( + item, + "ok_soundcloud", + **_without_status(direct), + **self._sc_fields(track), + verdict="source", + evidence={ + "duration_s": round( + float(track.get("full_duration_ms") or 0) / 1000.0, 1 + ) + }, + ) + return + if direct_status == "retryable": + self._record( + item, + "retryable", + **_without_status(direct), + **self._sc_fields(track), + stage="soundcloud", + ) + return + + youtube = self.backend.verify_youtube(item, track) + youtube_status = youtube.get("status") + common = self._sc_fields(track) if track else {"sc_id": None, "sc_url": None} + if youtube_status == "ok": + self._record( + item, "ok_youtube", **common, **_without_status(youtube) + ) + return + if youtube_status == "retryable": + self._record( + item, + "retryable", + **common, + **_without_status(youtube), + stage="youtube", + ) + return + if not track: + self._record( + item, "fallthrough", **common, **_without_status(youtube) + ) + return + + self._finish_capture(item, track, youtube.get("reason"), youtube) + + def _finish_capture( + self, + item: dict[str, Any], + track: dict[str, Any], + stage2_reason: str | None, + youtube: dict[str, Any] | None = None, + ) -> None: + capture = self.backend.capture(item, track) + capture_status = capture.get("status") + common = self._sc_fields(track) + if capture_status == "ok": + self._record( + item, + "ok_capture", + **common, + **_without_status(capture), + verdict="acoustic" if track.get("has_preview") else "capture", + ) + elif capture_status == "retryable": + self._record( + item, + "retryable", + **common, + **_without_status(capture), + stage2_reason=stage2_reason, + stage="capture", + ) + else: + fields = _without_status(capture) + fields.setdefault("reason", "capture_unavailable") + fields["stage2_reason"] = stage2_reason + if youtube and youtube.get("candidates"): + fields["stage2_candidates"] = youtube["candidates"] + self._record(item, "fallthrough", **common, **fields) + + def run(self, items: Iterable[dict[str, Any]], verb: str) -> int: + archived = self.state.archived() + attempted = 0 + for item in items: + if str(item["id"]) in archived: + continue + if self.limit and attempted >= self.limit: + break + if verb in {"bandcamp", "ytmusic"}: + self._direct_source(item, verb) + else: + self._cascade(item) + attempted += 1 + archived = self.state.archived() + return attempted + + +def _common_parser() -> argparse.ArgumentParser: + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--batch", default=argparse.SUPPRESS) + common.add_argument("--dry-run", action="store_true", default=argparse.SUPPRESS) + common.add_argument("--limit", type=int, default=argparse.SUPPRESS) + common.add_argument("--no-capture", action="store_true", default=argparse.SUPPRESS) + common.add_argument("--out", type=Path, default=argparse.SUPPRESS) + return common + + +def parser() -> argparse.ArgumentParser: + common = _common_parser() + root = argparse.ArgumentParser( + prog="acquire", + description="Land source audio only when its identity is recorded and defensible.", + parents=[common], + ) + root.add_argument("--version", action="version", version=__version__) + commands = root.add_subparsers(dest="verb", required=True) + + tracklist = commands.add_parser("tracklist", parents=[common]) + tracklist.add_argument("worklist", type=Path) + tracklist.add_argument( + "--source", choices=("goldcast", "synapson", "vent-2024") + ) + + soundcloud = commands.add_parser("soundcloud", parents=[common]) + soundcloud.add_argument("artist") + soundcloud.add_argument("--likes", action="store_true") + soundcloud.add_argument("--reposts", action="store_true") + soundcloud.add_argument("--sets", action="store_true") + + bandcamp = commands.add_parser("bandcamp", parents=[common]) + bandcamp.add_argument("artist") + bandcamp.add_argument("--alias") + + ytmusic = commands.add_parser("ytmusic", parents=[common]) + ytmusic.add_argument("artist") + + status = commands.add_parser("status", parents=[common]) + status.add_argument("--json", action="store_true") + + commands.add_parser("resume", parents=[common]) + return root + + +def _state_root(environ: dict[str, str]) -> Path: + return Path(environ.get("MUSIC_ACQUIRE_STATE_ROOT", str(DEFAULT_STATE_ROOT))).expanduser() + + +def _campaign_repo(environ: dict[str, str]) -> Path: + return Path( + environ.get("MUSIC_CONSOLIDATION_REPO", str(DEFAULT_CAMPAIGN_REPO)) + ).expanduser() + + +def _human_status(statuses: list[dict[str, Any]]) -> None: + if not statuses: + print("no music-acquire batches") + return + for value in statuses: + percent = 100.0 * value["completion"] + counts = " ".join( + f"{name}={count}" for name, count in value["dispositions"].items() + ) + estimate = value.get("estimate_seconds") + eta = f" eta={estimate}s" if estimate is not None and value["remaining"] else "" + print( + f"{value['batch']}: {value['archived']}/{value['total']} " + f"({percent:.1f}%) remaining={value['remaining']}{eta}" + ) + if counts: + print(f" {counts}") + + +def _default_batch(args: argparse.Namespace) -> str: + if hasattr(args, "batch"): + return slug(args.batch) + if args.verb == "tracklist": + return slug(args.source or args.worklist.stem) + if args.verb in {"soundcloud", "bandcamp", "ytmusic"}: + return slug(args.artist.rstrip("/").rsplit("/", 1)[-1]) + raise ValueError("--batch is required") + + +def _enumerate( + args: argparse.Namespace, backend: LiveBackend | None +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + if args.verb == "tracklist": + path = args.worklist.expanduser().resolve() + items = tracklist_items(path, args.source) + return items, { + "verb": "tracklist", + "source": str(path), + "input": str(path), + "source_filter": args.source, + } + if backend is None: + raise RuntimeError("a live backend is required for remote enumeration") + if args.verb == "soundcloud": + artist_url, items = backend.enumerate_soundcloud( + args.artist, likes=args.likes, reposts=args.reposts, sets=args.sets + ) + return items, { + "verb": "soundcloud", + "source": artist_url, + "input": args.artist, + "likes": args.likes, + "reposts": args.reposts, + "sets": args.sets, + } + if args.verb == "bandcamp": + items = backend.enumerate_bandcamp(args.artist, alias=args.alias) + return items, { + "verb": "bandcamp", + "source": args.artist, + "input": args.artist, + "alias": args.alias, + } + artist_url, items = backend.enumerate_ytmusic(args.artist) + return items, {"verb": "ytmusic", "source": artist_url, "input": args.artist} + + +def main(argv: list[str] | None = None, *, environ: dict[str, str] | None = None) -> int: + environ = dict(os.environ if environ is None else environ) + args = parser().parse_args(argv) + state_root = _state_root(environ) + + if args.verb == "status": + if hasattr(args, "batch"): + statuses = [BatchState(state_root, slug(args.batch)).status()] + else: + statuses = all_batch_statuses(state_root) + if args.json: + print(json.dumps({"batches": statuses}, ensure_ascii=False, sort_keys=True)) + else: + _human_status(statuses) + return 0 + + if args.verb == "resume" and not hasattr(args, "batch"): + parser().error("resume requires --batch NAME") + + batch = _default_batch(args) + state = BatchState(state_root, batch) + no_capture = bool(getattr(args, "no_capture", False)) + limit = max(0, int(getattr(args, "limit", 0))) + dry_run = bool(getattr(args, "dry_run", False)) + campaign_repo = _campaign_repo(environ) + backend: LiveBackend | None = None + + if args.verb == "resume": + request = state.request() + verb = str(request["verb"]) + items = state.worklist() + out = Path(getattr(args, "out", request["out"])).expanduser() + else: + verb = args.verb + existing_request = state.request() if state.request_path.exists() else None + if existing_request: + if existing_request.get("verb") != verb: + parser().error( + f"batch {batch!r} belongs to {existing_request.get('verb')!r}, " + f"not {verb!r}" + ) + requested_input = ( + str(args.worklist.expanduser().resolve()) + if verb == "tracklist" + else args.artist + ) + old_input = existing_request.get("input") or existing_request.get("source") + if old_input is not None and old_input != requested_input: + parser().error( + f"batch {batch!r} already belongs to source {old_input!r}" + ) + if verb == "tracklist" and existing_request.get("source_filter") != args.source: + parser().error( + f"batch {batch!r} already uses --source " + f"{existing_request.get('source_filter')!r}" + ) + request = existing_request + out = Path(getattr(args, "out", request["out"])).expanduser() + if verb == "tracklist": + fresh_items, _ = _enumerate(args, None) + state.merge_worklist(fresh_items) + items = state.worklist() + else: + out = Path( + getattr(args, "out", DEFAULT_STAGING / f"acquire-{batch}") + ).expanduser() + # Remote enumeration needs command/cookie mechanics but deliberately + # precedes prepare(): a dry run never probes entitlement. + if verb == "tracklist": + items, request = _enumerate(args, None) + else: + backend = LiveBackend( + state, + out, + campaign_repo, + no_capture=no_capture, + capture_host=environ.get("MUSIC_ACQUIRE_CAPTURE_HOST", "worker"), + environ=environ, + ) + items, request = _enumerate(args, backend) + request.update({"batch": batch, "out": str(out), "no_capture": no_capture}) + state.merge_worklist(items) + state.save_request(request) + items = state.worklist() + + archived = state.archived() + remaining = [item for item in items if str(item["id"]) not in archived] + if dry_run: + preview = { + "batch": batch, + "verb": verb, + "total": len(items), + "remaining": len(remaining), + "would_attempt": min(len(remaining), limit) if limit else len(remaining), + "out": str(out), + } + print(json.dumps(preview, ensure_ascii=False, sort_keys=True)) + return 0 + + reopenable = args.verb == "resume" and not no_capture and any( + row.get("disposition") == "fallthrough" + and row.get("reason") == "capture_unavailable" + for row in state.final_rows().values() + ) + if not remaining and not reopenable: + status = state.status() + print( + f"batch {batch}: attempted 0; " + f"{status['archived']}/{status['total']} archived" + ) + return 0 + + if backend is None: + backend = LiveBackend( + state, + out, + campaign_repo, + no_capture=no_capture, + capture_host=environ.get("MUSIC_ACQUIRE_CAPTURE_HOST", "worker"), + environ=environ, + ) + capture = backend.prepare() + if args.verb == "resume" and not no_capture and ( + capture.get("entitlement_active") and capture.get("host_reachable") + ): + state.reopen_capture_unavailable() + items = state.worklist() + state.append_header( + verb=verb, + source=request.get("source"), + host=socket.gethostname(), + out=str(out), + entitlement=capture.get("entitlement"), + capture=capture, + ) + attempted = Acquirer(state, backend, limit=limit).run(items, verb) + status = state.status() + print( + f"batch {batch}: attempted {attempted}; " + f"{status['archived']}/{status['total']} archived" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pkgs/music-acquire/music_acquire/state.py b/pkgs/music-acquire/music_acquire/state.py new file mode 100644 index 00000000..90b001fe --- /dev/null +++ b/pkgs/music-acquire/music_acquire/state.py @@ -0,0 +1,318 @@ +"""Durable batch state for :mod:`music_acquire`. + +The archive is intentionally the resume gate. The append-only ledger is the +audit trail; a retryable row is evidence about an attempt, never permission to +skip the item on the next run. +""" + +from __future__ import annotations + +import collections +import fcntl +import json +import os +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + + +TERMINAL_DISPOSITIONS = { + "ok_soundcloud", + "ok_youtube", + "ok_capture", + "ok_bandcamp", + "skipped_duplicate", + "gone", + "fallthrough", +} + + +def now() -> str: + return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + rows: list[dict[str, Any]] = [] + with path.open(encoding="utf-8") as handle: + for line in handle: + try: + row = json.loads(line) + except (json.JSONDecodeError, TypeError): + continue + if isinstance(row, dict): + rows.append(row) + return rows + + +def _atomic_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(value, handle, ensure_ascii=False, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, path) + finally: + try: + os.unlink(tmp_name) + except FileNotFoundError: + pass + + +def _atomic_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, path) + finally: + try: + os.unlink(tmp_name) + except FileNotFoundError: + pass + + +class BatchState: + """One batch's worklist, append-only evidence ledger, and resume gate.""" + + def __init__(self, root: Path, batch: str): + self.root = root.expanduser().resolve() + self.batch = batch + self.path = self.root / batch + self.worklist_path = self.path / "worklist.jsonl" + self.ledger_path = self.path / "ledger.jsonl" + self.archive_path = self.path / "archive.txt" + self.request_path = self.path / "request.json" + self.cookies_path = self.path / "cookies" + self.lock_path = self.path / ".lock" + + def ensure(self) -> None: + self.path.mkdir(parents=True, exist_ok=True, mode=0o700) + self.cookies_path.mkdir(parents=True, exist_ok=True, mode=0o700) + self.ledger_path.touch(exist_ok=True) + self.archive_path.touch(exist_ok=True) + os.chmod(self.path, 0o700) + os.chmod(self.cookies_path, 0o700) + + def _lock(self): + self.ensure() + handle = self.lock_path.open("a+") + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + return handle + + @staticmethod + def _merge_appearances( + old: list[dict[str, Any]], new: list[dict[str, Any]] + ) -> list[dict[str, Any]]: + seen: set[str] = set() + merged: list[dict[str, Any]] = [] + for appearance in old + new: + key = json.dumps(appearance, sort_keys=True, ensure_ascii=False) + if key not in seen: + seen.add(key) + merged.append(appearance) + return merged + + def merge_worklist(self, rows: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + """Create or extend a worklist without changing existing identities.""" + incoming = [dict(row) for row in rows] + for row in incoming: + if not row.get("id"): + raise ValueError("every worklist row needs a stable id") + + lock = self._lock() + try: + existing = _read_jsonl(self.worklist_path) + by_id = {str(row["id"]): row for row in existing if row.get("id")} + order = [str(row["id"]) for row in existing if row.get("id")] + changed = not self.worklist_path.exists() + for row in incoming: + item_id = str(row["id"]) + if item_id not in by_id: + by_id[item_id] = row + order.append(item_id) + changed = True + continue + current = by_id[item_id] + merged_appearances = self._merge_appearances( + list(current.get("appearances") or []), + list(row.get("appearances") or []), + ) + if merged_appearances != list(current.get("appearances") or []): + current = dict(current) + current["appearances"] = merged_appearances + by_id[item_id] = current + changed = True + result = [by_id[item_id] for item_id in order] + if changed or not self.worklist_path.exists(): + _atomic_jsonl(self.worklist_path, result) + return result + finally: + lock.close() + + def worklist(self) -> list[dict[str, Any]]: + return _read_jsonl(self.worklist_path) + + def save_request(self, request: dict[str, Any]) -> None: + lock = self._lock() + try: + if self.request_path.exists(): + return + _atomic_json(self.request_path, request) + finally: + lock.close() + + def request(self) -> dict[str, Any]: + if not self.request_path.exists(): + raise FileNotFoundError(f"missing batch request: {self.request_path}") + with self.request_path.open(encoding="utf-8") as handle: + value = json.load(handle) + if not isinstance(value, dict): + raise ValueError(f"invalid batch request: {self.request_path}") + return value + + def append(self, row: dict[str, Any]) -> None: + lock = self._lock() + try: + with self.ledger_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + handle.flush() + os.fsync(handle.fileno()) + finally: + lock.close() + + def append_header(self, **fields: Any) -> None: + self.append({"record": "batch", "ts": now(), "batch": self.batch, **fields}) + + def record(self, item: dict[str, Any], disposition: str, **fields: Any) -> dict[str, Any]: + if disposition not in TERMINAL_DISPOSITIONS and disposition != "retryable": + raise ValueError(f"unknown disposition: {disposition}") + row = { + "record": "item", + "ts": now(), + "id": str(item["id"]), + "query": item.get("query") or item.get("title") or item.get("url") or "", + "disposition": disposition, + "source": item.get("source"), + "appearances": list(item.get("appearances") or []), + **fields, + } + lock = self._lock() + try: + with self.ledger_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + handle.flush() + os.fsync(handle.fileno()) + if disposition in TERMINAL_DISPOSITIONS: + archived = self.archived() + if row["id"] not in archived: + with self.archive_path.open("a", encoding="utf-8") as handle: + handle.write(row["id"] + "\n") + handle.flush() + os.fsync(handle.fileno()) + finally: + lock.close() + return row + + def ledger(self) -> list[dict[str, Any]]: + return _read_jsonl(self.ledger_path) + + def final_rows(self) -> dict[str, dict[str, Any]]: + final: dict[str, dict[str, Any]] = {} + for row in self.ledger(): + if row.get("record") == "item" and row.get("id"): + final[str(row["id"])] = row + return final + + def archived(self) -> set[str]: + if not self.archive_path.exists(): + return set() + with self.archive_path.open(encoding="utf-8") as handle: + return {line.strip() for line in handle if line.strip()} + + def reopen_capture_unavailable(self) -> list[str]: + """Make no-capture holes runnable when a later resume has capture.""" + final = self.final_rows() + reopen = { + item_id + for item_id, row in final.items() + if row.get("disposition") == "fallthrough" + and row.get("reason") == "capture_unavailable" + } + if not reopen: + return [] + lock = self._lock() + try: + keep = sorted(self.archived() - reopen) + text = "".join(f"{item_id}\n" for item_id in keep) + fd, tmp_name = tempfile.mkstemp(prefix=".archive.", dir=self.path) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, self.archive_path) + finally: + try: + os.unlink(tmp_name) + except FileNotFoundError: + pass + finally: + lock.close() + return sorted(reopen) + + def status(self) -> dict[str, Any]: + work = self.worklist() + ids = {str(row["id"]) for row in work if row.get("id")} + archived = self.archived() & ids + final = self.final_rows() + dispositions = collections.Counter( + row.get("disposition", "unknown") for row in final.values() + ) + total = len(ids) + done = len(archived) + retryable = sum( + 1 for row in final.values() if row.get("disposition") == "retryable" + ) + + headers = [row for row in self.ledger() if row.get("record") == "batch"] + estimate_seconds = None + if done and headers: + try: + started = datetime.fromisoformat(headers[0]["ts"]) + elapsed = max(0.0, (datetime.now(started.tzinfo) - started).total_seconds()) + estimate_seconds = round((total - done) * elapsed / done) + except (KeyError, TypeError, ValueError): + estimate_seconds = None + + return { + "batch": self.batch, + "total": total, + "archived": done, + "remaining": max(0, total - done), + "completion": round(done / total, 6) if total else 1.0, + "dispositions": dict(sorted(dispositions.items())), + "retryable": retryable, + "estimate_seconds": estimate_seconds, + "last_header": headers[-1] if headers else None, + } + + +def all_batch_statuses(root: Path) -> list[dict[str, Any]]: + root = root.expanduser() + if not root.exists(): + return [] + statuses = [] + for path in sorted(root.iterdir()): + if path.is_dir() and (path / "worklist.jsonl").exists(): + statuses.append(BatchState(root, path.name).status()) + return statuses diff --git a/pkgs/music-acquire/music_acquire/verification.py b/pkgs/music-acquire/music_acquire/verification.py new file mode 100644 index 00000000..f56df8f1 --- /dev/null +++ b/pkgs/music-acquire/music_acquire/verification.py @@ -0,0 +1,222 @@ +"""Identity gates shared by source resolution and recorded verification.""" + +from __future__ import annotations + +import re +import unicodedata +from typing import Iterable + + +BER_ACCEPT = 0.15 +DUR_TOL_ACOUSTIC = 12.0 +DUR_TOL_ISRC = 3.0 +DUR_TOL_METADATA = 2.0 + +VERSION_WORDS = { + "remix", + "mix", + "edit", + "version", + "live", + "acoustic", + "instrumental", + "instrumentale", + "dub", + "vip", + "rework", + "bootleg", + "radio", + "extended", + "remaster", + "remastered", + "reprise", + "demo", + "cover", + "unplugged", + "session", + "interlude", + "intro", + "outro", + "club", + "original", +} + +NOISE_WORDS = { + "official", + "video", + "audio", + "music", + "lyric", + "lyrics", + "visualizer", + "visualiser", + "hd", + "hq", + "4k", + "1080p", + "720p", + "mv", + "clip", + "full", + "stream", + "streaming", + "new", + "out", + "now", + "free", + "download", + "premiere", + "topic", + "provided", + "youtube", + "records", + "recordings", + "release", + "explicit", +} + +FEAT_RE = re.compile(r"\b(feat|ft|featuring|avec|with)\b.*", re.I) + + +def norm(value: str | None) -> str: + if not value: + return "" + value = unicodedata.normalize("NFKD", value) + value = "".join(char for char in value if not unicodedata.combining(char)) + value = value.lower().replace("&", " and ").replace("’", "'") + value = re.sub(r"[^a-z0-9]+", " ", value) + return re.sub(r"\s+", " ", value).strip() + + +def core_and_version(title: str | None) -> tuple[list[str], set[str]]: + value = norm(FEAT_RE.sub("", title or "")) + tokens = [token for token in value.split() if token] + version = {token for token in tokens if token in VERSION_WORDS} + core = [ + token + for token in tokens + if token not in VERSION_WORDS + and token not in NOISE_WORDS + and not (len(token) == 1 and token.isdigit()) + ] + return core, version + + +def version_signature(markers: Iterable[str]) -> set[str]: + return set(markers) - {"original", "mix", "club", "version"} + + +def version_agrees(left: Iterable[str], right: Iterable[str]) -> bool: + return version_signature(left) == version_signature(right) + + +def core_agrees(left: Iterable[str], right: Iterable[str]) -> bool: + """Use the campaign's containment rule without fuzzy similarity.""" + left_set, right_set = set(left), set(right) + if not left_set or not right_set: + return False + shorter, longer = ( + (left_set, right_set) + if len(left_set) <= len(right_set) + else (right_set, left_set) + ) + missing = shorter - longer + return not missing or ( + len(missing) == 1 + and len(shorter) >= 4 + and all(len(token) <= 3 for token in missing) + ) + + +def title_agrees( + reference: str | None, + candidate: str | None, + allowed_context: Iterable[str] = (), +) -> bool: + """Require every candidate title token to be explained by identity context. + + Artist/album/label tokens may decorate an upload title. Unexplained title + tokens may name another movement or version (``Part Two``), so accepting + them merely because the shorter title is contained would violate the exact- + recording bar. + """ + ref_core, ref_version = core_and_version(reference) + got_core, got_version = core_and_version(candidate) + if not ref_core or not got_core or not version_agrees(ref_version, got_version): + return False + reference_tokens, candidate_tokens = set(ref_core), set(got_core) + if not reference_tokens.issubset(candidate_tokens): + return False + allowed_tokens: set[str] = set() + for value in allowed_context: + allowed_tokens.update(core_and_version(value)[0]) + return (candidate_tokens - reference_tokens).issubset(allowed_tokens) + + +def artist_agrees(artists: Iterable[str], candidate_text: str | None) -> bool: + haystack = norm(candidate_text) + for artist in artists: + for piece in re.split(r"[,&/]| x | and ", artist or ""): + needle = norm(piece) + if len(needle) >= 3 and needle in haystack: + return True + return False + + +def metadata_verdict( + *, + reference_titles: Iterable[str], + reference_artists: Iterable[str], + candidate_title: str, + candidate_channel: str, + candidate_duration_s: float, + source_duration_s: float | None, + mb_recording: dict | None, + reference_context: Iterable[str] = (), +) -> tuple[str | None, dict | str]: + """Apply the ISRC or metadata bar and return auditable evidence.""" + titles = [title for title in reference_titles if title] + artists = [artist for artist in reference_artists if artist] + if not any( + title_agrees(title, candidate_title, [*artists, *reference_context]) + for title in titles + ): + return None, "title_or_version_mismatch" + if not artist_agrees(artists, f"{candidate_title} {candidate_channel}"): + return None, "artist_mismatch" + + mb_length_ms = (mb_recording or {}).get("length_ms") + if ( + mb_recording + and mb_recording.get("isrc") + and mb_length_ms + and source_duration_s is not None + ): + mb_duration_s = float(mb_length_ms) / 1000.0 + if ( + abs(candidate_duration_s - source_duration_s) <= DUR_TOL_ISRC + and abs(candidate_duration_s - mb_duration_s) <= DUR_TOL_ISRC + ): + return "isrc", { + "isrc": (mb_recording or {}).get("isrc"), + "mb_recording": (mb_recording or {}).get("mbid"), + "mb_title": (mb_recording or {}).get("title"), + "mb_artist": (mb_recording or {}).get("artist"), + "dur_ref_s": round(source_duration_s, 1), + "dur_mb_s": round(mb_duration_s, 1), + "dur_got_s": round(candidate_duration_s, 1), + } + return None, "duration_outside_isrc_tolerance" + + reference_duration = source_duration_s + if reference_duration is None and mb_length_ms: + reference_duration = float(mb_length_ms) / 1000.0 + if reference_duration is None: + return None, "no_reference_duration" + if abs(candidate_duration_s - reference_duration) <= DUR_TOL_METADATA: + return "metadata", { + "matched_title": titles[0] if titles else "", + "dur_ref_s": round(reference_duration, 1), + "dur_got_s": round(candidate_duration_s, 1), + } + return None, "duration_outside_metadata_tolerance" diff --git a/pkgs/music-acquire/tests/fixtures/stage1-kirschberg.jsonl b/pkgs/music-acquire/tests/fixtures/stage1-kirschberg.jsonl new file mode 100644 index 00000000..74757fa0 --- /dev/null +++ b/pkgs/music-acquire/tests/fixtures/stage1-kirschberg.jsonl @@ -0,0 +1 @@ +{"key":"fixture-kirschberg","sc_id":"101000650","query":"Liftboi - Kirschberg (Original)","artist":"Liftboi","title":"Kirschberg (Original)","url":"https://soundcloud.com/liftboi/kirschberg","appearances":[{"source":"fixture","set":"stage1","segment":"main","position":1}]} diff --git a/pkgs/music-acquire/tests/fixtures/stage2-corrida.jsonl b/pkgs/music-acquire/tests/fixtures/stage2-corrida.jsonl new file mode 100644 index 00000000..f85dc700 --- /dev/null +++ b/pkgs/music-acquire/tests/fixtures/stage2-corrida.jsonl @@ -0,0 +1 @@ +{"key":"fixture-corrida","sc_id":"1010003194","query":"SCH - Corrida","artist":"SCH","title":"Corrida","url":"https://soundcloud.com/sch-official/corrida","appearances":[{"source":"fixture","set":"stage2","segment":"main","position":1}]} diff --git a/pkgs/music-acquire/tests/fixtures/stage2-isrc-pommade.jsonl b/pkgs/music-acquire/tests/fixtures/stage2-isrc-pommade.jsonl new file mode 100644 index 00000000..c8173cf7 --- /dev/null +++ b/pkgs/music-acquire/tests/fixtures/stage2-isrc-pommade.jsonl @@ -0,0 +1 @@ +{"key":"fixture-pommade","sc_id":"1016015629","query":"Lomepal - Pommade","artist":"Lomepal","title":"Pommade","url":"https://soundcloud.com/lomepal/pommade","appearances":[{"source":"fixture","set":"stage2-isrc","segment":"main","position":1}]} diff --git a/pkgs/music-acquire/tests/test_music_acquire.py b/pkgs/music-acquire/tests/test_music_acquire.py new file mode 100644 index 00000000..690060b7 --- /dev/null +++ b/pkgs/music-acquire/tests/test_music_acquire.py @@ -0,0 +1,589 @@ +from __future__ import annotations + +import collections +import json +import os +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from music_acquire.backend import LiveBackend +from music_acquire.cli import Acquirer, main, tracklist_items +from music_acquire.state import BatchState +from music_acquire.verification import core_and_version, title_agrees, version_agrees + + +def track(item_id: str) -> dict: + return { + "id": item_id, + "title": "Fixture", + "url": f"https://soundcloud.com/fixture/{item_id}", + "full_duration_ms": 170_000, + "has_preview": True, + } + + +class FakeBackend: + def __init__(self, scenarios: dict[str, dict], duplicates: dict[str, dict] | None = None): + self.scenarios = scenarios + self.duplicates = duplicates or {} + self.calls = collections.Counter() + self.remembered = [] + + def _scenario(self, item): + return self.scenarios[str(item["id"])] + + def duplicate(self, item, sc_track=None): + self.calls["duplicate"] += 1 + return self.duplicates.get(str(item["id"])) + + def resolve_soundcloud(self, item): + self.calls["resolve_soundcloud"] += 1 + return self._scenario(item).get( + "resolve", {"status": "found", "track": track(str(item["id"]))} + ) + + def download_soundcloud(self, item, sc_track): + self.calls["download_soundcloud"] += 1 + return self._scenario(item).get("soundcloud", {"status": "drm"}) + + def verify_youtube(self, item, sc_track): + self.calls["verify_youtube"] += 1 + return self._scenario(item).get( + "youtube", {"status": "unverified", "reason": "unverified"} + ) + + def retry_youtube_download(self, item, previous): + self.calls["retry_youtube_download"] += 1 + return self._scenario(item)["youtube_retry"] + + def capture(self, item, sc_track): + self.calls["capture"] += 1 + return self._scenario(item).get( + "capture", {"status": "unavailable", "reason": "capture_unavailable"} + ) + + def download_bandcamp(self, item): + self.calls["download_bandcamp"] += 1 + return self._scenario(item)["bandcamp"] + + def download_ytmusic(self, item): + self.calls["download_ytmusic"] += 1 + return self._scenario(item)["ytmusic"] + + def remember(self, item, row): + self.remembered.append((item["id"], row["disposition"])) + + +class BatchTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + + def tearDown(self): + self.temp.cleanup() + + def state(self, batch="fixtures"): + state = BatchState(self.root / "state", batch) + state.merge_worklist([]) + return state + + @staticmethod + def item(item_id, **values): + return { + "id": item_id, + "query": values.pop("query", item_id), + "source": values.pop("source", "tracklist:fixture"), + "appearances": values.pop("appearances", []), + **values, + } + + def run_one(self, item, scenario, duplicates=None): + state = self.state() + state.merge_worklist([item]) + backend = FakeBackend({item["id"]: scenario}, duplicates) + Acquirer(state, backend, backoff_seconds=0).run(state.worklist(), "tracklist") + return state, backend, state.final_rows()[item["id"]] + + def test_stage_one_stops_before_youtube(self): + item = self.item("liftboi-kirschberg") + state, backend, row = self.run_one( + item, + { + "resolve": { + "status": "found", + "track": { + **track("liftboi-kirschberg"), + "id": "289635627", + "full_duration_ms": 342_000, + }, + }, + "soundcloud": {"status": "ok", "path": "/music/kirschberg.mp3"}, + }, + ) + self.assertEqual(row["disposition"], "ok_soundcloud") + self.assertEqual(row["evidence"]["duration_s"], 342.0) + self.assertEqual(backend.calls["verify_youtube"], 0) + self.assertEqual(backend.calls["capture"], 0) + self.assertIn(item["id"], state.archived()) + + def test_recorded_acoustic_fixture(self): + item = self.item("1010003194") + _, _, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": { + "status": "ok", + "path": "/music/1010003194.opus", + "yt_id": "JR7DCavxVVM", + "verdict": "acoustic", + "evidence": { + "ber": 0.0381, + "dur_ref_s": 170.1, + "dur_got_s": 171.0, + }, + }, + }, + ) + self.assertEqual(row["disposition"], "ok_youtube") + self.assertEqual(row["yt_id"], "JR7DCavxVVM") + self.assertEqual(row["evidence"]["ber"], 0.0381) + + def test_recorded_isrc_fixture(self): + item = self.item("1016015629") + _, _, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": { + "status": "ok", + "path": "/music/1016015629.webm", + "yt_id": "a9rZFirpbyQ", + "verdict": "isrc", + "evidence": { + "isrc": "FR9W11708727", + "mb_recording": "7ddc6397-2247-48f2-b419-9d0a9969620e", + "dur_ref_s": 208.6, + "dur_mb_s": 208.0, + "dur_got_s": 209.0, + }, + }, + }, + ) + self.assertEqual(row["disposition"], "ok_youtube") + self.assertEqual(row["verdict"], "isrc") + + def test_recorded_capture_fixture(self): + item = self.item("1028566936") + _, backend, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": {"status": "unverified", "reason": "unverified"}, + "capture": { + "status": "ok", + "path": "/music/1028566936.flac", + "evidence": { + "captured_s": 137.5, + "ber_vs_preview": 0.0174, + "mean_volume_db": -10.6, + "artwork": "embedded", + }, + }, + }, + ) + self.assertEqual(row["disposition"], "ok_capture") + self.assertTrue(row["path"].endswith(".flac")) + self.assertEqual(backend.calls["capture"], 1) + + def test_wrong_version_stays_rejected(self): + item = self.item("252618029") + _, _, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": {"status": "unverified", "reason": "unverified"}, + "capture": { + "status": "rejected", + "reason": "capture_verification_failed", + "evidence": {"ber_attempts": [0.281, 0.285, 0.304]}, + }, + }, + ) + self.assertEqual(row["disposition"], "fallthrough") + self.assertEqual(row["reason"], "capture_verification_failed") + + def test_deleted_upstream_is_gone_without_retry_loop(self): + item = self.item("25858463") + state, backend, row = self.run_one( + item, + {"resolve": {"status": "gone", "reason": "track_gone_upstream"}}, + ) + self.assertEqual(row["disposition"], "gone") + self.assertIn(item["id"], state.archived()) + self.assertEqual(backend.calls["download_soundcloud"], 0) + self.assertEqual(backend.calls["verify_youtube"], 0) + + def test_verified_download_auth_failure_is_retryable_and_unarchived(self): + item = self.item("814429207") + state, _, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": { + "status": "retryable", + "reason": "verified_but_download_failed", + "detail": "use --cookies for authentication", + "yt_id": "M3sYTPAlHvA", + "verdict": "metadata", + }, + }, + ) + self.assertEqual(row["disposition"], "retryable") + self.assertNotIn(item["id"], state.archived()) + + def test_verified_download_retry_reuses_recorded_video_and_evidence(self): + item = self.item("814429207") + state = self.state() + state.merge_worklist([item]) + state.record( + item, + "retryable", + reason="verified_but_download_failed", + stage="youtube", + sc_id="814429207", + yt_id="M3sYTPAlHvA", + verdict="metadata", + evidence={"dur_ref_s": 256.1, "dur_got_s": 256.0}, + ) + backend = FakeBackend( + { + item["id"]: { + "youtube_retry": { + "status": "ok", + "path": "/music/814429207.opus", + "yt_id": "M3sYTPAlHvA", + "verdict": "metadata", + "evidence": {"dur_ref_s": 256.1, "dur_got_s": 256.0}, + } + } + } + ) + Acquirer(state, backend, backoff_seconds=0).run(state.worklist(), "tracklist") + row = state.final_rows()[item["id"]] + self.assertEqual(row["disposition"], "ok_youtube") + self.assertEqual(backend.calls["retry_youtube_download"], 1) + self.assertEqual(backend.calls["resolve_soundcloud"], 0) + self.assertEqual(backend.calls["verify_youtube"], 0) + + def test_capture_retry_skips_direct_and_youtube_stages(self): + item = self.item("capture-retry") + state = self.state() + state.merge_worklist([item]) + state.record( + item, + "retryable", + reason="capture_host_busy", + stage="capture", + stage2_reason="unverified", + sc_id="capture-retry", + ) + backend = FakeBackend( + { + item["id"]: { + "resolve": {"status": "found", "track": track(item["id"])}, + "capture": { + "status": "ok", + "path": "/music/capture-retry.flac", + "evidence": {"ber_vs_preview": 0.02}, + }, + } + } + ) + Acquirer(state, backend, backoff_seconds=0).run(state.worklist(), "tracklist") + self.assertEqual( + state.final_rows()[item["id"]]["disposition"], "ok_capture" + ) + self.assertEqual(backend.calls["download_soundcloud"], 0) + self.assertEqual(backend.calls["verify_youtube"], 0) + self.assertEqual(backend.calls["capture"], 1) + + def test_duplicate_suppression_performs_no_network_call(self): + item = self.item("101000650") + state, backend, row = self.run_one( + item, + {}, + duplicates={item["id"]: {"path": "/held/101000650.mp3"}}, + ) + self.assertEqual(row["disposition"], "skipped_duplicate") + self.assertEqual(backend.calls["resolve_soundcloud"], 0) + self.assertIn(item["id"], state.archived()) + + def test_no_capture_is_first_class_and_can_be_reopened(self): + item = self.item("needs-capture") + state, _, row = self.run_one( + item, + { + "soundcloud": {"status": "drm"}, + "youtube": {"status": "unverified", "reason": "unverified"}, + "capture": {"status": "unavailable", "reason": "capture_unavailable"}, + }, + ) + self.assertEqual(row["disposition"], "fallthrough") + self.assertIn(item["id"], state.archived()) + self.assertEqual(state.reopen_capture_unavailable(), [item["id"]]) + self.assertNotIn(item["id"], state.archived()) + + def test_rerun_of_completed_batch_has_zero_network_calls(self): + item = self.item("done") + state, _, _ = self.run_one( + item, + {"soundcloud": {"status": "ok", "path": "/music/done.mp3"}}, + ) + backend = FakeBackend({item["id"]: {}}) + attempted = Acquirer(state, backend, backoff_seconds=0).run( + state.worklist(), "tracklist" + ) + self.assertEqual(attempted, 0) + self.assertEqual(sum(backend.calls.values()), 0) + + def test_completed_cli_rerun_does_not_prepare_network_backend(self): + worklist = self.root / "completed.jsonl" + worklist.write_text( + json.dumps( + {"key": "done", "query": "Artist - Done", "appearances": []} + ) + + "\n" + ) + state_root = self.root / "cli-complete" + state = BatchState(state_root, "completed") + items = tracklist_items(worklist) + state.merge_worklist(items) + state.save_request( + { + "verb": "tracklist", + "input": str(worklist.resolve()), + "source": str(worklist.resolve()), + "source_filter": None, + "batch": "completed", + "out": str(self.root / "out"), + } + ) + state.record(items[0], "ok_youtube", path="/music/done.opus") + environment = { + "MUSIC_ACQUIRE_STATE_ROOT": str(state_root), + "MUSIC_CONSOLIDATION_REPO": str(self.root / "campaign"), + } + with mock.patch.object( + LiveBackend, "prepare", side_effect=AssertionError("network prepare called") + ): + self.assertEqual( + main( + ["tracklist", str(worklist), "--batch", "completed"], + environ=environment, + ), + 0, + ) + + def test_last_write_wins_but_retryable_never_enters_archive(self): + item = self.item("retry") + state = self.state() + state.merge_worklist([item]) + state.record(item, "retryable", reason="throttle") + self.assertNotIn(item["id"], state.archived()) + state.record(item, "ok_youtube", path="/music/retry.opus") + self.assertEqual(state.final_rows()[item["id"]]["disposition"], "ok_youtube") + self.assertIn(item["id"], state.archived()) + + def test_tracklist_merges_cross_source_and_repeat_appearances(self): + worklist = self.root / "worklist.jsonl" + rows = [ + { + "key": "dire straits six blade knife", + "query": "Dire Straits - Six Blade Knife (THE ODDNESS Re-work)", + "artist": "Dire Straits", + "title": "Six Blade Knife (THE ODDNESS Re-work)", + "appearances": [ + { + "source": "goldcast", + "set": "027", + "segment": "guestmix:The Oddness", + "position": 10, + } + ], + }, + { + "key": "dire straits six blade knife", + "query": "Dire Straits - Six Blade Knife (THE ODDNESS Re-work)", + "artist": "Dire Straits", + "title": "Six Blade Knife (THE ODDNESS Re-work)", + "appearances": [ + { + "source": "vent-2024", + "set": "2024", + "segment": "main", + "position": 95, + } + ], + }, + { + "key": "chambord wonderland", + "query": "Chambord - Wonderland (Maga Remix)", + "artist": "Chambord", + "title": "Wonderland (Maga Remix)", + "appearances": [ + {"source": "goldcast", "set": str(i), "segment": "main", "position": 2} + for i in range(4) + ], + }, + ] + worklist.write_text("".join(json.dumps(row) + "\n" for row in rows)) + items = tracklist_items(worklist) + self.assertEqual(len(items), 2) + self.assertEqual(len(items[0]["appearances"]), 2) + self.assertEqual(len(items[1]["appearances"]), 4) + + def test_status_json_has_totals_dispositions_and_estimate(self): + worklist = self.root / "worklist.jsonl" + worklist.write_text( + json.dumps({"key": "one", "query": "Artist - One", "appearances": []}) + + "\n" + ) + environment = { + "MUSIC_ACQUIRE_STATE_ROOT": str(self.root / "cli-state"), + "MUSIC_CONSOLIDATION_REPO": str(self.root / "campaign"), + } + self.assertEqual( + main( + ["tracklist", str(worklist), "--dry-run", "--out", str(self.root / "out")], + environ=environment, + ), + 0, + ) + self.assertEqual(main(["status", "--batch", "worklist", "--json"], environ=environment), 0) + + +class VerificationTest(unittest.TestCase): + def test_version_markers_are_sets_not_similarity(self): + original = core_and_version("Lucy") + instrumental = core_and_version("Lucy (Instrumentale)") + remix = core_and_version("Lucy (Kaytranada Remix)") + self.assertFalse(version_agrees(original[1], instrumental[1])) + self.assertFalse(version_agrees(original[1], remix[1])) + self.assertTrue(title_agrees("Lucy", "Lucy (Original Mix)")) + self.assertFalse(title_agrees("Chapter One", "Chapter One Part Two")) + self.assertTrue(title_agrees("Corrida", "SCH - Corrida", ["SCH"])) + + +class NativeDownloadTest(unittest.TestCase): + def test_final_download_command_never_transcodes_or_overwrites(self): + with tempfile.TemporaryDirectory() as temp_name: + root = Path(temp_name) + state = BatchState(root / "state", "native") + state.merge_worklist([]) + backend = LiveBackend( + state, + root / "out", + root / "campaign", + no_capture=True, + environ={"MUSIC_ACQUIRE_STAGING": str(root / "staging")}, + ) + commands = [] + + def fake_run(command, **_kwargs): + commands.append(command) + template = Path(command[command.index("-o") + 1]) + output = Path(str(template).replace("%(ext)s", "webm")) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(b"native codec bytes") + return subprocess.CompletedProcess(command, 0, "", "") + + backend._run = fake_run + result = backend._download_native( + url="https://youtube.example/fixture", + item_id="fixture", + cookies=None, + format_selector="bestaudio/best", + ) + self.assertEqual(result["status"], "ok") + command = commands[0] + self.assertNotIn("-x", command) + self.assertNotIn("--extract-audio", command) + self.assertNotIn("--audio-format", command) + self.assertNotIn("--recode-video", command) + self.assertIn("--no-overwrites", command) + + def test_official_long_form_tracklist_is_resolved_and_stream_copied(self): + description = """Tracklist: +01 - Visage [00:00] +02 - In A Search Of Touch [01:38] +07 - My Personality Shaped By Curves & Angles [24:09] +""" + segments = LiveBackend._description_segments(description, 1_863_053) + self.assertEqual(segments[-1]["start_s"], 1_449.0) + self.assertAlmostEqual(segments[-1]["end_s"], 1_863.053) + + with tempfile.TemporaryDirectory() as temp_name: + root = Path(temp_name) + state = BatchState(root / "state", "segment") + state.merge_worklist([]) + backend = LiveBackend( + state, + root / "out", + root / "campaign", + no_capture=True, + environ={"MUSIC_ACQUIRE_STAGING": str(root / "staging")}, + ) + backend._soundcloud_get = lambda *_args, **_kwargs: { + "collection": [ + { + "id": 738396898, + "title": "HAKIMONU - Music For Children [Full Album]", + "description": description, + "duration": 1_863_053, + "full_duration": 1_863_053, + "permalink_url": "https://soundcloud.example/full-album", + "user": {"username": "HAKIMONU", "permalink": "hakimonu"}, + } + ] + } + item = { + "id": "hakimonu my personality shaped by curves angles", + "query": "Hakimonu - My Personality Shaped by Curves & Angles", + "artist": "Hakimonu", + "title": "My Personality Shaped by Curves & Angles", + "album": None, + "source": "tracklist:goldcast", + } + resolved = backend.resolve_soundcloud(item) + self.assertEqual(resolved["status"], "found") + self.assertEqual(resolved["track"]["segment"]["start_s"], 1_449.0) + self.assertIsNone(backend.duplicate(item, resolved["track"])) + + parent = backend.work / "sc-parent-738396898.m4a" + parent.write_bytes(b"native parent") + commands = [] + + def fake_run(command, **_kwargs): + commands.append(command) + Path(command[-1]).write_bytes(b"native segment") + return subprocess.CompletedProcess(command, 0, "", "") + + backend._run = fake_run + backend.sc_cookie.write_text("cookie") + outcome = backend.download_soundcloud(item, resolved["track"]) + self.assertEqual(outcome["status"], "ok") + self.assertEqual(outcome["segment"]["parent_sc_id"], "738396898") + command = commands[0] + self.assertEqual(Path(command[-1]).parent, backend.out) + self.assertIn(".segment.part", Path(command[-1]).name) + self.assertIn("copy", command) + self.assertNotIn("libopus", command) + self.assertNotIn("aac", command) + + +if __name__ == "__main__": + unittest.main() From 4af2255bcdad09078203de563b2249fe9712c75b Mon Sep 17 00:00:00 2001 From: mecattaf Date: Thu, 13 Aug 2026 14:51:30 +0200 Subject: [PATCH 2/3] tally: deploy 52eff4db and retire git-ai --- flake.lock | 53 +++----------------------------------------------- flake.nix | 13 ------------- home/home.nix | 3 --- home/tally.nix | 11 ----------- 4 files changed, 3 insertions(+), 77 deletions(-) diff --git a/flake.lock b/flake.lock index 63576a35..a7f605d7 100644 --- a/flake.lock +++ b/flake.lock @@ -467,31 +467,6 @@ "type": "github" } }, - "git-ai": { - "inputs": { - "flake-utils": [ - "tally", - "flake-utils" - ], - "nixpkgs": [ - "nixpkgs" - ], - "rust-overlay": "rust-overlay" - }, - "locked": { - "lastModified": 1785028852, - "narHash": "sha256-qbc3xEHIG2SSDU66Bc7xqHJw6JailMGVDBs9i41gP9w=", - "owner": "git-ai-project", - "repo": "git-ai", - "rev": "e78d10d8e2139cb0981d707a5f9f247d4cd8512e", - "type": "github" - }, - "original": { - "owner": "git-ai-project", - "repo": "git-ai", - "type": "github" - } - }, "home-manager": { "inputs": { "nixpkgs": [ @@ -1008,7 +983,6 @@ "apple-fonts": "apple-fonts", "deploy-rs": "deploy-rs", "disko": "disko", - "git-ai": "git-ai", "home-manager": "home-manager", "llm-agents": "llm-agents", "microvm": "microvm", @@ -1027,27 +1001,6 @@ "zmx": "zmx" } }, - "rust-overlay": { - "inputs": { - "nixpkgs": [ - "git-ai", - "nixpkgs" - ] - }, - "locked": { - "lastModified": 1770174315, - "narHash": "sha256-GUaMxDmJB1UULsIYpHtfblskVC6zymAaQ/Zqfo+13jc=", - "owner": "oxalica", - "repo": "rust-overlay", - "rev": "095c394bb91342882f27f6c73f64064fb9de9f2a", - "type": "github" - }, - "original": { - "owner": "oxalica", - "repo": "rust-overlay", - "type": "github" - } - }, "sf-arabic": { "flake": false, "locked": { @@ -1283,11 +1236,11 @@ ] }, "locked": { - "lastModified": 1786552755, - "narHash": "sha256-Zc7pohM+S0m2HjEo2X7ZKMBjxdQWbs9xmJhyDjr75Tk=", + "lastModified": 1786622428, + "narHash": "sha256-9NVSU3JuYBJ+mi9TXZQVTLFg5BfdxdHlVYzqj+RFKro=", "owner": "mecattaf", "repo": "tally.nix", - "rev": "78dd4871c97b72fa87c9bc12083392780675c8d1", + "rev": "52eff4db28b2b3f06c096715d1a9c1de8559ed92", "type": "github" }, "original": { diff --git a/flake.nix b/flake.nix index b242dadc..2ccf1474 100644 --- a/flake.nix +++ b/flake.nix @@ -90,19 +90,6 @@ # while attached, and a persistent session survives disconnects server-side. zmx.url = "github:neurosnap/zmx"; - # git-ai — AI-authorship tracking CLI (github.com/git-ai-project/git-ai). - # Consume its flake package directly and pin it in flake.lock. The Home - # Manager profile installs upstream's `minimal` output, which provides - # `git-ai` and `git-og` without replacing the `git` binary already owned - # by programs.git. Following nixpkgs keeps the Rust build on our one package - # pin instead of adding another nixpkgs universe to the lock. - # `nix flake update git-ai` bumps to the latest pushed commit. - git-ai = { - url = "github:git-ai-project/git-ai"; - inputs.nixpkgs.follows = "nixpkgs"; - inputs.flake-utils.follows = "tally/flake-utils"; - }; - # llm-agents.nix — numtide's daily-rebuilt catalog of ~100 AI coding agents # and tooling (claude-code, codex, gemini-cli, opencode, crush, goose, amp, # ...). Its `overlays.default` exposes the whole set, prebuilt against its OWN diff --git a/home/home.nix b/home/home.nix index 38136291..92b72f6d 100644 --- a/home/home.nix +++ b/home/home.nix @@ -460,9 +460,6 @@ in # seed via modules/secrets.nix, and DISABLE_UPDATES=1 keeps the native # updater from clobbering ~/.local/bin. llmAgentsSelected - # Upstream's minimal flake output: git-ai + git-og, while programs.git below - # remains the sole provider of the real git binary. - inputs.git-ai.packages.${pkgs.stdenv.hostPlatform.system}.minimal huggingface-cli # metadata CLI; agenix authentication is coordinator-only gh google-cloud-sdk diff --git a/home/tally.nix b/home/tally.nix index 3604f289..15eff205 100644 --- a/home/tally.nix +++ b/home/tally.nix @@ -155,17 +155,6 @@ in }; }; - # E5 (dotfiles#138): bind code-result revisions to git-ai authorship - # notes, advisory posture first — an unprovisioned host and a squash that - # lost its attribution produce identical evidence, so advisory is how the - # binding proves itself before anything is allowed to fail on it. git-ai - # 1.6.17 is externally provisioned (verified on the estate 2026-08-03); - # tally.nix does not ship the binary. - gitAi = { - enable = true; - mode = "advisory"; - }; - # One low-priority durable row replaces the old 02:00/03:30/04:30/06:00 chain. # It holds the build and coordinator GPU lanes end-to-end, making the measured # single-node build plus activation one exclusive maintenance window. From c4124369044550e38a0ff8ecd42199f8cfee8c7a Mon Sep 17 00:00:00 2001 From: mecattaf Date: Thu, 13 Aug 2026 14:57:30 +0200 Subject: [PATCH 3/3] tally: guard deploys on campaign quiescence --- home/tally.nix | 8 ++++++++ hosts/coordinator/fleet-deploy.nix | 4 ++++ 2 files changed, 12 insertions(+) diff --git a/home/tally.nix b/home/tally.nix index 15eff205..e7f1600b 100644 --- a/home/tally.nix +++ b/home/tally.nix @@ -251,4 +251,12 @@ in }; }; + # Freeze the nightly deploy at its admission boundary while any campaign is + # registered. This replaces the dated, hand-maintained ExecCondition + # drop-ins used during the silent-factory ladder. + systemd.user.services = lib.optionalAttrs isCoordinator { + "tally-producer-nightly-fleet-deploy".Service.ExecCondition = + "${tallyPackage}/bin/tally --config /home/tom/.config/tally/config.json campaign quiescent --state-dir /home/tom/.local/state/tally"; + }; + } diff --git a/hosts/coordinator/fleet-deploy.nix b/hosts/coordinator/fleet-deploy.nix index f30f7941..5d78e15c 100644 --- a/hosts/coordinator/fleet-deploy.nix +++ b/hosts/coordinator/fleet-deploy.nix @@ -14,6 +14,7 @@ let system = pkgs.stdenv.hostPlatform.system; deployPackage = inputs.deploy-rs.packages.${system}.deploy-rs; + tallyPackage = inputs.tally.packages.${system}.tally; failureMarker = "/var/lib/fleet-deploy/fleet-deploy.service.fail"; rollingResolution = lib.concatMapStringsSep "\n" (input: '' resolved="$(resolve_flake ${lib.escapeShellArg input.url})" @@ -192,6 +193,9 @@ in Type = "oneshot"; User = "tom"; Group = "users"; + # Re-check at activation time as well as producer admission so a queued + # or manually started deploy cannot move the pin under an armed campaign. + ExecCondition = "${tallyPackage}/bin/tally --config /home/tom/.config/tally/config.json campaign quiescent --state-dir /home/tom/.local/state/tally"; ExecStart = "${fleetDeploy}/bin/fleet-deploy"; StateDirectory = "fleet-deploy"; StateDirectoryMode = "0755";