guba-indicator/config_manager.py

"""
配置管理模块 - 负责配置的读取、验证和持久化
"""
import json
import os
from typing import Any, Dict
from pathlib import Path
from loguru import logger


class ConfigManager:
    """配置管理器"""

    DEFAULT_CONFIG = {
        "llm_api": {
            "base_url": "https://integrate.api.nvidia.com/v1",
            "api_key": "",
            "model": "deepseek-ai/deepseek-r1",
            "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,
            "playwright_dir": ""
        },
        "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"):
        self.config_path = Path(config_path)
        self.config = self._load_config()
        logger.info(f"配置管理器初始化完成，配置文件: {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("配置加载成功")
                    return merged
            except (json.JSONDecodeError, IOError) as e:
                logger.error(f"配置文件加载失败，使用默认配置: {e}")
                return self.DEFAULT_CONFIG.copy()
        else:
            logger.warning(f"配置文件不存在: {self.config_path}，使用默认配置")
            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, base_url: str = None, api_key: str = None,
                       model: str = None, timeout: int = None, retry_times: int = None):
        """更新LLM API配置"""
        if base_url:
            self.config["llm_api"]["base_url"] = base_url
        if api_key:
            self.config["llm_api"]["api_key"] = api_key
        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,
                      playwright_dir: 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 playwright_dir:
            self.config["spider"]["playwright_dir"] = playwright_dir
        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"]