349 lines
13 KiB
Python
349 lines
13 KiB
Python
"""
|
|
配置管理
|
|
|
|
- config.toml: 运行时配置
|
|
- config.defaults.toml: 默认配置基线
|
|
"""
|
|
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
import tomllib
|
|
|
|
from app.core.logger import logger
|
|
|
|
DEFAULT_CONFIG_FILE = Path(__file__).parent.parent.parent / "config.defaults.toml"
|
|
|
|
|
|
def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""深度合并字典: override 覆盖 base."""
|
|
if not isinstance(base, dict):
|
|
return deepcopy(override) if isinstance(override, dict) else deepcopy(base)
|
|
|
|
result = deepcopy(base)
|
|
if not isinstance(override, dict):
|
|
return result
|
|
|
|
for key, val in override.items():
|
|
if isinstance(val, dict) and isinstance(result.get(key), dict):
|
|
result[key] = _deep_merge(result[key], val)
|
|
else:
|
|
result[key] = val
|
|
return result
|
|
|
|
|
|
def _migrate_deprecated_config(
|
|
config: Dict[str, Any], valid_sections: set
|
|
) -> tuple[Dict[str, Any], set]:
|
|
"""
|
|
迁移废弃的配置节到新配置结构
|
|
|
|
Returns:
|
|
(迁移后的配置, 废弃的配置节集合)
|
|
"""
|
|
# 配置映射规则：旧配置 -> 新配置
|
|
MIGRATION_MAP = {
|
|
# grok.* -> 对应的新配置节
|
|
"grok.temporary": "app.temporary",
|
|
"grok.disable_memory": "app.disable_memory",
|
|
"grok.stream": "app.stream",
|
|
"grok.thinking": "app.thinking",
|
|
"grok.dynamic_statsig": "app.dynamic_statsig",
|
|
"grok.filter_tags": "app.filter_tags",
|
|
"chat.capture_enabled": "app.chat_capture_enabled",
|
|
"chat.capture_file": "app.chat_capture_file",
|
|
"grok.timeout": "voice.timeout",
|
|
"grok.base_proxy_url": "proxy.base_proxy_url",
|
|
"grok.asset_proxy_url": "proxy.asset_proxy_url",
|
|
"network.base_proxy_url": "proxy.base_proxy_url",
|
|
"network.asset_proxy_url": "proxy.asset_proxy_url",
|
|
"grok.cf_clearance": "proxy.cf_clearance",
|
|
"grok.browser": "proxy.browser",
|
|
"grok.user_agent": "proxy.user_agent",
|
|
"security.cf_clearance": "proxy.cf_clearance",
|
|
"security.browser": "proxy.browser",
|
|
"security.user_agent": "proxy.user_agent",
|
|
"grok.max_retry": "retry.max_retry",
|
|
"grok.retry_status_codes": "retry.retry_status_codes",
|
|
"grok.retry_backoff_base": "retry.retry_backoff_base",
|
|
"grok.retry_backoff_factor": "retry.retry_backoff_factor",
|
|
"grok.retry_backoff_max": "retry.retry_backoff_max",
|
|
"grok.retry_budget": "retry.retry_budget",
|
|
"grok.video_idle_timeout": "video.stream_timeout",
|
|
"grok.image_ws_nsfw": "image.nsfw",
|
|
"grok.image_ws_blocked_seconds": "image.final_timeout",
|
|
"grok.image_ws_final_min_bytes": "image.final_min_bytes",
|
|
"grok.image_ws_medium_min_bytes": "image.medium_min_bytes",
|
|
# legacy sections
|
|
"network.base_proxy_url": "proxy.base_proxy_url",
|
|
"network.asset_proxy_url": "proxy.asset_proxy_url",
|
|
"network.timeout": [
|
|
"chat.timeout",
|
|
"image.timeout",
|
|
"video.timeout",
|
|
"voice.timeout",
|
|
],
|
|
"security.cf_clearance": "proxy.cf_clearance",
|
|
"security.browser": "proxy.browser",
|
|
"security.user_agent": "proxy.user_agent",
|
|
"timeout.stream_idle_timeout": [
|
|
"chat.stream_timeout",
|
|
"image.stream_timeout",
|
|
"video.stream_timeout",
|
|
],
|
|
"timeout.video_idle_timeout": "video.stream_timeout",
|
|
"image.image_ws_nsfw": "image.nsfw",
|
|
"image.image_ws_blocked_seconds": "image.final_timeout",
|
|
"image.image_ws_final_min_bytes": "image.final_min_bytes",
|
|
"image.image_ws_medium_min_bytes": "image.medium_min_bytes",
|
|
"performance.assets_max_concurrent": [
|
|
"asset.upload_concurrent",
|
|
"asset.download_concurrent",
|
|
"asset.list_concurrent",
|
|
"asset.delete_concurrent",
|
|
],
|
|
"performance.assets_delete_batch_size": "asset.delete_batch_size",
|
|
"performance.assets_batch_size": "asset.list_batch_size",
|
|
"performance.media_max_concurrent": ["chat.concurrent", "video.concurrent"],
|
|
"performance.usage_max_concurrent": "usage.concurrent",
|
|
"performance.usage_batch_size": "usage.batch_size",
|
|
"performance.nsfw_max_concurrent": "nsfw.concurrent",
|
|
"performance.nsfw_batch_size": "nsfw.batch_size",
|
|
}
|
|
|
|
deprecated_sections = set(config.keys()) - valid_sections
|
|
if not deprecated_sections:
|
|
return config, set()
|
|
|
|
result = {k: deepcopy(v) for k, v in config.items() if k in valid_sections}
|
|
migrated_count = 0
|
|
|
|
# 处理废弃配置节或旧配置键
|
|
for old_section, old_values in config.items():
|
|
if not isinstance(old_values, dict):
|
|
continue
|
|
for old_key, old_value in old_values.items():
|
|
old_path = f"{old_section}.{old_key}"
|
|
new_paths = MIGRATION_MAP.get(old_path)
|
|
if not new_paths:
|
|
continue
|
|
if isinstance(new_paths, str):
|
|
new_paths = [new_paths]
|
|
for new_path in new_paths:
|
|
try:
|
|
new_section, new_key = new_path.split(".", 1)
|
|
if new_section not in result:
|
|
result[new_section] = {}
|
|
if new_key not in result[new_section]:
|
|
result[new_section][new_key] = old_value
|
|
migrated_count += 1
|
|
logger.debug(
|
|
f"Migrated config: {old_path} -> {new_path} = {old_value}"
|
|
)
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"Skip config migration for {old_path}: {e}"
|
|
)
|
|
continue
|
|
if isinstance(result.get(old_section), dict):
|
|
result[old_section].pop(old_key, None)
|
|
|
|
# 兼容旧 chat.* 配置键迁移到 app.*
|
|
legacy_chat_map = {
|
|
"temporary": "temporary",
|
|
"disable_memory": "disable_memory",
|
|
"stream": "stream",
|
|
"thinking": "thinking",
|
|
"dynamic_statsig": "dynamic_statsig",
|
|
"filter_tags": "filter_tags",
|
|
"capture_enabled": "chat_capture_enabled",
|
|
"capture_file": "chat_capture_file",
|
|
}
|
|
chat_section = config.get("chat")
|
|
if isinstance(chat_section, dict):
|
|
app_section = result.setdefault("app", {})
|
|
for old_key, new_key in legacy_chat_map.items():
|
|
if old_key in chat_section:
|
|
if new_key not in app_section:
|
|
app_section[new_key] = chat_section[old_key]
|
|
if isinstance(result.get("chat"), dict):
|
|
result["chat"].pop(old_key, None)
|
|
migrated_count += 1
|
|
logger.debug(
|
|
f"Migrated config: chat.{old_key} -> app.{new_key} = {chat_section[old_key]}"
|
|
)
|
|
|
|
if migrated_count > 0:
|
|
logger.info(
|
|
f"Migrated {migrated_count} config items from deprecated/legacy sections"
|
|
)
|
|
|
|
return result, deprecated_sections
|
|
|
|
|
|
def _load_defaults() -> Dict[str, Any]:
|
|
"""加载默认配置文件"""
|
|
if not DEFAULT_CONFIG_FILE.exists():
|
|
return {}
|
|
try:
|
|
with DEFAULT_CONFIG_FILE.open("rb") as f:
|
|
return tomllib.load(f)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load defaults from {DEFAULT_CONFIG_FILE}: {e}")
|
|
return {}
|
|
|
|
|
|
class Config:
|
|
"""配置管理器"""
|
|
|
|
_instance = None
|
|
_config = {}
|
|
|
|
def __init__(self):
|
|
self._config = {}
|
|
self._defaults = {}
|
|
self._code_defaults = {}
|
|
self._defaults_loaded = False
|
|
|
|
def register_defaults(self, defaults: Dict[str, Any]):
|
|
"""注册代码中定义的默认值"""
|
|
self._code_defaults = _deep_merge(self._code_defaults, defaults)
|
|
|
|
def _ensure_defaults(self):
|
|
if self._defaults_loaded:
|
|
return
|
|
file_defaults = _load_defaults()
|
|
# 合并文件默认值和代码默认值（代码默认值优先级更低）
|
|
self._defaults = _deep_merge(self._code_defaults, file_defaults)
|
|
self._defaults_loaded = True
|
|
|
|
async def load(self):
|
|
"""显式加载配置"""
|
|
try:
|
|
from app.core.storage import get_storage, LocalStorage
|
|
|
|
self._ensure_defaults()
|
|
|
|
storage = get_storage()
|
|
config_data = await storage.load_config()
|
|
from_remote = True
|
|
|
|
# 从本地 data/config.toml 初始化后端
|
|
if config_data is None:
|
|
local_storage = LocalStorage()
|
|
from_remote = False
|
|
try:
|
|
# 尝试读取本地配置
|
|
config_data = await local_storage.load_config()
|
|
except Exception as e:
|
|
logger.info(f"Failed to auto-init config from local: {e}")
|
|
config_data = {}
|
|
|
|
config_data = config_data or {}
|
|
|
|
# 检查是否有废弃的配置节
|
|
valid_sections = set(self._defaults.keys())
|
|
config_data, deprecated_sections = _migrate_deprecated_config(
|
|
config_data, valid_sections
|
|
)
|
|
if deprecated_sections:
|
|
logger.info(
|
|
f"Cleaned deprecated config sections: {deprecated_sections}"
|
|
)
|
|
|
|
merged = _deep_merge(self._defaults, config_data)
|
|
|
|
# 自动回填缺失配置到存储
|
|
# 或迁移了配置后需要更新
|
|
should_persist = (
|
|
(not from_remote) or (merged != config_data) or deprecated_sections
|
|
)
|
|
if should_persist:
|
|
async with storage.acquire_lock("config_save", timeout=10):
|
|
await storage.save_config(merged)
|
|
if not from_remote:
|
|
logger.info(
|
|
f"Initialized remote storage ({storage.__class__.__name__}) with config baseline."
|
|
)
|
|
if deprecated_sections:
|
|
logger.info("Configuration automatically migrated and cleaned.")
|
|
|
|
self._config = merged
|
|
except Exception as e:
|
|
logger.error(f"Error loading config: {e}")
|
|
self._config = {}
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""
|
|
获取配置值
|
|
|
|
Args:
|
|
key: 配置键，格式 "section.key"
|
|
default: 默认值
|
|
"""
|
|
if "." in key:
|
|
try:
|
|
section, attr = key.split(".", 1)
|
|
return self._config.get(section, {}).get(attr, default)
|
|
except (ValueError, AttributeError):
|
|
return default
|
|
|
|
return self._config.get(key, default)
|
|
|
|
async def update(self, new_config: dict):
|
|
"""更新配置"""
|
|
from app.core.storage import get_storage
|
|
|
|
storage = get_storage()
|
|
async with storage.acquire_lock("config_save", timeout=10):
|
|
self._ensure_defaults()
|
|
previous = deepcopy(self._config or {})
|
|
base = _deep_merge(self._defaults, self._config or {})
|
|
merged = _deep_merge(base, new_config or {})
|
|
await storage.save_config(merged)
|
|
self._config = merged
|
|
self._apply_runtime_updates(previous, self._config)
|
|
|
|
def _apply_runtime_updates(self, previous: Dict[str, Any], current: Dict[str, Any]):
|
|
"""配置变更后立即应用可热更新的运行时设置。"""
|
|
prev_app = previous.get("app") if isinstance(previous, dict) else {}
|
|
curr_app = current.get("app") if isinstance(current, dict) else {}
|
|
prev_app = prev_app if isinstance(prev_app, dict) else {}
|
|
curr_app = curr_app if isinstance(curr_app, dict) else {}
|
|
|
|
log_related_keys = {
|
|
"app_log_enabled",
|
|
"chat_capture_enabled",
|
|
"chat_capture_file",
|
|
}
|
|
if any(prev_app.get(key) != curr_app.get(key) for key in log_related_keys):
|
|
try:
|
|
from app.core.logger import setup_logging
|
|
import os
|
|
|
|
setup_logging(
|
|
level=os.getenv("LOG_LEVEL", "INFO"),
|
|
json_console=False,
|
|
file_logging=bool(curr_app.get("app_log_enabled", True)),
|
|
)
|
|
logger.info("Applied runtime log configuration update.")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to hot-apply log configuration: {e}")
|
|
|
|
|
|
# 全局配置实例
|
|
config = Config()
|
|
|
|
|
|
def get_config(key: str, default: Any = None) -> Any:
|
|
"""获取配置"""
|
|
return config.get(key, default)
|
|
|
|
|
|
def register_defaults(defaults: Dict[str, Any]):
|
|
"""注册默认配置"""
|
|
config.register_defaults(defaults)
|
|
|
|
|
|
__all__ = ["Config", "config", "get_config", "register_defaults"]
|