alembic/app/services/credential_service.py

import re
import time
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Secret
from app.security import crypto
from app.settings import settings

# Which fields exist per scope, and whether each is safe to display back to
# the UI once saved (short opaque strings need never be shown again).
SCOPE_FIELDS = {
    "spotify": ["client_id", "client_secret"],
    "soulseek": ["username", "password"],
    "navidrome": ["base_url", "admin_user", "admin_pass"],
    "bandcamp": ["username", "format_pref", "cookies_txt"],
    "azuracast": ["api_key"],
    "qobuz": ["token", "app_id", "region"],
    "telegram": ["bot_token", "chat_id"],
}


def _upsert(db: Session, scope: str, key: str, value: str) -> None:
    row = db.execute(
        select(Secret).where(Secret.scope == scope, Secret.key == key)
    ).scalar_one_or_none()
    encrypted = crypto.encrypt(value)
    if row is None:
        db.add(Secret(scope=scope, key=key, value_encrypted=encrypted, updated_at=time.time()))
    else:
        row.value_encrypted = encrypted
        row.updated_at = time.time()


def set_credential(db: Session, scope: str, key: str, value: str) -> None:
    """Set a single field and re-render immediately. For scopes with more
    than one field, prefer set_credentials() so the render sees every field
    at once instead of a transiently half-populated scope."""
    if scope not in SCOPE_FIELDS or key not in SCOPE_FIELDS[scope]:
        raise ValueError(f"unknown credential {scope}.{key}")
    _upsert(db, scope, key, value)
    db.commit()
    render_scope(db, scope)


def set_credentials(db: Session, scope: str, values: dict[str, str]) -> None:
    """Set every given field for a scope in one transaction, then render
    once — the way a UI form submission covering multiple fields should
    call this, rather than looping set_credential() per field."""
    if scope not in SCOPE_FIELDS:
        raise ValueError(f"unknown credential scope: {scope}")
    unknown = set(values) - set(SCOPE_FIELDS[scope])
    if unknown:
        raise ValueError(f"unknown fields for {scope}: {unknown}")
    for key, value in values.items():
        _upsert(db, scope, key, value)
    db.commit()
    render_scope(db, scope)


def get_credential(db: Session, scope: str, key: str) -> str | None:
    row = db.execute(
        select(Secret).where(Secret.scope == scope, Secret.key == key)
    ).scalar_one_or_none()
    return crypto.decrypt(row.value_encrypted) if row else None


def get_scope(db: Session, scope: str) -> dict[str, str]:
    rows = db.execute(select(Secret).where(Secret.scope == scope)).scalars()
    return {row.key: crypto.decrypt(row.value_encrypted) for row in rows}


def _patch_conf_field(path: Path, key: str, value: str) -> None:
    """Rewrite a single `key = value` line in an sldl .conf file, leaving
    every other line (including PLAYLIST_NAME/SPOTIFY_URL substitutions
    regen.sh already applied) untouched."""
    text = path.read_text()
    pattern = re.compile(rf"^{re.escape(key)}\s*=.*$", re.MULTILINE)
    new_line = f"{key} = {value}"
    if pattern.search(text):
        text = pattern.sub(new_line, text)
    else:
        text = text.rstrip("\n") + f"\n{new_line}\n"
    path.write_text(text)


def _every_playlist_conf() -> list[Path]:
    return sorted(settings.pipeline_config_dir.glob("*.conf"))


def _write_env_file(path: Path, values: dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [f"{k}='{v}'" for k, v in values.items()]
    path.write_text("\n".join(lines) + "\n")
    path.chmod(0o600)


def render_scope(db: Session, scope: str) -> None:
    values = get_scope(db, scope)
    if not values:
        return  # nothing saved yet for this scope — nothing to render

    if scope == "spotify":
        cid = values.get("client_id", "")
        csec = values.get("client_secret", "")
        for conf in _every_playlist_conf():
            _patch_conf_field(conf, "spotify-id", cid)
            _patch_conf_field(conf, "spotify-secret", csec)
        _write_env_file(
            settings.pipeline_config_dir / "_spotify.env",
            {"SPOTIFY_CLIENT_ID": cid, "SPOTIFY_CLIENT_SECRET": csec},
        )

    elif scope == "soulseek":
        user = values.get("username", "")
        pw = values.get("password", "")
        for conf in _every_playlist_conf():
            _patch_conf_field(conf, "user", user)
            _patch_conf_field(conf, "pass", pw)

    elif scope == "navidrome":
        _write_env_file(
            settings.pipeline_config_dir / "navidrome" / "admin.env",
            {
                "ND_BASE": values.get("base_url", "http://navidrome:4533"),
                "ND_USER": values.get("admin_user", "andrew"),
                "ND_PASS": values.get("admin_pass", ""),
            },
        )

    elif scope == "bandcamp":
        bandcamp_dir = settings.pipeline_config_dir / "bandcamp"
        bandcamp_dir.mkdir(parents=True, exist_ok=True)
        _write_env_file(
            bandcamp_dir / "config.env",
            {
                "BANDCAMP_USERNAME": values.get("username", ""),
                "BANDCAMP_FORMAT_PREF": values.get("format_pref", "flac"),
                "BANDCAMP_COOKIES": str(bandcamp_dir / "cookies.txt"),
                "BANDCAMP_STATE": str(bandcamp_dir / "state.json"),
                "BANDCAMP_STAGING": str(
                    settings.music_data_dir / "sldl-dropbox" / "_bandcamp-staging"
                ),
            },
        )
        cookies_txt = values.get("cookies_txt", "")
        if cookies_txt:
            cookies_path = bandcamp_dir / "cookies.txt"
            cookies_path.write_text(cookies_txt)
            cookies_path.chmod(0o600)

    elif scope == "azuracast":
        az_dir = settings.pipeline_config_dir / "azuracast"
        az_dir.mkdir(parents=True, exist_ok=True)
        key_path = az_dir / "api_key"
        key_path.write_text(values.get("api_key", ""))
        key_path.chmod(0o600)

    elif scope == "qobuz":
        qobuz_dir = settings.pipeline_config_dir / "qobuz"
        qobuz_dir.mkdir(parents=True, exist_ok=True)
        for field, filename in (
            ("token", "token"),
            ("app_id", "app_id"),
            ("region", "region"),
        ):
            if field in values:
                path = qobuz_dir / filename
                path.write_text(values[field])
                path.chmod(0o600)

    elif scope == "telegram":
        _write_env_file(
            settings.pipeline_config_dir / "telegram" / "notify.env",
            {
                "TG_BOT_TOKEN": values.get("bot_token", ""),
                "TG_CHAT_ID": values.get("chat_id", ""),
            },
        )

    else:
        raise ValueError(f"unknown credential scope: {scope}")