alembic/app/services/credential_service.py

import re
import shlex
import time
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import AppSetting, 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"],
}

# One line explaining what each service is actually used for, shown on the
# credentials page so "should I bother setting this up" is answerable
# without reading the pipeline scripts.
SCOPE_DESCRIPTIONS = {
    "spotify": "Reads playlist tracks and metadata (artist, title, album, genre) for every synced playlist, tag rewriting, and the genre refill job. Required.",
    "soulseek": "Logs into Soulseek so sldl can search for and download tracks. Required.",
    "navidrome": "Triggers a library rescan after downloads/imports/edits, and lets the daily status report check Navidrome's scan health. Required.",
    "bandcamp": "Pulls new purchases from your Bandcamp collection, imports them automatically, and stamps a buy link into the file tag.",
    "qobuz": "A buy-link source (alongside Bandcamp) for tracks you didn't purchase there -- DRM-free hi-res links when available, written directly into the file tag.",
    "azuracast": "Only relevant if you run an AzuraCast radio station off the same library. Not a buy-link source itself (that's Bandcamp/Qobuz, written straight to the file tag) -- this lets alembic tell AzuraCast to immediately reprocess touched files or fix a playlist assignment, instead of waiting for its periodic scan.",
    "telegram": "Sends the daily pipeline health digest to a chat instead of you having to check the dashboard.",
}

# spotify/soulseek/navidrome are load-bearing for every playlist sync --
# there's no clean single place to "pause" them without silently breaking
# the whole pipeline, so they're always on and not shown with a toggle.
# The rest are optional add-ons that render to their own standalone
# credential file(s), which the scripts that read them already treat a
# missing/empty file as "not configured, skip this" -- see _clear_rendered().
CORE_SCOPES = {"spotify", "soulseek", "navidrome"}
OPTIONAL_SCOPES = {"bandcamp", "azuracast", "qobuz", "telegram"}


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}


# A playlist name is also its .conf filename; it is validated to this charset
# at creation (playlist_service.validate_name). Re-checked here as defense in
# depth so a bad name can never escape the config dir or inject config text.
_SAFE_CONF_NAME = re.compile(r"^[A-Za-z0-9 _-]{1,64}$")

# The credential lines the renderer fills in from the encrypted store. Everything
# else in a rendered .conf comes verbatim from _template.conf.
_CONF_CRED_KEYS = ("user", "pass", "spotify-id", "spotify-secret")


def _set_conf_field(text: str, key: str, value: str) -> str:
    """Return `text` with the single `key = ...` line set to `value`,
    appending the line if absent. Newlines in the value are stripped so a
    credential can't inject an extra config line."""
    value = value.replace("\r", "").replace("\n", "")
    pattern = re.compile(rf"^{re.escape(key)}\s*=.*$", re.MULTILINE)
    new_line = f"{key} = {value}"
    if pattern.search(text):
        return pattern.sub(new_line, text)
    return text.rstrip("\n") + f"\n{new_line}\n"


def render_playlist_confs(db: Session) -> None:
    """Render every playlist's sldl .conf directly from _template.conf, with
    the path placeholders and the Soulseek/Spotify credentials substituted in
    one pass, written 0600.

    This is the single source of .conf rendering. It replaces the older
    DB -> playlists.json -> regen.sh -> regex-patch-back chain: the app owns
    every input (playlists in its DB, credentials in its encrypted store), so
    there is no reason to round-trip through an intermediate file and a
    subprocess. Called on any playlist change (playlist_service) and on any
    Spotify/Soulseek credential change (render_scope)."""
    from app.services import playlist_service

    template = (settings.pipeline_dir / "configs" / "_template.conf").read_text()
    dropbox_root = str(settings.music_data_dir / "sldl-dropbox")
    spotify = get_scope(db, "spotify")
    soulseek = get_scope(db, "soulseek")
    creds = {
        "user": soulseek.get("username", ""),
        "pass": soulseek.get("password", ""),
        "spotify-id": spotify.get("client_id", ""),
        "spotify-secret": spotify.get("client_secret", ""),
    }
    # Path placeholders are substituted as LITERAL text in a single left-to-right
    # pass (never re-scanned), so a value can't be reinterpreted as another
    # placeholder. Credential values are set as whole lines afterwards.
    token_re = re.compile("|".join(map(re.escape, ("PLAYLIST_NAME", "SPOTIFY_URL", "SLDL_DROPBOX_ROOT"))))

    settings.pipeline_config_dir.mkdir(parents=True, exist_ok=True)
    for p in playlist_service.list_all(db):
        if not _SAFE_CONF_NAME.match(p.name):
            continue
        mapping = {"PLAYLIST_NAME": p.name, "SPOTIFY_URL": p.spotify_url, "SLDL_DROPBOX_ROOT": dropbox_root}
        text = token_re.sub(lambda m: mapping[m.group(0)], template)
        for key in _CONF_CRED_KEYS:
            text = _set_conf_field(text, key, creds[key])
        out = settings.pipeline_config_dir / f"{p.name}.conf"
        out.write_text(text)
        out.chmod(0o600)


def _write_env_file(path: Path, values: dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    # shlex.quote so a credential value containing a quote, space, or shell
    # metacharacter (e.g. a password like `p'a$(id)ss`) stays a single literal
    # value when a pipeline script `source`s this file, instead of breaking out
    # and executing. shlex.quote also handles the empty string correctly ('').
    lines = [f"{k}={shlex.quote(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", "")
        # _spotify.env is read by the pipeline python scripts (spotify-genre etc.).
        _write_env_file(
            settings.pipeline_config_dir / "_spotify.env",
            {"SPOTIFY_CLIENT_ID": cid, "SPOTIFY_CLIENT_SECRET": csec},
        )
        # Re-render every playlist .conf so the new Spotify creds land in them.
        render_playlist_confs(db)

    elif scope == "soulseek":
        render_playlist_confs(db)

    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", ""),
                "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}")


def _enabled_key(scope: str) -> str:
    return f"cred_enabled:{scope}"


def is_scope_enabled(db: Session, scope: str) -> bool:
    """Core scopes are always enabled. Optional scopes default to enabled
    (matches pre-existing behavior for anyone who already had them
    configured before this toggle existed) until explicitly turned off."""
    if scope in CORE_SCOPES:
        return True
    row = db.execute(select(AppSetting).where(AppSetting.key == _enabled_key(scope))).scalar_one_or_none()
    return row is None or row.value == "1"


def _clear_rendered(scope: str) -> None:
    """Remove the standalone rendered credential file(s) for a disabled
    optional scope, without touching the encrypted Secret rows -- the
    script that reads each of these already treats a missing file as
    "not configured, skip this source" (see enrich-buy-url.py's docstring
    for qobuz/azuracast, and pipeline-status.sh's handling of a failed
    notify-telegram.sh call). Re-enabling calls render_scope() again to
    recreate the file from the still-stored values."""
    if scope == "azuracast":
        (settings.pipeline_config_dir / "azuracast" / "api_key").unlink(missing_ok=True)
    elif scope == "qobuz":
        qobuz_dir = settings.pipeline_config_dir / "qobuz"
        for name in ("token", "app_id", "region"):
            (qobuz_dir / name).unlink(missing_ok=True)
    elif scope == "telegram":
        (settings.pipeline_config_dir / "telegram" / "notify.env").unlink(missing_ok=True)


def set_scope_enabled(db: Session, scope: str, enabled: bool) -> None:
    """Toggle an optional integration on/off without discarding its saved
    credentials. Bandcamp is the one optional scope with its own dedicated
    scheduled job (maintenance:sync_bandcamp) -- that job hard-fails if its
    cookies file goes missing (see sync-bandcamp.sh), so bandcamp is gated
    by pausing the job itself (the same enable/disable machinery the Jobs
    page already uses) rather than by removing its rendered files. The
    other three scopes have no job of their own; the script that reads
    them already skips gracefully on a missing file, so removing the
    rendered file is enough."""
    if scope not in OPTIONAL_SCOPES:
        raise ValueError(f"{scope} cannot be toggled")

    key = _enabled_key(scope)
    row = db.execute(select(AppSetting).where(AppSetting.key == key)).scalar_one_or_none()
    if row is None:
        db.add(AppSetting(key=key, value="1" if enabled else "0"))
    else:
        row.value = "1" if enabled else "0"
    db.commit()

    if scope == "bandcamp":
        from app.services import scheduler_service

        scheduler = scheduler_service.get_scheduler()
        if scheduler is not None:
            scheduler_service.set_maintenance_enabled(scheduler, "maintenance:sync_bandcamp", enabled)
    elif enabled:
        render_scope(db, scope)
    else:
        _clear_rendered(scope)


def _bandcamp_cookie_expiry(cookies_txt: str) -> float | None:
    """Parse the Netscape-format cookie jar for the 'identity' cookie's
    expiry (Unix timestamp) -- same field pipeline-status.sh already reads
    via `awk -F'\\t' '$6=="identity" {print $5}'`. Purely local string
    parsing, no network call, safe to run on every dashboard load."""
    for line in cookies_txt.splitlines():
        fields = line.split("\t")
        if len(fields) >= 6 and fields[5] == "identity":
            try:
                return float(fields[4])
            except ValueError:
                return None
    return None


def auth_states(db: Session) -> list[dict]:
    """Per-scope credential health for the dashboard: whether it's
    configured at all, plus (bandcamp only) cookie expiry -- computed
    locally from the stored cookie jar, not a live network probe (unlike
    pipeline-status.sh's Qobuz check, which hits Qobuz's API on every run;
    deliberately not replicated here since the dashboard renders on every
    page load)."""
    states = []
    for scope in SCOPE_FIELDS:
        if scope in OPTIONAL_SCOPES and not is_scope_enabled(db, scope):
            continue  # turned off -- don't clutter the dashboard with it
        values = get_scope(db, scope)
        configured = bool(values)
        entry = {"scope": scope, "configured": configured}
        if scope == "bandcamp" and values.get("cookies_txt"):
            expiry = _bandcamp_cookie_expiry(values["cookies_txt"])
            if expiry is not None:
                entry["expires_at"] = expiry
                entry["days_left"] = int((expiry - time.time()) / 86400)
        states.append(entry)
    return states