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