Commit bd3a680f by PLN (Algolia)

feat(tide-table): build_delete_manifest — the verdict becomes a delete allowlist

`sc_generation_lens` already knows which OPAL-26 uploads are the Aug-12 first
cut; nothing turned that verdict into something the new `sc delete --manifest`
could execute. This does, and it is the last checkpoint before a destructive
call, so it refuses rather than warns.

In: verdicts starting STALE/ORPHAN (right permalink, wrong cut) plus every
stray (an older cut under another slug). Out, and asserted out: verdict OK
(that IS the ear-verified cut), NOT UPLOADED (nothing to delete), anything
without an sc_id, and everything in `coincidences_discarded` — the lens finds
strays by duration and on a 132-track account two unrelated tracks will share
a length to 50 ms, which is how /au-revoir-lord-toyota (2020, and PUBLIC) and
/saria-demo (2025) both matched an OPAL render. The date floor is re-applied
here instead of trusting the file's own bucket names: two chances to catch a
mistake whose cost is permanent.

Every entry carries a reason citing the evidence, the artwork the pull saved,
and the local render the upload came from — all fourteen resolve to a FLAC in
Opal26_master/tracks/, so deleting these loses a URL and not a recording.

One bug found while writing it, and it is the reason the reason strings matter:
the first version lined a stale upload up against the newest cut BY TRACK
NUMBER, and the Aug-12 cut had fifteen tracks where v4 has fourteen. Dropping
Desire renumbered REVOLUTION from 15 to 14, so the manifest claimed
"#14 Desire, 388.1s vs v4 205.6s" — 205.6s is REVOLUTION's length. A
measurement is a comparison; an unnamed reference cites the wrong track. The
lookup is keyed on the title slug now, and a track absent from the newest cut
says so instead of borrowing a neighbour's duration.

Dry-run against the live account: 14/14 resolved, both directions, all private,
nothing refused. Nothing deleted — that is PLN's call.
parent ec9c7d4b
#!/usr/bin/env python3
"""build_delete_manifest — turn a generation verdict into a delete allowlist.
## Why a manifest and not a pattern
SoundCloud will not replace the audio behind an existing permalink — that is a
paid feature — and our uploader is idempotent by permalink, so it SKIPS a
permalink that already exists rather than replacing it. A stale upload is
therefore permanent until somebody deletes it, and the whole Aug-12 first cut
of OPAL-26 is currently sitting on the account beside the six correct v4
uploads under ad-hoc slugs.
Those slugs are the reason this is a list and not a rule. The stale first cut
lives at `/opal26-wap`, `/desire`, `/12-mafia-12`; the good v4 cut lives at
`/wap` and `/mafia`. No pattern permissive enough to reach the first set fails
to reach the second. So `tidal-ears master sc delete` takes an explicit
manifest naming every track by **permalink AND sc_id**, verifies both against
the live account, and refuses on any disagreement. This script writes that
manifest, and it is the only thing standing between `sc_generation_lens`'s
verdict and a destructive call — so it fails loudly rather than emitting a
questionable entry.
## What gets in, and what can never
IN: verdicts whose verdict starts with `STALE` or `ORPHAN` — the right
permalink holding the wrong cut — and every `strays[]` entry, an older
cut of this set living under some other permalink.
OUT: anything with verdict `OK` (that IS the ear-verified cut),
`NOT UPLOADED` (there is nothing there to delete), anything in
`coincidences_discarded[]`, and anything without an `sc_id`.
`coincidences_discarded` is the one that earns its assertion. The lens finds
strays by duration, and on a 132-track account two unrelated tracks will
occasionally share a length to 50 ms: `/au-revoir-lord-toyota` (2020, and
PUBLIC) and `/saria-demo` (2025) both matched an OPAL render. The lens already
drops anything uploaded before the oldest render on disk, because an upload
that predates the render cannot be a cut of it. This script re-checks that
decision against the file it reads instead of trusting the file's own bucket
names — one parser per concept, but two chances to notice a mistake, because
the cost here is asymmetric and permanent.
## Provenance
Every entry carries a `reason` citing the evidence (which generation matched,
the duration gap against the cut that supersedes it), the artwork pulled from
that track, and the local render the upload came from — `reason` is required by
the delete tool and refused if absent, because this is PLN's archive and a
deletion with no recorded why is a hole nobody can later explain.
## Usage
./build_delete_manifest.py [VERDICTS.json] [-o MANIFEST.json]
[--pull DIR] [--renders DIR] [--account NAME]
"""
from __future__ import annotations
import argparse
import json
import re
import sys
import unicodedata
from datetime import date
from pathlib import Path
PROD = Path("/home/pln/Work/Sound/Prod")
DEFAULT_VERDICTS = PROD / "sc_pull_2026-09-05" / "opal26_generation_verdicts.json"
DEFAULT_OUT = PROD / "sc_pull_2026-09-05" / "opal26_delete_manifest.json"
DEFAULT_RENDERS = PROD / "Opal26_master"
AUDIO_EXT = {".flac", ".wav", ".mp3", ".m4a", ".aiff"}
SET_SLUG = "opal-festival-2026"
"""The gig, as the site's canonical content names it.
`../../Web/www/next/content/lives/2026/opal-festival-2026.md` is where the gig
metadata lives; nothing here invents any of it. The slug is quoted in each
reason only so a reader of the receipt knows which set is meant.
"""
def slugify(title: str) -> str:
"""`scform.permalink_of`, mirrored — the key that survives a renumbering.
Used here to line a stale upload up against its counterpart in the newest
cut BY TITLE. Track numbers cannot do that job across generations: the
Aug-12 cut had fifteen tracks, the v4 cut has fourteen, and dropping
Desire renumbered REVOLUTION from 15 to 14. Comparing by number therefore
printed "#14 Desire, 388.1s vs v4 205.6s" — 205.6s is REVOLUTION's length.
A measurement is a comparison, and an unnamed reference is how a
provenance string ends up citing the wrong track (`feedback_name_the_reference`).
"""
s = unicodedata.normalize("NFKD", title or "")
s = s.encode("ascii", "ignore").decode("ascii").lower()
return re.sub(r"[^a-z0-9]+", "-", s).strip("-")
def artwork_for(pull: Path, sc_id, title: str) -> str | None:
"""The pulled original artwork for a track, if the pull got one.
Named `<id>_<slug>.jpg` by `sc pull`, but matched on the id alone: the
slug half is derived from a title that may since have changed, and the id
half is the part that cannot.
"""
d = pull / "artwork"
if not d.is_dir():
return None
hits = sorted(d.glob(f"{sc_id}_*"))
return str(hits[0]) if hits else None
def render_for(renders: Path, generation: str, n: int) -> str | None:
"""The local file for track `n` of a render generation.
This is what makes the deletion reversible in fact rather than only in
principle: the Aug-12 cut is still on disk as
`Opal26_master/tracks/07 - Piment Bresilien.flac`, so removing its upload
loses a URL, not a recording. Matched by the leading track number, the
same key `sc_generation_lens` indexes a generation by.
"""
d = renders / generation
if not d.is_dir():
return None
for f in sorted(d.iterdir()):
if f.suffix.lower() not in AUDIO_EXT or not f.is_file():
continue
m = re.match(r"\s*(\d{1,2})\s*[-_. ]", f.name)
if m and int(m.group(1)) == n:
return str(f)
return None
def reason_for(kind: str, *, title: str, n: int, generation: str,
newest: str, sc_dur, other_dur, created: str, pull: Path,
art: str | None) -> str:
"""One sentence of evidence per entry, in the delete tool's own voice.
Cites what was measured (which generation the upload's duration matches,
and by how much it differs from the cut that supersedes it), when it was
uploaded, what supersedes it, and where the bytes were kept.
"""
if isinstance(sc_dur, (int, float)) and isinstance(other_dur, (int, float)):
gap = f", {sc_dur:.1f}s vs {other_dur:.1f}s in the newest cut"
elif isinstance(sc_dur, (int, float)):
gap = f", {sc_dur:.1f}s, and this track is not in the newest cut at all"
else:
gap = ""
where = "stray permalink" if kind == "stray" else "correct permalink"
kept = (f"artwork preserved in {pull.name}/artwork/"
if art else "no artwork on this upload; nothing lost")
return (f"{created} cut of {SET_SLUG} (#{n} {title}{gap}) on a {where}; "
f"matches render generation `{generation}`, superseded by the "
f"ear-verified `{newest}`; {kept}")
def build(v: dict, pull: Path, renders: Path, account: str) -> tuple[list, dict]:
"""Verdict document -> manifest entries. Pure apart from the disk probes."""
newest = v.get("newest_generation") or "newest"
floor = v.get("stray_floor_date") or ""
# Keyed by title-slug, NOT by track number: see `slugify`.
newest_dur = {slugify(e.get("title") or ""): e.get("expected_duration_s")
for e in v.get("verdicts", [])}
# Everything the manifest must never contain, collected as ids so the
# assertions below can be stated as set arithmetic rather than as trust.
forbidden: dict[int, str] = {}
for c in v.get("coincidences_discarded", []):
if c.get("sc_id"):
forbidden[c["sc_id"]] = (
f"duration coincidence, uploaded {str(c.get('sc_created_at'))[:10]} "
f"before the oldest render ({floor})")
for e in v.get("verdicts", []):
if e.get("verdict") == "OK" and e.get("sc_id"):
forbidden[e["sc_id"]] = "verdict OK — this is the cut to keep"
entries = []
for e in v.get("verdicts", []):
if not str(e.get("verdict", "")).startswith(("STALE", "ORPHAN")):
continue
if not e.get("sc_id"):
print(f" ! {e.get('verdict')} at /{e.get('permalink')} has no "
f"sc_id — skipped, there is nothing to address", file=sys.stderr)
continue
gen = (e["verdict"].split(":", 1) + [""])[1] or "unknown"
art = artwork_for(pull, e["sc_id"], e.get("title") or "")
entries.append({
"permalink": e["permalink"],
"sc_id": e["sc_id"],
"reason": reason_for("stale", title=e.get("title") or "",
n=e["n"], generation=gen, newest=newest,
sc_dur=e.get("sc_duration_s"),
other_dur=newest_dur.get(slugify(e.get("title") or "")),
created=str(e.get("sc_created_at"))[:10],
pull=pull, art=art),
"title": e.get("title"),
"sc_url": e.get("sc_url"),
"sc_created_at": e.get("sc_created_at"),
"sc_duration_s": e.get("sc_duration_s"),
"sc_public": e.get("sc_public"),
"verdict": e["verdict"],
"source_generation": gen,
"source_track_n": e["n"],
"artwork_archived": art,
"local_source": render_for(renders, gen, e["n"]),
})
for s in v.get("strays", []):
if not s.get("sc_id"):
print(f" ! stray /{s.get('permalink')} has no sc_id — skipped",
file=sys.stderr)
continue
gen = s.get("matches_generation") or "unknown"
n = s.get("matches_track")
art = artwork_for(pull, s["sc_id"], s.get("matches_title") or "")
entries.append({
"permalink": s["permalink"],
"sc_id": s["sc_id"],
"reason": reason_for("stray", title=s.get("matches_title") or "",
n=n, generation=gen, newest=newest,
sc_dur=s.get("sc_duration_s"),
other_dur=newest_dur.get(slugify(s.get("matches_title") or "")),
created=str(s.get("sc_created_at"))[:10],
pull=pull, art=art),
"title": s.get("matches_title"),
"sc_url": s.get("sc_url"),
"sc_created_at": s.get("sc_created_at"),
"sc_duration_s": s.get("sc_duration_s"),
"sc_public": s.get("sc_public"),
"verdict": f"STRAY:{gen}",
"source_generation": gen,
"source_track_n": n,
"artwork_archived": art,
"local_source": render_for(renders, gen, n) if n else None,
})
assert_clean(entries, forbidden, floor)
return entries, {
"account": account,
"set": SET_SLUG,
"generated_by": "armada/tide-table/build_delete_manifest.py",
"generated_at": date.today().isoformat(),
"source": str(v.get("_source_path") or ""),
"source_pulled_at": v.get("pulled_at"),
"newest_generation": newest,
"tolerance_s": v.get("tolerance_s"),
"stray_floor_date": floor,
"artwork_archive": str(pull / "artwork"),
"render_archive": str(renders),
}
def assert_clean(entries: list, forbidden: dict[int, str], floor: str) -> None:
"""Fail loudly rather than emit a manifest that could delete the wrong thing.
These are assertions and not warnings because the output feeds a
destructive call. A warning printed above 14 lines of table is a warning
nobody reads; a non-zero exit with no file written cannot be missed.
"""
bad = []
seen: dict = {}
for e in entries:
if e["sc_id"] in forbidden:
bad.append(f"/{e['permalink']} (sc_id {e['sc_id']}) is excluded: "
f"{forbidden[e['sc_id']]}")
if not e.get("sc_id"):
bad.append(f"/{e['permalink']} has no sc_id")
if not (e.get("reason") or "").strip():
bad.append(f"/{e['permalink']} has no reason")
if e.get("sc_public"):
bad.append(f"/{e['permalink']} is PUBLIC — a public track must not "
f"reach a delete manifest without a human saying so")
# An entry created before the oldest render on disk cannot be a cut of
# this set, whichever list the verdict file filed it under.
created = str(e.get("sc_created_at") or "")[:10]
if floor and created and created < floor:
bad.append(f"/{e['permalink']} was uploaded {created}, before the "
f"oldest render ({floor}) — it cannot be a cut of this set")
for key in ("permalink", "sc_id"):
k = (key, e[key])
if k in seen:
bad.append(f"duplicate {key} {e[key]!r}")
seen[k] = True
if bad:
print("REFUSING to write the manifest:", file=sys.stderr)
for b in bad:
print(f" ! {b}", file=sys.stderr)
raise SystemExit(f"{len(bad)} problem(s); no file written")
def main() -> int:
ap = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("verdicts", nargs="?", type=Path, default=DEFAULT_VERDICTS,
help=f"sc_generation_lens output (default: {DEFAULT_VERDICTS})")
ap.add_argument("-o", "--out", type=Path, default=DEFAULT_OUT)
ap.add_argument("--pull", type=Path, default=None,
help="the `sc pull` directory holding the archived artwork "
"(default: the verdict file's own directory)")
ap.add_argument("--renders", type=Path, default=DEFAULT_RENDERS,
help="directory holding the tracks*/ render generations")
ap.add_argument("--account", default="parvagues",
help="the SoundCloud account the manifest is FOR; the "
"delete tool refuses to run it against another")
a = ap.parse_args()
if not a.verdicts.exists():
raise SystemExit(f"not found: {a.verdicts}")
v = json.loads(a.verdicts.read_text())
v["_source_path"] = str(a.verdicts)
pull = a.pull or a.verdicts.parent
entries, meta = build(v, pull, a.renders, a.account)
print(f"{a.verdicts.name}: {len(v.get('verdicts', []))} roll-call verdicts, "
f"{len(v.get('strays', []))} strays, "
f"{len(v.get('coincidences_discarded', []))} coincidences dropped")
print(f"newest generation: {meta['newest_generation']} · "
f"stray floor: {meta['stray_floor_date']}\n")
print(f"{'permalink':38} {'sc_id':>10} {'dur':>8} {'uploaded':10} "
f"{'gen':14} art local render")
for e in entries:
print(f"/{e['permalink'][:36]:37} {e['sc_id']:>10} "
f"{(e['sc_duration_s'] or 0):8.1f} {str(e['sc_created_at'])[:10]:10} "
f"{e['source_generation'][:14]:14} "
f"{'yes' if e['artwork_archived'] else ' - ':3} "
f"{Path(e['local_source']).name if e['local_source'] else 'MISSING'}")
no_render = [e for e in entries if not e["local_source"]]
if no_render:
print(f"\n ! {len(no_render)} entr"
f"{'y' if len(no_render) == 1 else 'ies'} have no local render — "
f"deleting those loses the only copy:")
for e in no_render:
print(f" /{e['permalink']}")
a.out.parent.mkdir(parents=True, exist_ok=True)
a.out.write_text(json.dumps({**meta, "tracks": entries},
indent=2, ensure_ascii=False) + "\n")
print(f"\n{len(entries)} entries -> {a.out}")
print(f"\nDry-run it before anyone types --go:\n"
f" cd ~/Work/Sound/tidal-ears && .venv/bin/python -m "
f"tidal_ears.master sc delete --manifest {a.out}")
return 0
if __name__ == "__main__":
sys.exit(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment