280 lines
9.0 KiB
Python
280 lines
9.0 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"
|
|||
|
|
|
|||
|
|
# 默认配置
|
|||
|
|
self.database = DatabaseConfig()
|
|||
|
|
self.llm = LLMConfig()
|
|||
|
|
self.server = ServerConfig()
|
|||
|
|
self.feishu = FeishuConfig()
|
|||
|
|
self.ai_accuracy = AIAccuracyConfig()
|
|||
|
|
self.system = SystemConfig()
|
|||
|
|
|
|||
|
|
# 加载配置
|
|||
|
|
self.load_config()
|
|||
|
|
|
|||
|
|
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()
|