"""Upload reviewed songs to the NK server and create their catalog rows. Reads songs_manifest.csv (from scan_songs.py, after you fixed it in Excel), uploads each included file to the `audio` bucket and creates the album, singers and track. Safe to run again: songs already imported are skipped. Usage: python tools/import_songs.py --dry-run # check the CSV, upload nothing python tools/import_songs.py # import as pending (approve in Studio) python tools/import_songs.py --approve # import straight to approved Environment (from the server's .env, never put these in the app): NK_URL e.g. https://app.nkbmusic.com NK_SERVICE_KEY SERVICE_ROLE_KEY from the server .env Only the standard library is used. """ import argparse import csv import json import os import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path OCCASIONS = {"huttu", "habba", "madhuvae", "devotional", "folk", "saavu", "instrumental"} MIME = {".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".aac": "audio/aac"} SINGER_SPLIT = re.compile(r"\s*(?:,|&&|&|/|\band\b)\s*", re.I) class Api: def __init__(self, url, key): self.url = url.rstrip("/") self.key = key def _call(self, method, path, body=None, headers=None, raw=None): h = {"apikey": self.key, "Authorization": f"Bearer {self.key}"} h.update(headers or {}) data = raw if body is not None: data = json.dumps(body).encode() h["Content-Type"] = "application/json" req = urllib.request.Request(self.url + path, data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=300) as r: text = r.read().decode() return json.loads(text) if text else None except urllib.error.HTTPError as e: raise RuntimeError(f"{method} {path} -> {e.code}: {e.read().decode()[:300]}") from None def select(self, table, query): return self._call("GET", f"/rest/v1/{table}?{query}") def insert(self, table, row): return self._call("POST", f"/rest/v1/{table}", row, {"Prefer": "return=representation"})[0] def upload(self, key, path): mime = MIME[path.suffix.lower()] self._call("POST", f"/storage/v1/object/audio/{urllib.parse.quote(key)}", raw=path.read_bytes(), headers={"Content-Type": mime, "x-upsert": "true"}) def load_rows(manifest, songs_dir): rows, problems = [], [] with open(manifest, encoding="utf-8-sig") as f: for r in csv.DictReader(f): if r["include"].strip().lower() not in ("yes", "y", "1"): continue path = songs_dir / r["file"] occasion = (r["occasion"] or "folk").strip().lower() if not r["title"].strip(): problems.append(f"{r['id']}: no title ({r['file']})") elif occasion not in OCCASIONS: problems.append(f"{r['id']}: occasion '{occasion}' is not one of {sorted(OCCASIONS)}") elif path.suffix.lower() not in MIME: problems.append(f"{r['id']}: {path.suffix} must be converted to mp3 first") elif not path.is_file(): problems.append(f"{r['id']}: file not found {path}") else: r["occasion"] = occasion rows.append((r, path)) return rows, problems def main(): ap = argparse.ArgumentParser() ap.add_argument("--manifest", default="songs_manifest.csv") ap.add_argument("--songs", default="songs") ap.add_argument("--approve", action="store_true", help="publish straight away") ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() rows, problems = load_rows(args.manifest, Path(args.songs)) print(f"{len(rows)} songs ready, {len(problems)} skipped") for p in problems[:30]: print(" skip", p) if len(problems) > 30: print(f" ... and {len(problems) - 30} more") if args.dry_run or not rows: return url, key = os.environ.get("NK_URL"), os.environ.get("NK_SERVICE_KEY") if not url or not key: sys.exit("Set NK_URL and NK_SERVICE_KEY first (see the top of this file).") api = Api(url, key) # Re-runs: source_note starts with "import:". done = {t["source_note"].split()[0][7:] for t in api.select("tracks", "select=source_note&source_note=like.import:*")} artists = {a["name_latin"].lower(): a["id"] for a in api.select("artists", "select=id,name_latin")} albums = {a["title_latin"].lower(): a["id"] for a in api.select("albums", "select=id,title_latin")} def artist_id(name): if name.lower() not in artists: artists[name.lower()] = api.insert("artists", {"name_latin": name})["id"] return artists[name.lower()] def album_id(title): if not title: return None if title.lower() not in albums: albums[title.lower()] = api.insert("albums", {"title_latin": title})["id"] return albums[title.lower()] imported = failed = 0 for i, (r, path) in enumerate(rows, 1): if r["id"] in done: continue key_name = f"legacy/{r['id']}{path.suffix.lower()}" try: api.upload(key_name, path) track = api.insert("tracks", { "title_latin": r["title"].strip(), "album_id": album_id(r["album"].strip()), "occasion": r["occasion"], "year": int(r["year"]) if r["year"].strip().isdigit() else None, "duration_ms": int(float(r["duration_s"] or 0) * 1000) or None, "audio_key_128": key_name, "status": "approved" if args.approve else "pending", "rights_status": "legacy_unverified", "tier": "free", "source_note": f"import:{r['id']} {r['file']}", }) names = [n for n in SINGER_SPLIT.split(r["singers"].strip()) if n] for pos, name in enumerate(names): api.insert("track_artists", {"track_id": track["id"], "artist_id": artist_id(name), "role": "singer", "position": pos}) imported += 1 print(f"[{i}/{len(rows)}] {r['id']} {r['title']}") except Exception as e: failed += 1 print(f"[{i}/{len(rows)}] FAILED {r['id']}: {e}") print(f"Imported {imported}, failed {failed}, already there {len(done)}.") if not args.approve and imported: print("They are pending. Approve in Studio: update tracks set status='approved' where status='pending';") if __name__ == "__main__": main()