Spaces:
Running
Running
| """配置数据模型: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"} | |
| 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"} | |
| 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 | |