"""配置数据模型:Provider、AppConfig。 向后兼容原 Netlify 版字段名(驼峰与下划线都接受)。 """ from __future__ import annotations from datetime import datetime, timezone from typing import Any, List, Literal, Optional from pydantic import BaseModel, Field, model_validator def _split_key_list(value: Any) -> List[str]: if not value: return [] if isinstance(value, list): out = [] for item in value: s = str(item or "").strip() if s: out.append(s) return list(dict.fromkeys(out)) out = [] for item in str(value).replace("\n", ",").replace(";", ",").split(","): s = item.strip() if s: out.append(s) return list(dict.fromkeys(out)) class Provider(BaseModel): """单个上游厂商配置。""" id: str name: str = "" type: Literal["gemini", "openai"] = "gemini" base_url: Optional[str] = None api_keys: List[str] = Field(default_factory=list) api_key: Optional[str] = None # 兼容字段,等价于 api_keys[0] model_prefix: str = "" enabled: bool = True model_policy_mode: Literal["allowlist", "denylist"] = "denylist" allow_models: List[str] = Field(default_factory=list) deny_models: List[str] = Field(default_factory=list) model_config = {"extra": "ignore"} @model_validator(mode="before") @classmethod def _normalize(cls, data: Any) -> Any: if not isinstance(data, dict): return data # 兼容 alias out = dict(data) if "baseUrl" in out and "base_url" not in out: out["base_url"] = out.pop("baseUrl") if "modelPrefix" in out and "model_prefix" not in out: out["model_prefix"] = out.pop("modelPrefix") if "modelPolicyMode" in out and "model_policy_mode" not in out: out["model_policy_mode"] = out.pop("modelPolicyMode") if "allowModels" in out and "allow_models" not in out: out["allow_models"] = out.pop("allowModels") if "denyModels" in out and "deny_models" not in out: out["deny_models"] = out.pop("denyModels") # enabled_models / disabled_models 别名 if "enabledModels" in out and "allow_models" not in out: out["allow_models"] = out.pop("enabledModels") if "disabledModels" in out and "deny_models" not in out: out["deny_models"] = out.pop("disabledModels") # id 默认从 name if not out.get("id"): out["id"] = str(out.get("name") or "").strip() # 名称默认从 id if not out.get("name"): out["name"] = str(out.get("id") or "").strip() # apiKeys / api_keys / apiKey 合并去重(兼容多种命名) keys = _split_key_list(out.get("api_keys")) keys += _split_key_list(out.get("apiKeys")) keys += _split_key_list(out.get("api_key")) keys += _split_key_list(out.get("apiKey")) out["api_keys"] = list(dict.fromkeys(keys)) out["api_key"] = out["api_keys"][0] if out["api_keys"] else None # base_url 兜底 if not out.get("base_url"): ptype = str(out.get("type") or "").lower() out["base_url"] = None if ptype == "gemini" else "https://api.openai.com" else: out["base_url"] = str(out["base_url"]).rstrip("/") return out def masked(self) -> "Provider": """返回脱敏副本(用于后台展示,密钥只显示首尾)。""" def mask(k: str) -> str: k = str(k or "") return f"{k[:6]}***{k[-4:]}" if len(k) > 10 else "***" return Provider( id=self.id, name=self.name, type=self.type, base_url=self.base_url, api_keys=[mask(k) for k in self.api_keys], api_key=mask(self.api_key) if self.api_key else None, model_prefix=self.model_prefix, enabled=self.enabled, model_policy_mode=self.model_policy_mode, allow_models=list(self.allow_models), deny_models=list(self.deny_models), ) def public(self) -> dict: """返回给手表端 /v1/xtc/providers 的精简信息。""" return { "id": self.id, "name": self.name, "type": self.type, "modelPrefix": self.model_prefix or "", } class AppConfig(BaseModel): """完整配置(厂商列表 + 默认厂商 + 更新时间)。""" providers: List[Provider] = Field(default_factory=list) default_provider_id: Optional[str] = None updated_at: Optional[str] = None model_config = {"extra": "ignore"} @model_validator(mode="before") @classmethod def _normalize(cls, data: Any) -> Any: if not isinstance(data, dict): return data out = dict(data) if "defaultProviderId" in out and "default_provider_id" not in out: out["default_provider_id"] = out.pop("defaultProviderId") if "updatedAt" in out and "updated_at" not in out: out["updated_at"] = out.pop("updatedAt") return out def normalize(self) -> "AppConfig": """确保 default_provider_id 指向启用的厂商。""" enabled = [p for p in self.providers if p.enabled] if self.default_provider_id and any(p.id == self.default_provider_id for p in enabled): return self self.default_provider_id = enabled[0].id if enabled else None self.updated_at = datetime.now(timezone.utc).isoformat() return self def find_provider(self, provider_id: Optional[str], *, enabled_only: bool = True) -> Optional[Provider]: if not provider_id: return None for p in self.providers: if p.id == provider_id and (not enabled_only or p.enabled): return p return None def find_default_provider(self) -> Optional[Provider]: if self.default_provider_id: p = self.find_provider(self.default_provider_id) if p: return p for p in self.providers: if p.enabled: return p return None