""" 配置管理模块 支持多配置项、默认值回退、自动保存 """ import json import os import logging logger = logging.getLogger(__name__) CONFIG_FILE = "config.json" OUTPUT_DIR = "xhs_workspace" DEFAULT_CONFIG = { "api_key": "", "base_url": "https://api.openai.com/v1", "sd_url": "http://127.0.0.1:7860", "mcp_url": "http://localhost:18060/mcp", "model": "gpt-3.5-turbo", "persona": "温柔知性的时尚博主", "auto_reply_enabled": False, "schedule_enabled": False, "my_user_id": "", "active_llm": "", "llm_providers": [], } class ConfigManager: """配置管理器 - 单例模式""" _instance = None _config = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): if self._config is None: self._config = self._load() def _load(self) -> dict: """从文件加载配置,缺失项用默认值填充""" config = DEFAULT_CONFIG.copy() if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: saved = json.load(f) config.update(saved) except (json.JSONDecodeError, IOError) as e: logger.warning("配置文件读取失败,使用默认值: %s", e) return config def save(self): """保存配置到文件""" try: with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(self._config, f, indent=4, ensure_ascii=False) except IOError as e: logger.error("配置保存失败: %s", e) def get(self, key: str, default=None): """获取配置项""" return self._config.get(key, default) def set(self, key: str, value): """设置配置项并自动保存""" self._config[key] = value self.save() def update(self, data: dict): """批量更新配置""" self._config.update(data) self.save() @property def all(self) -> dict: """返回全部配置(副本)""" return self._config.copy() def ensure_workspace(self): """确保工作空间目录存在""" os.makedirs(OUTPUT_DIR, exist_ok=True) # ---------- 多 LLM 提供商管理 ---------- def get_llm_providers(self) -> list[dict]: """获取所有 LLM 提供商配置""" providers = self._config.get("llm_providers", []) # 兼容旧配置: 如果 providers 为空但有 api_key,自动迁移 if not providers and self._config.get("api_key"): default_provider = { "name": "默认", "api_key": self._config["api_key"], "base_url": self._config.get("base_url", "https://api.openai.com/v1"), } providers = [default_provider] self._config["llm_providers"] = providers self._config["active_llm"] = "默认" self.save() return providers def get_llm_provider_names(self) -> list[str]: """获取所有提供商名称列表""" return [p["name"] for p in self.get_llm_providers()] def get_active_llm(self) -> dict | None: """获取当前激活的 LLM 提供商配置""" active_name = self._config.get("active_llm", "") for p in self.get_llm_providers(): if p["name"] == active_name: return p # 没找到就返回第一个 providers = self.get_llm_providers() return providers[0] if providers else None def add_llm_provider(self, name: str, api_key: str, base_url: str) -> str: """添加一个 LLM 提供商,返回状态消息""" name = name.strip() if not name: return "❌ 名称不能为空" if not api_key.strip(): return "❌ API Key 不能为空" providers = self.get_llm_providers() for p in providers: if p["name"] == name: return f"❌ 名称「{name}」已存在,请换一个" providers.append({ "name": name, "api_key": api_key.strip(), "base_url": (base_url or "https://api.openai.com/v1").strip().rstrip("/"), }) self._config["llm_providers"] = providers if not self._config.get("active_llm"): self._config["active_llm"] = name self.save() return f"✅ 已添加「{name}」" def remove_llm_provider(self, name: str) -> str: """删除一个 LLM 提供商""" providers = self.get_llm_providers() new_providers = [p for p in providers if p["name"] != name] if len(new_providers) == len(providers): return f"⚠️ 未找到「{name}」" self._config["llm_providers"] = new_providers if self._config.get("active_llm") == name: self._config["active_llm"] = new_providers[0]["name"] if new_providers else "" self.save() return f"✅ 已删除「{name}」" def set_active_llm(self, name: str): """切换当前激活的 LLM 提供商""" self._config["active_llm"] = name # 同步到兼容字段 p = self.get_active_llm() if p: self._config["api_key"] = p["api_key"] self._config["base_url"] = p["base_url"] self.save()