alembic/scripts/import_legacy_credentials.py

#!/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 <playlist>.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()