Files
iot-device-management-service/app/config.py
16337 51dbad6794 fix: 告警列表500错误修复 + 摄像头查询容错 + COS认证重构
- 添加缺失的 import httpx(修复 _notify_ops_platform NameError)
- 摄像头批量查询加 try/except 容错,失败时降级使用 device_id
- 摄像头查询超时从 5 秒提升到 15 秒
- COS 存储重构:支持 CVM 角色认证 + 密钥认证双模式
- STS 接口改为返回提示(CVM 模式不支持 STS)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-02 13:33:28 +08:00

136 lines
4.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
配置管理
"""
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 存储配置(通过 CVM 角色认证,无需密钥)"""
region: str = "ap-beijing"
bucket: str = "" # 格式: bucketname-appid
upload_prefix: str = "alerts" # 对象 Key 前缀
presign_expire: int = 1800 # 预签名URL有效期默认30分钟
enabled: bool = False # 是否启用 COS
@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 RedisConfig:
"""Redis 配置"""
host: str = "localhost"
port: int = 6379
password: str = ""
db: int = 0
max_connections: int = 50
decode_responses: bool = True
enabled: bool = True
@dataclass
class CameraNameConfig:
"""摄像头名称格式化配置"""
# WVP API基础URL
wvp_api_base: str = "http://localhost:18080"
# 显示格式模板(支持变量:{camera_code}, {name}, {stream}
# 可选格式:
# - "{name}" - 仅名称(推荐,告警列表使用)
# - "{camera_code} {name}/{stream}" - cam_xxx 名称/流id完整格式
# - "{name}/{stream}" - 名称/流id
# - "{camera_code}" - 仅code
display_format: str = "{name}"
# 名称字段优先级(从高到低)
# 从StreamProxy对象中提取名称时的字段优先级
# 注意gb_name 可能包含 "/" 后缀,会自动去除
name_field_priority: list = None
# 查询超时(秒)
query_timeout: int = 15
def __post_init__(self):
if self.name_field_priority is None:
# 默认优先级gb_name > app > stream
self.name_field_priority = ["gbName", "app", "stream"]
class Settings(BaseModel):
"""全局配置"""
database: DatabaseConfig = DatabaseConfig()
cos: COSConfig = COSConfig()
app: AppConfig = AppConfig()
ai_model: AIModelConfig = AIModelConfig()
redis: RedisConfig = RedisConfig()
camera_name: CameraNameConfig = CameraNameConfig()
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(
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")),
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", ""),
),
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",
),
camera_name=CameraNameConfig(
wvp_api_base=os.getenv("WVP_API_BASE", "http://localhost:18080"),
display_format=os.getenv("CAMERA_NAME_FORMAT", "{name}"),
query_timeout=int(os.getenv("CAMERA_QUERY_TIMEOUT", "15")),
),
)
settings = load_settings()