#!/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@jeason.online/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 = "qwen" api_key: str = "sk-c0dbefa1718d46eaa897199135066f00" base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1" model: str = "qwen-plus-latest" 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()