File size: 7,262 Bytes
1a9e2c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | """配置管理器 - 管理应用配置的读写"""
import toml
from pathlib import Path
from typing import Dict, Any, Optional, Literal
# 默认配置
DEFAULT_GROK = {
"api_key": "",
"proxy_url": "",
"proxy_pool_url": "",
"proxy_pool_interval": 300,
"cache_proxy_url": "",
"cf_clearance": "",
"x_statsig_id": "",
"dynamic_statsig": True,
"filtered_tags": "xaiartifact,xai:tool_usage_card",
"show_thinking": True,
"temporary": False,
"max_upload_concurrency": 20,
"max_request_concurrency": 100,
"stream_first_response_timeout": 30,
"stream_chunk_timeout": 120,
"stream_total_timeout": 600,
"retry_status_codes": [401, 429], # 可重试的HTTP状态码
}
DEFAULT_GLOBAL = {
"base_url": "http://localhost:8000",
"log_level": "INFO",
"image_mode": "url",
"admin_password": "admin",
"admin_username": "admin",
"image_cache_max_size_mb": 512,
"video_cache_max_size_mb": 1024,
"max_upload_concurrency": 20, # 最大并发上传数
"max_request_concurrency": 50, # 最大并发请求数
"batch_save_interval": 1.0, # 批量保存间隔（秒）
"batch_save_threshold": 10 # 触发批量保存的变更数阈值
}
class ConfigManager:
"""配置管理器"""
def __init__(self) -> None:
"""初始化配置"""
self.config_path: Path = Path(__file__).parents[2] / "data" / "setting.toml"
self._storage: Optional[Any] = None
self._ensure_exists()
self.global_config: Dict[str, Any] = self.load("global")
self.grok_config: Dict[str, Any] = self.load("grok")
def _ensure_exists(self) -> None:
"""确保配置存在"""
if not self.config_path.exists():
self.config_path.parent.mkdir(parents=True, exist_ok=True)
self._create_default()
def _create_default(self) -> None:
"""创建默认配置"""
default = {"grok": DEFAULT_GROK.copy(), "global": DEFAULT_GLOBAL.copy()}
with open(self.config_path, "w", encoding="utf-8") as f:
toml.dump(default, f)
def _normalize_proxy(self, proxy: str) -> str:
"""标准化代理URL（sock5/socks5 → socks5h://）"""
if not proxy:
return proxy
proxy = proxy.strip()
if proxy.startswith("sock5h://"):
proxy = proxy.replace("sock5h://", "socks5h://", 1)
if proxy.startswith("sock5://"):
proxy = proxy.replace("sock5://", "socks5://", 1)
if proxy.startswith("socks5://"):
return proxy.replace("socks5://", "socks5h://", 1)
return proxy
def _normalize_cf(self, cf: str) -> str:
"""标准化CF Clearance（自动添加前缀）"""
if cf and not cf.startswith("cf_clearance="):
return f"cf_clearance={cf}"
return cf
def set_storage(self, storage: Any) -> None:
"""设置存储实例"""
self._storage = storage
def load(self, section: Literal["global", "grok"]) -> Dict[str, Any]:
"""加载配置节"""
try:
with open(self.config_path, "r", encoding="utf-8") as f:
config = toml.load(f)[section]
# 标准化Grok配置
if section == "grok":
if "proxy_url" in config:
config["proxy_url"] = self._normalize_proxy(config["proxy_url"])
if "cache_proxy_url" in config:
config["cache_proxy_url"] = self._normalize_proxy(config["cache_proxy_url"])
if "cf_clearance" in config:
config["cf_clearance"] = self._normalize_cf(config["cf_clearance"])
return config
except Exception as e:
raise Exception(f"[Setting] 配置加载失败: {e}") from e
async def reload(self) -> None:
"""重新加载配置"""
self.global_config = self.load("global")
self.grok_config = self.load("grok")
async def _save_file(self, updates: Dict[str, Dict[str, Any]]) -> None:
"""保存到文件"""
import aiofiles
async with aiofiles.open(self.config_path, "r", encoding="utf-8") as f:
config = toml.loads(await f.read())
for section, data in updates.items():
if section in config:
config[section].update(data)
async with aiofiles.open(self.config_path, "w", encoding="utf-8") as f:
await f.write(toml.dumps(config))
async def _save_storage(self, updates: Dict[str, Dict[str, Any]]) -> None:
"""保存到存储"""
config = await self._storage.load_config()
for section, data in updates.items():
if section in config:
config[section].update(data)
await self._storage.save_config(config)
def _prepare_grok(self, grok: Dict[str, Any]) -> Dict[str, Any]:
"""准备Grok配置（移除前缀）"""
processed = grok.copy()
if "cf_clearance" in processed:
cf = processed["cf_clearance"]
if cf and cf.startswith("cf_clearance="):
processed["cf_clearance"] = cf.replace("cf_clearance=", "", 1)
return processed
async def save(self, global_config: Optional[Dict[str, Any]] = None, grok_config: Optional[Dict[str, Any]] = None) -> None:
"""保存配置"""
updates = {}
if global_config:
updates["global"] = global_config
if grok_config:
updates["grok"] = self._prepare_grok(grok_config)
# 选择存储方式
if self._storage:
await self._save_storage(updates)
else:
await self._save_file(updates)
await self.reload()
async def get_proxy_async(self, proxy_type: Literal["service", "cache"] = "service") -> str:
"""异步获取代理URL（支持代理池）
Args:
proxy_type: 代理类型
- service: 服务代理（client/upload）
- cache: 缓存代理（cache）
"""
from app.core.proxy_pool import proxy_pool
if proxy_type == "cache":
cache_proxy = self.grok_config.get("cache_proxy_url", "")
if cache_proxy:
return cache_proxy
# 从代理池获取
return await proxy_pool.get_proxy() or ""
def get_proxy(self, proxy_type: Literal["service", "cache"] = "service") -> str:
"""获取代理URL（同步方法，用于向后兼容）
Args:
proxy_type: 代理类型
- service: 服务代理（client/upload）
- cache: 缓存代理（cache）
"""
from app.core.proxy_pool import proxy_pool
if proxy_type == "cache":
cache_proxy = self.grok_config.get("cache_proxy_url", "")
if cache_proxy:
return cache_proxy
# 返回当前代理（如果是代理池，返回最后一次获取的）
return proxy_pool.get_current_proxy() or self.grok_config.get("proxy_url", "")
# 全局实例
setting = ConfigManager()
|