andrew
68cb007e4c
Moves all ~30 pipeline scripts, configs, and the vendored sldl binary into this repo (source /opt/sldl left untouched). Removes all docker exec/docker compose dependencies now that beets and sldl run in-process/as a subprocess of this container instead of via soulbeet/on-demand sldl containers. Replaces hardcoded host paths, Navidrome credentials, and Spotify credential sourcing with env-var-driven paths and shared credential loaders. Adds Dockerfile, entrypoint.sh, requirements.txt, docker-compose.snippet.yml, and the initial app DB schema.
441 lines
16 KiB
Python
Executable File
441 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
sync-bandcamp.py — download new items from a Bandcamp purchase history.
|
|
|
|
Reads a Netscape-format cookies file, queries Bandcamp's collection API to
|
|
find every item the authenticated user has purchased, then downloads each
|
|
new item (one not in the state file yet) as a ZIP, extracts the audio files,
|
|
and records the item in the state file.
|
|
|
|
Stdlib only — no third-party deps.
|
|
|
|
Usage:
|
|
sync-bandcamp.py <username> <cookies_file> <state_file> <output_dir> [<format_pref>]
|
|
|
|
format_pref is a space-separated preference list (default: "flac mp3-320").
|
|
First match wins.
|
|
|
|
Exit code 0 even if some items failed (we want cron to keep trying); details
|
|
are logged. Exit code !=0 only on auth/setup problems where retrying without
|
|
human intervention is pointless.
|
|
"""
|
|
import sys, os, re, json, time, html, subprocess
|
|
import urllib.request, urllib.parse
|
|
from http.cookiejar import MozillaCookieJar
|
|
from pathlib import Path
|
|
import zipfile, io
|
|
import traceback
|
|
|
|
|
|
COLLECTION_ITEMS_URL = "https://bandcamp.com/api/fancollection/1/collection_items"
|
|
UA = "Mozilla/5.0 (X11; Linux x86_64) sync-bandcamp/1.0"
|
|
REQUEST_TIMEOUT = 60
|
|
DOWNLOAD_TIMEOUT = 600 # ZIPs can be large
|
|
SLEEP_BETWEEN = 2.0 # be polite
|
|
|
|
|
|
def opener_for(jar):
|
|
return urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(jar)
|
|
)
|
|
|
|
|
|
def http_get(url, jar, timeout=REQUEST_TIMEOUT):
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
return opener_for(jar).open(req, timeout=timeout)
|
|
|
|
|
|
def http_post_json(url, body, jar, timeout=REQUEST_TIMEOUT):
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(body).encode(),
|
|
headers={
|
|
"User-Agent": UA,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"Referer": "https://bandcamp.com/",
|
|
},
|
|
)
|
|
return opener_for(jar).open(req, timeout=timeout)
|
|
|
|
|
|
def get_fan_id(username, jar):
|
|
"""The fan_id is inside the data-blob JSON on the user's collection page.
|
|
Bandcamp HTML-escapes the JSON inside the attribute, so we have to
|
|
unescape before matching."""
|
|
resp = http_get(f"https://bandcamp.com/{username}", jar)
|
|
page = resp.read().decode("utf-8", errors="replace")
|
|
page_unescaped = html.unescape(page)
|
|
m = re.search(r'"fan_id"\s*:\s*"?(\d+)', page_unescaped)
|
|
if not m:
|
|
sys.stderr.write("[bandcamp] Could not find fan_id on profile page.\n")
|
|
sys.stderr.write("[bandcamp] Likely your cookies file is missing or expired,\n")
|
|
sys.stderr.write("[bandcamp] or the username in config.env is wrong.\n")
|
|
sys.exit(2)
|
|
return int(m.group(1))
|
|
|
|
|
|
def fetch_collection(fan_id, jar):
|
|
"""Page through the collection. Returns (items, redownload_urls).
|
|
|
|
items: list of dicts (Bandcamp's per-item JSON)
|
|
redownload_urls: dict like {"p12345": "https://bandcamp.com/download?...", ...}
|
|
keyed by <type-letter><sale_item_id>.
|
|
"""
|
|
all_items = []
|
|
all_redownload = {}
|
|
older_than_token = "9999999999::a::" # paginate from newest
|
|
|
|
while True:
|
|
body = {
|
|
"fan_id": fan_id,
|
|
"older_than_token": older_than_token,
|
|
"count": 100,
|
|
}
|
|
resp = http_post_json(COLLECTION_ITEMS_URL, body, jar)
|
|
data = json.loads(resp.read())
|
|
|
|
page_items = data.get("items", [])
|
|
page_redownload = data.get("redownload_urls", {}) or {}
|
|
all_items.extend(page_items)
|
|
all_redownload.update(page_redownload)
|
|
|
|
if not data.get("more_available") or not page_items:
|
|
break
|
|
|
|
last_token = data.get("last_token")
|
|
if not last_token or last_token == older_than_token:
|
|
break
|
|
older_than_token = last_token
|
|
|
|
return all_items, all_redownload
|
|
|
|
|
|
def parse_data_blob(html_text):
|
|
"""Bandcamp embeds a JSON 'data-blob' attribute on <div id="pagedata">.
|
|
Extract and decode it."""
|
|
m = re.search(r'id="pagedata"[^>]*data-blob="([^"]+)"', html_text)
|
|
if not m:
|
|
return None
|
|
raw = m.group(1)
|
|
raw = html.unescape(raw)
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def get_format_url(redownload_url, format_pref, jar):
|
|
"""Load the download landing page and return the signed URL for the
|
|
preferred format. Returns (format_chosen, url) or (None, None)."""
|
|
resp = http_get(redownload_url, jar, timeout=REQUEST_TIMEOUT)
|
|
page = resp.read().decode("utf-8", errors="replace")
|
|
blob = parse_data_blob(page)
|
|
if not blob:
|
|
return None, None
|
|
|
|
download_items = blob.get("download_items") or []
|
|
if not download_items:
|
|
return None, None
|
|
|
|
item = download_items[0]
|
|
downloads = item.get("downloads") or {}
|
|
for fmt in format_pref:
|
|
if fmt in downloads and downloads[fmt].get("url"):
|
|
return fmt, downloads[fmt]["url"]
|
|
return None, None
|
|
|
|
|
|
def stat_url_until_ready(stat_url, jar, max_wait_sec=120):
|
|
"""Bandcamp sometimes returns a 'stat' URL that has to be polled until
|
|
download is signed. Returns the final download_url, or None if it
|
|
never becomes ready."""
|
|
deadline = time.time() + max_wait_sec
|
|
last = None
|
|
while time.time() < deadline:
|
|
# The stat endpoint expects ?.rand=<num>
|
|
url = stat_url + ("&" if "?" in stat_url else "?") + ".rand=" + str(int(time.time() * 1000))
|
|
try:
|
|
resp = http_get(url, jar, timeout=30)
|
|
text = resp.read().decode("utf-8", errors="replace")
|
|
# Strip JSONP wrapper if present
|
|
text = re.sub(r"^\s*[A-Za-z_]\w*\s*\(", "", text)
|
|
text = re.sub(r"\)\s*;?\s*$", "", text)
|
|
data = json.loads(text)
|
|
for k, v in (data.get("download_url_stat") or {}).items():
|
|
last = v
|
|
if last and last.get("download_ready"):
|
|
return last.get("download_url")
|
|
except Exception:
|
|
pass
|
|
time.sleep(3)
|
|
return None
|
|
|
|
|
|
def download_payload(url, jar):
|
|
"""Returns (bytes, suggested_filename) — Bandcamp returns either a ZIP
|
|
(for albums / multi-track purchases) or a single audio file (for single-
|
|
track purchases). The Content-Disposition header tells us which."""
|
|
resp = http_get(url, jar, timeout=DOWNLOAD_TIMEOUT)
|
|
cd = resp.headers.get("Content-Disposition", "")
|
|
m = re.search(r'filename\*?=(?:UTF-8\'\')?"?([^";]+)"?', cd)
|
|
suggested = m.group(1) if m else ""
|
|
suggested = urllib.parse.unquote(suggested).strip()
|
|
return resp.read(), suggested
|
|
|
|
|
|
def safe_name(s):
|
|
"""Make a string safe for filesystem use (and not too long)."""
|
|
s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", s)
|
|
s = s.strip().rstrip(".")
|
|
return s[:80] if len(s) > 80 else s
|
|
|
|
|
|
_AUDIO_EXTS = (".flac", ".mp3", ".m4a", ".aac", ".ogg", ".oga", ".alac",
|
|
".wav", ".aiff", ".aif")
|
|
_IMAGE_EXTS = (".jpg", ".jpeg", ".png", ".webp")
|
|
|
|
# --- Buy-link tagging -------------------------------------------------------
|
|
# We stamp the release's Bandcamp page into the file so AzuraCast can show a
|
|
# "Buy" button on ephemeral.club. AzuraCast only auto-assigns custom fields
|
|
# from its *known* tag enum, so a bespoke "BUY_URL" tag would be ignored
|
|
# (it lands in extraTags). Instead we use the "Commercial Information" tag —
|
|
# ID3 frame WCOM, getID3 key `commercial_information`, semantically "a webpage
|
|
# with information such as where the album can be bought". AzuraCast's custom
|
|
# field `buy_url` has auto_assign=commercial_information, so this surfaces in
|
|
# now_playing.song.custom_fields.buy_url. See README "Buy links" section.
|
|
BUY_URL_TAG = "COMMERCIAL_INFORMATION" # FLAC Vorbis comment name
|
|
|
|
|
|
def tag_buy_url(filepath, url):
|
|
"""Best-effort: stamp the purchasable URL into the file's commercial-
|
|
information tag. Never fails the download over a tagging error."""
|
|
if not url:
|
|
return
|
|
low = filepath.lower()
|
|
try:
|
|
if low.endswith(".flac"):
|
|
subprocess.run(["metaflac", f"--remove-tag={BUY_URL_TAG}", filepath],
|
|
capture_output=True, check=False)
|
|
subprocess.run(["metaflac", f"--set-tag={BUY_URL_TAG}={url}", filepath],
|
|
capture_output=True, check=False)
|
|
elif low.endswith(".mp3"):
|
|
# WCOM = "Commercial information" URL link frame.
|
|
from mutagen.id3 import ID3, WCOM, ID3NoHeaderError
|
|
try:
|
|
tags = ID3(filepath)
|
|
except ID3NoHeaderError:
|
|
tags = ID3()
|
|
tags.delall("WCOM")
|
|
tags.add(WCOM(url=url))
|
|
tags.save(filepath)
|
|
except Exception as e:
|
|
sys.stderr.write(f"[bandcamp] WARN: buy_url tag failed on {filepath}: {e}\n")
|
|
|
|
|
|
def save_payload(payload, suggested_name, dest_dir, item_title=None, buy_url=None):
|
|
"""Save a Bandcamp download — either ZIP (extract audio) or raw audio
|
|
(write directly). For raw single-track audio, ensure an ALBUM tag is set
|
|
so the file doesn't land as [Unknown Album] in Navidrome. Each audio file
|
|
is stamped with the release's buy_url (see tag_buy_url).
|
|
|
|
Returns count of audio files written."""
|
|
# Detect ZIP by magic bytes
|
|
is_zip = payload[:4] == b"PK\x03\x04" or payload[:4] == b"PK\x05\x06"
|
|
if is_zip:
|
|
count = 0
|
|
with zipfile.ZipFile(io.BytesIO(payload)) as z:
|
|
for n in z.namelist():
|
|
low = n.lower()
|
|
if low.endswith(_AUDIO_EXTS):
|
|
outp = z.extract(n, dest_dir)
|
|
tag_buy_url(outp, buy_url)
|
|
count += 1
|
|
elif low.endswith(_IMAGE_EXTS):
|
|
z.extract(n, dest_dir) # cover art for beets fetchart
|
|
return count
|
|
|
|
# Raw audio file (single-track purchase) — Bandcamp serves these without
|
|
# any ALBUM tag, so we tag them ourselves before they hit the pipeline.
|
|
name = suggested_name or "track"
|
|
low = name.lower()
|
|
if not low.endswith(_AUDIO_EXTS):
|
|
if payload[:4] == b"fLaC":
|
|
name = (name or "track") + ".flac"
|
|
elif payload[:3] == b"ID3" or payload[:2] == b"\xff\xfb":
|
|
name = (name or "track") + ".mp3"
|
|
else:
|
|
return 0 # unknown format
|
|
name = safe_name(name)
|
|
out = os.path.join(dest_dir, name)
|
|
with open(out, "wb") as f:
|
|
f.write(payload)
|
|
|
|
# Set ALBUM = "{item_title} - Single" if the file has no ALBUM tag.
|
|
# Matches Spotify/Apple Music's convention for single-track releases.
|
|
if item_title:
|
|
_ensure_album_tag(out, f"{item_title} - Single")
|
|
tag_buy_url(out, buy_url)
|
|
return 1
|
|
|
|
|
|
def _ensure_album_tag(filepath, fallback_album):
|
|
"""If the file has no ALBUM tag, set it. No-op if one already exists."""
|
|
low = filepath.lower()
|
|
try:
|
|
if low.endswith(".flac"):
|
|
r = subprocess.run(
|
|
["metaflac", "--show-tag=ALBUM", filepath],
|
|
capture_output=True, text=True
|
|
)
|
|
if "ALBUM=" in r.stdout:
|
|
return
|
|
subprocess.run(
|
|
["metaflac", f"--set-tag=ALBUM={fallback_album}", filepath],
|
|
check=True, stderr=subprocess.DEVNULL,
|
|
)
|
|
elif low.endswith(".mp3"):
|
|
r = subprocess.run(
|
|
["id3v2", "-l", filepath],
|
|
capture_output=True, text=True
|
|
)
|
|
if re.search(r"^TALB\b", r.stdout, re.MULTILINE):
|
|
return
|
|
subprocess.run(
|
|
["id3v2", "--TALB", fallback_album, filepath],
|
|
check=True, stderr=subprocess.DEVNULL,
|
|
)
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
pass # best-effort; don't fail the whole download over this
|
|
|
|
|
|
def load_state(path):
|
|
if not os.path.exists(path):
|
|
return {}
|
|
try:
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
except json.JSONDecodeError:
|
|
sys.stderr.write(f"[bandcamp] WARN: {path} is corrupt; starting fresh\n")
|
|
return {}
|
|
|
|
|
|
def save_state(path, state):
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(state, f, indent=2, sort_keys=True)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 5:
|
|
sys.exit(f"Usage: {sys.argv[0]} <username> <cookies_file> <state_file> <output_dir> [<format_pref>]")
|
|
|
|
username = sys.argv[1]
|
|
cookies_file = sys.argv[2]
|
|
state_file = sys.argv[3]
|
|
output_dir = sys.argv[4]
|
|
format_pref = sys.argv[5].split() if len(sys.argv) > 5 else ["flac", "mp3-320"]
|
|
|
|
if not os.path.exists(cookies_file):
|
|
sys.stderr.write(f"[bandcamp] ERROR: cookies file not found: {cookies_file}\n")
|
|
sys.stderr.write("[bandcamp] Export cookies for bandcamp.com from a logged-in browser tab\n")
|
|
sys.stderr.write("[bandcamp] (e.g. via the 'Get cookies.txt LOCALLY' extension) and save here.\n")
|
|
sys.exit(2)
|
|
|
|
jar = MozillaCookieJar(cookies_file)
|
|
try:
|
|
jar.load(ignore_discard=True, ignore_expires=True)
|
|
except Exception as e:
|
|
sys.stderr.write(f"[bandcamp] ERROR: cookies file unreadable: {e}\n")
|
|
sys.exit(2)
|
|
|
|
Path(output_dir).mkdir(parents=True, exist_ok=True)
|
|
state = load_state(state_file)
|
|
|
|
print(f"[bandcamp] auth as {username!r}, format pref={format_pref}")
|
|
fan_id = get_fan_id(username, jar)
|
|
print(f"[bandcamp] fan_id={fan_id}")
|
|
|
|
items, redownload = fetch_collection(fan_id, jar)
|
|
print(f"[bandcamp] collection has {len(items)} items, {len(redownload)} download URLs")
|
|
|
|
new_count = skipped = failed = 0
|
|
for item in items:
|
|
sale_id = item.get("sale_item_id")
|
|
if not sale_id:
|
|
continue
|
|
sale_id_str = str(sale_id)
|
|
if sale_id_str in state:
|
|
skipped += 1
|
|
continue
|
|
|
|
band = item.get("band_name") or "Unknown Artist"
|
|
title = item.get("item_title") or f"item-{item.get('item_id')}"
|
|
buy_url = item.get("item_url") # the release's Bandcamp page (purchasable)
|
|
item_type = item.get("item_type", "package") # "track" or "package" (album)
|
|
type_prefix = "p" if item_type in ("package", "album") else "t"
|
|
url_key = f"{type_prefix}{sale_id}"
|
|
|
|
download_landing = redownload.get(url_key)
|
|
if not download_landing:
|
|
# Sometimes the key uses the OTHER prefix
|
|
alt_prefix = "t" if type_prefix == "p" else "p"
|
|
download_landing = redownload.get(f"{alt_prefix}{sale_id}")
|
|
if not download_landing:
|
|
print(f"[bandcamp] NO_URL {band} — {title}")
|
|
failed += 1
|
|
continue
|
|
|
|
try:
|
|
fmt, signed_url = get_format_url(download_landing, format_pref, jar)
|
|
if not signed_url:
|
|
print(f"[bandcamp] NO_FORMAT {band} — {title} (preferred: {format_pref})")
|
|
failed += 1
|
|
continue
|
|
|
|
# Some signed_urls are direct, some are "stat" URLs that need polling
|
|
actual_url = signed_url
|
|
if "/statdownload/" in signed_url:
|
|
polled = stat_url_until_ready(signed_url, jar)
|
|
if polled:
|
|
actual_url = polled
|
|
else:
|
|
print(f"[bandcamp] STAT_TIMEOUT {band} — {title}")
|
|
failed += 1
|
|
continue
|
|
|
|
payload, suggested = download_payload(actual_url, jar)
|
|
dest = Path(output_dir) / safe_name(band) / safe_name(title)
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
n = save_payload(payload, suggested, str(dest), item_title=title,
|
|
buy_url=buy_url)
|
|
if n == 0:
|
|
print(f"[bandcamp] EMPTY_PAYLOAD {band} — {title} (got {len(payload)} bytes, type unknown)")
|
|
failed += 1
|
|
continue
|
|
|
|
state[sale_id_str] = {
|
|
"downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
|
"band": band,
|
|
"title": title,
|
|
"format": fmt,
|
|
"files": n,
|
|
"path": str(dest),
|
|
"buy_url": buy_url,
|
|
}
|
|
save_state(state_file, state) # save after each — survives interruption
|
|
print(f"[bandcamp] OK ({fmt}, {n} files) {band} — {title}")
|
|
new_count += 1
|
|
time.sleep(SLEEP_BETWEEN)
|
|
except Exception as e:
|
|
sys.stderr.write(f"[bandcamp] FAIL {band} — {title}: {e}\n")
|
|
traceback.print_exc(file=sys.stderr)
|
|
failed += 1
|
|
|
|
print(f"\n[bandcamp] === {new_count} new, {skipped} already had, {failed} failed ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|