Maas-Model-platform/backend/app/services/local_config_service.py

import os
import sqlite3
from typing import Optional

# DB path: backend/data/local_config.db（相对于本文件所在的 services 目录往上两级）
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
_BACKEND_ROOT = os.path.abspath(os.path.join(_THIS_DIR, "..", ".."))
DB_PATH = os.path.join(_BACKEND_ROOT, "data", "local_config.db")

_SSO_KEYS = [
    "sso_enable_redirect",
    "cas_base_url",
    "app_public_base_url",
    "frontend_base_url",
    "sso_service_path",
    "sso_login_url",
    "sso_logout_url",
]

_ADMIN_API_KEY_KEYS = ["admin_api_key"]


def _init_db() -> None:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS local_config (
                key        TEXT PRIMARY KEY,
                value      TEXT NOT NULL,
                updated_at DATETIME DEFAULT (datetime('now'))
            )
            """
        )
        conn.commit()


def get_all() -> dict:
    rows: dict[str, str] = {}
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.execute("SELECT key, value FROM local_config")
        for key, value in cursor.fetchall():
            rows[key] = value

    sso: dict = {}
    for k in _SSO_KEYS:
        raw = rows.get(k)  # None 表示 db 中没有该 key
        if raw is None:
            sso[k] = None  # 让调用方自行决定默认值
        elif k == "sso_enable_redirect":
            sso[k] = raw == "true"
        else:
            sso[k] = raw

    admin_api_key_raw = rows.get("admin_api_key", "")
    admin_api_key_section = {
        "admin_api_key": "***" if admin_api_key_raw else ""
    }

    return {"sso": sso, "admin_api_key": admin_api_key_section}


def get_admin_api_key() -> Optional[str]:
    """返回原始（未脱敏）的 admin_api_key，不存在则返回 None。"""
    try:
        with sqlite3.connect(DB_PATH) as conn:
            row = conn.execute("SELECT value FROM local_config WHERE key='admin_api_key'").fetchone()
        return row[0] if row else None
    except Exception:
        return None


def upsert(conn: sqlite3.Connection, key: str, value: str) -> None:
    conn.execute(
        "INSERT OR REPLACE INTO local_config (key, value, updated_at) VALUES (?, ?, datetime('now'))",
        (key, value),
    )


def upsert_many(data: dict) -> None:
    with sqlite3.connect(DB_PATH) as conn:
        for key, value in data.items():
            upsert(conn, key, value)
        conn.commit()


# Ensure table exists on first import
_init_db()