""" 配置管理模块 - 负责配置的读取、验证和持久化 """ import json import os from typing import Any, Dict from pathlib import Path from loguru import logger class ConfigManager: """配置管理器""" DEFAULT_CONFIG = { "llm_api": { "api_key": "", "base_url": "https://open.bigmodel.cn/api/paas/v4", "model": "glm-4.7-flash", "timeout": 120, "retry_times": 3 }, "spider": { "target_url": "https://example.com", "xpath": "//a[contains(@class, 'linkblack')]", "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "fetch_interval": 60, "retry_times": 3, "retry_interval": 5, "chrome_path": "" }, "ui": { "opacity": 0.9, "is_on_top": True, "thresholds": { "cold": 30, "warm": 70 } }, "database": { "path": "guba.db" }, "logging": { "level": "INFO", "path": "guba.log" } } def __init__(self, config_path: str = "config.json"): import sys # 确定配置文件的正确路径 if getattr(sys, 'frozen', False): # 打包后的环境 current_dir = Path(sys.executable).parent self.config_path = current_dir / config_path else: # 开发环境 self.config_path = Path(config_path) self.config = self._load_config() logger.info(f"配置管理器初始化完成,配置文件: {self.config_path}") def _load_config(self) -> Dict[str, Any]: """加载配置文件""" if self.config_path.exists(): try: logger.info(f"从文件加载配置: {self.config_path}") with open(self.config_path, 'r', encoding='utf-8') as f: loaded_config = json.load(f) # 合并默认配置,确保所有键都存在 merged = self._merge_config(self.DEFAULT_CONFIG, loaded_config) logger.info(f"配置加载成功,目标URL: {merged.get('spider', {}).get('target_url', '未设置')}") return merged except (json.JSONDecodeError, IOError) as e: logger.error(f"配置文件加载失败,使用默认配置: {e}") return self.DEFAULT_CONFIG.copy() else: logger.warning(f"配置文件不存在: {self.config_path},使用默认配置") logger.warning(f"默认配置目标URL: {self.DEFAULT_CONFIG.get('spider', {}).get('target_url', '未设置')}") return self.DEFAULT_CONFIG.copy() def _merge_config(self, default: Dict, loaded: Dict) -> Dict: """递归合并配置""" result = default.copy() for key, value in loaded.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = self._merge_config(result[key], value) else: result[key] = value return result def save_config(self) -> bool: """保存配置到文件""" try: logger.debug(f"保存配置到文件: {self.config_path}") with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(self.config, f, ensure_ascii=False, indent=4) logger.info("配置保存成功") return True except IOError as e: logger.error(f"配置保存失败: {e}") return False def get(self, *keys: str, default: Any = None) -> Any: """获取嵌套配置值""" value = self.config for key in keys: if isinstance(value, dict) and key in value: value = value[key] else: return default return value def set(self, value: Any, *keys: str) -> bool: """设置嵌套配置值""" if len(keys) < 1: return False current = self.config for key in keys[:-1]: if key not in current: current[key] = {} current = current[key] current[keys[-1]] = value return self.save_config() def update_llm_api(self, api_key: str = None, base_url: str = None, model: str = None, timeout: int = None, retry_times: int = None): """更新LLM API配置""" if api_key: self.config["llm_api"]["api_key"] = api_key if base_url: self.config["llm_api"]["base_url"] = base_url if model: self.config["llm_api"]["model"] = model if timeout: self.config["llm_api"]["timeout"] = timeout if retry_times: self.config["llm_api"]["retry_times"] = retry_times logger.info("LLM API配置已更新") self.save_config() def update_spider(self, target_url: str = None, xpath: str = None, user_agent: str = None, fetch_interval: int = None, retry_times: int = None, retry_interval: int = None, chrome_path: str = None): """更新爬虫配置""" if target_url: self.config["spider"]["target_url"] = target_url if xpath: self.config["spider"]["xpath"] = xpath if user_agent: self.config["spider"]["user_agent"] = user_agent if fetch_interval: self.config["spider"]["fetch_interval"] = fetch_interval if retry_times: self.config["spider"]["retry_times"] = retry_times if retry_interval: self.config["spider"]["retry_interval"] = retry_interval if chrome_path: self.config["spider"]["chrome_path"] = chrome_path logger.info("爬虫配置已更新") self.save_config() def update_ui(self, opacity: float = None, is_on_top: bool = None, cold_threshold: int = None, warm_threshold: int = None): """更新UI配置""" if opacity is not None: self.config["ui"]["opacity"] = max(0.3, min(1.0, opacity)) if is_on_top is not None: self.config["ui"]["is_on_top"] = is_on_top if cold_threshold is not None: self.config["ui"]["thresholds"]["cold"] = cold_threshold if warm_threshold is not None: self.config["ui"]["thresholds"]["warm"] = warm_threshold logger.info("UI配置已更新") self.save_config() @property def llm_api_config(self) -> Dict: return self.config["llm_api"] @property def spider_config(self) -> Dict: return self.config["spider"] @property def ui_config(self) -> Dict: return self.config["ui"] @property def database_config(self) -> Dict: return self.config["database"] @property def logging_config(self) -> Dict: return self.config["logging"]