#!/usr/bin/env python3 """ backfill-buy-url.py — stamp the Bandcamp purchase URL onto library files for albums/tracks already imported from a Bandcamp purchase. The forward path (sync-bandcamp.py) tags new purchases at download time. This one-shot covers everything bought *before* that change landed. It: 1. Pulls the current Bandcamp collection (cookies auth) and builds an index keyed by (album_artist, album) -> release URL. (Bandcamp's embedded ALBUM tag matches the collection item_title exactly, and ALBUMARTIST matches the band name — verified against the library.) 2. Walks the FLAC library, reads ALBUMARTIST/ALBUM, and where a match exists, writes the COMMERCIAL_INFORMATION Vorbis comment (AzuraCast maps this to the buy_url custom field — see sync-bandcamp.py BUY_URL_TAG). 3. Optionally tells AzuraCast to reprocess each touched file so the custom field repopulates without waiting for the periodic media scan. Dry-run by default. Stdlib only (mutagen NOT required — uses metaflac). Usage: backfill-buy-url.py --cookies F --state F [--library DIR] [--apply] [--azuracast-key ID:SECRET] [--azuracast-base URL] [--station N] [--force] """ import sys, os, re, json, time, html, argparse, subprocess import urllib.request from http.cookiejar import MozillaCookieJar from pathlib import Path COLLECTION_ITEMS_URL = "https://bandcamp.com/api/fancollection/1/collection_items" UA = "Mozilla/5.0 (X11; Linux x86_64) backfill-buy-url/1.0" BUY_URL_TAG = "COMMERCIAL_INFORMATION" # ---- Bandcamp collection --------------------------------------------------- def opener(jar): return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) def get_fan_id(username, jar): req = urllib.request.Request(f"https://bandcamp.com/{username}", headers={"User-Agent": UA}) page = html.unescape(opener(jar).open(req, timeout=60).read().decode("utf-8", "replace")) m = re.search(r'"fan_id"\s*:\s*"?(\d+)', page) if not m: sys.exit("[backfill] could not find fan_id (cookies expired / wrong user?)") return int(m.group(1)) def fetch_collection(fan_id, jar): items = [] tok = "9999999999::a::" while True: body = json.dumps({"fan_id": fan_id, "older_than_token": tok, "count": 100}).encode() req = urllib.request.Request( COLLECTION_ITEMS_URL, data=body, headers={"User-Agent": UA, "Content-Type": "application/json", "Accept": "application/json", "Referer": "https://bandcamp.com/"}) data = json.loads(opener(jar).open(req, timeout=60).read()) page = data.get("items", []) items.extend(page) if not data.get("more_available") or not page: break nt = data.get("last_token") if not nt or nt == tok: break tok = nt return items # ---- matching -------------------------------------------------------------- def norm(s): return re.sub(r"\s+", " ", (s or "").strip()).casefold() def build_index(items): """(album_artist, album) -> url. Index both the bare title and the " - Single" form (single-track purchases get that ALBUM tag).""" idx = {} for it in items: url = it.get("item_url") band = it.get("band_name") title = it.get("item_title") if not (url and band and title): continue for album in (title, f"{title} - Single"): idx.setdefault((norm(band), norm(album)), url) return idx def flac_tag(path, name): r = subprocess.run(["metaflac", f"--show-tag={name}", path], capture_output=True, text=True) for line in r.stdout.splitlines(): if "=" in line: return line.split("=", 1)[1] return "" def set_flac_tag(path, name, value): subprocess.run(["metaflac", f"--remove-tag={name}", path], capture_output=True) return subprocess.run(["metaflac", f"--set-tag={name}={value}", path], capture_output=True).returncode == 0 # ---- AzuraCast reprocess --------------------------------------------------- def az_reprocess(base, key, station, host_paths, log): """Find each host path in AzuraCast and batch-reprocess so the custom field repopulates. host_path -> relative AzuraCast path is library-root stripped.""" root = f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/Library/" want = {p[len(root):] for p in host_paths if p.startswith(root)} headers = {"X-API-Key": key, "Accept": "application/json"} def get(path): req = urllib.request.Request(base + path, headers=headers) return json.loads(urllib.request.urlopen(req, timeout=60).read()) unique_ids, page = [], 1 while True: d = get(f"/api/station/{station}/files?per_page=500&page={page}") for row in d["rows"]: if row["path"] in want: unique_ids.append(row["unique_id"]) if page >= d["total_pages"]: break page += 1 if not unique_ids: log(" no matching AzuraCast files to reprocess") return body = json.dumps({"do": "reprocess", "files": unique_ids}).encode() req = urllib.request.Request(base + f"/api/station/{station}/files/batch", data=body, method="PUT", headers={**headers, "Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=120).read() log(f" queued reprocess for {len(unique_ids)} AzuraCast file(s)") def main(): ap = argparse.ArgumentParser() ap.add_argument("--cookies", required=True) ap.add_argument("--state", required=True, help="state.json (for reference/logging)") ap.add_argument("--library", default=f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/Library") ap.add_argument("--apply", action="store_true") ap.add_argument("--force", action="store_true", help="overwrite an existing buy-link tag") ap.add_argument("--azuracast-key", default=None) ap.add_argument("--azuracast-base", default="https://admin.ephemeral.club") ap.add_argument("--station", type=int, default=1) ap.add_argument("--username", default=None, help="Bandcamp username; default read from config.env if present") args = ap.parse_args() user = args.username if not user: cfg = f"{os.environ.get('ALEMBIC_CONFIG_DIR', '/config')}/pipeline/bandcamp/config.env" if os.path.exists(cfg): for line in Path(cfg).read_text().splitlines(): if line.startswith("BANDCAMP_USERNAME="): user = line.split("=", 1)[1].strip().strip('"') if not user: sys.exit("[backfill] --username required (no config.env BANDCAMP_USERNAME)") print(f"[backfill] mode={'APPLY' if args.apply else 'DRY RUN'} user={user!r}") jar = MozillaCookieJar(args.cookies) jar.load(ignore_discard=True, ignore_expires=True) fan_id = get_fan_id(user, jar) items = fetch_collection(fan_id, jar) idx = build_index(items) print(f"[backfill] collection: {len(items)} items, {len(idx)} match keys") flacs = list(Path(args.library).rglob("*.flac")) print(f"[backfill] scanning {len(flacs)} FLAC files...\n") matched = tagged = already = unmatched = 0 touched = [] for p in flacs: sp = str(p) albumartist = flac_tag(sp, "ALBUMARTIST") or flac_tag(sp, "ARTIST") album = flac_tag(sp, "ALBUM") url = idx.get((norm(albumartist), norm(album))) if not url: unmatched += 1 continue matched += 1 existing = flac_tag(sp, BUY_URL_TAG) if existing and not args.force: already += 1 continue rel = sp[len(args.library):].lstrip("/") print(f" + {rel}\n -> {url}") if args.apply: if set_flac_tag(sp, BUY_URL_TAG, url): tagged += 1 touched.append(sp) else: print(" ! metaflac write failed") else: tagged += 1 touched.append(sp) print(f"\n[backfill] matched={matched} {'tagged' if args.apply else 'would-tag'}={tagged}" f" already-had={already} unmatched-files={unmatched}") if args.apply and touched and args.azuracast_key: print("[backfill] telling AzuraCast to reprocess touched files...") az_reprocess(args.azuracast_base, args.azuracast_key, args.station, touched, lambda m: print(m)) elif not args.apply: print("[backfill] re-run with --apply to write (add --azuracast-key to reprocess)") if __name__ == "__main__": main()