#!/usr/bin/env python3 """ import_legacy_credentials.py — one-time Stage 1/2 migration step. Reads every existing plaintext credential from its legacy scattered location and populates alembic's encrypted credential store via credential_service, which immediately re-renders everything fresh under $ALEMBIC_CONFIG_DIR/pipeline/. Never writes to or modifies the legacy files -- read-only against them. Idempotent (safe to re-run; each credential is just an upsert). Run playlist_service.seed_legacy() BEFORE this script, not after: the soulseek/spotify credential render patches "every existing .conf" in $ALEMBIC_CONFIG_DIR/pipeline/ -- if no .conf files exist yet (no playlists seeded), that render is a silent no-op. Expects the legacy paths bind-mounted read-only at the /legacy/* paths below -- see docs/MIGRATION.md for the one-off `docker run` invocation. """ import re import sys from pathlib import Path sys.path.insert(0, "/app") from app.db import SessionLocal, init_db # noqa: E402 from app.services import credential_service # noqa: E402 LEGACY_SLDL = Path("/legacy/opt-sldl") LEGACY_BANDCAMP = Path("/legacy/bandcampconfig") LEGACY_AZURACAST = Path("/legacy/azuracastconfig") LEGACY_QOBUZ = Path("/legacy/qobuzconfig") LEGACY_TELEGRAM = Path("/legacy/telegramconfig") def parse_env_file(path: Path) -> dict[str, str]: values = {} if not path.exists(): return values for line in path.read_text().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") values[key.strip()] = value.strip().strip("'").strip('"') return values def import_spotify_and_soulseek(db) -> None: # Every playlist .conf carries the same spotify-id/spotify-secret and # user/pass (regen.sh templates them identically into each one) -- any # single one is a representative source. sample_conf = next(LEGACY_SLDL.glob("configs/*.conf"), None) if sample_conf is None or sample_conf.name == "_template.conf": sample_conf = LEGACY_SLDL / "configs" / "y2k.conf" if not sample_conf.exists(): print(f"[import] SKIP spotify/soulseek: no playlist .conf found at {sample_conf}") return body = sample_conf.read_text() def field(pattern): m = re.search(pattern, body, re.MULTILINE) return m.group(1).strip() if m else None cid = field(r"^spotify-id\s*=\s*(\S+)") csec = field(r"^spotify-secret\s*=\s*(\S+)") if cid and csec: credential_service.set_credentials(db, "spotify", {"client_id": cid, "client_secret": csec}) print("[import] spotify: OK") else: print("[import] SKIP spotify: fields not found") user = field(r"^user\s*=\s*(\S+)") pw = field(r"^pass\s*=\s*(\S+)") if user and pw: credential_service.set_credentials(db, "soulseek", {"username": user, "password": pw}) print("[import] soulseek: OK") else: print("[import] SKIP soulseek: fields not found") def import_navidrome(db) -> None: # Never lived in a file -- hardcoded across 8 legacy scripts. Extract # the password from one of them rather than re-transcribing it by hand. # base_url is deliberately NOT the old hardcoded LAN IP (192.168.0.72): # those scripts ran on the bare host, but alembic runs on the compose # network, so the container DNS name is the correct address now. script = LEGACY_SLDL / "lib" / "navidrome-scan.sh" if not script.exists(): print(f"[import] SKIP navidrome: {script} not found") return body = script.read_text() m = re.search(r"^ND_PASS=['\"]?([^'\"\n]+)['\"]?", body, re.MULTILINE) if not m: print("[import] SKIP navidrome: ND_PASS not found in navidrome-scan.sh") return credential_service.set_credentials( db, "navidrome", {"base_url": "http://navidrome:4533", "admin_user": "andrew", "admin_pass": m.group(1)}, ) print("[import] navidrome: OK") def import_bandcamp(db) -> None: config = parse_env_file(LEGACY_BANDCAMP / "config.env") cookies_path = LEGACY_BANDCAMP / "cookies.txt" values = {} if "BANDCAMP_USERNAME" in config: values["username"] = config["BANDCAMP_USERNAME"] if "BANDCAMP_FORMAT_PREF" in config: values["format_pref"] = config["BANDCAMP_FORMAT_PREF"] if cookies_path.exists(): values["cookies_txt"] = cookies_path.read_text() if values: credential_service.set_credentials(db, "bandcamp", values) print(f"[import] bandcamp: OK ({', '.join(k for k in values if k != 'cookies_txt')}" f"{', cookies_txt' if 'cookies_txt' in values else ''})") else: print("[import] SKIP bandcamp: nothing found") def import_azuracast(db) -> None: key_path = LEGACY_AZURACAST / "api_key" if not key_path.exists(): print(f"[import] SKIP azuracast: {key_path} not found") return credential_service.set_credentials(db, "azuracast", {"api_key": key_path.read_text().strip()}) print("[import] azuracast: OK") def import_qobuz(db) -> None: token_path = LEGACY_QOBUZ / "token" if not token_path.exists(): print(f"[import] SKIP qobuz: {token_path} not found") return values = {"token": token_path.read_text().strip()} # app_id/region are optional -- enrich-buy-url.py auto-scrapes app_id and # defaults region if these files don't exist (they didn't, in this deployment). for field, filename in (("app_id", "app_id"), ("region", "region")): p = LEGACY_QOBUZ / filename if p.exists(): values[field] = p.read_text().strip() credential_service.set_credentials(db, "qobuz", values) print(f"[import] qobuz: OK ({', '.join(values)})") def import_telegram(db) -> None: config = parse_env_file(LEGACY_TELEGRAM / "notify.env") values = {} if "TG_BOT_TOKEN" in config: values["bot_token"] = config["TG_BOT_TOKEN"] if "TG_CHAT_ID" in config: values["chat_id"] = config["TG_CHAT_ID"] if values: credential_service.set_credentials(db, "telegram", values) print("[import] telegram: OK") else: print("[import] SKIP telegram: nothing found") def main(): init_db() db = SessionLocal() try: import_spotify_and_soulseek(db) import_navidrome(db) import_bandcamp(db) import_azuracast(db) import_qobuz(db) import_telegram(db) finally: db.close() print("[import] done.") if __name__ == "__main__": main()