xhs_factory/config_manager.py

"""
配置管理模块
支持多配置项、默认值回退、自动保存
"""
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()