Spotify integration added
This commit is contained in:
Executable
+392
@@ -0,0 +1,392 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
bandcamp-sync — Download new Bandcamp purchases (WAV) to the local music library,
|
||||
then invoke bandcamp-tag.py to fill in metadata.
|
||||
|
||||
Usage:
|
||||
bandcamp-sync.py [MUSIC_DIR] # sync new purchases to MUSIC_DIR
|
||||
bandcamp-sync.py --dry-run # show what would be downloaded, no writes
|
||||
bandcamp-sync.py --list # list full collection, no download
|
||||
|
||||
Credentials (first match wins):
|
||||
1. Env vars BANDCAMP_EMAIL / BANDCAMP_PASSWORD
|
||||
2. bandcamp-api service at http://localhost:8091/credentials
|
||||
3. JSON file ~/.config/audioserver/bandcamp.json { "email": ..., "password": ... }
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote
|
||||
|
||||
import requests
|
||||
|
||||
# ── Config ────────────────────────────────────────────────────────────────────
|
||||
|
||||
DEFAULT_MUSIC_DIR = Path("/mnt/local/music")
|
||||
SCRIPT_DIR = Path(__file__).parent
|
||||
TAGGER = SCRIPT_DIR / "bandcamp-tag.py"
|
||||
REQUEST_DELAY = 1.0 # seconds between Bandcamp requests
|
||||
COLLECTION_BATCH = 50
|
||||
|
||||
SESSION = requests.Session()
|
||||
SESSION.headers["User-Agent"] = (
|
||||
"Mozilla/5.0 (X11; Linux aarch64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
|
||||
)
|
||||
|
||||
# ── Credentials ───────────────────────────────────────────────────────────────
|
||||
|
||||
def load_credentials() -> tuple[str, str]:
|
||||
email = os.environ.get("BANDCAMP_EMAIL", "")
|
||||
password = os.environ.get("BANDCAMP_PASSWORD", "")
|
||||
if email and password:
|
||||
return email, password
|
||||
|
||||
try:
|
||||
r = requests.get("http://localhost:8091/credentials", timeout=3)
|
||||
if r.ok:
|
||||
data = r.json()
|
||||
api_email = data.get("email", "")
|
||||
# password_set=True but we can't retrieve the plaintext from GET;
|
||||
# the service exposes /test-login but not the raw password.
|
||||
# Fall through to the JSON file which stores plaintext.
|
||||
if api_email:
|
||||
email = api_email
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
cfg_file = Path.home() / ".config" / "audioserver" / "bandcamp.json"
|
||||
if cfg_file.exists():
|
||||
with open(cfg_file) as f:
|
||||
cfg = json.load(f)
|
||||
email = cfg.get("email", email)
|
||||
password = cfg.get("password", password)
|
||||
|
||||
# Also try the Docker volume path (running on host alongside Docker)
|
||||
docker_cfg = Path("/var/lib/docker/volumes/audioserver_bandcamp_config/_data/bandcamp.json")
|
||||
if not password and docker_cfg.exists():
|
||||
try:
|
||||
with open(docker_cfg) as f:
|
||||
cfg = json.load(f)
|
||||
email = cfg.get("email", email)
|
||||
password = cfg.get("password", password)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return email, password
|
||||
|
||||
|
||||
# ── Bandcamp auth ─────────────────────────────────────────────────────────────
|
||||
|
||||
def bc_login(email: str, password: str) -> dict:
|
||||
"""Login to Bandcamp. Returns account dict with fan_id."""
|
||||
r = SESSION.get("https://bandcamp.com/login", timeout=15)
|
||||
csrf = unquote(r.cookies.get("csrf_token", ""))
|
||||
if not csrf:
|
||||
m = re.search(r'"csrf_token"\s*:\s*"([^"]+)"', r.text)
|
||||
if m:
|
||||
csrf = unquote(m.group(1))
|
||||
if not csrf:
|
||||
raise RuntimeError("Could not get CSRF token from Bandcamp login page")
|
||||
|
||||
time.sleep(REQUEST_DELAY)
|
||||
r = SESSION.post(
|
||||
"https://bandcamp.com/login_cb",
|
||||
data={"username": email, "password": password, "csrf_token": csrf},
|
||||
headers={
|
||||
"X-CSRFToken": csrf,
|
||||
"Referer": "https://bandcamp.com/login",
|
||||
"Accept": "application/json, text/javascript, */*",
|
||||
"Origin": "https://bandcamp.com",
|
||||
},
|
||||
timeout=15,
|
||||
)
|
||||
payload = r.json()
|
||||
if payload.get("ok") != 1:
|
||||
err = payload.get("error") or payload.get("error_message") or json.dumps(payload)
|
||||
raise RuntimeError(f"Bandcamp login failed: {err}")
|
||||
|
||||
return payload.get("account", {})
|
||||
|
||||
|
||||
def get_fan_id(account: dict) -> int:
|
||||
fan_id = account.get("fan_id") or account.get("id")
|
||||
if fan_id:
|
||||
return int(fan_id)
|
||||
|
||||
username = account.get("username", "")
|
||||
if username:
|
||||
time.sleep(REQUEST_DELAY)
|
||||
r = SESSION.get(f"https://bandcamp.com/{username}", timeout=15)
|
||||
m = re.search(r'"fan_id"\s*:\s*(\d+)', r.text)
|
||||
if m:
|
||||
return int(m.group(1))
|
||||
m = re.search(r'data-fan-id="(\d+)"', r.text)
|
||||
if m:
|
||||
return int(m.group(1))
|
||||
|
||||
raise RuntimeError("Could not determine fan_id — check credentials")
|
||||
|
||||
|
||||
# ── Collection ────────────────────────────────────────────────────────────────
|
||||
|
||||
def fetch_collection(fan_id: int) -> list[dict]:
|
||||
"""Fetch all purchased items from the fan collection."""
|
||||
items = []
|
||||
token = "9999999999:0:a::"
|
||||
|
||||
while True:
|
||||
time.sleep(REQUEST_DELAY)
|
||||
r = SESSION.post(
|
||||
"https://bandcamp.com/api/fancollection/1/collection_items",
|
||||
json={"fan_id": fan_id, "older_than_token": token, "count": COLLECTION_BATCH},
|
||||
timeout=15,
|
||||
)
|
||||
r.raise_for_status()
|
||||
data = r.json()
|
||||
batch = data.get("items") or data.get("redownload_urls") or []
|
||||
items.extend(batch)
|
||||
if not data.get("more_available") or not batch:
|
||||
break
|
||||
token = data.get("last_token", "")
|
||||
if not token:
|
||||
break
|
||||
|
||||
return items
|
||||
|
||||
|
||||
# ── Local library index ───────────────────────────────────────────────────────
|
||||
|
||||
def _norm(s: str) -> str:
|
||||
"""Normalize for loose matching (lowercase, strip punctuation/spaces)."""
|
||||
return re.sub(r"[^a-z0-9]", "", s.lower())
|
||||
|
||||
|
||||
def index_local(music_dir: Path) -> set[str]:
|
||||
"""Return a set of normalized 'artist||album' keys for existing folders."""
|
||||
keys = set()
|
||||
for entry in music_dir.iterdir():
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
name = entry.name
|
||||
if " - " in name:
|
||||
artist, album = name.split(" - ", 1)
|
||||
keys.add(f"{_norm(artist)}||{_norm(album)}")
|
||||
else:
|
||||
keys.add(f"||{_norm(name)}")
|
||||
return keys
|
||||
|
||||
|
||||
def item_key(item: dict) -> str:
|
||||
artist = item.get("band_name", "")
|
||||
album = item.get("album_title", "") or item.get("item_title", "")
|
||||
return f"{_norm(artist)}||{_norm(album)}"
|
||||
|
||||
|
||||
def item_label(item: dict) -> str:
|
||||
artist = item.get("band_name", "")
|
||||
album = item.get("album_title", "") or item.get("item_title", "")
|
||||
return f"{artist} - {album}"
|
||||
|
||||
|
||||
# ── Download ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def _parse_blob(html_text: str) -> dict:
|
||||
"""Extract and parse the data-blob JSON from a Bandcamp page."""
|
||||
m = re.search(r'data-blob="([^"]+)"', html_text)
|
||||
if not m:
|
||||
m = re.search(r"data-blob='([^']+)'", html_text)
|
||||
if not m:
|
||||
raise RuntimeError("data-blob not found on download page")
|
||||
return json.loads(html.unescape(m.group(1)))
|
||||
|
||||
|
||||
def get_wav_url(sale_item_id: int) -> str | None:
|
||||
"""Return the WAV download URL for a purchase, or None if unavailable."""
|
||||
time.sleep(REQUEST_DELAY)
|
||||
r = SESSION.get(
|
||||
"https://bandcamp.com/download",
|
||||
params={
|
||||
"from": "collection",
|
||||
"payment_id": sale_item_id,
|
||||
"stp": "gen",
|
||||
"type": "album",
|
||||
},
|
||||
timeout=15,
|
||||
)
|
||||
if r.status_code != 200:
|
||||
return None
|
||||
|
||||
try:
|
||||
blob = _parse_blob(r.text)
|
||||
except Exception as exc:
|
||||
print(f" ✗ Could not parse data-blob: {exc}")
|
||||
return None
|
||||
|
||||
for dl_item in blob.get("download_items", []):
|
||||
downloads = dl_item.get("downloads", {})
|
||||
for fmt in ("wav", "WAV"):
|
||||
entry = downloads.get(fmt)
|
||||
if entry and entry.get("url"):
|
||||
return entry["url"]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def download_and_extract(url: str, dest_dir: Path, label: str) -> Path | None:
|
||||
"""Download a ZIP from url, extract to dest_dir/<Artist - Album>/."""
|
||||
tmp_zip = dest_dir / f"_download_{os.getpid()}.zip"
|
||||
tmp_extract = dest_dir / f"_extract_{os.getpid()}"
|
||||
|
||||
try:
|
||||
print(f" Downloading {label}…", end="", flush=True)
|
||||
with SESSION.get(url, stream=True, timeout=120, allow_redirects=True) as r:
|
||||
r.raise_for_status()
|
||||
total = int(r.headers.get("Content-Length", 0))
|
||||
downloaded = 0
|
||||
with open(tmp_zip, "wb") as f:
|
||||
for chunk in r.iter_content(chunk_size=1024 * 256):
|
||||
f.write(chunk)
|
||||
downloaded += len(chunk)
|
||||
print(f" {downloaded // 1024 // 1024} MB")
|
||||
|
||||
tmp_extract.mkdir(parents=True, exist_ok=True)
|
||||
with zipfile.ZipFile(tmp_zip) as z:
|
||||
z.extractall(tmp_extract)
|
||||
|
||||
# Find the album folder inside the extracted ZIP
|
||||
subdirs = [p for p in tmp_extract.iterdir() if p.is_dir()]
|
||||
if subdirs:
|
||||
album_src = subdirs[0]
|
||||
else:
|
||||
album_src = tmp_extract # files in root of ZIP
|
||||
|
||||
# Build destination folder name from the label
|
||||
safe_label = re.sub(r'[<>:"/\\|?*]', "-", label)
|
||||
final_dest = dest_dir / safe_label
|
||||
|
||||
if final_dest.exists():
|
||||
shutil.rmtree(final_dest)
|
||||
shutil.move(str(album_src), str(final_dest))
|
||||
return final_dest
|
||||
|
||||
except Exception as exc:
|
||||
print(f" ✗ Download failed: {exc}")
|
||||
return None
|
||||
finally:
|
||||
if tmp_zip.exists():
|
||||
tmp_zip.unlink()
|
||||
if tmp_extract.exists():
|
||||
shutil.rmtree(tmp_extract, ignore_errors=True)
|
||||
|
||||
|
||||
def run_tagger(album_dir: Path):
|
||||
"""Run bandcamp-tag.py on a single album folder."""
|
||||
if not TAGGER.exists():
|
||||
return
|
||||
print(f" Tagging…")
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(TAGGER), "--album", album_dir.name, str(album_dir.parent)],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
for line in result.stdout.strip().splitlines():
|
||||
print(f" {line}")
|
||||
if result.returncode != 0:
|
||||
for line in result.stderr.strip().splitlines():
|
||||
print(f" ERR {line}")
|
||||
|
||||
|
||||
# ── Main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Sync Bandcamp purchases to local library")
|
||||
parser.add_argument("music_dir", nargs="?", type=Path, default=DEFAULT_MUSIC_DIR)
|
||||
parser.add_argument("--dry-run", action="store_true", help="Show plan only, no downloads")
|
||||
parser.add_argument("--list", action="store_true", help="List collection, exit")
|
||||
parser.add_argument("--no-tag", action="store_true", help="Skip bandcamp-tag.py after download")
|
||||
args = parser.parse_args()
|
||||
|
||||
music_dir = args.music_dir
|
||||
|
||||
# Load credentials
|
||||
email, password = load_credentials()
|
||||
if not email or not password:
|
||||
print("✗ No credentials found. Save them in the Settings → Bandcamp section, or set")
|
||||
print(" BANDCAMP_EMAIL / BANDCAMP_PASSWORD environment variables.")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Logging in as {email}…")
|
||||
account = bc_login(email, password)
|
||||
fan_id = get_fan_id(account)
|
||||
username = account.get("username", str(fan_id))
|
||||
print(f"✓ Logged in — {username} (fan_id={fan_id})")
|
||||
|
||||
print("Fetching collection…")
|
||||
items = fetch_collection(fan_id)
|
||||
print(f" {len(items)} purchases found")
|
||||
|
||||
if args.list:
|
||||
for item in items:
|
||||
print(f" {item_label(item)}")
|
||||
return
|
||||
|
||||
if not music_dir.exists():
|
||||
print(f"✗ Music directory not found: {music_dir}")
|
||||
sys.exit(1)
|
||||
|
||||
local = index_local(music_dir)
|
||||
print(f" {len(local)} albums already in {music_dir}")
|
||||
|
||||
# Only consider album purchases (not singles / merch)
|
||||
album_items = [i for i in items if i.get("item_type") in ("album", None)]
|
||||
new_items = [i for i in album_items if item_key(i) not in local]
|
||||
|
||||
if not new_items:
|
||||
print("✓ Library is up to date — nothing to download")
|
||||
return
|
||||
|
||||
print(f"\n{len(new_items)} new album(s) to download:")
|
||||
for item in new_items:
|
||||
print(f" • {item_label(item)}")
|
||||
|
||||
if args.dry_run:
|
||||
return
|
||||
|
||||
print()
|
||||
ok = 0
|
||||
for item in new_items:
|
||||
label = item_label(item)
|
||||
sale_id = item.get("sale_item_id") or item.get("payment_id")
|
||||
if not sale_id:
|
||||
print(f" [{label}] — no sale_item_id, skipping")
|
||||
continue
|
||||
|
||||
print(f" [{label}]")
|
||||
wav_url = get_wav_url(int(sale_id))
|
||||
if not wav_url:
|
||||
print(f" ✗ No WAV download available (check Bandcamp — may need to choose format)")
|
||||
continue
|
||||
|
||||
album_dir = download_and_extract(wav_url, music_dir, label)
|
||||
if album_dir is None:
|
||||
continue
|
||||
|
||||
if not args.no_tag and TAGGER.exists():
|
||||
run_tagger(album_dir)
|
||||
|
||||
ok += 1
|
||||
|
||||
print(f"\n✓ {ok}/{len(new_items)} album(s) downloaded")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,597 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
bandcamp-tag — Tag Bandcamp WAV downloads using Bandcamp metadata.
|
||||
|
||||
For each 'Artist - Album' folder in MUSIC_DIR:
|
||||
1. Checks which ID3 tags are missing (year, genre, track total, art)
|
||||
2. Searches Bandcamp for the release (or uses --url)
|
||||
3. Fetches TralbumData JSON from the Bandcamp album page
|
||||
4. Writes missing tags and embeds album art using mutagen
|
||||
|
||||
Usage:
|
||||
bandcamp-tag.py [MUSIC_DIR] # scan all albums
|
||||
bandcamp-tag.py --album "Artist - Title" # one album
|
||||
bandcamp-tag.py --album "..." --url https://artist.bandcamp.com/album/slug
|
||||
bandcamp-tag.py --dry-run # show plan without writing
|
||||
bandcamp-tag.py --force # rewrite tags even if present
|
||||
bandcamp-tag.py --scan # just list what needs updating
|
||||
|
||||
Dependencies: mutagen, requests (both already installed)
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
from mutagen.id3 import (
|
||||
APIC, TALB, TCON, TDRC, TIT2, TPOS, TPE1, TPE2, TRCK,
|
||||
)
|
||||
from mutagen.wave import WAVE
|
||||
|
||||
# ── Config ────────────────────────────────────────────────────────────────────
|
||||
|
||||
DEFAULT_MUSIC_DIR = Path('/mnt/local/music')
|
||||
REQUEST_DELAY = 0.8 # seconds between Bandcamp requests
|
||||
ART_MAX_BYTES = 3 * 1024 * 1024 # skip art larger than 3 MB
|
||||
|
||||
SESSION = requests.Session()
|
||||
SESSION.headers['User-Agent'] = (
|
||||
'Mozilla/5.0 (X11; Linux aarch64) AppleWebKit/537.36 '
|
||||
'(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||||
)
|
||||
|
||||
# ── Bandcamp fetch ────────────────────────────────────────────────────────────
|
||||
|
||||
def _bc_slug(s: str) -> str:
|
||||
"""Convert a string to a Bandcamp-style URL slug."""
|
||||
s = s.lower().strip()
|
||||
s = re.sub(r'[^a-z0-9 -]', '', s)
|
||||
s = re.sub(r'\s+', '-', s)
|
||||
return s.strip('-')
|
||||
|
||||
|
||||
def _bc_subdomains(artist: str) -> list[str]:
|
||||
"""Generate plausible Bandcamp subdomain candidates for an artist name."""
|
||||
base = re.sub(r'[^a-z0-9]', '', artist.lower()) # no-spaces, no-hyphens
|
||||
hyph = _bc_slug(artist) # with hyphens
|
||||
candidates = []
|
||||
for v in (base, hyph, base + 'music', base + 'dnb',
|
||||
hyph + 'music', base.replace('-', '')):
|
||||
if v and v not in candidates:
|
||||
candidates.append(v)
|
||||
return candidates
|
||||
|
||||
|
||||
def bandcamp_search(artist: str, album: str) -> str | None:
|
||||
"""
|
||||
Return the URL of the best Bandcamp album match, or None.
|
||||
|
||||
Strategy: generate plausible subdomain + album-slug combinations and
|
||||
probe them with HEAD requests (fast, no JS rendering needed).
|
||||
"""
|
||||
album_slug = _bc_slug(album)
|
||||
subdomains = _bc_subdomains(artist)
|
||||
|
||||
for sub in subdomains:
|
||||
url = f'https://{sub}.bandcamp.com/album/{album_slug}'
|
||||
try:
|
||||
r = SESSION.head(url, timeout=8, allow_redirects=True)
|
||||
if r.status_code == 200:
|
||||
return str(r.url)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def fetch_tralbum(page_url: str) -> dict | None:
|
||||
"""Fetch and parse TralbumData JSON from a Bandcamp album page."""
|
||||
try:
|
||||
r = SESSION.get(page_url, timeout=20)
|
||||
r.raise_for_status()
|
||||
except Exception as e:
|
||||
print(f' [fetch error] {e}')
|
||||
return None
|
||||
|
||||
text = r.text
|
||||
data = None
|
||||
|
||||
# Primary: data-tralbum attribute (HTML-entity-encoded JSON)
|
||||
m = re.search(r'data-tralbum="([^"]*)"', text)
|
||||
if m:
|
||||
try:
|
||||
data = json.loads(html.unescape(m.group(1)))
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
# Fallback: inline TralbumData variable
|
||||
if data is None:
|
||||
m = re.search(r'TralbumData\s*=\s*(\{.*?\});\s*(?:\n|$)', text, re.DOTALL)
|
||||
if m:
|
||||
try:
|
||||
data = json.loads(m.group(1))
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
# Fallback: JSON-LD
|
||||
if data is None:
|
||||
for m in re.finditer(
|
||||
r'<script[^>]+type="application/ld\+json"[^>]*>(.*?)</script>',
|
||||
text, re.DOTALL,
|
||||
):
|
||||
try:
|
||||
ld = json.loads(m.group(1))
|
||||
if ld.get('@type') == 'MusicAlbum':
|
||||
data = _ld_to_tralbum(ld)
|
||||
break
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
pass
|
||||
|
||||
if data is None:
|
||||
return None
|
||||
|
||||
# If tags are missing from TralbumData, scrape them from the page HTML
|
||||
if not (data.get('tags') or (data.get('current') or {}).get('tags')):
|
||||
page_tags = re.findall(r'class="tag"[^>]*>([^<]+)<', text)
|
||||
if page_tags:
|
||||
data['tags'] = [t.strip() for t in page_tags if t.strip()]
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def _ld_to_tralbum(ld: dict) -> dict:
|
||||
"""Convert JSON-LD MusicAlbum to a minimal TralbumData-style dict."""
|
||||
tracks = []
|
||||
for item in (ld.get('track') or {}).get('itemListElement') or []:
|
||||
rec = item.get('item', {})
|
||||
tracks.append({
|
||||
'title': rec.get('name', ''),
|
||||
'track_num': item.get('position', len(tracks) + 1),
|
||||
'duration': None,
|
||||
'disc_number': 1,
|
||||
})
|
||||
pub = ld.get('datePublished', '')
|
||||
return {
|
||||
'title': ld.get('name', ''),
|
||||
'artist': (ld.get('byArtist') or {}).get('name', ''),
|
||||
'art_id': None,
|
||||
'current': {
|
||||
'publish_date': pub,
|
||||
'tags': [],
|
||||
},
|
||||
'tags': [],
|
||||
'trackinfo': tracks,
|
||||
}
|
||||
|
||||
|
||||
def fetch_art(art_id: int) -> tuple[bytes | None, str]:
|
||||
"""Download album art from Bandcamp CDN. Returns (bytes, mime) or (None, '')."""
|
||||
for size in ('10', '0'): # 10 = 1200px max, 0 = original
|
||||
url = f'https://f4.bcbits.com/img/a{art_id}_{size}.jpg'
|
||||
try:
|
||||
r = SESSION.get(url, timeout=30, stream=True)
|
||||
if not r.ok:
|
||||
continue
|
||||
mime = r.headers.get('content-type', 'image/jpeg').split(';')[0].strip()
|
||||
if not mime.startswith('image/'):
|
||||
continue
|
||||
data = r.content
|
||||
if len(data) > ART_MAX_BYTES:
|
||||
continue
|
||||
if len(data) > 1000:
|
||||
return data, mime
|
||||
except Exception:
|
||||
pass
|
||||
return None, ''
|
||||
|
||||
|
||||
# ── NFO helper ────────────────────────────────────────────────────────────────
|
||||
|
||||
def parse_nfo(nfo_path: Path) -> dict:
|
||||
"""Parse a Kodi/Jellyfin album.nfo into a minimal TralbumData-style dict."""
|
||||
try:
|
||||
tree = ET.parse(nfo_path)
|
||||
root = tree.getroot()
|
||||
|
||||
def text(tag):
|
||||
el = root.find(tag)
|
||||
return el.text.strip() if el is not None and el.text else ''
|
||||
|
||||
tracks = []
|
||||
for t in root.findall('track'):
|
||||
pos_el = t.find('position')
|
||||
title_el = t.find('title')
|
||||
pos = int(pos_el.text) if pos_el is not None and pos_el.text else len(tracks) + 1
|
||||
title = title_el.text.strip() if title_el is not None and title_el.text else ''
|
||||
tracks.append({'title': title, 'track_num': pos, 'duration': None, 'disc_number': 1})
|
||||
|
||||
return {
|
||||
'title': text('title'),
|
||||
'artist': text('albumartist') or text('artist'),
|
||||
'art_id': None,
|
||||
'current': {'publish_date': text('year'), 'tags': [text('genre')] if text('genre') else []},
|
||||
'tags': [text('genre')] if text('genre') else [],
|
||||
'trackinfo': tracks,
|
||||
}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
# ── File helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
def parse_folder_name(name: str) -> tuple[str, str] | tuple[None, None]:
|
||||
"""Split 'Artist - Album' folder name into (artist, album)."""
|
||||
parts = name.split(' - ', 1)
|
||||
if len(parts) == 2:
|
||||
return parts[0].strip(), parts[1].strip()
|
||||
return None, None
|
||||
|
||||
|
||||
def parse_track_filename(stem: str, folder_name: str) -> tuple[int | None, str]:
|
||||
"""
|
||||
Return (track_number, title_guess) from a WAV filename stem.
|
||||
Input examples:
|
||||
'ALIX PEREZ - Gloom EP - 01 Psychosis'
|
||||
'01 Psychosis'
|
||||
"""
|
||||
# Strip leading 'Artist - Album - '
|
||||
prefix = folder_name + ' - '
|
||||
if stem.startswith(prefix):
|
||||
stem = stem[len(prefix):]
|
||||
|
||||
m = re.match(r'^(\d{1,3})\s+(.+)$', stem)
|
||||
if m:
|
||||
return int(m.group(1)), m.group(2).strip()
|
||||
return None, stem.strip()
|
||||
|
||||
|
||||
def tag_completeness(wav_path: Path) -> dict:
|
||||
"""Return dict of missing tag names for a WAV file."""
|
||||
missing = {}
|
||||
try:
|
||||
f = WAVE(wav_path)
|
||||
tags = f.tags or {}
|
||||
|
||||
def has(prefix):
|
||||
return any(k.startswith(prefix) for k in tags.keys())
|
||||
|
||||
def val(key):
|
||||
v = tags.get(key)
|
||||
return str(v[0]) if v else ''
|
||||
|
||||
if not val('TIT2'):
|
||||
missing['title'] = True
|
||||
if not val('TALB'):
|
||||
missing['album'] = True
|
||||
if not val('TPE1'):
|
||||
missing['artist'] = True
|
||||
if not val('TDRC'):
|
||||
missing['year'] = True
|
||||
if not val('TCON'):
|
||||
missing['genre'] = True
|
||||
if not has('APIC'):
|
||||
missing['art'] = True
|
||||
trck = val('TRCK')
|
||||
if trck and '/' not in trck:
|
||||
missing['track_total'] = True
|
||||
except Exception:
|
||||
missing['read_error'] = True
|
||||
return missing
|
||||
|
||||
|
||||
def load_folder_art(folder: Path) -> tuple[bytes | None, str]:
|
||||
"""Try to load cover art from a folder."""
|
||||
candidates = ['cover.jpg', 'cover.jpeg', 'cover.png', 'folder.jpg', 'folder.png']
|
||||
for name in candidates:
|
||||
p = folder / name
|
||||
if p.exists():
|
||||
try:
|
||||
data = p.read_bytes()
|
||||
mime = 'image/jpeg' if name.endswith(('.jpg', '.jpeg')) else 'image/png'
|
||||
return data, mime
|
||||
except Exception:
|
||||
pass
|
||||
return None, ''
|
||||
|
||||
|
||||
def write_tags(
|
||||
wav_path: Path,
|
||||
bc_track: dict,
|
||||
bc_album: dict,
|
||||
art_bytes: bytes | None,
|
||||
art_mime: str,
|
||||
track_total: int,
|
||||
missing: dict,
|
||||
dry_run: bool,
|
||||
force: bool,
|
||||
) -> None:
|
||||
title = bc_track.get('title', '')
|
||||
track_num = bc_track.get('track_num') or 1
|
||||
disc_num = bc_track.get('disc_number') or 1
|
||||
|
||||
current = bc_album.get('current') or {}
|
||||
album_str = current.get('title') or bc_album.get('title', '')
|
||||
artist = bc_track.get('artist') or bc_album.get('artist', '')
|
||||
tags_list = current.get('tags') or bc_album.get('tags') or []
|
||||
genre = tags_list[0] if tags_list else ''
|
||||
|
||||
pub = current.get('publish_date', '')
|
||||
year = re.search(r'\d{4}', pub).group(0) if re.search(r'\d{4}', pub) else ''
|
||||
|
||||
if dry_run:
|
||||
updates = []
|
||||
if force or missing.get('title'): updates.append(f'title={title!r}')
|
||||
if force or missing.get('album'): updates.append(f'album={album_str!r}')
|
||||
if force or missing.get('artist'): updates.append(f'artist={artist!r}')
|
||||
if force or missing.get('year'): updates.append(f'year={year!r}')
|
||||
if force or missing.get('genre'): updates.append(f'genre={genre!r}')
|
||||
if force or missing.get('track_total'): updates.append(f'trck={track_num}/{track_total}')
|
||||
if (force or missing.get('art')) and art_bytes: updates.append('art')
|
||||
print(f' [dry] {wav_path.name}: {", ".join(updates) or "no changes"}')
|
||||
return
|
||||
|
||||
try:
|
||||
f = WAVE(wav_path)
|
||||
if f.tags is None:
|
||||
f.add_tags()
|
||||
t = f.tags
|
||||
|
||||
def set_tag(key, cls, **kw):
|
||||
for k in list(t.keys()):
|
||||
if k == key or k.startswith(key + ':'):
|
||||
del t[k]
|
||||
t[key] = cls(**kw)
|
||||
|
||||
if force or missing.get('title') and title:
|
||||
set_tag('TIT2', TIT2, text=[title])
|
||||
if force or missing.get('album') and album_str:
|
||||
set_tag('TALB', TALB, text=[album_str])
|
||||
if force or missing.get('artist') and artist:
|
||||
set_tag('TPE1', TPE1, text=[artist])
|
||||
set_tag('TPE2', TPE2, text=[bc_album.get('artist', artist)])
|
||||
if (force or missing.get('year')) and year:
|
||||
set_tag('TDRC', TDRC, text=[year])
|
||||
if (force or missing.get('genre')) and genre:
|
||||
set_tag('TCON', TCON, text=[genre])
|
||||
if force or missing.get('track_total'):
|
||||
set_tag('TRCK', TRCK, text=[f'{track_num}/{track_total}'])
|
||||
if disc_num > 1:
|
||||
set_tag('TPOS', TPOS, text=[str(disc_num)])
|
||||
if (force or missing.get('art')) and art_bytes:
|
||||
for k in list(t.keys()):
|
||||
if k.startswith('APIC'):
|
||||
del t[k]
|
||||
t['APIC:cover'] = APIC(
|
||||
mime=art_mime, type=3, desc='cover', data=art_bytes,
|
||||
)
|
||||
|
||||
f.save()
|
||||
print(f' OK {wav_path.name}')
|
||||
except Exception as e:
|
||||
print(f' ERR {wav_path.name}: {e}')
|
||||
|
||||
|
||||
# ── Title similarity (for fuzzy track matching) ───────────────────────────────
|
||||
|
||||
def _jaccard(a: str, b: str) -> float:
|
||||
wa = set(re.findall(r'\w+', a.lower()))
|
||||
wb = set(re.findall(r'\w+', b.lower()))
|
||||
if not wa or not wb:
|
||||
return 0.0
|
||||
return len(wa & wb) / len(wa | wb)
|
||||
|
||||
|
||||
# ── Album processing ──────────────────────────────────────────────────────────
|
||||
|
||||
def process_album(
|
||||
folder: Path,
|
||||
url: str | None = None,
|
||||
force: bool = False,
|
||||
dry_run: bool = False,
|
||||
scan_only: bool = False,
|
||||
) -> bool:
|
||||
"""Tag all WAV files in one album folder. Returns True if any work was done."""
|
||||
wav_files = sorted(folder.glob('*.wav'))
|
||||
if not wav_files:
|
||||
return False
|
||||
|
||||
folder_name = folder.name
|
||||
artist, album = parse_folder_name(folder_name)
|
||||
if not artist:
|
||||
print(f'SKIP {folder_name} (can\'t parse artist/album)')
|
||||
return False
|
||||
|
||||
# Check which tags are missing across all WAVs
|
||||
all_missing: dict[str, bool] = {}
|
||||
for wf in wav_files:
|
||||
for k, v in tag_completeness(wf).items():
|
||||
if v:
|
||||
all_missing[k] = True
|
||||
|
||||
if not force and not all_missing:
|
||||
if not scan_only:
|
||||
print(f'OK {folder_name}')
|
||||
return False
|
||||
|
||||
missing_str = ', '.join(sorted(all_missing)) or 'none'
|
||||
print(f'NEED {folder_name} [{missing_str}]')
|
||||
|
||||
if scan_only:
|
||||
return True
|
||||
|
||||
# ── Load metadata source ──────────────────────────────────────────────────
|
||||
|
||||
bc_data: dict | None = None
|
||||
|
||||
# Try .nfo first (no network needed for title/genre/tracks)
|
||||
nfo_path = folder / 'album.nfo'
|
||||
if nfo_path.exists():
|
||||
nfo = parse_nfo(nfo_path)
|
||||
if nfo.get('trackinfo'):
|
||||
print(f' Using .nfo: {nfo_path.name}')
|
||||
bc_data = nfo
|
||||
|
||||
# Fetch from Bandcamp if still missing year, genre, or we have no data yet
|
||||
still_need_bc = (
|
||||
bc_data is None
|
||||
or (force or all_missing.get('year')) and not (bc_data.get('current') or {}).get('publish_date', '').strip()
|
||||
or (force or all_missing.get('genre')) and not ((bc_data.get('current') or {}).get('tags') or bc_data.get('tags'))
|
||||
or all_missing.get('art') and bc_data.get('art_id') is None
|
||||
)
|
||||
|
||||
if still_need_bc:
|
||||
if not url:
|
||||
print(f' Searching Bandcamp for {artist!r} — {album!r} …')
|
||||
url = bandcamp_search(artist, album)
|
||||
time.sleep(REQUEST_DELAY)
|
||||
if not url:
|
||||
print(f' No Bandcamp result. Use --url to provide one.')
|
||||
if bc_data is None:
|
||||
return False
|
||||
else:
|
||||
print(f' Found: {url}')
|
||||
|
||||
if url:
|
||||
fetched = fetch_tralbum(url)
|
||||
time.sleep(REQUEST_DELAY)
|
||||
if fetched:
|
||||
# Merge: prefer Bandcamp data but keep .nfo track list if Bandcamp has none
|
||||
if bc_data and bc_data.get('trackinfo') and not fetched.get('trackinfo'):
|
||||
fetched['trackinfo'] = bc_data['trackinfo']
|
||||
bc_data = fetched
|
||||
elif bc_data is None:
|
||||
print(f' Could not fetch album data.')
|
||||
return False
|
||||
|
||||
if bc_data is None:
|
||||
return False
|
||||
|
||||
# ── Album art ─────────────────────────────────────────────────────────────
|
||||
|
||||
art_bytes: bytes | None = None
|
||||
art_mime = 'image/jpeg'
|
||||
|
||||
if force or all_missing.get('art'):
|
||||
art_id = bc_data.get('art_id') or (bc_data.get('current') or {}).get('art_id')
|
||||
if art_id:
|
||||
print(f' Downloading art (id={art_id}) …')
|
||||
art_bytes, art_mime = fetch_art(art_id)
|
||||
time.sleep(REQUEST_DELAY)
|
||||
if art_bytes:
|
||||
print(f' Art: {len(art_bytes) // 1024} KB ({art_mime})')
|
||||
else:
|
||||
print(f' Art download failed, falling back to folder art')
|
||||
|
||||
if not art_bytes:
|
||||
art_bytes, art_mime = load_folder_art(folder)
|
||||
if art_bytes:
|
||||
print(f' Using folder art ({art_mime}, {len(art_bytes)//1024} KB)')
|
||||
|
||||
# ── Build track map ───────────────────────────────────────────────────────
|
||||
|
||||
bc_tracks_by_num: dict[int, dict] = {}
|
||||
for t in bc_data.get('trackinfo') or []:
|
||||
n = t.get('track_num')
|
||||
if n:
|
||||
bc_tracks_by_num[n] = t
|
||||
|
||||
total = len(bc_data.get('trackinfo') or []) or len(wav_files)
|
||||
|
||||
# ── Tag each file ─────────────────────────────────────────────────────────
|
||||
|
||||
for wav_file in wav_files:
|
||||
stem = wav_file.stem
|
||||
track_num, title_guess = parse_track_filename(stem, folder_name)
|
||||
file_missing = tag_completeness(wav_file)
|
||||
|
||||
if not force and not file_missing:
|
||||
print(f' -- {wav_file.name} (complete)')
|
||||
continue
|
||||
|
||||
# Match to Bandcamp track
|
||||
if track_num and track_num in bc_tracks_by_num:
|
||||
bc_track = bc_tracks_by_num[track_num]
|
||||
elif bc_tracks_by_num:
|
||||
# Fuzzy match by title
|
||||
best = max(
|
||||
bc_tracks_by_num.values(),
|
||||
key=lambda t: _jaccard(t.get('title', ''), title_guess),
|
||||
)
|
||||
bc_track = best
|
||||
else:
|
||||
bc_track = {'title': title_guess, 'track_num': track_num or 1, 'disc_number': 1}
|
||||
|
||||
write_tags(
|
||||
wav_file, bc_track, bc_data,
|
||||
art_bytes, art_mime, total,
|
||||
file_missing, dry_run, force,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ── Main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(
|
||||
description='Tag Bandcamp WAV downloads with metadata and album art.',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
)
|
||||
ap.add_argument(
|
||||
'music_dir', nargs='?', default=str(DEFAULT_MUSIC_DIR),
|
||||
help=f'Music directory to scan (default: {DEFAULT_MUSIC_DIR})',
|
||||
)
|
||||
ap.add_argument('--album', metavar='NAME',
|
||||
help='Process only this folder name (relative to MUSIC_DIR or absolute)')
|
||||
ap.add_argument('--url', metavar='URL',
|
||||
help='Bandcamp album URL (use with --album to skip search)')
|
||||
ap.add_argument('--force', action='store_true',
|
||||
help='Rewrite all tags even if already present')
|
||||
ap.add_argument('--dry-run', action='store_true',
|
||||
help='Show planned changes without writing anything')
|
||||
ap.add_argument('--scan', action='store_true',
|
||||
help='List albums that need updating and exit')
|
||||
args = ap.parse_args()
|
||||
|
||||
music_dir = Path(args.music_dir)
|
||||
if not music_dir.is_dir():
|
||||
sys.exit(f'Error: {music_dir} is not a directory')
|
||||
|
||||
if args.album:
|
||||
target = Path(args.album)
|
||||
if not target.is_absolute():
|
||||
target = music_dir / args.album
|
||||
if not target.is_dir():
|
||||
sys.exit(f'Error: {target} is not a directory')
|
||||
process_album(target, url=args.url, force=args.force,
|
||||
dry_run=args.dry_run, scan_only=args.scan)
|
||||
return
|
||||
|
||||
folders = sorted(d for d in music_dir.iterdir() if d.is_dir())
|
||||
wav_folders = [d for d in folders if any(d.glob('*.wav'))]
|
||||
print(f'Scanning {len(wav_folders)} album folders in {music_dir}\n')
|
||||
|
||||
needs_work = 0
|
||||
for folder in wav_folders:
|
||||
worked = process_album(folder, force=args.force,
|
||||
dry_run=args.dry_run, scan_only=args.scan)
|
||||
if worked:
|
||||
needs_work += 1
|
||||
if not args.scan:
|
||||
time.sleep(0.3)
|
||||
|
||||
if args.scan:
|
||||
print(f'\n{needs_work} of {len(wav_folders)} albums need updating.')
|
||||
else:
|
||||
print(f'\nDone. {needs_work} albums processed.')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user