297 lines
9.8 KiB
Python
297 lines
9.8 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
统一配置管理模块
|
||
整合所有配置，提供统一的配置接口
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import logging
|
||
from typing import Dict, Any, Optional
|
||
from dataclasses import dataclass, asdict
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
@dataclass
|
||
class DatabaseConfig:
|
||
"""数据库配置"""
|
||
url: str = "mysql+pymysql://tsp_assistant:password@43.134.68.207/tsp_assistant?charset=utf8mb4"
|
||
pool_size: int = 10
|
||
max_overflow: int = 20
|
||
pool_timeout: int = 30
|
||
pool_recycle: int = 3600
|
||
|
||
@dataclass
|
||
class LLMConfig:
|
||
"""LLM配置"""
|
||
provider: str = "openai"
|
||
api_key: str = ""
|
||
base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
|
||
model: str = "qwen-turbo"
|
||
temperature: float = 0.7
|
||
max_tokens: int = 2000
|
||
timeout: int = 30
|
||
|
||
@dataclass
|
||
class ServerConfig:
|
||
"""服务器配置"""
|
||
host: str = "0.0.0.0"
|
||
port: int = 5000
|
||
websocket_port: int = 8765
|
||
debug: bool = False
|
||
log_level: str = "INFO"
|
||
|
||
@dataclass
|
||
class FeishuConfig:
|
||
"""飞书配置"""
|
||
app_id: str = ""
|
||
app_secret: str = ""
|
||
app_token: str = ""
|
||
table_id: str = ""
|
||
status: str = "active"
|
||
sync_limit: int = 10
|
||
auto_sync_interval: int = 0
|
||
|
||
@dataclass
|
||
class AIAccuracyConfig:
|
||
"""AI准确率配置"""
|
||
auto_approve_threshold: float = 0.95
|
||
use_human_resolution_threshold: float = 0.90
|
||
manual_review_threshold: float = 0.80
|
||
ai_suggestion_confidence: float = 0.95
|
||
human_resolution_confidence: float = 0.90
|
||
prefer_human_when_low_accuracy: bool = True
|
||
enable_auto_approval: bool = True
|
||
enable_human_fallback: bool = True
|
||
|
||
@dataclass
|
||
class SystemConfig:
|
||
"""系统配置"""
|
||
backup_enabled: bool = True
|
||
backup_interval: int = 24 # 小时
|
||
max_backup_files: int = 7
|
||
cache_enabled: bool = True
|
||
cache_ttl: int = 3600 # 秒
|
||
monitoring_enabled: bool = True
|
||
|
||
class UnifiedConfig:
|
||
"""统一配置管理器"""
|
||
|
||
def __init__(self, config_dir: str = "config"):
|
||
self.config_dir = Path(config_dir)
|
||
self.config_file = self.config_dir / "unified_config.json"
|
||
|
||
# 默认配置 - 从config/llm_config.py加载默认LLM配置
|
||
self.database = DatabaseConfig()
|
||
self.llm = self._load_default_llm_config()
|
||
self.server = ServerConfig()
|
||
self.feishu = FeishuConfig()
|
||
self.ai_accuracy = AIAccuracyConfig()
|
||
self.system = SystemConfig()
|
||
|
||
# 加载配置
|
||
self.load_config()
|
||
|
||
def _load_default_llm_config(self) -> LLMConfig:
|
||
"""加载默认LLM配置"""
|
||
try:
|
||
from config.llm_config import DEFAULT_CONFIG
|
||
# 将config/llm_config.py中的配置转换为统一配置的格式
|
||
return LLMConfig(
|
||
provider=DEFAULT_CONFIG.provider,
|
||
api_key=DEFAULT_CONFIG.api_key,
|
||
base_url=DEFAULT_CONFIG.base_url,
|
||
model=DEFAULT_CONFIG.model,
|
||
temperature=DEFAULT_CONFIG.temperature,
|
||
max_tokens=DEFAULT_CONFIG.max_tokens
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"无法加载默认LLM配置，使用内置默认值: {e}")
|
||
return LLMConfig()
|
||
|
||
def load_config(self):
|
||
"""加载配置文件"""
|
||
try:
|
||
if self.config_file.exists():
|
||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||
config_data = json.load(f)
|
||
|
||
# 更新配置
|
||
if 'database' in config_data:
|
||
self.database = DatabaseConfig(**config_data['database'])
|
||
if 'llm' in config_data:
|
||
self.llm = LLMConfig(**config_data['llm'])
|
||
if 'server' in config_data:
|
||
self.server = ServerConfig(**config_data['server'])
|
||
if 'feishu' in config_data:
|
||
self.feishu = FeishuConfig(**config_data['feishu'])
|
||
if 'ai_accuracy' in config_data:
|
||
self.ai_accuracy = AIAccuracyConfig(**config_data['ai_accuracy'])
|
||
if 'system' in config_data:
|
||
self.system = SystemConfig(**config_data['system'])
|
||
|
||
logger.info("配置文件加载成功")
|
||
else:
|
||
logger.info("配置文件不存在，使用默认配置")
|
||
self.save_config()
|
||
|
||
except Exception as e:
|
||
logger.error(f"加载配置文件失败: {e}")
|
||
|
||
def save_config(self):
|
||
"""保存配置文件"""
|
||
try:
|
||
self.config_dir.mkdir(exist_ok=True)
|
||
|
||
config_data = {
|
||
'database': asdict(self.database),
|
||
'llm': asdict(self.llm),
|
||
'server': asdict(self.server),
|
||
'feishu': asdict(self.feishu),
|
||
'ai_accuracy': asdict(self.ai_accuracy),
|
||
'system': asdict(self.system)
|
||
}
|
||
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(config_data, f, indent=2, ensure_ascii=False)
|
||
|
||
logger.info("配置文件保存成功")
|
||
|
||
except Exception as e:
|
||
logger.error(f"保存配置文件失败: {e}")
|
||
|
||
def load_from_env(self):
|
||
"""从环境变量加载配置"""
|
||
# 数据库配置
|
||
if os.getenv('DATABASE_URL'):
|
||
self.database.url = os.getenv('DATABASE_URL')
|
||
|
||
# LLM配置
|
||
if os.getenv('LLM_PROVIDER'):
|
||
self.llm.provider = os.getenv('LLM_PROVIDER')
|
||
if os.getenv('LLM_API_KEY'):
|
||
self.llm.api_key = os.getenv('LLM_API_KEY')
|
||
if os.getenv('LLM_MODEL'):
|
||
self.llm.model = os.getenv('LLM_MODEL')
|
||
|
||
# 服务器配置
|
||
if os.getenv('SERVER_PORT'):
|
||
self.server.port = int(os.getenv('SERVER_PORT'))
|
||
if os.getenv('LOG_LEVEL'):
|
||
self.server.log_level = os.getenv('LOG_LEVEL')
|
||
|
||
# 飞书配置
|
||
if os.getenv('FEISHU_APP_ID'):
|
||
self.feishu.app_id = os.getenv('FEISHU_APP_ID')
|
||
if os.getenv('FEISHU_APP_SECRET'):
|
||
self.feishu.app_secret = os.getenv('FEISHU_APP_SECRET')
|
||
if os.getenv('FEISHU_APP_TOKEN'):
|
||
self.feishu.app_token = os.getenv('FEISHU_APP_TOKEN')
|
||
if os.getenv('FEISHU_TABLE_ID'):
|
||
self.feishu.table_id = os.getenv('FEISHU_TABLE_ID')
|
||
|
||
def get_database_url(self) -> str:
|
||
"""获取数据库连接URL"""
|
||
return self.database.url
|
||
|
||
def get_llm_config(self) -> Dict[str, Any]:
|
||
"""获取LLM配置"""
|
||
return asdict(self.llm)
|
||
|
||
def get_server_config(self) -> Dict[str, Any]:
|
||
"""获取服务器配置"""
|
||
return asdict(self.server)
|
||
|
||
def get_feishu_config(self) -> Dict[str, Any]:
|
||
"""获取飞书配置"""
|
||
return asdict(self.feishu)
|
||
|
||
def get_ai_accuracy_config(self) -> Dict[str, Any]:
|
||
"""获取AI准确率配置"""
|
||
return asdict(self.ai_accuracy)
|
||
|
||
def get_system_config(self) -> Dict[str, Any]:
|
||
"""获取系统配置"""
|
||
return asdict(self.system)
|
||
|
||
def update_config(self, section: str, config_data: Dict[str, Any]):
|
||
"""更新配置"""
|
||
try:
|
||
if section == 'database':
|
||
self.database = DatabaseConfig(**config_data)
|
||
elif section == 'llm':
|
||
self.llm = LLMConfig(**config_data)
|
||
elif section == 'server':
|
||
self.server = ServerConfig(**config_data)
|
||
elif section == 'feishu':
|
||
self.feishu = FeishuConfig(**config_data)
|
||
elif section == 'ai_accuracy':
|
||
self.ai_accuracy = AIAccuracyConfig(**config_data)
|
||
elif section == 'system':
|
||
self.system = SystemConfig(**config_data)
|
||
else:
|
||
raise ValueError(f"未知的配置节: {section}")
|
||
|
||
self.save_config()
|
||
logger.info(f"配置节 {section} 更新成功")
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新配置失败: {e}")
|
||
raise
|
||
|
||
def validate_config(self) -> bool:
|
||
"""验证配置有效性"""
|
||
try:
|
||
# 验证数据库配置
|
||
if not self.database.url:
|
||
logger.error("数据库URL未配置")
|
||
return False
|
||
|
||
# 验证LLM配置
|
||
if not self.llm.api_key:
|
||
logger.warning("LLM API密钥未配置")
|
||
|
||
# 验证飞书配置
|
||
if self.feishu.status == "active":
|
||
if not all([self.feishu.app_id, self.feishu.app_secret,
|
||
self.feishu.app_token, self.feishu.table_id]):
|
||
logger.warning("飞书配置不完整")
|
||
|
||
logger.info("配置验证通过")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"配置验证失败: {e}")
|
||
return False
|
||
|
||
def get_all_config(self) -> Dict[str, Any]:
|
||
"""获取所有配置"""
|
||
return {
|
||
'database': asdict(self.database),
|
||
'llm': asdict(self.llm),
|
||
'server': asdict(self.server),
|
||
'feishu': asdict(self.feishu),
|
||
'ai_accuracy': asdict(self.ai_accuracy),
|
||
'system': asdict(self.system)
|
||
}
|
||
|
||
# 全局配置实例
|
||
_config_instance = None
|
||
|
||
def get_config() -> UnifiedConfig:
|
||
"""获取全局配置实例"""
|
||
global _config_instance
|
||
if _config_instance is None:
|
||
_config_instance = UnifiedConfig()
|
||
_config_instance.load_from_env()
|
||
return _config_instance
|
||
|
||
def reload_config():
|
||
"""重新加载配置"""
|
||
global _config_instance
|
||
_config_instance = None
|
||
return get_config()
|