Initial commit: 股吧人气指示器 - PySide6桌面悬浮工具
This commit is contained in:
172
config_manager.py
Normal file
172
config_manager.py
Normal file
@@ -0,0 +1,172 @@
|
||||
"""
|
||||
配置管理模块 - 负责配置的读取、验证和持久化
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Dict
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class ConfigManager:
|
||||
"""配置管理器"""
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"llm_api": {
|
||||
"base_url": "https://api.openai.com/v1",
|
||||
"api_key": "",
|
||||
"model": "gpt-3.5-turbo",
|
||||
"timeout": 30,
|
||||
"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
|
||||
},
|
||||
"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()
|
||||
|
||||
def _load_config(self) -> Dict[str, Any]:
|
||||
"""加载配置文件"""
|
||||
if self.config_path.exists():
|
||||
try:
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
loaded_config = json.load(f)
|
||||
# 合并默认配置，确保所有键都存在
|
||||
return self._merge_config(self.DEFAULT_CONFIG, loaded_config)
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
print(f"配置文件加载失败，使用默认配置: {e}")
|
||||
return self.DEFAULT_CONFIG.copy()
|
||||
else:
|
||||
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:
|
||||
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.config, f, ensure_ascii=False, indent=4)
|
||||
return True
|
||||
except IOError as e:
|
||||
print(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
|
||||
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):
|
||||
"""更新爬虫配置"""
|
||||
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
|
||||
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
|
||||
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"]
|
||||
Reference in New Issue
Block a user