"""Admin-editable system configuration: a DB-override layer over the env/config defaults. The registry (SPECS) is the single source of truth for which config keys can be overridden from the admin Configuration page, plus each key's type, admin-UI group, env/config default, bounds, and whether it's a secret. The resolver returns the DB override if one is set, else ``settings.`` (same DB-override-or-default pattern as app.state). Secrets are stored Fernet-encrypted at rest (requires TOKEN_ENCRYPTION_KEY; without it, secrets can't be stored in the DB and stay env-only). """ from dataclasses import dataclass from sqlalchemy.orm import Session from app import security from app.config import settings from app.models import SystemConfig @dataclass(frozen=True) class ConfigSpec: key: str type: str # "int" | "str" | "bool" group: str # admin-UI grouping (logically related keys stay together) default_attr: str # attribute on `settings` holding the env/config default secret: bool = False min: int | None = None max: int | None = None # Registry of DB-overridable keys. Add a key here + wire its read through get_*() to move it # off env into the admin UI; the admin page renders it automatically from this list. SPECS: tuple[ConfigSpec, ...] = ( # --- Email / SMTP (one logical group; the password is the only secret) --- ConfigSpec("smtp_host", "str", "email", "smtp_host"), ConfigSpec("smtp_port", "int", "email", "smtp_port", min=1, max=65535), ConfigSpec("smtp_user", "str", "email", "smtp_user"), ConfigSpec("smtp_from", "str", "email", "smtp_from"), ConfigSpec("smtp_password", "str", "email", "smtp_password", secret=True), # --- Quota (shared YouTube Data API daily budget) --- ConfigSpec("quota_daily_budget", "int", "quota", "quota_daily_budget", min=1, max=1_000_000), ConfigSpec("backfill_quota_reserve", "int", "quota", "backfill_quota_reserve", min=0, max=1_000_000), # --- Backfill (recent-first first pass per channel) --- ConfigSpec("backfill_recent_max_videos", "int", "backfill", "backfill_recent_max_videos", min=1, max=100_000), ConfigSpec("backfill_recent_max_days", "int", "backfill", "backfill_recent_max_days", min=1, max=36_500), # --- Shorts probe --- ConfigSpec("shorts_probe_max_seconds", "int", "shorts", "shorts_probe_max_seconds", min=1, max=3_600), ConfigSpec("shorts_probe_batch", "int", "shorts", "shorts_probe_batch", min=1, max=10_000), # --- Batch sizes --- ConfigSpec("enrich_batch_size", "int", "batch", "enrich_batch_size", min=1, max=50), ConfigSpec("autotag_title_sample", "int", "batch", "autotag_title_sample", min=1, max=1_000), # --- YouTube Data API key (secret; optional — public reads prefer it over OAuth) --- ConfigSpec("youtube_api_key", "str", "youtube", "youtube_api_key", secret=True), # --- Access / registration --- ConfigSpec("allow_registration", "bool", "access", "allow_registration"), ) _BY_KEY: dict[str, ConfigSpec] = {s.key: s for s in SPECS} def spec(key: str) -> ConfigSpec | None: return _BY_KEY.get(key) def secrets_manageable() -> bool: """Secret values can only be stored when Fernet is configured (TOKEN_ENCRYPTION_KEY).""" return bool(settings.token_encryption_key) def _default(s: ConfigSpec): return getattr(settings, s.default_attr) def _coerce(s: ConfigSpec, raw: str): if s.type == "int": return int(raw) if s.type == "bool": return raw == "true" return raw def _raw_override(db: Session, key: str) -> str | None: """Stored (decrypted) override string for `key`, or None if no override is set.""" row = db.get(SystemConfig, key) if row is None or row.value is None: return None return security.decrypt(row.value) if row.encrypted else row.value def get(db: Session, key: str): """Effective value: the DB override (typed) if set, else the env/config default.""" s = _BY_KEY[key] raw = _raw_override(db, key) if raw is None or raw == "": return _default(s) try: return _coerce(s, raw) except (TypeError, ValueError): return _default(s) def get_str(db: Session, key: str) -> str: v = get(db, key) return "" if v is None else str(v) def get_int(db: Session, key: str) -> int: return int(get(db, key)) def get_bool(db: Session, key: str) -> bool: return bool(get(db, key)) def set_value(db: Session, key: str, value) -> None: """Validate, coerce, and persist a DB override for `key`. Secrets are encrypted.""" s = _BY_KEY[key] if s.type == "int": v = int(value) if s.min is not None and v < s.min: raise ValueError(f"{key} must be >= {s.min}") if s.max is not None and v > s.max: raise ValueError(f"{key} must be <= {s.max}") raw = str(v) elif s.type == "bool": raw = "true" if value else "false" else: raw = "" if value is None else str(value) stored, encrypted = raw, False if s.secret: if not secrets_manageable(): raise RuntimeError("TOKEN_ENCRYPTION_KEY is not configured") stored, encrypted = security.encrypt(raw) or "", True row = db.get(SystemConfig, key) if row is None: row = SystemConfig(key=key) db.add(row) row.value = stored row.encrypted = encrypted db.commit() def reset(db: Session, key: str) -> None: """Remove the DB override for `key` (fall back to the env/config default).""" row = db.get(SystemConfig, key) if row is not None: db.delete(row) db.commit() def describe(db: Session) -> dict: """Admin-UI metadata + current values, grouped. Secret values are never echoed — only whether an override is set and whether an env default exists.""" groups: dict[str, list[dict]] = {} for s in SPECS: is_set = db.get(SystemConfig, s.key) is not None item: dict = { "key": s.key, "type": s.type, "group": s.group, "secret": s.secret, "min": s.min, "max": s.max, "is_set": is_set, # an override row exists } if s.secret: item["default_is_set"] = bool(_default(s)) else: item["value"] = get(db, s.key) item["default"] = _default(s) groups.setdefault(s.group, []).append(item) return {"groups": groups, "secrets_manageable": secrets_manageable()}