import os from pathlib import Path from typing import Any from pydantic import BaseModel, Field def _load_dotenv_file() -> None: """环境加载:轻量读取项目根目录 `.env`,避免强依赖 python-dotenv。""" env_path = next( (parent / ".env" for parent in Path(__file__).resolve().parents if (parent / ".env").exists()), None, ) if env_path is None: return for line in env_path.read_text(encoding="utf-8").splitlines(): if not line or line.strip().startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip()) _load_dotenv_file() def _env_first(*keys: str, default: str = "") -> str: """环境读取:按优先级读取多个环境变量。""" for key in keys: value = os.getenv(key) if value: return value return default def _normalize_sync_database_url(url: str) -> str: """数据库连接:将异步 MySQL URL 转换为当前同步 ORM 可用的 URL。""" if url.startswith("mysql+aiomysql://"): return url.replace("mysql+aiomysql://", "mysql+pymysql://", 1) if url.startswith("mysql://"): return url.replace("mysql://", "mysql+pymysql://", 1) return url def _env_bool(key: str, default: bool) -> bool: """布尔配置:统一解析环境变量中的 true/false 开关。""" return os.getenv(key, str(default)).lower() == "true" def _env_csv(key: str, default: str = "") -> list[str]: """列表配置:把逗号分隔的环境变量转换为去空白列表。""" return [item.strip() for item in os.getenv(key, default).split(",") if item.strip()] class Settings(BaseModel): """系统配置:集中管理数据库、DeepSeek、报告和短期 memory 配置。""" app_name: str = Field(default_factory=lambda: os.getenv("APP_NAME", "Medical Consultation Agent Demo")) app_env: str = Field(default_factory=lambda: os.getenv("APP_ENV", "local")) app_debug: bool = Field(default_factory=lambda: _env_bool("APP_DEBUG", True)) app_root_path: str = Field(default_factory=lambda: os.getenv("APP_ROOT_PATH", "")) api_v1_prefix: str = Field(default_factory=lambda: os.getenv("API_V1_PREFIX", "/api/v1")) cors_allow_origins: list[str] = Field( default_factory=lambda: _env_csv( "CORS_ALLOW_ORIGINS", "http://127.0.0.1:5173,http://localhost:5173,http://127.0.0.1:5174,http://localhost:5174", ) ) cors_allow_origin_regex: str = Field( default_factory=lambda: os.getenv( "CORS_ALLOW_ORIGIN_REGEX", r"^http://(127\.0\.0\.1|localhost|192\.168\.\d+\.\d+):\d+$", ) ) mysql_url: str = Field(default_factory=lambda: os.getenv("MYSQL_URL", "")) database_url: str = Field( default_factory=lambda: _normalize_sync_database_url( _env_first("DATABASE_URL", "MYSQL_URL", default="sqlite:///./storage/demo.db") ) ) llm_api_key: str = Field(default_factory=lambda: _env_first("LLM_API_KEY", "DEEPSEEK_API_KEY", default="")) llm_base_url: str = Field( default_factory=lambda: _env_first("LLM_BASE_URL", "DEEPSEEK_BASE_URL", default="https://api.deepseek.com") ) llm_model: str = Field(default_factory=lambda: _env_first("LLM_MODEL", "DEEPSEEK_FAST_MODEL", default="deepseek-chat")) llm_fast_model: str = Field(default_factory=lambda: _env_first("LLM_FAST_MODEL", "LLM_MODEL", "DEEPSEEK_FAST_MODEL", default="deepseek-chat")) llm_reason_model: str = Field( default_factory=lambda: _env_first("LLM_REASON_MODEL", "LLM_MODEL", "DEEPSEEK_REASON_MODEL", default="deepseek-reasoner") ) llm_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_TIMEOUT_SECONDS", "45"))) llm_chat_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_CHAT_TIMEOUT_SECONDS", "20"))) llm_stream_first_token_timeout_seconds: int = Field( default_factory=lambda: int(os.getenv("LLM_STREAM_FIRST_TOKEN_TIMEOUT_SECONDS", "15")) ) llm_stream_total_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_STREAM_TOTAL_TIMEOUT_SECONDS", "45"))) llm_stream_enabled: bool = Field(default_factory=lambda: _env_bool("LLM_STREAM_ENABLED", True)) llm_mock_enabled: bool = Field(default_factory=lambda: _env_bool("LLM_MOCK_ENABLED", True)) llm_fallback_to_mock: bool = Field(default_factory=lambda: _env_bool("LLM_FALLBACK_TO_MOCK", True)) llm_fast_thinking_enabled: bool = Field(default_factory=lambda: _env_bool("LLM_FAST_THINKING_ENABLED", False)) llm_reason_thinking_enabled: bool = Field(default_factory=lambda: _env_bool("LLM_REASON_THINKING_ENABLED", False)) llm_reasoning_effort: str = Field(default_factory=lambda: os.getenv("LLM_REASONING_EFFORT", "low")) llm_fast_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_FAST_MAX_TOKENS", "512"))) llm_hint_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_HINT_MAX_TOKENS", "1200"))) llm_scoring_json_response: bool = Field(default_factory=lambda: _env_bool("LLM_SCORING_JSON_RESPONSE", True)) llm_scoring_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_SCORING_MAX_TOKENS", "4096"))) report_storage_dir: str = Field(default_factory=lambda: os.getenv("REPORT_STORAGE_DIR", "./storage/reports")) runtime_memory_ttl_seconds: int = Field(default_factory=lambda: int(os.getenv("RUNTIME_MEMORY_TTL_SECONDS", "7200"))) runtime_memory_backend: str = Field(default_factory=lambda: os.getenv("RUNTIME_MEMORY_BACKEND", "memory")) runtime_memory_fallback_enabled: bool = Field( default_factory=lambda: _env_bool("RUNTIME_MEMORY_FALLBACK_ENABLED", True) ) redis_url: str = Field(default_factory=lambda: os.getenv("REDIS_URL", "redis://redis:6379/0")) auth_validate_enabled: bool = Field(default_factory=lambda: _env_bool("AUTH_VALIDATE_ENABLED", True)) auth_user_me_url: str = Field(default_factory=lambda: os.getenv("AUTH_USER_ME_URL", "")) auth_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("AUTH_TIMEOUT_SECONDS", "5"))) auth_cache_ttl_seconds: int = Field(default_factory=lambda: int(os.getenv("AUTH_CACHE_TTL_SECONDS", "300"))) @property def is_production(self) -> bool: """环境判断:标识当前是否运行在生产环境。""" return self.app_env.lower() == "production" def deployment_config_errors(self) -> list[str]: """部署检查:返回会导致生产核心链路不可用的配置问题。""" errors: list[str] = [] if self.database_url.startswith("sqlite"): errors.append("DATABASE_URL must use MySQL") if "CHANGE_ME" in self.database_url: errors.append("DATABASE_URL still contains a placeholder") if self.runtime_memory_backend.lower() != "redis": errors.append("RUNTIME_MEMORY_BACKEND must be redis") if not self.redis_url: errors.append("REDIS_URL is required") if not self.auth_user_me_url: errors.append("AUTH_USER_ME_URL is required") if self.llm_api_key in {"", "CHANGE_ME"} and not self.llm_mock_enabled: errors.append("LLM_API_KEY is required when mock mode is disabled") return errors def as_public_dict(self) -> dict[str, Any]: """配置展示:返回允许暴露给 Demo 前端的功能开关。""" mock_enabled = self.llm_mock_enabled or not self.llm_api_key return { "stream_chat": self.llm_stream_enabled, "score_types": ["percentage", "five_point"], "pdf_export": True, "knowledge_search": True, "llm_mock_enabled": mock_enabled, "llm_mode": "mock" if mock_enabled else "real", "llm_fallback_to_mock": self.llm_fallback_to_mock, "llm_fast_model": self.llm_fast_model, "llm_reason_model": self.llm_reason_model, "llm_fast_thinking_enabled": self.llm_fast_thinking_enabled, "llm_reason_thinking_enabled": self.llm_reason_thinking_enabled, "llm_reasoning_effort": self.llm_reasoning_effort, "llm_fast_max_tokens": self.llm_fast_max_tokens, "runtime_memory_backend": self.runtime_memory_backend, "auth_validate_enabled": True, "auth_source": "django_user_center", } settings = Settings()