fix(edge): EDGE_DEVICE_ID从edge-001改为edge，与WVP标准值一致

Edge的config_sync过滤Stream事件时要求device_id匹配，
WVP端标准device_id为"edge"，此前不匹配导致Edge静默忽略
所有配置推送事件。

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.env

@@ -0,0 +1,23 @@
+# Local debug overrides
+DEBUG_SERVER_HOST=0.0.0.0
+
+# Alarm upload - cloud API URL (Service backend)
+CLOUD_API_URL=http://localhost:8000
+
+# ===== 边缘设备 ID（必须与 WVP 数据库 ROI 表的 device_id 一致） =====
+EDGE_DEVICE_ID=edge
+
+# ===== 配置同步模式（REDIS=监听云端Stream, LOCAL=仅本地SQLite） =====
+CONFIG_SYNC_MODE=REDIS
+
+# ===== 云端 Redis（截图处理器 + 配置同步，db=1） =====
+CLOUD_REDIS_HOST=sh-crs-6upea3zn.sql.tencentcdb.com
+CLOUD_REDIS_PORT=24637
+CLOUD_REDIS_DB=1
+CLOUD_REDIS_PASSWORD=HkVZkVnn1
+
+# ===== 腾讯云 COS（截图上传） =====
+COS_REGION=ap-shanghai
+COS_BUCKET=xhwkzx-1-1389966313
+COS_SECRET_ID=AKIDVxPiqmVhYv7FCwVqytdAVddQ2TJySt9I
+COS_SECRET_KEY=1rVyEI8mMVWs21xfBUjy4BE6DA4z7KWb

config/settings.py

@@ -48,7 +48,7 @@ class CloudRedisConfig:
     """云端 Redis 配置（三层权威模型 - 云端层）"""
     host: str = "localhost"
     port: int = 6379
-    db: int = 0
+    db: int = 1
     password: Optional[str] = None
     decode_responses: bool = True
     max_connections: int = 20
@@ -227,7 +227,7 @@ class Settings:
         self.cloud_redis = CloudRedisConfig(
             host=os.getenv("CLOUD_REDIS_HOST", "localhost"),
             port=int(os.getenv("CLOUD_REDIS_PORT", "6379")),
-            db=int(os.getenv("CLOUD_REDIS_DB", "0")),
+            db=int(os.getenv("CLOUD_REDIS_DB", "1")),
             password=os.getenv("CLOUD_REDIS_PASSWORD"),
         )
 

core/screenshot_handler.py

@@ -0,0 +1,320 @@
+"""
+截图处理模块 - 监听云端 Redis Stream 截图请求，截图后上传 COS
+
+数据流：
+  WVP → XADD edge_snap_request → Edge 消费 → capture frame → upload COS → HTTP 回调 WVP
+  （回调失败时降级写 Redis snap:result:{id}）
+"""
+
+import io
+import json
+import logging
+import threading
+import time
+import uuid
+from typing import Optional
+
+import cv2
+import numpy as np
+import requests
+
+from config.settings import get_settings, COSConfig
+
+logger = logging.getLogger(__name__)
+
+# Redis 常量
+SNAP_REQUEST_STREAM = "edge_snap_request"
+SNAP_CONSUMER_GROUP = "edge_snap_group"
+SNAP_CONSUMER_NAME = "edge_snap_worker"
+SNAP_RESULT_KEY_PREFIX = "snap:result:"
+SNAP_RESULT_TTL = 60       # 降级结果 key 60s 过期
+
+# HTTP 回调
+SNAP_CALLBACK_PATH = "/api/ai/roi/snap/callback"
+SNAP_CALLBACK_TIMEOUT = 5  # 回调 HTTP 超时（秒）
+
+
+class ScreenshotHandler:
+    """截图处理器
+
+    从云端 Redis Stream 消费截图请求，通过 MultiStreamManager 获取最新帧，
+    JPEG 编码后直传 COS，最后通过 HTTP 回调 WVP 返回结果 URL。
+    回调失败时降级写 Redis。
+    """
+
+    def __init__(self, cloud_redis, stream_manager):
+        """
+        Args:
+            cloud_redis: 云端 Redis 客户端 (redis.Redis)
+            stream_manager: MultiStreamManager 实例
+        """
+        self._cloud_redis = cloud_redis
+        self._stream_manager = stream_manager
+        self._settings = get_settings()
+        self._cos_client = None
+        self._cos_config: COSConfig = self._settings.cos
+
+        self._thread: Optional[threading.Thread] = None
+        self._stop_event = threading.Event()
+
+    # ==================== 生命周期 ====================
+
+    def start(self):
+        """启动截图监听线程"""
+        if not self._cloud_redis:
+            logger.warning("[截图] 云端 Redis 不可用，截图处理器未启动")
+            return
+        if not self._cos_config.secret_id or not self._cos_config.bucket:
+            logger.warning("[截图] COS 配置不完整，截图处理器未启动")
+            return
+
+        self._init_cos_client()
+        self._ensure_consumer_group()
+
+        self._stop_event.clear()
+        self._thread = threading.Thread(
+            target=self._listen_loop,
+            name="ScreenshotHandler",
+            daemon=True,
+        )
+        self._thread.start()
+        logger.info("[截图] 截图处理器已启动")
+
+    def stop(self):
+        """停止截图监听线程"""
+        self._stop_event.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5)
+        logger.info("[截图] 截图处理器已停止")
+
+    # ==================== COS 初始化 ====================
+
+    def _init_cos_client(self):
+        """初始化 COS 客户端"""
+        try:
+            from qcloud_cos import CosConfig, CosS3Client
+
+            config = CosConfig(
+                Region=self._cos_config.region,
+                SecretId=self._cos_config.secret_id,
+                SecretKey=self._cos_config.secret_key,
+            )
+            self._cos_client = CosS3Client(config)
+            logger.info("[截图] COS 客户端初始化成功: bucket=%s, region=%s",
+                        self._cos_config.bucket, self._cos_config.region)
+        except Exception as e:
+            logger.error("[截图] COS 客户端初始化失败: %s", e)
+
+    # ==================== Consumer Group ====================
+
+    def _ensure_consumer_group(self):
+        """确保 Redis Consumer Group 存在"""
+        try:
+            self._cloud_redis.xgroup_create(
+                SNAP_REQUEST_STREAM, SNAP_CONSUMER_GROUP, id="0", mkstream=True
+            )
+            logger.info("[截图] 创建 consumer group: %s", SNAP_CONSUMER_GROUP)
+        except Exception as e:
+            # BUSYGROUP = group already exists, safe to ignore
+            if "BUSYGROUP" in str(e):
+                logger.debug("[截图] consumer group 已存在")
+            else:
+                logger.error("[截图] 创建 consumer group 失败: %s", e)
+
+    # ==================== 主循环 ====================
+
+    def _listen_loop(self):
+        """XREADGROUP 循环消费截图请求"""
+        backoff = 5
+        max_backoff = 60
+
+        while not self._stop_event.is_set():
+            try:
+                results = self._cloud_redis.xreadgroup(
+                    SNAP_CONSUMER_GROUP,
+                    SNAP_CONSUMER_NAME,
+                    {SNAP_REQUEST_STREAM: ">"},
+                    count=1,
+                    block=5000,
+                )
+                if not results:
+                    continue
+
+                backoff = 5  # 重置退避
+
+                for stream_name, messages in results:
+                    for msg_id, fields in messages:
+                        try:
+                            self._handle_request(fields)
+                        except Exception as e:
+                            logger.error("[截图] 处理请求失败: %s", e)
+                        finally:
+                            # ACK 消息
+                            try:
+                                self._cloud_redis.xack(
+                                    SNAP_REQUEST_STREAM, SNAP_CONSUMER_GROUP, msg_id
+                                )
+                            except Exception:
+                                pass
+
+            except Exception as e:
+                if self._stop_event.is_set():
+                    return
+                logger.warning("[截图] 监听异常: %s, %ds 后重试", e, backoff)
+                self._stop_event.wait(backoff)
+                backoff = min(backoff * 2, max_backoff)
+
+    # ==================== 请求处理 ====================
+
+    def _handle_request(self, fields: dict):
+        """处理单个截图请求"""
+        request_id = fields.get("request_id", "")
+        camera_code = fields.get("camera_code", "")
+        cos_path = fields.get("cos_path", "")
+        callback_url = fields.get("callback_url", "")
+
+        if not request_id or not camera_code or not cos_path:
+            logger.warning("[截图] 请求字段不完整: %s", fields)
+            self._send_result(callback_url, request_id, camera_code, {
+                "status": "error",
+                "message": "请求字段不完整",
+            })
+            return
+
+        logger.info("[截图] 收到截图请求: request_id=%s, camera=%s, callback=%s",
+                    request_id, camera_code, callback_url or "(无)")
+
+        # 1. 抓帧
+        jpeg_bytes = self._capture_frame(camera_code)
+        if jpeg_bytes is None:
+            self._send_result(callback_url, request_id, camera_code, {
+                "status": "error",
+                "message": f"摄像头流未连接: {camera_code}",
+            })
+            return
+
+        # 2. 上传 COS（失败重试 1 次）
+        url = self._upload_to_cos(jpeg_bytes, cos_path)
+        if url is None:
+            # 重试一次
+            logger.info("[截图] COS 上传失败，重试一次...")
+            url = self._upload_to_cos(jpeg_bytes, cos_path)
+
+        if url is None:
+            self._send_result(callback_url, request_id, camera_code, {
+                "status": "error",
+                "message": "COS 上传失败",
+            })
+            return
+
+        # 3. 发送结果
+        self._send_result(callback_url, request_id, camera_code, {
+            "status": "ok",
+            "url": url,
+        })
+
+        logger.info("[截图] 截图完成: camera=%s, url=%s", camera_code, url)
+
+    # ==================== 抓帧 ====================
+
+    def _capture_frame(self, camera_code: str) -> Optional[bytes]:
+        """从 MultiStreamManager 获取最新帧并编码为 JPEG"""
+        stream = self._stream_manager.get_stream(camera_code)
+        if stream is None:
+            logger.warning("[截图] 未找到摄像头流: %s", camera_code)
+            return None
+
+        if not stream.is_connected:
+            logger.warning("[截图] 摄像头未连接: %s", camera_code)
+            return None
+
+        frame = stream.get_latest_frame(timeout=2.0)
+        if frame is None:
+            # 回退：直接从缓冲区读一帧
+            frame = stream.read(timeout=2.0)
+
+        if frame is None:
+            logger.warning("[截图] 获取帧超时: %s", camera_code)
+            return None
+
+        # JPEG 编码
+        try:
+            encode_params = [cv2.IMWRITE_JPEG_QUALITY, 85]
+            success, buffer = cv2.imencode(".jpg", frame.image, encode_params)
+            if not success:
+                logger.error("[截图] JPEG 编码失败: %s", camera_code)
+                return None
+            return buffer.tobytes()
+        except Exception as e:
+            logger.error("[截图] 帧编码异常: %s", e)
+            return None
+
+    # ==================== COS 上传 ====================
+
+    # 预签名 URL 有效期（秒），必须大于 WVP 缓存 TTL（300s）
+    PRESIGNED_URL_EXPIRES = 3600
+
+    def _upload_to_cos(self, jpeg_bytes: bytes, cos_path: str) -> Optional[str]:
+        """上传 JPEG 到 COS，返回预签名访问 URL"""
+        if not self._cos_client:
+            logger.error("[截图] COS 客户端未初始化")
+            return None
+
+        try:
+            body = io.BytesIO(jpeg_bytes)
+            self._cos_client.put_object(
+                Bucket=self._cos_config.bucket,
+                Body=body,
+                Key=cos_path,
+                ContentType="image/jpeg",
+            )
+            # 生成预签名 URL（不附加额外 Params，保持 URL 简洁）
+            url = self._cos_client.get_presigned_url(
+                Method="GET",
+                Bucket=self._cos_config.bucket,
+                Key=cos_path,
+                Expired=self.PRESIGNED_URL_EXPIRES,
+            )
+            return url
+        except Exception as e:
+            logger.error("[截图] COS 上传失败: %s", e)
+            return None
+
+    # ==================== 结果发送（回调 + 降级） ====================
+
+    def _send_result(self, callback_url: str, request_id: str, camera_code: str, data: dict):
+        """发送截图结果：优先 HTTP 回调，失败则降级写 Redis"""
+        if not request_id:
+            return
+
+        # 构造回调 body
+        body = {
+            "request_id": request_id,
+            "camera_code": camera_code,
+            **data,
+        }
+
+        # 优先尝试 HTTP 回调
+        if callback_url:
+            try:
+                url = callback_url.rstrip("/") + SNAP_CALLBACK_PATH
+                resp = requests.post(url, json=body, timeout=SNAP_CALLBACK_TIMEOUT)
+                if resp.status_code < 300:
+                    logger.info("[截图] HTTP 回调成功: request_id=%s", request_id)
+                    return
+                else:
+                    logger.warning("[截图] HTTP 回调返回 %d: request_id=%s", resp.status_code, request_id)
+            except Exception as e:
+                logger.warning("[截图] HTTP 回调失败: %s, 降级写 Redis", e)
+
+        # 降级：写 Redis result key
+        self._write_result_redis(request_id, body)
+
+    def _write_result_redis(self, request_id: str, data: dict):
+        """降级：将结果写入 Redis key（TTL 60s）"""
+        key = SNAP_RESULT_KEY_PREFIX + request_id
+        try:
+            self._cloud_redis.set(key, json.dumps(data), ex=SNAP_RESULT_TTL)
+            logger.info("[截图] 降级写 Redis 成功: request_id=%s", request_id)
+        except Exception as e:
+            logger.error("[截图] 降级写 Redis 也失败: %s", e)

main.py

@@ -21,6 +21,7 @@ from core.tensorrt_engine import TensorRTEngine, EngineManager
 from core.postprocessor import PostProcessor
 from core.result_reporter import ResultReporter
 from core.alarm_upload_worker import AlarmUploadWorker
+from core.screenshot_handler import ScreenshotHandler
 from algorithms import AlgorithmManager
 from utils.logger import get_logger, StructuredLogger
 from utils.version_control import get_version_control
@@ -48,6 +49,7 @@ class EdgeInferenceService:
         self._postprocessor: Optional[PostProcessor] = None
         self._reporter: Optional[ResultReporter] = None
         self._alarm_worker: Optional[AlarmUploadWorker] = None
+        self._screenshot_handler: Optional[ScreenshotHandler] = None
         self._algorithm_manager: Optional[AlgorithmManager] = None
         self._debug_reload_thread: Optional[threading.Thread] = None
         self._debug_http_server = None
@@ -94,7 +96,7 @@ class EdgeInferenceService:
         try:
             self._config_manager = get_config_sync_manager()
             self._config_manager.start_config_subscription()
-            if self._settings.config_sync_mode == "LOCAL" and self._config_manager:
+            if self._config_manager:
                 def _on_config_update(topic, data):
                     if self._algorithm_manager:
                         # 保留状态地更新参数，避免告警重复
@@ -178,6 +180,45 @@ class EdgeInferenceService:
             self._logger.info("算法管理器初始化成功")
         except Exception as e:
             self._logger.error(f"算法管理器初始化失败: {e}")
+
+    def _init_screenshot_handler(self):
+        """初始化截图处理器"""
+        try:
+            # 优先从 config_manager 获取已有的云端 Redis 连接
+            cloud_redis = getattr(self._config_manager, '_cloud_redis', None)
+
+            # LOCAL 模式下 config_manager 不初始化云端 Redis，需要独立创建
+            if cloud_redis is None:
+                try:
+                    import redis
+                    cfg = self._settings.cloud_redis
+                    cloud_redis = redis.Redis(
+                        host=cfg.host,
+                        port=cfg.port,
+                        db=cfg.db,
+                        password=cfg.password,
+                        decode_responses=cfg.decode_responses,
+                        socket_connect_timeout=5,
+                        socket_timeout=5,
+                        retry_on_timeout=True,
+                    )
+                    cloud_redis.ping()
+                    self._logger.info(f"截图处理器独立连接云端 Redis 成功: {cfg.host}:{cfg.port}/{cfg.db}")
+                except Exception as e:
+                    self._logger.warning(f"截图处理器无法连接云端 Redis: {e}")
+                    cloud_redis = None
+
+            if cloud_redis and self._stream_manager:
+                self._screenshot_handler = ScreenshotHandler(
+                    cloud_redis=cloud_redis,
+                    stream_manager=self._stream_manager,
+                )
+                self._screenshot_handler.start()
+                self._logger.info("截图处理器初始化成功")
+            else:
+                self._logger.warning("截图处理器跳过初始化（云端 Redis 不可达或流管理器不可用）")
+        except Exception as e:
+            self._logger.error(f"截图处理器初始化失败: {e}")
     
     def _start_debug_reload_watcher(self):
         """本地调试：监听文件触发同步"""
@@ -257,6 +298,7 @@ class EdgeInferenceService:
         self._init_postprocessor()
         self._init_reporter()
         self._init_algorithm_manager()
+        self._init_screenshot_handler()
         self._start_debug_reload_watcher()
         self._start_debug_http_server()
         
@@ -848,6 +890,9 @@ class EdgeInferenceService:
         if self._alarm_worker:
             self._alarm_worker.stop()
 
+        if self._screenshot_handler:
+            self._screenshot_handler.stop()
+
         if self._reporter:
             self._reporter.close()
 

requirements.txt

@@ -32,6 +32,9 @@ paho-mqtt==1.6.1
 # Redis客户端 - Redis 4.6.0，4.x最终稳定版
 redis==4.6.0
 
+# 腾讯云COS SDK - 用于截图上传
+cos-python-sdk-v5>=1.9.30
+
 # ============================================================
 # 工具库
 # ============================================================