Add playlist_service and credential_service

playlist_service: CRUD for the playlists table, playlists.json export,
regen.sh invocation, and a one-time seed_legacy() importing the 17-entry
legacy array. Creating/updating a playlist automatically re-renders confs
and re-patches credentials into any newly-generated .conf file.

credential_service: encrypted credential storage (via security/crypto.py)
and per-scope rendering to the exact files the pipeline scripts read --
every <playlist>.conf (spotify/soulseek), navidrome/admin.env,
bandcamp/config.env+cookies.txt, azuracast/api_key, qobuz/{token,app_id,region},
telegram/notify.env. set_credentials() batches multi-field saves so a render
never sees a half-populated scope.

Also fixes a real bug found via integration testing: regen.sh's embedded
python3 -c snippet used backslash-escaped quotes inside a single-quoted
bash string, which is invalid Python syntax (backslash passed through
literally) -- switched to double-quoted bash wrapper + single-quoted Python
strings.

Verified end-to-end: seeded 17 playlists, regenerated all 17 .conf files,
saved spotify/soulseek/navidrome credentials and confirmed correct
patching into rendered files (including bash-sourcing admin.env with
special characters in the password), and confirmed delete/create both
correctly add/remove .conf files with credentials pre-patched on create.
This commit is contained in:
andrew
2026-07-08 13:32:37 -06:00
parent bd56c8153f
commit 5eaae8e813
6 changed files with 352 additions and 3 deletions
+184
View File
@@ -0,0 +1,184 @@
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}")