Kansho/backend/provider_settings.py

"""Admin-configurable generate/detect endpoints. Keys stay in env, never in the DB."""
from __future__ import annotations

import json
import os
from pathlib import Path
from urllib.parse import urlparse

from db import get_db
from env_loader import (
    allows_remote_plaintext_detect,
    remote_plaintext_detect_reason,
    runtime_env,
    upsert_env_value,
)

SEED_PATH = Path(__file__).resolve().parent / "config" / "providers.seed.json"
ROLES = ("generate", "detect")
_KEY_ENV = {
    "generate": "KANSHO_PROVIDER_KEY",
    "detect": "KANSHO_DETECT_PROVIDER_KEY",
}
# Interne Rollen bleiben generate/detect. Fachlich: Sprachmodell vs. Maskierung.
ROLE_META = {
    "generate": {
        "title": "Sprachmodell",
        "task": "Dialogzug und Journalentwurf. Nur maskierter Kontext. Wählt die Operation im Dialogzug mit.",
    },
    "detect": {
        "title": "Maskierung",
        "task": "Eigener Detect-Endpunkt und eigenes Modell. Vollständige semantische Detection des persönlichen Egress. Ziel: lokales Detect-Modell. Übergang: externes Klartext-Detect nur mit Operator-Freigabe. Kein Pattern-Fallback.",
    },
}


class SettingsError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        super().__init__(message)
        self.code = code
        self.message = message
        self.status_code = status_code


def seed_provider_settings(conn) -> None:
    items = json.loads(SEED_PATH.read_text(encoding="utf-8"))
    for item in items:
        role = item["role"]
        if role not in ROLES:
            continue
        existing = conn.execute("SELECT role FROM provider_settings WHERE role = ?", (role,)).fetchone()
        if existing:
            continue
        conn.execute(
            """
            INSERT INTO provider_settings (role, name, url, model, zdr, no_train)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                role,
                item.get("name") or "",
                item.get("url") or "",
                item.get("model") or "",
                1 if item.get("zdr", True) else 0,
                1 if item.get("no_train", True) else 0,
            ),
        )


def get_setting(role: str) -> dict | None:
    if role not in ROLES:
        return None
    with get_db() as conn:
        row = conn.execute("SELECT * FROM provider_settings WHERE role = ?", (role,)).fetchone()
    return dict(row) if row else None


def upsert_setting(role: str, *, name: str, url: str, model: str, zdr: bool, no_train: bool) -> dict:
    if role not in ROLES:
        raise SettingsError("unknown_provider_role", "Unbekannte Provider-Rolle.")
    url = (url or "").strip()
    model = (model or "").strip()
    name = (name or "").strip()
    if role == "generate" and not url:
        raise SettingsError("provider_url_required", "Das Sprachmodell braucht eine URL.")
    if url:
        parsed = urlparse(url)
        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
            raise SettingsError("provider_url_invalid", "URL muss http(s) sein.")
    if role == "generate" and not model:
        raise SettingsError("provider_model_required", "Das Sprachmodell braucht ein Modell.")
    with get_db() as conn:
        conn.execute(
            """
            INSERT INTO provider_settings (role, name, url, model, zdr, no_train, updated)
            VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
            ON CONFLICT(role) DO UPDATE SET
              name = excluded.name,
              url = excluded.url,
              model = excluded.model,
              zdr = excluded.zdr,
              no_train = excluded.no_train,
              updated = datetime('now')
            """,
            (role, name, url, model, 1 if zdr else 0, 1 if no_train else 0),
        )
    return get_setting(role) or {}


def set_role_key(role: str, key: str) -> None:
    if role not in ROLES:
        raise SettingsError("unknown_provider_role", "Unbekannte Provider-Rolle.")
    value = (key or "").strip()
    if not value:
        raise SettingsError("provider_key_empty", "Leerer Key wird nicht gespeichert.")
    upsert_env_value(_KEY_ENV[role], value)


def public_role_status(role: str) -> dict:
    from providers import _is_local_url, detect_provider, generate_provider

    row = get_setting(role) or {}
    config = generate_provider() if role == "generate" else detect_provider()
    local = _is_local_url(row.get("url") or "")
    own_key = bool((os.environ.get(_KEY_ENV[role]) or "").strip())
    generate_key = bool((os.environ.get(_KEY_ENV["generate"]) or "").strip())
    if own_key:
        key_source = "own"
    elif role == "detect" and generate_key and not local:
        key_source = "generate"
    else:
        key_source = "none"
    meta = ROLE_META.get(role, {})
    return {
        "role": role,
        "title": meta.get("title") or role,
        "task": meta.get("task") or "",
        "name": row.get("name") or (config.name if config else ""),
        "url": row.get("url") or "",
        "model": row.get("model") or "",
        "zdr": bool(row.get("zdr", 1)),
        "no_train": bool(row.get("no_train", 1)),
        "local": local,
        "key_present": key_source != "none",
        "key_source": key_source,
        "ready": config is not None,
        "mode": config.mode if config else "unconfigured",
        "remote_plaintext_detect": bool(config and not config.local and config.mode == "http" and role == "detect"),
        "remote_plaintext_allowed": allows_remote_plaintext_detect() if role == "detect" else None,
        "remote_plaintext_reason": remote_plaintext_detect_reason() if role == "detect" else None,
        "updated": row.get("updated") or "",
    }


def public_status() -> dict:
    return {
        "roles": [public_role_status(role) for role in ROLES],
        "runtime_env": runtime_env(),
        "note": "Zwei Stufen, zwei Modelle: Maskierung und Sprachmodell. Semantische Detection ist Pflicht vor Generate. Ziel bleibt lokales Detect. Externes Klartext-Detect in Production nur mit KANSHO_ALLOW_REMOTE_DETECT. Keys nur in .env.",
        "remote_plaintext_reason": remote_plaintext_detect_reason(),
        "remote_plaintext_allowed": allows_remote_plaintext_detect(),
    }