#!/usr/bin/env python # -*- coding: utf-8 -*- """ 统一配置管理模块 从环境变量加载所有配置,提供统一的配置接口 """ import os import logging from typing import Dict, Any, Optional from dataclasses import dataclass, asdict from dotenv import load_dotenv # 在模块加载时,自动从.env文件加载环境变量 # 这使得所有后续的os.getenv调用都能获取到.env中定义的值 load_dotenv() logger = logging.getLogger(__name__) # --- 数据类定义 --- # 这些类定义了配置的结构,但不包含敏感的默认值。 # 默认值只用于那些不敏感或在大多数环境中都相同的值。 @dataclass class DatabaseConfig: """数据库配置""" url: str pool_size: int = 10 max_overflow: int = 20 pool_timeout: int = 30 pool_recycle: int = 600 # 改为 10 分钟回收连接,避免连接超时 @dataclass class LLMConfig: """LLM配置""" provider: str api_key: str model: str base_url: Optional[str] = None 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: Optional[str] = None app_secret: Optional[str] = None app_token: Optional[str] = None verification_token: Optional[str] = None encrypt_key: Optional[str] = None table_id: Optional[str] = None @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 # --- 统一配置管理器 --- class UnifiedConfig: """ 统一配置管理器 在实例化时,从环境变量中加载所有配置。 """ def __init__(self): logger.info("Initializing unified configuration from environment variables...") self.database = self._load_database_from_env() self.llm = self._load_llm_from_env() self.server = self._load_server_from_env() self.feishu = self._load_feishu_from_env() self.ai_accuracy = self._load_ai_accuracy_from_env() self.validate_config() def _load_database_from_env(self) -> DatabaseConfig: db_url = os.getenv("DATABASE_URL") if not db_url: raise ValueError("FATAL: DATABASE_URL environment variable is not set.") logger.info("Database config loaded.") return DatabaseConfig(url=db_url) def _load_llm_from_env(self) -> LLMConfig: api_key = os.getenv("LLM_API_KEY") if not api_key: logger.warning("LLM_API_KEY is not set. LLM functionality will be disabled or fail.") config = LLMConfig( provider=os.getenv("LLM_PROVIDER", "qwen"), api_key=api_key, model=os.getenv("LLM_MODEL", "qwen-plus-latest"), base_url=os.getenv("LLM_BASE_URL"), temperature=float(os.getenv("LLM_TEMPERATURE", 0.7)), max_tokens=int(os.getenv("LLM_MAX_TOKENS", 2000)), timeout=int(os.getenv("LLM_TIMEOUT", 30)) ) logger.info("LLM config loaded.") return config def _load_server_from_env(self) -> ServerConfig: config = ServerConfig( host=os.getenv("SERVER_HOST", "0.0.0.0"), port=int(os.getenv("SERVER_PORT", 5000)), websocket_port=int(os.getenv("WEBSOCKET_PORT", 8765)), debug=os.getenv("DEBUG_MODE", "False").lower() in ('true', '1', 't'), log_level=os.getenv("LOG_LEVEL", "INFO").upper() ) logger.info("Server config loaded.") return config def _load_feishu_from_env(self) -> FeishuConfig: config = FeishuConfig( app_id=os.getenv("FEISHU_APP_ID"), app_secret=os.getenv("FEISHU_APP_SECRET"), app_token=os.getenv("FEISHU_APP_TOKEN"), verification_token=os.getenv("FEISHU_VERIFICATION_TOKEN"), encrypt_key=os.getenv("FEISHU_ENCRYPT_KEY"), table_id=os.getenv("FEISHU_TABLE_ID") ) logger.info("Feishu config loaded.") return config def _load_ai_accuracy_from_env(self) -> AIAccuracyConfig: config = AIAccuracyConfig( auto_approve_threshold=float(os.getenv("AI_AUTO_APPROVE_THRESHOLD", 0.95)), use_human_resolution_threshold=float(os.getenv("AI_USE_HUMAN_RESOLUTION_THRESHOLD", 0.90)), manual_review_threshold=float(os.getenv("AI_MANUAL_REVIEW_THRESHOLD", 0.80)), ai_suggestion_confidence=float(os.getenv("AI_SUGGESTION_CONFIDENCE", 0.95)), human_resolution_confidence=float(os.getenv("AI_HUMAN_RESOLUTION_CONFIDENCE", 0.90)) ) logger.info("AI Accuracy config loaded.") return config def validate_config(self): """在启动时验证关键配置""" if not self.database.url: raise ValueError("Database URL is missing.") if not self.llm.api_key: logger.warning("LLM API key is not configured. AI features may fail.") if self.feishu.app_id and not self.feishu.app_secret: logger.warning("FEISHU_APP_ID is set, but FEISHU_APP_SECRET is missing.") logger.info("Configuration validation passed (warnings may exist).") # --- Public Getters --- 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), } # --- 全局单例模式 --- _config_instance: Optional[UnifiedConfig] = None def get_config() -> UnifiedConfig: """ 获取全局统一配置实例。 第一次调用时,它会创建并加载配置。后续调用将返回缓存的实例。 """ global _config_instance if _config_instance is None: _config_instance = UnifiedConfig() return _config_instance def reload_config() -> UnifiedConfig: """强制重新加载配置""" global _config_instance # 强制重新读取 .env 文件并覆盖当前环境变量 load_dotenv(override=True) _config_instance = None return get_config()