SlabUploader/backend/settings.py

"""S1 Settings service (DATA-MODEL §6, API.md §10, SETTINGS-5/15/20/25/30/75/85/90).

Secrets (value_enc) are AES-GCM write-only; GET returns only {configured: bool}.
woo_base_url is env read-only. Flags default false. Partial PUT. Decrypt fail -> configured:false (no 500, no leak).
"""
from __future__ import annotations

import os
from typing import Any

from fastapi import HTTPException

from .crypto import decrypt, encrypt, DecryptError
from .db import get_connection

SECRET_KEYS: set[str] = {
    "woo_app_password",
    "inference_base_url",
    "inference_api_key",
}
PLAIN_KEYS: set[str] = {
    "woo_wp_username",
    "inference_model",
    "inference_enabled",
    "content_llm_enabled",
    "brand_voice",
    "geo_context",
    "woo_create_status",
    "taxonomy_last_sync_at",
    "taxonomy_anchor",
    "crop_fill",
}
ALL_KEYS = SECRET_KEYS | PLAIN_KEYS
# Plain keys surfaced as numbers in the API (stored as TEXT in `settings`).
NUMERIC_KEYS: set[str] = {"crop_fill"}

DEFAULTS: dict[str, Any] = {
    "inference_enabled": False,
    "content_llm_enabled": False,
    "woo_create_status": "draft",
    # #99: global crop fill %, one source of truth for the ref line + PNG crop.
    "crop_fill": 90,
}


def _row_to_value(row: dict) -> Any:
    if row is None:
        return None
    v = row.get("value")
    if v is None:
        return None
    if isinstance(v, str):
        if v.lower() in ("true", "false"):
            return v.lower() == "true"
    return v


def get_settings() -> dict[str, Any]:
    conn = get_connection()
    rows = {}
    for r in conn.execute("SELECT key, value, value_enc FROM settings").fetchall():
        rows[r["key"]] = {"value": r["value"], "value_enc": r["value_enc"]}

    result: dict[str, Any] = {}
    result["woo_base_url"] = os.environ.get("WOO_BASE_URL", "")

    # TEMPORARY UAT override (owner-directed, LAN/HTTP only): when set, secrets are
    # returned as their decrypted plaintext so they can be inspected on the dev-test
    # box. Default OFF -> SETTINGS-75 masking intact. Remove before prod.
    unmask = os.environ.get("SLAB_UAT_UNMASK_SECRETS") == "1"

    for k in sorted(ALL_KEYS):
        row = rows.get(k, {})
        if k in SECRET_KEYS:
            enc = row.get("value_enc")
            if unmask:
                if enc:
                    try:
                        result[k] = decrypt(enc)
                    except Exception:
                        result[k] = ""
                else:
                    result[k] = ""
            else:
                if enc:
                    try:
                        decrypt(enc)  # verify roundtrip possible
                        result[k] = {"configured": True}
                    except Exception:
                        result[k] = {"configured": False}
                else:
                    result[k] = {"configured": False}
        else:
            if k in rows:
                val = _row_to_value(row)
                result[k] = val if val is not None else DEFAULTS.get(k)
            else:
                result[k] = DEFAULTS.get(k)
            if k in NUMERIC_KEYS:
                try:
                    f = float(result[k])
                    result[k] = int(f) if f.is_integer() else f
                except (TypeError, ValueError):
                    result[k] = DEFAULTS.get(k)

    return result


def update_settings(payload: dict[str, Any]) -> dict[str, Any]:
    if "woo_base_url" in payload:
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "code": "validation",
                    "message": "woo_base_url is read-only (SETTINGS-85: env only)",
                    "detail": {"field": "woo_base_url"},
                }
            },
        )

    # #99: rules live here in the 4xx layer, not in the DB (structural only).
    if "crop_fill" in payload:
        raw = payload["crop_fill"]
        try:
            pct = float(raw)
        except (TypeError, ValueError):
            pct = None
        if pct is None or not (1 <= pct <= 100):
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "code": "validation",
                        "message": "crop_fill must be a number between 1 and 100 (percent)",
                        "detail": {"field": "crop_fill"},
                    }
                },
            )

    conn = get_connection()
    for k, v in payload.items():
        if k not in ALL_KEYS:
            continue  # ignore unknown for partial
        if k in SECRET_KEYS:
            if v in (None, "", False):
                conn.execute(
                    "INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,NULL)",
                    (k, None),
                )
            else:
                try:
                    enc = encrypt(str(v))
                except Exception:
                    enc = None
                conn.execute(
                    "INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,?)",
                    (k, None, enc),
                )
        else:
            # plain
            if v in (None, ""):
                conn.execute("DELETE FROM settings WHERE key=?", (k,))
            else:
                if k in ("inference_enabled", "content_llm_enabled"):
                    val_str = "true" if bool(v) else "false"
                else:
                    val_str = str(v)
                conn.execute(
                    "INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,NULL)",
                    (k, val_str),
                )
    conn.commit()
    return get_settings()