import base64 import hashlib import json import os import secrets from pathlib import Path from typing import Any, Literal from urllib.parse import urljoin, urlparse import boto3 import httpx import anyio import psycopg import redis from botocore.config import Config as BotoConfig from cryptography.fernet import Fernet, InvalidToken from pydantic import BaseModel, Field from app.config import Settings, get_settings from app.settings_schema import CATEGORIES, EDITABLE_FIELDS, FIELDS, SECRET_FIELDS, SettingCategory class SettingsReadiness(BaseModel): ready: bool completed_required: int total_required: int missing: list[str] = Field(default_factory=list) untested: list[str] = Field(default_factory=list) class RuntimeSettingsResponse(BaseModel): categories: list[SettingCategory] values: dict[str, Any] configured: dict[str, bool] tests: dict[str, dict[str, Any]] readiness: SettingsReadiness class RuntimeSettingsUpdate(BaseModel): values: dict[str, Any] class RuntimeSettingsTestRequest(BaseModel): target: Literal[ "infrastructure", "storage", "baidu_ocr", "ai_models", "gpu", "langfuse", "sentry", ] values: dict[str, Any] = Field(default_factory=dict) class RuntimeSettingsTestResult(BaseModel): target: str ok: bool message: str details: dict[str, Any] = Field(default_factory=dict) class SecretGenerateRequest(BaseModel): kind: Literal["hex24", "hex32", "base64_32"] class SecretGenerateResponse(BaseModel): value: str REQUIRED_GROUPS: dict[str, list[str]] = { "基础运行": ["database_status", "redis_status", "encryption_status"], "文件存储": [ "s3_endpoint", "s3_public_endpoint", "s3_access_key_id", "s3_secret_access_key", "s3_bucket_inputs", "s3_bucket_derived", "s3_bucket_renders", ], "AI 模型": [ "ai_provider", "ai_base_url", "ai_api_key", "orchestrator_model", "vision_model", "image_model", ], } REQUIRED_TESTS = { "基础运行": "infrastructure", "文件存储": "storage", "AI 模型": "ai_models", } TEST_FIELDS: dict[str, set[str]] = { "infrastructure": set(), "storage": {key for key in FIELDS if key.startswith("s3_")}, "baidu_ocr": {key for key in FIELDS if key.startswith("baidu_ocr_")}, "ai_models": {key for key in FIELDS if key.startswith("ai_")} | { "orchestrator_model", "vision_model", "image_model", }, "gpu": {key for key in FIELDS if key.startswith("gpu_")}, "langfuse": {key for key in FIELDS if key.startswith("langfuse_")}, "sentry": {key for key in FIELDS if key.startswith("sentry_")}, } def _is_configured(value: Any) -> bool: if value is None or value is False: return False if isinstance(value, str): normalized = value.strip().lower() return bool(normalized) and not normalized.startswith(("replace_", "__auto_")) return True def _join_api_url(base_url: str, path: str) -> str: return urljoin(base_url.rstrip("/") + "/", path.lstrip("/")) class EncryptedSettingsStore: def __init__(self, bootstrap: Settings | None = None) -> None: self.bootstrap = bootstrap or get_settings() self.root = Path(self.bootstrap.runtime_settings_dir) if not self.root.is_absolute(): self.root = Path.cwd() / self.root self.data_path = self.root / "settings.enc" self.key_path = self.root / ".master-key" def _fernet(self) -> Fernet: self.root.mkdir(parents=True, exist_ok=True) configured_key = self.bootstrap.settings_master_key.strip() if configured_key: try: return Fernet(configured_key.encode("ascii")) except (ValueError, TypeError): derived = base64.urlsafe_b64encode(hashlib.sha256(configured_key.encode()).digest()) return Fernet(derived) if self.key_path.exists(): return Fernet(self.key_path.read_bytes().strip()) key = Fernet.generate_key() self.key_path.write_bytes(key) try: os.chmod(self.key_path, 0o600) except OSError: pass return Fernet(key) def defaults(self) -> dict[str, Any]: return { key: field.default for key, field in FIELDS.items() if key in EDITABLE_FIELDS and field.default is not None } def load_document(self) -> dict[str, Any]: if not self.data_path.exists(): return {"values": self.defaults(), "tests": {}} try: decrypted = self._fernet().decrypt(self.data_path.read_bytes()) document = json.loads(decrypted.decode("utf-8")) except (InvalidToken, ValueError, json.JSONDecodeError) as exc: raise RuntimeError("运行期配置无法解密，请检查主密钥是否发生变化。") from exc document.setdefault("values", {}) document.setdefault("tests", {}) return document def save_document(self, document: dict[str, Any]) -> None: self.root.mkdir(parents=True, exist_ok=True) payload = json.dumps(document, ensure_ascii=False, sort_keys=True).encode("utf-8") encrypted = self._fernet().encrypt(payload) temporary = self.data_path.with_suffix(".tmp") temporary.write_bytes(encrypted) temporary.replace(self.data_path) def merged_values(self, pending: dict[str, Any] | None = None) -> dict[str, Any]: document = self.load_document() values = {**self.defaults(), **document["values"]} for key, value in (pending or {}).items(): if key not in EDITABLE_FIELDS: continue if key in SECRET_FIELDS and not _is_configured(value): continue values[key] = self._coerce(key, value) return values def update(self, patch: dict[str, Any]) -> None: document = self.load_document() changed: set[str] = set() for key, raw_value in patch.items(): if key not in EDITABLE_FIELDS: continue if key in SECRET_FIELDS and not _is_configured(raw_value): continue value = self._coerce(key, raw_value) if document["values"].get(key) != value: document["values"][key] = value changed.add(key) for target, fields in TEST_FIELDS.items(): if changed & fields: document["tests"].pop(target, None) self.save_document(document) def record_test(self, result: RuntimeSettingsTestResult, values: dict[str, Any] | None = None) -> None: document = self.load_document() document["tests"][result.target] = { "ok": result.ok, "message": result.message, "fingerprint": self.test_fingerprint(result.target, values or self.merged_values()), } self.save_document(document) def public_response(self) -> RuntimeSettingsResponse: document = self.load_document() values = {**self.defaults(), **document["values"]} public_values = {key: value for key, value in values.items() if key not in SECRET_FIELDS} configured = {key: _is_configured(values.get(key)) for key in FIELDS} configured.update(self.bootstrap_statuses()) current_tests = self.current_tests(values, document["tests"]) return RuntimeSettingsResponse( categories=CATEGORIES, values=public_values, configured=configured, tests=current_tests, readiness=self.readiness(values, current_tests), ) def bootstrap_statuses(self) -> dict[str, bool]: return { "database_status": _is_configured(self.bootstrap.database_url), "redis_status": _is_configured(self.bootstrap.redis_url), "encryption_status": self._master_key_available(), } def _master_key_available(self) -> bool: if self.bootstrap.settings_master_key.strip() or self.key_path.exists(): return True try: self._fernet() return True except OSError: return False def readiness( self, values: dict[str, Any] | None = None, tests: dict[str, dict[str, Any]] | None = None, ) -> SettingsReadiness: values = values or self.merged_values() if tests is None: tests = self.current_tests(values, self.load_document()["tests"]) statuses = self.bootstrap_statuses() all_values = {**values, **statuses} missing: list[str] = [] untested: list[str] = [] completed = 0 for group, keys in REQUIRED_GROUPS.items(): group_missing = [key for key in keys if not _is_configured(all_values.get(key))] if group_missing: labels = [FIELDS[key].label for key in group_missing] missing.append(f"{group}：{'、'.join(labels)}") continue target = REQUIRED_TESTS[group] if not tests.get(target, {}).get("ok"): untested.append(f"{group}：尚未通过连接测试") continue completed += 1 return SettingsReadiness( ready=not missing and not untested, completed_required=completed, total_required=len(REQUIRED_GROUPS), missing=missing, untested=untested, ) def current_tests( self, values: dict[str, Any], tests: dict[str, dict[str, Any]], ) -> dict[str, dict[str, Any]]: current: dict[str, dict[str, Any]] = {} for target, result in tests.items(): item = dict(result) if item.get("fingerprint") != self.test_fingerprint(target, values): item["ok"] = False item["message"] = "配置已变更，请重新测试。" current[target] = item return current def test_fingerprint(self, target: str, values: dict[str, Any]) -> str: if target == "infrastructure": relevant: dict[str, Any] = { "database_url": self.bootstrap.database_url, "redis_url": self.bootstrap.redis_url, } else: relevant = {key: values.get(key) for key in sorted(TEST_FIELDS.get(target, set()))} payload = json.dumps(relevant, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") return hashlib.sha256(payload).hexdigest() @staticmethod def _coerce(key: str, value: Any) -> Any: field = FIELDS[key] if field.kind == "toggle": if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "on"} return bool(value) if field.kind == "number": return int(value) return value.strip() if isinstance(value, str) else value def get_runtime_store() -> EncryptedSettingsStore: return EncryptedSettingsStore(get_settings()) def generate_secret(kind: str) -> str: if kind == "hex24": return secrets.token_hex(24) if kind == "hex32": return secrets.token_hex(32) if kind == "base64_32": return base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii") raise ValueError("不支持的密钥类型。") async def test_runtime_settings( target: str, values: dict[str, Any], bootstrap: Settings, ) -> RuntimeSettingsTestResult: try: if target == "infrastructure": return await _test_infrastructure(bootstrap) if target == "storage": return await _test_storage(values) if target == "baidu_ocr": return await _test_baidu_ocr(values) if target == "ai_models": return await _test_ai_models(values) if target == "gpu": return await _test_gpu(values) if target == "langfuse": return await _test_langfuse(values) if target == "sentry": return await _test_sentry(values) except Exception as exc: # Integration boundaries return a safe, user-readable failure. return RuntimeSettingsTestResult(target=target, ok=False, message=f"连接失败：{exc}") return RuntimeSettingsTestResult(target=target, ok=False, message="未知测试类型。") async def _test_infrastructure(bootstrap: Settings) -> RuntimeSettingsTestResult: return await anyio.to_thread.run_sync(_test_infrastructure_sync, bootstrap) def _test_infrastructure_sync(bootstrap: Settings) -> RuntimeSettingsTestResult: with psycopg.connect(bootstrap.database_url, connect_timeout=5) as connection: with connection.cursor() as cursor: cursor.execute("SELECT 1") cursor.fetchone() redis_client = redis.Redis.from_url( bootstrap.redis_url, socket_connect_timeout=5, socket_timeout=5, ) redis_client.ping() return RuntimeSettingsTestResult( target="infrastructure", ok=True, message="PostgreSQL 与 Redis 均连接正常。", ) async def _test_storage(values: dict[str, Any]) -> RuntimeSettingsTestResult: return await anyio.to_thread.run_sync(_test_storage_sync, values) def _test_storage_sync(values: dict[str, Any]) -> RuntimeSettingsTestResult: client = boto3.client( "s3", endpoint_url=values["s3_endpoint"], aws_access_key_id=values["s3_access_key_id"], aws_secret_access_key=values["s3_secret_access_key"], region_name="us-east-1", config=BotoConfig( connect_timeout=5, read_timeout=5, retries={"max_attempts": 1}, s3={"addressing_style": "path" if values.get("s3_force_path_style", True) else "virtual"}, ), ) existing = {bucket["Name"] for bucket in client.list_buckets().get("Buckets", [])} required = [ values["s3_bucket_inputs"], values["s3_bucket_derived"], values["s3_bucket_renders"], ] missing = [bucket for bucket in required if bucket not in existing] return RuntimeSettingsTestResult( target="storage", ok=not missing, message="MinIO 连接正常，三个存储空间均可用。" if not missing else "MinIO 可连接，但需要先创建部分存储空间。", details={"missing_buckets": missing}, ) async def _test_baidu_ocr(values: dict[str, Any]) -> RuntimeSettingsTestResult: if not values.get("baidu_ocr_enabled"): return RuntimeSettingsTestResult(target="baidu_ocr", ok=True, message="百度 OCR 当前未启用。") async with httpx.AsyncClient(timeout=8) as client: response = await client.post( "https://aip.baidubce.com/oauth/2.0/token", params={ "grant_type": "client_credentials", "client_id": values.get("baidu_ocr_api_key", ""), "client_secret": values.get("baidu_ocr_secret_key", ""), }, ) response.raise_for_status() payload = response.json() ok = bool(payload.get("access_token")) return RuntimeSettingsTestResult( target="baidu_ocr", ok=ok, message="百度 OCR 凭证有效。" if ok else "百度 OCR 未返回访问令牌。", ) async def _test_ai_models(values: dict[str, Any]) -> RuntimeSettingsTestResult: url = _join_api_url(values["ai_base_url"], values.get("ai_models_path", "/models")) async with httpx.AsyncClient(timeout=12) as client: response = await client.get(url, headers={"Authorization": f"Bearer {values['ai_api_key']}"}) response.raise_for_status() payload = response.json() model_ids = { item.get("id") for item in payload.get("data", []) if isinstance(item, dict) and item.get("id") } selected = [values.get("orchestrator_model"), values.get("vision_model"), values.get("image_model")] missing = [model for model in selected if model and model_ids and model not in model_ids] return RuntimeSettingsTestResult( target="ai_models", ok=not missing, message="API Key 有效，三个模型均可用。" if not missing else "API 可以连接，但部分模型名不在账号模型列表中。", details={"missing_models": missing, "model_count": len(model_ids)}, ) async def _test_gpu(values: dict[str, Any]) -> RuntimeSettingsTestResult: mode = values.get("gpu_mode", "disabled") if mode == "disabled": return RuntimeSettingsTestResult(target="gpu", ok=True, message="本地 GPU 当前未启用。") local_host = "host.docker.internal" if Path("/.dockerenv").exists() else "127.0.0.1" url = f"http://{local_host}:8100" if mode == "local" else values.get("gpu_service_url", "") async with httpx.AsyncClient(timeout=8) as client: response = await client.get( _join_api_url(url, "/health"), headers={"Authorization": f"Bearer {values.get('gpu_service_token', '')}"}, ) response.raise_for_status() return RuntimeSettingsTestResult(target="gpu", ok=True, message="GPU Worker 连接正常。") async def _test_langfuse(values: dict[str, Any]) -> RuntimeSettingsTestResult: if not values.get("langfuse_enabled"): return RuntimeSettingsTestResult(target="langfuse", ok=True, message="Langfuse 当前未启用。") async with httpx.AsyncClient(timeout=8) as client: response = await client.get( _join_api_url(values.get("langfuse_host", ""), "/api/public/projects"), auth=( str(values.get("langfuse_public_key", "")), str(values.get("langfuse_secret_key", "")), ), ) response.raise_for_status() return RuntimeSettingsTestResult(target="langfuse", ok=True, message="Langfuse 服务与项目密钥均有效。") async def _test_sentry(values: dict[str, Any]) -> RuntimeSettingsTestResult: if not values.get("sentry_enabled"): return RuntimeSettingsTestResult(target="sentry", ok=True, message="Sentry 当前未启用。") dsn = values.get("sentry_dsn", "") parsed = urlparse(dsn) if parsed.scheme not in {"http", "https"} or not parsed.hostname or not parsed.username: return RuntimeSettingsTestResult(target="sentry", ok=False, message="Sentry DSN 格式不正确。") async with httpx.AsyncClient(timeout=8) as client: response = await client.get(f"{parsed.scheme}://{parsed.netloc.split('@')[-1]}") return RuntimeSettingsTestResult( target="sentry", ok=response.status_code < 500, message="Sentry 地址可访问，DSN 格式正确。" if response.status_code < 500 else "Sentry 服务返回异常。", )