xiaji
f256bd0852
重构配置管理器以支持打包环境路径处理 将LLM分析器从OpenAI迁移至智谱AI API 替换Playwright截图功能为Selenium实现 更新默认配置中的API端点和模型
197 lines
6.9 KiB
Python
197 lines
6.9 KiB
Python
"""
|
||
配置管理模块 - 负责配置的读取、验证和持久化
|
||
"""
|
||
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"]
|