""" 配置管理 """ 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 VLMConfig: """VLM 视觉语言模型配置""" api_key: str = "" base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1" model: str = "qwen3-vl-flash-2026-01-22" timeout: int = 10 enabled: bool = False enable_thinking: bool = False @dataclass class WeChatConfig: """企微通知配置""" corp_id: str = "" agent_id: str = "" secret: str = "" token: str = "" encoding_aes_key: str = "" enabled: bool = False test_uids: str = "" # 演示模式:逗号分隔的企微userid,如 "zhangsan,lisi" service_base_url: str = "" # 公网地址,如 https://vsp.viewshanghai.com group_chat_id: str = "" # 告警群聊ID(通过企微API创建或手动指定) @dataclass class AgentConfig: """交互Agent配置""" vlm_api_key: str = "" vlm_base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1" vlm_model: str = "qwen3-vl-flash-2026-01-22" # 视觉模型(图片分析保留) model: str = "qwen3.5-plus" # 文本模型(Function Calling) vlm_timeout: int = 15 timeout: int = 30 # FC 多轮超时 enabled: bool = False @dataclass class WorkOrderConfig: """安保工单开放接口配置""" base_url: str = "" # 工单系统地址,如 http://aiot-platform.viewsh.com:48080 app_id: str = "" # 应用ID app_secret: str = "" # 应用密钥 tenant_id: str = "1" # 租户编号 timeout: int = 10 # 请求超时(秒) 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 @dataclass class EdgeAuthConfig: """Edge 设备认证配置""" token: str = "" # 共享 Token,Edge 请求时携带 Authorization: Bearer {token} enabled: bool = False # 是否启用 token 校验 @dataclass class DailyReportConfig: """日报配置""" enabled: bool = False send_hour: int = 9 send_minute: int = 0 @dataclass class CameraNameConfig: """摄像头名称配置""" wvp_api_base: str = "http://localhost:18080" query_timeout: int = 15 class Settings(BaseModel): """全局配置""" database: DatabaseConfig = DatabaseConfig() cos: COSConfig = COSConfig() app: AppConfig = AppConfig() ai_model: AIModelConfig = AIModelConfig() vlm: VLMConfig = VLMConfig() wechat: WeChatConfig = WeChatConfig() agent: AgentConfig = AgentConfig() work_order: WorkOrderConfig = WorkOrderConfig() redis: RedisConfig = RedisConfig() daily_report: DailyReportConfig = DailyReportConfig() camera_name: CameraNameConfig = CameraNameConfig() edge_auth: EdgeAuthConfig = EdgeAuthConfig() 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", ""), ), vlm=VLMConfig( api_key=os.getenv("DASHSCOPE_API_KEY", ""), base_url=os.getenv("VLM_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"), model=os.getenv("VLM_MODEL", "qwen3-vl-flash-2026-01-22"), timeout=int(os.getenv("VLM_TIMEOUT", "10")), enabled=os.getenv("VLM_ENABLED", "false").lower() == "true", enable_thinking=os.getenv("VLM_ENABLE_THINKING", "false").lower() == "true", ), wechat=WeChatConfig( corp_id=os.getenv("WECHAT_CORP_ID", ""), agent_id=os.getenv("WECHAT_AGENT_ID", ""), secret=os.getenv("WECHAT_SECRET", ""), token=os.getenv("WECHAT_TOKEN", ""), encoding_aes_key=os.getenv("WECHAT_ENCODING_AES_KEY", ""), enabled=os.getenv("WECHAT_ENABLED", "false").lower() == "true", test_uids=os.getenv("WECHAT_TEST_UIDS", ""), service_base_url=os.getenv("SERVICE_BASE_URL", ""), group_chat_id=os.getenv("WECHAT_GROUP_CHAT_ID", ""), ), agent=AgentConfig( vlm_api_key=os.getenv("DASHSCOPE_API_KEY", ""), vlm_base_url=os.getenv("AGENT_VLM_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"), vlm_model=os.getenv("AGENT_VLM_MODEL", os.getenv("VLM_MODEL", "qwen3-vl-flash-2026-01-22")), model=os.getenv("AGENT_MODEL", "qwen3.5-plus"), vlm_timeout=int(os.getenv("AGENT_VLM_TIMEOUT", "15")), timeout=int(os.getenv("AGENT_TIMEOUT", "30")), enabled=os.getenv("AGENT_ENABLED", "false").lower() == "true", ), work_order=WorkOrderConfig( base_url=os.getenv("WORK_ORDER_BASE_URL", ""), app_id=os.getenv("WORK_ORDER_APP_ID", ""), app_secret=os.getenv("WORK_ORDER_APP_SECRET", ""), tenant_id=os.getenv("WORK_ORDER_TENANT_ID", "1"), timeout=int(os.getenv("WORK_ORDER_TIMEOUT", "10")), enabled=os.getenv("WORK_ORDER_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", ), camera_name=CameraNameConfig( wvp_api_base=os.getenv("WVP_API_BASE", "http://localhost:18080"), query_timeout=int(os.getenv("CAMERA_QUERY_TIMEOUT", "15")), ), daily_report=DailyReportConfig( enabled=os.getenv("DAILY_REPORT_ENABLED", "false").lower() == "true", send_hour=int(os.getenv("DAILY_REPORT_HOUR", "9")), send_minute=int(os.getenv("DAILY_REPORT_MINUTE", "0")), ), edge_auth=EdgeAuthConfig( token=os.getenv("EDGE_AUTH_TOKEN", ""), enabled=os.getenv("EDGE_AUTH_ENABLED", "false").lower() == "true", ), ) settings = load_settings()