""" 配置管理 """ 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 COSConfig: """腾讯云 COS 存储配置""" secret_id: str = "" secret_key: str = "" region: str = "ap-beijing" bucket: str = "" # 格式: bucketname-appid upload_prefix: str = "alerts" # 对象 Key 前缀 presign_expire: int = 1800 # 预签名URL有效期(秒),默认30分钟 sts_expire: int = 1800 # STS 临时凭证有效期(秒) enabled: bool = False # 是否启用 COS(False 时使用本地存储) @dataclass class AppConfig: """应用配置""" host: str = "0.0.0.0" port: int = 8000 debug: bool = True dev_mode: bool = True # 开发模式:跳过认证验证,返回超级管理员权限 @dataclass class AIModelConfig: """AI 模型配置""" endpoint: str = "" api_key: str = "" @dataclass class MQTTConfig: """MQTT 配置 (已废弃 - 告警上报已改为 HTTP + COS)""" 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 = False # 默认禁用 @dataclass class RedisConfig: """Redis 配置""" host: str = "localhost" port: int = 6379 password: str = "" db: int = 0 max_connections: int = 50 decode_responses: bool = True enabled: bool = True class Settings(BaseModel): """全局配置""" database: DatabaseConfig = DatabaseConfig() cos: COSConfig = COSConfig() app: AppConfig = AppConfig() ai_model: AIModelConfig = AIModelConfig() mqtt: MQTTConfig = MQTTConfig() redis: RedisConfig = RedisConfig() 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"), ), cos=COSConfig( secret_id=os.getenv("COS_SECRET_ID", ""), secret_key=os.getenv("COS_SECRET_KEY", ""), region=os.getenv("COS_REGION", "ap-beijing"), bucket=os.getenv("COS_BUCKET", ""), upload_prefix=os.getenv("COS_UPLOAD_PREFIX", "alerts"), presign_expire=int(os.getenv("COS_PRESIGN_EXPIRE", "1800")), sts_expire=int(os.getenv("COS_STS_EXPIRE", "1800")), enabled=os.getenv("COS_ENABLED", "false").lower() == "true", ), 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", dev_mode=os.getenv("DEV_MODE", "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", "false").lower() == "true", ), redis=RedisConfig( host=os.getenv("REDIS_HOST", "localhost"), port=int(os.getenv("REDIS_PORT", "6379")), password=os.getenv("REDIS_PASSWORD", ""), db=int(os.getenv("REDIS_DB", "0")), max_connections=int(os.getenv("REDIS_MAX_CONNECTIONS", "50")), enabled=os.getenv("REDIS_ENABLED", "true").lower() == "true", ), ) settings = load_settings()