193 lines
6.1 KiB
Python
193 lines
6.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
系统配置模块
|
|
功能：
|
|
1. 支持多平台路径适配
|
|
2. 环境变量与配置文件优先级管理
|
|
3. 敏感信息加密存储
|
|
4. 配置热更新检测
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import platform
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
import dotenv
|
|
from cryptography.fernet import Fernet
|
|
|
|
class ConfigManager:
|
|
def __init__(self, app_name: str = "intelligence_system"):
|
|
"""
|
|
初始化配置管理器
|
|
|
|
参数：
|
|
app_name: 应用名称（用于生成配置目录）
|
|
"""
|
|
self.system = platform.system().lower()
|
|
self.app_name = app_name
|
|
self._config = {}
|
|
self._secret_key = None
|
|
|
|
# 初始化配置路径
|
|
self.config_dir = self._get_config_dir()
|
|
os.makedirs(self.config_dir, exist_ok=True)
|
|
|
|
# 加载配置顺序
|
|
self._load_defaults()
|
|
self._load_env_file()
|
|
self._load_user_config()
|
|
|
|
# 初始化加密模块
|
|
self._init_encryption()
|
|
|
|
def _get_config_dir(self) -> str:
|
|
"""获取适合当前平台的配置目录"""
|
|
if self.system == 'windows':
|
|
return os.path.join(os.environ['APPDATA'], self.app_name)
|
|
elif self.system == 'darwin': # macOS
|
|
return os.path.expanduser(f"~/Library/Application Support/{self.app_name}")
|
|
else: # Linux及其他Unix-like
|
|
return os.path.expanduser(f"~/.config/{self.app_name}")
|
|
|
|
def _load_defaults(self):
|
|
"""加载默认配置"""
|
|
self._config = {
|
|
"system": {
|
|
"log_level": "INFO",
|
|
"max_threads": os.cpu_count() or 4
|
|
},
|
|
"api": {
|
|
"newsapi": {"endpoint": "https://newsapi.org/v2"},
|
|
"weibo": {"version": "2"}
|
|
},
|
|
"paths": {
|
|
"data_dir": os.path.join(self.config_dir, "data"),
|
|
"cache_dir": os.path.join(self.config_dir, "cache")
|
|
}
|
|
}
|
|
|
|
def _load_env_file(self):
|
|
"""加载.env环境变量文件"""
|
|
env_path = Path(self.config_dir) / ".env"
|
|
if env_path.exists():
|
|
dotenv.load_dotenv(env_path)
|
|
|
|
# 环境变量覆盖配置
|
|
if os.getenv("LOG_LEVEL"):
|
|
self._config["system"]["log_level"] = os.getenv("LOG_LEVEL")
|
|
|
|
def _load_user_config(self):
|
|
"""加载用户自定义配置"""
|
|
config_file = Path(self.config_dir) / "config.json"
|
|
try:
|
|
if config_file.exists():
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
user_config = json.load(f)
|
|
self._deep_update(self._config, user_config)
|
|
except Exception as e:
|
|
print(f"加载用户配置失败: {str(e)}")
|
|
|
|
def _init_encryption(self):
|
|
"""初始化配置加密模块"""
|
|
key_file = Path(self.config_dir) / ".secret.key"
|
|
if key_file.exists():
|
|
with open(key_file, 'rb') as f:
|
|
self._secret_key = f.read()
|
|
else:
|
|
self._secret_key = Fernet.generate_key()
|
|
with open(key_file, 'wb') as f:
|
|
f.write(self._secret_key)
|
|
key_file.chmod(0o600) # 设置密钥文件权限
|
|
|
|
def _deep_update(self, original: Dict, update: Dict) -> Dict:
|
|
"""深度合并字典"""
|
|
for key, value in update.items():
|
|
if isinstance(value, dict) and key in original:
|
|
original[key] = self._deep_update(original.get(key, {}), value)
|
|
else:
|
|
original[key] = value
|
|
return original
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""
|
|
获取配置项（支持点分路径）
|
|
示例: get("api.newsapi.endpoint")
|
|
"""
|
|
keys = key.split('.')
|
|
value = self._config
|
|
try:
|
|
for k in keys:
|
|
value = value[k]
|
|
return value
|
|
except (KeyError, TypeError):
|
|
return default
|
|
|
|
def set(self, key: str, value: Any, persist: bool = False):
|
|
"""
|
|
设置配置项
|
|
参数：
|
|
persist: 是否保存到用户配置文件
|
|
"""
|
|
keys = key.split('.')
|
|
config_ref = self._config
|
|
|
|
for k in keys[:-1]:
|
|
if k not in config_ref:
|
|
config_ref[k] = {}
|
|
config_ref = config_ref[k]
|
|
|
|
config_ref[keys[-1]] = value
|
|
|
|
if persist:
|
|
self._save_user_config()
|
|
|
|
def encrypt_value(self, plaintext: str) -> str:
|
|
"""加密敏感信息"""
|
|
fernet = Fernet(self._secret_key)
|
|
return fernet.encrypt(plaintext.encode()).decode()
|
|
|
|
def decrypt_value(self, ciphertext: str) -> str:
|
|
"""解密敏感信息"""
|
|
fernet = Fernet(self._secret_key)
|
|
return fernet.decrypt(ciphertext.encode()).decode()
|
|
|
|
def _save_user_config(self):
|
|
"""保存用户配置到文件"""
|
|
config_file = Path(self.config_dir) / "config.json"
|
|
with open(config_file, 'w', encoding='utf-8') as f:
|
|
json.dump(self._config, f, indent=2, ensure_ascii=False)
|
|
|
|
def reload(self):
|
|
"""重新加载所有配置"""
|
|
self._config = {}
|
|
self._load_defaults()
|
|
self._load_env_file()
|
|
self._load_user_config()
|
|
|
|
# 全局配置实例
|
|
config = ConfigManager()
|
|
|
|
# 快捷访问方法（兼容旧代码）
|
|
def get_config(key: str, default: Optional[Any] = None) -> Any:
|
|
return config.get(key, default)
|
|
|
|
def set_config(key: str, value: Any, persist: bool = False):
|
|
config.set(key, value, persist)
|
|
|
|
# 测试代码
|
|
if __name__ == "__main__":
|
|
# 设置并保存API密钥（自动加密）
|
|
api_key = "your_newsapi_key_here"
|
|
encrypted_key = config.encrypt_value(api_key)
|
|
config.set("api.newsapi.key", encrypted_key, persist=True)
|
|
|
|
# 获取配置示例
|
|
print(f"日志级别: {config.get('system.log_level')}")
|
|
print(f"NewsAPI端点: {config.get('api.newsapi.endpoint')}")
|
|
|
|
# 解密敏感信息
|
|
stored_key = config.get("api.newsapi.key")
|
|
if stored_key:
|
|
print(f"解密后的API密钥: {config.decrypt_value(stored_key)}") |