xtc-backend / app /models /config.py
a3216's picture
deploy: initial release to Hugging Face Spaces
36fae79
Raw
History Blame Contribute Delete
6.25 kB
"""配置数据模型: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