feat(tide-table): sc_generation_lens — which cut is actually on the account

PLN opened the day hearing it: 'our private cuts of OPAL are badly cut
begin/end unaligned'. The board had one theory (a 24s truncation) and it was
wrong. This lens found the real shape: the ENTIRE Aug-12 first cut is still
on SoundCloud — all 15 tracks — living beside the 6 correct v4 ones under a
litter of ad-hoc permalinks (/opal26-wap, /12-mafia-12, /ceci-nest-pas-une-
bombe), and eight of the stale ones are the ones wearing artwork. So the
wrong cut is the one that looks finished. Of course he heard it.

Duration is the discriminator, and it is far sharper than assumed: measured
on this account, SoundCloud's full_duration reproduces the source to the
millisecond (477.600 uploaded, 477.600 reported). Tolerance is 50ms, not the
seconds you would allow for a transcode — a loose tolerance here buys no
safety, it only invents ambiguity between cuts that differ plainly.

Three things I got wrong first, each left in as a guard:
  - filenames are not authoritative. 'Perfect <3' uploads to /perfect-3, so a
    filename-derived slug reported two live tracks as NOT UPLOADED. Titles
    come from the plan now, which is what the uploader actually used.
  - streaming and club masters share every duration, so offering both as
    rival candidates marked all 14 AMBIGUOUS. Identical generations collapse.
  - a duration collision is not provenance. /au-revoir-lord-toyota (2020) and
    /saria-demo (2025) both matched OPAL to 50ms. Anything uploaded before
    the oldest render on disk cannot be a cut of it, and a coincidence must
    never reach a delete manifest — so the date floor is a hard gate and the
    discards are printed, not silently dropped.

The second pass is the one that earned its keep. Roll call only looks where
it expects a track to be, so it can never see a stale cut hiding under a
different slug. Strays scans the whole account. Absence needs proof.

    fifteen ghosts of a set played once,
    each wearing the cover art of the real thing --
    the lens does not listen, it only counts seconds,
    and seconds were enough.

armada/tide-table/sc_generation_lens.py

+#!/usr/bin/env python3
+"""sc_generation_lens — which render generation is actually on SoundCloud?
+
+## Why this exists
+
+A live set gets re-cut. OPAL-26 was cut three times: an Aug-12 first pass, a
+v3 on measured boundaries, and the ear-verified v4 PLN signed off. Every pass
+writes the same fourteen filenames into a different directory and renders the
+same *titles* — so nothing in a filename, a permalink or a title tells you
+which cut a listener is hearing.
+
+SoundCloud will not replace the audio behind a permalink (that is a paid
+feature), and our uploader is idempotent by permalink, so a stale upload is
+invisible and permanent until someone deletes it. The failure is silent in the
+worst way: the track plays, it is the right song, it is simply cut wrong.
+
+PLN heard it before any tool did — *"they are badly cut begin/end unaligned"*.
+This is the lens that turns that into a per-track verdict.
+
+## What it does
+
+Duration is the discriminator, for the same reason `check_permalinks.py` uses
+it. Measured on this account: SoundCloud's `full_duration` reproduces the
+source to the **millisecond** (477.600 s uploaded, 477.600 s reported), so the
+tolerance is 50 ms rather than the seconds you would allow for a transcode.
+A loose tolerance here does not buy safety — it invents ambiguity between
+generations that are genuinely distinguishable.
+
+Two passes, and the second is the one that found the truth:
+
+1. **Roll call** — for each track the newest generation says should be up,
+   which generation does the upload at that permalink actually match?
+
+       OK             only the newest fits
+       OK?            the newest fits, but so does an older cut — duration
+                      cannot prove which one is up
+       STALE:<gen>    only an older generation fits: re-upload this
+       ORPHAN         nothing on disk fits
+       NOT UPLOADED   no track at that permalink
+
+2. **Strays** — scan *every* track on the account for durations matching an
+   older generation of this set under some *other* permalink. A re-upload
+   under a fresh slug does not replace the old one, it sits beside it; and the
+   old one may be the one wearing the artwork, so it looks more finished than
+   the correct cut. Roll call alone cannot see these, because it only ever
+   looks where it expects a track to be (`feedback_absence_needs_proof`).
+
+Generations whose durations are identical on every track (a streaming and a
+club master of the same cut) are collapsed into one entry — they are one set
+of boundaries, and reporting them as rival candidates is noise.
+
+## Usage
+
+    ./sc_generation_lens.py <sc_catalog.json> --master <dir> [--plan <plan.json>]
+    ./sc_generation_lens.py <sc_catalog.json> --gen v4=<dir> --gen v3=<dir>
+
+`--master` auto-discovers sibling `tracks*` directories, oldest first by mtime.
+Pass `--plan` whenever one exists: the plan holds the titles the uploader
+actually used, and filenames are not authoritative — a title of `Perfect <3`
+uploads to `/perfect-3`, which a filename-derived slug will never find.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+import unicodedata
+from pathlib import Path
+
+AUDIO_EXT = {".flac", ".wav", ".mp3", ".m4a", ".aiff", ".ogg"}
+
+
+def _iso(mtime: float) -> str:
+    """The date floor a stray must clear, as YYYY-MM-DD."""
+    import datetime as _dt
+    return _dt.date.fromtimestamp(mtime).isoformat()
+
+
+def slugify(title: str) -> str:
+    """The uploader's permalink rule, kept identical on purpose.
+
+    `scform.permalink_of` is the authority; this mirrors it so the lens runs
+    without the tidal-ears venv. If the two drift apart the lens reports a
+    spurious miss, so keep them in step.
+    """
+    s = unicodedata.normalize("NFKD", title)
+    s = s.encode("ascii", "ignore").decode("ascii").lower()
+    s = re.sub(r"[^a-z0-9]+", "-", s)
+    return s.strip("-")
+
+
+def duration_of(path: Path) -> float | None:
+    try:
+        out = subprocess.run(
+            ["ffprobe", "-v", "0", "-show_entries", "format=duration",
+             "-of", "csv=p=0", str(path)],
+            capture_output=True, text=True, timeout=60,
+        ).stdout.strip()
+        return float(out) if out else None
+    except (ValueError, subprocess.SubprocessError):
+        return None
+
+
+def track_number(name: str) -> int | None:
+    m = re.match(r"\s*(\d{1,2})\s*[-_. ]", name)
+    return int(m.group(1)) if m else None
+
+
+def index_generation(d: Path) -> dict[int, dict]:
+    """One render directory: track number -> title, duration, filename.
+
+    Keyed by the leading track number, which survives a retitle; the title
+    rides along for the report and as a slug fallback when there is no plan.
+    """
+    out: dict[int, dict] = {}
+    for f in sorted(d.iterdir()):
+        if f.suffix.lower() not in AUDIO_EXT or not f.is_file():
+            continue
+        n = track_number(f.name)
+        if n is None:
+            continue
+        dur = duration_of(f)
+        if dur is None:
+            print(f"  ! unreadable, skipped: {f.name}", file=sys.stderr)
+            continue
+        out[n] = {
+            "title": re.sub(r"^\s*\d{1,2}\s*[-_. ]+", "", f.stem),
+            "duration": dur,
+            "file": f.name,
+        }
+    return out
+
+
+def discover(master: Path) -> list[tuple[str, Path]]:
+    """Sibling generation dirs, oldest first — the order a reader expects."""
+    cands = [
+        p for p in master.iterdir()
+        if p.is_dir() and p.name.startswith("tracks")
+        and any(f.suffix.lower() in AUDIO_EXT for f in p.iterdir() if f.is_file())
+    ]
+    cands.sort(key=lambda p: p.stat().st_mtime)
+    return [(p.name, p) for p in cands]
+
+
+def collapse(indexed: list[tuple[str, dict]], tol: float) -> list[tuple[str, dict]]:
+    """Merge generations that agree on every track — one cut, two masters.
+
+    A streaming and a club master of the same boundaries have identical
+    durations, so duration can never tell them apart and offering both as
+    candidates would mark every single track AMBIGUOUS. They are one
+    generation as far as this lens is concerned; the name records both.
+    """
+    merged: list[tuple[str, dict]] = []
+    for name, idx in indexed:
+        for i, (prev_name, prev_idx) in enumerate(merged):
+            if set(prev_idx) != set(idx):
+                continue
+            if all(abs(prev_idx[n]["duration"] - idx[n]["duration"]) <= tol
+                   for n in idx):
+                merged[i] = (f"{prev_name}≡{name.split('_')[-1]}", prev_idx)
+                break
+        else:
+            merged.append((name, idx))
+    return merged
+
+
+def plan_titles(plan_path: Path) -> dict[int, dict]:
+    """Titles and permalinks the uploader actually used, keyed by track number.
+
+    The plan is authoritative over filenames. Track number comes from the
+    audio path's leading digits; the continuous-mix entry has none and is
+    skipped, since it is not part of the per-track roll call.
+    """
+    plan = json.loads(plan_path.read_text())
+    items = plan.get("tracks") if isinstance(plan, dict) else plan
+    out: dict[int, dict] = {}
+    for it in items or []:
+        audio = it.get("audio") or it.get("_audio") or it.get("path") or ""
+        n = track_number(Path(audio).name)
+        if n is None or it.get("_is_continuous_mix"):
+            continue
+        title = it.get("title") or ""
+        out[n] = {
+            "title": title,
+            "permalink": it.get("permalink") or slugify(title),
+            "duration": it.get("_duration_s"),
+        }
+    return out
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("catalog", type=Path, help="catalog.json from `master sc pull`")
+    ap.add_argument("--master", type=Path,
+                    help="master dir holding tracks*/ generation subdirs")
+    ap.add_argument("--gen", action="append", default=[], metavar="NAME=DIR",
+                    help="explicit generation (repeatable), oldest first")
+    ap.add_argument("--plan", type=Path,
+                    help="release plan: the authoritative titles/permalinks "
+                         "(filenames are not — 'Perfect <3' uploads to /perfect-3)")
+    ap.add_argument("--newest", metavar="NAME",
+                    help="which generation is the reference (default: newest mtime)")
+    ap.add_argument("--tolerance", type=float, default=0.05,
+                    help="seconds of slack (default 0.05; SoundCloud's full_duration "
+                         "reproduces the source to the millisecond)")
+    ap.add_argument("--not-before", metavar="YYYY-MM-DD",
+                    help="ignore stray matches uploaded before this date "
+                         "(default: the oldest generation's mtime) — a track "
+                         "that predates the render cannot be a cut of it")
+    ap.add_argument("--out", type=Path, help="write the verdicts as JSON")
+    args = ap.parse_args()
+
+    gens: list[tuple[str, Path]] = []
+    for spec in args.gen:
+        name, _, path = spec.partition("=")
+        if not path:
+            ap.error(f"--gen wants NAME=DIR, got {spec!r}")
+        gens.append((name, Path(path)))
+    if args.master:
+        gens.extend(discover(args.master))
+    if not gens:
+        ap.error("give --master or at least one --gen")
+
+    print("generations, oldest first:")
+    indexed: list[tuple[str, dict]] = []
+    for name, d in gens:
+        if not d.is_dir():
+            print(f"  ! missing: {d}", file=sys.stderr)
+            continue
+        idx = index_generation(d)
+        indexed.append((name, idx))
+        print(f"  {name:24} {len(idx):2} tracks  {d}")
+    if not indexed:
+        print("no readable generations", file=sys.stderr)
+        return 2
+
+    before = len(indexed)
+    indexed = collapse(indexed, args.tolerance)
+    if len(indexed) < before:
+        print(f"  collapsed {before} -> {len(indexed)}: "
+              f"{', '.join(n for n, _ in indexed)}")
+
+    if args.newest:
+        hit = [i for i, (n, _) in enumerate(indexed) if args.newest in n]
+        if not hit:
+            ap.error(f"--newest {args.newest!r} matches none of "
+                     f"{[n for n, _ in indexed]}")
+        indexed.append(indexed.pop(hit[0]))
+    newest, newest_idx = indexed[-1]
+
+    cat = json.loads(args.catalog.read_text())
+    tracks = cat.get("tracks", [])
+    by_permalink = {t["permalink"]: t for t in tracks}
+    titles = plan_titles(args.plan) if args.plan else {}
+
+    def sc_seconds(t: dict) -> float:
+        return (t.get("full_duration") or t.get("duration") or 0) / 1000.0
+
+    print(f"\nnewest = {newest}   tolerance = ±{args.tolerance}s"
+          f"   titles from {'plan' if titles else 'filenames'}\n")
+    header = (f'{"#":>2} {"title":30} {"permalink":32} {"on SC":>8} '
+              + " ".join(f"{n[:11]:>11}" for n, _ in indexed) + "   verdict")
+    print(header)
+    print("-" * len(header))
+
+    verdicts = []
+    for n in sorted(newest_idx):
+        entry = newest_idx[n]
+        meta = titles.get(n, {})
+        title = meta.get("title") or entry["title"]
+        slug = meta.get("permalink") or slugify(title)
+        sc = by_permalink.get(slug)
+        cells = [f'{idx[n]["duration"]:11.2f}' if n in idx else " " * 11
+                 for _, idx in indexed]
+
+        if sc is None:
+            verdict, note, sc_dur = "NOT UPLOADED", "", float("nan")
+        else:
+            sc_dur = sc_seconds(sc)
+            fits = [nm for nm, idx in indexed
+                    if n in idx and abs(idx[n]["duration"] - sc_dur) <= args.tolerance]
+            if not fits:
+                verdict, note = "ORPHAN", "no generation on disk fits"
+            elif newest in fits:
+                others = [f for f in fits if f != newest]
+                verdict = "OK" if not others else "OK?"
+                note = "" if not others else "also fits " + "+".join(others)
+            else:
+                verdict = f"STALE:{fits[0]}"
+                note = f'{sc_dur - newest_idx[n]["duration"]:+.1f}s vs {newest}'
+
+        print(f'{n:>2} {title[:30]:30} {("/" + slug)[:32]:32} {sc_dur:8.2f} '
+              + " ".join(cells) + f"   {verdict}" + (f"  ({note})" if note else ""))
+        verdicts.append({
+            "n": n, "title": title, "permalink": slug,
+            "sc_id": sc.get("id") if sc else None,
+            "sc_url": sc.get("permalink_url") if sc else None,
+            "sc_public": sc.get("public") if sc else None,
+            "sc_has_artwork": bool(sc.get("artwork_url")) if sc else None,
+            "sc_created_at": sc.get("created_at") if sc else None,
+            "sc_duration_s": None if sc is None else round(sc_dur, 3),
+            "expected_duration_s": round(entry["duration"], 3),
+            "verdict": verdict, "note": note,
+        })
+
+    # ---- pass 2: strays under some other permalink -------------------------
+    # Roll call only looks where it expects a track. A re-upload under a fresh
+    # slug leaves the old cut sitting beside it, and the old one often carries
+    # the artwork — so it is the one that looks finished.
+    # A duration collision is not provenance. Two unrelated tracks on a
+    # 132-track account will occasionally share a length to 50 ms, and an
+    # upload that PREDATES the render it supposedly came from cannot be a cut
+    # of this set at all — /au-revoir-lord-toyota (2020) and /saria-demo
+    # (2025) both matched OPAL on duration alone. Anything older than the
+    # oldest generation on disk is a coincidence, and a coincidence must never
+    # reach a delete manifest.
+    claimed = {v["permalink"] for v in verdicts}
+    older = [(nm, idx) for nm, idx in indexed[:-1]]
+    floor = args.not_before or min(
+        (d.stat().st_mtime for _, d in gens if d.is_dir()), default=0)
+    floor_iso = _iso(floor) if not args.not_before else args.not_before
+    strays = []
+    coincidences = []
+    for t in tracks:
+        if t["permalink"] in claimed:
+            continue
+        created = (t.get("created_at") or "")[:10]
+        d = sc_seconds(t)
+        for nm, idx in older:
+            for n, e in idx.items():
+                if abs(e["duration"] - d) <= args.tolerance:
+                    hit = {
+                        "permalink": t["permalink"], "sc_id": t["id"],
+                        "sc_url": t.get("permalink_url"),
+                        "sc_created_at": t.get("created_at"),
+                        "sc_public": t.get("public"),
+                        "sc_has_artwork": bool(t.get("artwork_url")),
+                        "sc_duration_s": round(d, 3),
+                        "matches_generation": nm, "matches_track": n,
+                        "matches_title": e["title"],
+                    }
+                    (strays if created >= floor_iso else coincidences).append(hit)
+                    break
+            else:
+                continue
+            break
+
+    ok = [v for v in verdicts if v["verdict"] == "OK"]
+    weak = [v for v in verdicts if v["verdict"] == "OK?"]
+    bad = [v for v in verdicts if v["verdict"].startswith(("STALE", "ORPHAN"))]
+    missing = [v for v in verdicts if v["verdict"] == "NOT UPLOADED"]
+
+    print(f"\n  OK {len(ok)}  ·  OK-but-unproven {len(weak)}  ·  "
+          f"stale {len(bad)}  ·  never uploaded {len(missing)}  ·  "
+          f"strays {len(strays)}  ·  coincidences dropped {len(coincidences)}")
+
+    if bad:
+        print("\n  STALE at the right permalink — delete, then re-upload:")
+        for v in bad:
+            art = " [HAS ARTWORK]" if v["sc_has_artwork"] else ""
+            print(f'    /{v["permalink"]:34} {v["verdict"]:16} {v["note"]}{art}')
+    if strays:
+        print("\n  STRAYS — an older cut of this set, living under another permalink:")
+        for s in strays:
+            art = " [HAS ARTWORK]" if s["sc_has_artwork"] else ""
+            print(f'    /{s["permalink"]:34} #{s["matches_track"]:<2} '
+                  f'{s["matches_title"][:22]:22} = {s["matches_generation"]}'
+                  f'  {s["sc_created_at"][:10]}{art}')
+    if missing:
+        print("\n  NEVER UPLOADED:")
+        for v in missing:
+            print(f'    /{v["permalink"]:34} {v["expected_duration_s"]:8.2f}s')
+    if coincidences:
+        print(f"\n  discarded {len(coincidences)} duration coincidence(s) "
+              f"uploaded before {floor_iso} (cannot be a cut of this set):")
+        for c in coincidences:
+            print(f'    /{c["permalink"]:34} {c["sc_created_at"][:10]}  '
+                  f'= {c["sc_duration_s"]:.2f}s')
+    if weak:
+        print("\n  OK but not provable by duration alone:")
+        for v in weak:
+            print(f'    /{v["permalink"]:34} {v["note"]}')
+
+    if args.out:
+        args.out.write_text(json.dumps({
+            "catalog": str(args.catalog),
+            "pulled_at": cat.get("pulled_at"),
+            "newest_generation": newest,
+            "tolerance_s": args.tolerance,
+            "generations": [n for n, _ in indexed],
+            "titles_from": "plan" if titles else "filenames",
+            "stray_floor_date": floor_iso,
+            "verdicts": verdicts,
+            "strays": strays,
+            "coincidences_discarded": coincidences,
+        }, indent=2, ensure_ascii=False) + "\n")
+        print(f"\n  -> {args.out}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())