xhs_factory/services/config_manager.py

"""
配置管理模块
支持多配置项、默认值回退、自动保存
"""
import json
import os
import tempfile
import logging

logger = logging.getLogger(__name__)

# 敏感字段 keyring 存储常量
_KEYRING_PLACEHOLDER = "[keyring]"
_KEYRING_SERVICE = "autobot_xhs"

_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CONFIG_FILE = os.path.join(_PROJECT_ROOT, "config", "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": [],
    "use_smart_weights": True,
    # SD 图片生成参数
    "quality_mode": "标准 (约1分钟)",
    "sd_steps": 20,
    "sd_cfg_scale": 5.5,
    "sd_negative_prompt": "",
    # 自动运营调度参数
    "sched_comment_on": True,
    "sched_like_on": True,
    "sched_fav_on": True,
    "sched_reply_on": True,
    "sched_publish_on": True,
    "sched_c_min": 15,
    "sched_c_max": 45,
    "sched_l_min": 10,
    "sched_l_max": 30,
    "sched_like_count": 5,
    "sched_fav_min": 12,
    "sched_fav_max": 35,
    "sched_fav_count": 3,
    "sched_r_min": 20,
    "sched_r_max": 60,
    "sched_reply_max": 3,
    "sched_p_min": 60,
    "sched_p_max": 180,
    "sched_start_hour": 8,
    "sched_end_hour": 23,
    "auto_like_count": 5,
    "auto_fav_count": 3,
    "auto_reply_max": 5,
    # 智能学习参数
    "learn_interval": 6,
    # 内容排期参数
    "queue_gen_count": 3,
    # 热点自动采集参数
    "hotspot_auto_collect": {
        "enabled": False,
        "keywords": ["穿搭", "美妆", "好物"],
        "interval_hours": 4,
    },
}


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):
        """原子写：临时文件 + os.replace，防止写中断导致数据损坏"""
        import time
        config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE)) or "."
        try:
            fd, tmp_path = tempfile.mkstemp(dir=config_dir, suffix=".tmp", prefix="config_")
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as f:
                    json.dump(self._config, f, indent=4, ensure_ascii=False)
                # Windows 下目标文件被占用时 os.replace 会抛 PermissionError(WinError 5)，重试最多 3 次
                for attempt in range(3):
                    try:
                        os.replace(tmp_path, CONFIG_FILE)
                        break
                    except PermissionError:
                        if attempt == 2:
                            raise
                        time.sleep(0.1)
            except Exception:
                try:
                    os.remove(tmp_path)
                except OSError:
                    pass
                raise
        except IOError as e:
            logger.error("配置保存失败: %s", e)

    def get_secure(self, key: str, default: str = "") -> str:
        """读取敏感配置，优先级: 环境变量 AUTOBOT_<KEY> > keyring > config.json 明文(自动迁移)"""
        env_val = os.environ.get(f"AUTOBOT_{key.upper()}")
        if env_val:
            return env_val
        try:
            import keyring
            kr_val = keyring.get_password(_KEYRING_SERVICE, key)
            if kr_val is not None:
                return kr_val
        except Exception as e:
            logger.warning("keyring 读取失败，使用明文回退: %s", e)
        current = self._config.get(key, default)
        if current and current != _KEYRING_PLACEHOLDER:
            # 有明文值，尝试自动迁移到 keyring
            try:
                import keyring
                keyring.set_password(_KEYRING_SERVICE, key, current)
                self._config[key] = _KEYRING_PLACEHOLDER
                self.save()
                logger.info("已将 '%s' 自动迁移至系统 keyring", key)
            except Exception as e:
                logger.warning("keyring 迁移失败，继续使用明文: %s", e)
            return current
        return default

    def set_secure(self, key: str, value: str):
        """将敏感值存入 keyring（不可用时降级为明文），config.json 中写入占位符"""
        try:
            import keyring
            keyring.set_password(_KEYRING_SERVICE, key, value)
            self._config[key] = _KEYRING_PLACEHOLDER
            self.save()
        except Exception as e:
            logger.warning("keyring 不可用，明文保存 '%s': %s", key, e)
            self._config[key] = value
            self.save()

    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 提供商配置（自动解析 keyring 中的 api_key）"""
        active_name = self._config.get("active_llm", "")
        providers = self.get_llm_providers()
        provider = None
        for p in providers:
            if p["name"] == active_name:
                provider = p
                break
        if provider is None:
            provider = providers[0] if providers else None
        if provider is None:
            return None
        # 解析 api_key：占位符时从 keyring 读取真实值
        api_key = provider.get("api_key", "")
        if api_key == _KEYRING_PLACEHOLDER:
            secure_key = f"provider_{provider['name']}_api_key"
            api_key = self.get_secure(secure_key, "")
        return {**provider, "api_key": api_key}

    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}」已存在，请换一个"
        # 安全存储 api_key
        stored_key = api_key.strip()
        secure_key = f"provider_{name}_api_key"
        try:
            import keyring
            keyring.set_password(_KEYRING_SERVICE, secure_key, stored_key)
            stored_key = _KEYRING_PLACEHOLDER
        except Exception as e:
            logger.warning("keyring 不可用，api_key 明文存储: %s", e)
        providers.append({
            "name": name,
            "api_key": stored_key,
            "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()