""" 配置管理 """ import os from pathlib import Path from typing import Optional from dataclasses import dataclass from pydantic import BaseModel @dataclass class DatabaseConfig: """数据库配置""" url: str = "sqlite:///./data/alert_platform.db" @dataclass class OSSConfig: """OSS 存储配置""" access_key_id: str = "" access_key_secret: str = "" endpoint: str = "oss-cn-hangzhou.aliyuncs.com" bucket_name: str = "" url_prefix: str = "" @dataclass class AppConfig: """应用配置""" host: str = "0.0.0.0" port: int = 8000 debug: bool = True @dataclass class AIModelConfig: """AI 模型配置""" endpoint: str = "" api_key: str = "" @dataclass class MQTTConfig: """MQTT 配置""" broker_host: str = "localhost" broker_port: int = 1883 client_id: str = "alert_platform" username: str = "" password: str = "" alert_topic: str = "edge/alert/#" heartbeat_topic: str = "edge/alert/heartbeat/#" qos: int = 1 enabled: bool = True class Settings(BaseModel): """全局配置""" database: DatabaseConfig = DatabaseConfig() oss: OSSConfig = OSSConfig() app: AppConfig = AppConfig() ai_model: AIModelConfig = AIModelConfig() mqtt: MQTTConfig = MQTTConfig() def load_settings() -> Settings: """从环境变量加载配置""" from dotenv import load_dotenv load_dotenv() return Settings( database=DatabaseConfig( url=os.getenv("DATABASE_URL", "sqlite:///./data/alert_platform.db"), ), oss=OSSConfig( access_key_id=os.getenv("OSS_ACCESS_KEY_ID", ""), access_key_secret=os.getenv("OSS_ACCESS_KEY_SECRET", ""), endpoint=os.getenv("OSS_ENDPOINT", "oss-cn-hangzhou.aliyuncs.com"), bucket_name=os.getenv("OSS_BUCKET_NAME", ""), url_prefix=os.getenv("OSS_URL_PREFIX", ""), ), app=AppConfig( host=os.getenv("APP_HOST", "0.0.0.0"), port=int(os.getenv("APP_PORT", "8000")), debug=os.getenv("DEBUG", "true").lower() == "true", ), ai_model=AIModelConfig( endpoint=os.getenv("AI_MODEL_ENDPOINT", ""), api_key=os.getenv("AI_MODEL_API_KEY", ""), ), mqtt=MQTTConfig( broker_host=os.getenv("MQTT_BROKER_HOST", "localhost"), broker_port=int(os.getenv("MQTT_BROKER_PORT", "1883")), client_id=os.getenv("MQTT_CLIENT_ID", "alert_platform"), username=os.getenv("MQTT_USERNAME", ""), password=os.getenv("MQTT_PASSWORD", ""), alert_topic=os.getenv("MQTT_ALERT_TOPIC", "edge/alert/#"), heartbeat_topic=os.getenv("MQTT_HEARTBEAT_TOPIC", "edge/alert/heartbeat/#"), qos=int(os.getenv("MQTT_QOS", "1")), enabled=os.getenv("MQTT_ENABLED", "true").lower() == "true", ), ) settings = load_settings()