iot-device-management-service/app/config.py

"""
配置管理
"""
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()