修复：P0+P1 生产稳定性和性能优化（6项）

P0 稳定性修复：
- 告警去重字典添加惰性清理机制，防止长时间运行内存溢出
- Redis 连接断开时显式 close() 后再置 None，防止文件描述符泄漏
- 截图消息 ACK 移至成功路径，失败消息留在 pending list 自动重试

P1 性能优化：
- GPU NMS 添加 torch.no_grad() + 显式释放临时张量，减少显存碎片
- 截图存储改为 Redis 原始 bytes，去掉 Base64 编解码开销（兼容旧格式）
- ROI 配置查询 N+1 改为 get_all_bindings() 单次 JOIN 查询

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

algorithms.py

@@ -16,7 +16,42 @@ logger = logging.getLogger(__name__)
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 
-class LeavePostAlgorithm:
+class BaseAlgorithm:
+    """
+    算法基类，提取各算法共同逻辑。
+
+    子类须定义自己的状态常量和 process() 方法。
+    """
+
+    def __init__(self):
+        # 状态变量（子类构造函数中可覆盖 self.state 的初始值）
+        self.state: str = ""
+        self.state_start_time: Optional[datetime] = None
+
+        # 告警追踪
+        self._last_alarm_id: Optional[str] = None
+
+    # ---- 公共工具方法 ----
+
+    @staticmethod
+    def _check_detection_in_roi(detection: Dict, roi_id: str) -> bool:
+        """检查检测结果是否在ROI内"""
+        matched_rois = detection.get("matched_rois", [])
+        return any(roi.get("roi_id") == roi_id for roi in matched_rois)
+
+    @staticmethod
+    def _check_target_class(detection: Dict, target_class: Optional[str]) -> bool:
+        """检查是否为目标类别"""
+        if not target_class:
+            return True
+        return detection.get("class") == target_class
+
+    def set_last_alarm_id(self, alarm_id: str):
+        """由 main.py 在告警生成后回填 alarm_id"""
+        self._last_alarm_id = alarm_id
+
+
+class LeavePostAlgorithm(BaseAlgorithm):
     """
     离岗检测算法（优化版 v2.0）
 
@@ -46,6 +81,12 @@ class LeavePostAlgorithm:
     # 告警级别常量（默认值，可通过 params 覆盖）
     DEFAULT_ALARM_LEVEL = 2  # 普通
 
+    # Step 2: 阈值常量
+    RATIO_ON_DUTY_CONFIRM = 0.6       # 上岗确认命中率阈值
+    RATIO_OFF_DUTY_TRIGGER = 0.2      # 离岗触发阈值（低于此值进入离岗确认）
+    RATIO_RETURN_CONFIRM = 0.5        # 回岗确认命中率阈值
+    RATIO_OFF_DUTY_CONFIRM = 0.2      # 离岗确认完成阈值
+
     def __init__(
         self,
         confirm_on_duty_sec: int = 10,      # 上岗确认窗口（持续检测到人的时长）
@@ -59,6 +100,8 @@ class LeavePostAlgorithm:
         # 兼容旧参数名（向后兼容）
         confirm_leave_sec: Optional[int] = None,
     ):
+        super().__init__()
+
         # 时间参数（处理向后兼容）
         self.confirm_on_duty_sec = confirm_on_duty_sec
         self.confirm_off_duty_sec = confirm_leave_sec if confirm_leave_sec is not None else confirm_off_duty_sec
@@ -75,12 +118,11 @@ class LeavePostAlgorithm:
         self.state: str = self.STATE_INIT
         self.state_start_time: Optional[datetime] = None
 
-        # 滑动窗口（用于平滑检测结果）
+        # 滑动窗口（用于平滑检测结果）— Step 3: maxlen=1000
-        self.detection_window: deque = deque()  # [(timestamp, has_person), ...]
+        self.detection_window: deque = deque(maxlen=1000)  # [(timestamp, has_person), ...]
         self.window_size_sec = 10  # 滑动窗口大小：10秒
 
         # 告警追踪
-        self._last_alarm_id: Optional[str] = None
         self._leave_start_time: Optional[datetime] = None  # 人员离开时间（用于计算持续时长）
         self._alarm_triggered_time: Optional[datetime] = None  # 告警触发时间
         self.alert_cooldowns: Dict[str, datetime] = {}
@@ -123,17 +165,6 @@ class LeavePostAlgorithm:
         except:
             return 0
 
-    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
-        """检查检测结果是否在ROI内"""
-        matched_rois = detection.get("matched_rois", [])
-        return any(roi.get("roi_id") == roi_id for roi in matched_rois)
-
-    def _check_target_class(self, detection: Dict, target_class: str) -> bool:
-        """检查是否为目标类别"""
-        if not target_class:
-            return True
-        return detection.get("class") == target_class
-
     def _get_latest_bbox(self, tracks: List[Dict], roi_id: str) -> List[float]:
         """获取ROI内最新的检测框"""
         for det in tracks:
@@ -191,6 +222,8 @@ class LeavePostAlgorithm:
 
         # 更新滑动窗口
         self._update_detection_window(current_time, roi_has_person)
+
+        # Step 4: 计算一次比率，后续分支复用
         detection_ratio = self._get_detection_ratio()
 
         # 检查工作时间
@@ -246,8 +279,8 @@ class LeavePostAlgorithm:
                 self.state_start_time = current_time
                 self.detection_window.clear()
                 logger.debug(f"ROI {roi_id}: CONFIRMING_ON_DUTY → INIT (人消失)")
-            elif elapsed >= self.confirm_on_duty_sec and detection_ratio >= 0.6:
+            elif elapsed >= self.confirm_on_duty_sec and detection_ratio >= self.RATIO_ON_DUTY_CONFIRM:
-                # 上岗确认成功（命中率>=70%）
+                # 上岗确认成功
                 self.state = self.STATE_ON_DUTY
                 self.state_start_time = current_time
                 self.alert_cooldowns.clear()  # 确认在岗后清除冷却记录
@@ -255,7 +288,7 @@ class LeavePostAlgorithm:
 
         elif self.state == self.STATE_ON_DUTY:
             # 在岗状态：监控是否离岗
-            if detection_ratio < 0.2:
+            if detection_ratio < self.RATIO_OFF_DUTY_TRIGGER:
                 # 滑动窗口内 80% 以上帧无人，进入离岗确认
                 self.state = self.STATE_CONFIRMING_OFF_DUTY
                 self.state_start_time = current_time
@@ -265,12 +298,12 @@ class LeavePostAlgorithm:
             # 离岗确认中：需要持续未检测到人
             elapsed = (current_time - self.state_start_time).total_seconds()
 
-            if detection_ratio >= 0.5:
+            if detection_ratio >= self.RATIO_RETURN_CONFIRM:
                 # 窗口内检测率恢复到 50% 以上，人确实回来了
                 self.state = self.STATE_ON_DUTY
                 self.state_start_time = current_time
                 logger.debug(f"ROI {roi_id}: CONFIRMING_OFF_DUTY → ON_DUTY (人回来了, ratio={detection_ratio:.2f})")
-            elif elapsed >= self.confirm_off_duty_sec and detection_ratio < 0.2:
+            elif elapsed >= self.confirm_off_duty_sec and detection_ratio < self.RATIO_OFF_DUTY_CONFIRM:
                 # 离岗确认成功，进入倒计时
                 self.state = self.STATE_OFF_DUTY_COUNTDOWN
                 self.state_start_time = current_time
@@ -295,6 +328,9 @@ class LeavePostAlgorithm:
 
                     bbox = self._get_latest_bbox(tracks, roi_id)
 
+                    # Bug fix: _leave_start_time None guard
+                    first_frame_time = self._leave_start_time.strftime('%Y-%m-%d %H:%M:%S') if self._leave_start_time else current_time.strftime('%Y-%m-%d %H:%M:%S')
+
                     alerts.append({
                         "track_id": roi_id,
                         "camera_id": camera_id,
@@ -302,7 +338,7 @@ class LeavePostAlgorithm:
                         "alert_type": "leave_post",
                         "alarm_level": self._alarm_level,
                         "message": "人员离岗告警",
-                        "first_frame_time": self._leave_start_time.strftime('%Y-%m-%d %H:%M:%S'),
+                        "first_frame_time": first_frame_time,
                     })
 
                     self.alert_cooldowns[cooldown_key] = current_time
@@ -321,8 +357,13 @@ class LeavePostAlgorithm:
 
         # 特殊处理：从CONFIRMING_ON_DUTY再次确认上岗时，如果有未结束的告警，发送resolve
         if self.state == self.STATE_ON_DUTY and self._last_alarm_id:
-            # 回岗确认成功，发送resolve事件
+            # Bug fix: _leave_start_time None guard for resolve event
+            if self._leave_start_time is not None:
                 duration_ms = int((current_time - self._leave_start_time).total_seconds() * 1000)
+            else:
+                duration_ms = 0
+
+            # 回岗确认成功，发送resolve事件
             alerts.append({
                 "alert_type": "alarm_resolve",
                 "resolve_alarm_id": self._last_alarm_id,
@@ -339,10 +380,6 @@ class LeavePostAlgorithm:
 
         return alerts
 
-    def set_last_alarm_id(self, alarm_id: str):
-        """由 main.py 在告警生成后回填 alarm_id"""
-        self._last_alarm_id = alarm_id
-
     def reset(self):
         """重置算法状态"""
         self.state = self.STATE_INIT
@@ -355,10 +392,13 @@ class LeavePostAlgorithm:
 
     def get_state(self, roi_id: str) -> Dict[str, Any]:
         """获取当前状态（用于调试和监控）"""
+        # Step 4: 缓存比率计算
+        detection_ratio = self._get_detection_ratio()
+
         state_info = {
             "state": self.state,
             "state_start_time": self.state_start_time.isoformat() if self.state_start_time else None,
-            "detection_ratio": self._get_detection_ratio(),
+            "detection_ratio": detection_ratio,
             "window_size": len(self.detection_window),
         }
 
@@ -375,7 +415,7 @@ class LeavePostAlgorithm:
         return state_info
 
 
-class IntrusionAlgorithm:
+class IntrusionAlgorithm(BaseAlgorithm):
     """
     周界入侵检测算法（状态机版本 v3.0）
 
@@ -410,6 +450,8 @@ class IntrusionAlgorithm:
         target_class: Optional[str] = None,
         alarm_level: Optional[int] = None,
     ):
+        super().__init__()
+
         self.cooldown_seconds = cooldown_seconds
 
         # 参数兼容处理
@@ -427,7 +469,6 @@ class IntrusionAlgorithm:
         self.state_start_time: Optional[datetime] = None
 
         # 告警追踪
-        self._last_alarm_id: Optional[str] = None
         self._intrusion_start_time: Optional[datetime] = None
 
         # CONFIRMING_CLEAR状态下检测到人的时间（用于判断是否持续5秒）
@@ -441,24 +482,57 @@ class IntrusionAlgorithm:
         self.alert_triggered: Dict[str, bool] = {}
         self.detection_start: Dict[str, Optional[datetime]] = {}
 
-    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
-        matched_rois = detection.get("matched_rois", [])
-        for roi in matched_rois:
-            if roi.get("roi_id") == roi_id:
-                return True
-        return False
-
-    def _check_target_class(self, detection: Dict, target_class: Optional[str]) -> bool:
-        if not target_class:
-            return True
-        return detection.get("class") == target_class
-
     def _get_latest_bbox(self, tracks: List[Dict], roi_id: str) -> List[float]:
+        """获取ROI内最新的检测框 - IntrusionAlgorithm 不过滤 target_class"""
         for det in tracks:
             if self._check_detection_in_roi(det, roi_id):
                 return det.get("bbox", [])
         return []
 
+    # Step 7: CONFIRMING_CLEAR 逻辑拆分
+    def _handle_clear_person_detected(self, roi_id: str, current_time: datetime):
+        """CONFIRMING_CLEAR 状态下检测到人的处理"""
+        if self._person_detected_in_clear_time is None:
+            # 第一次检测到人，记录时间
+            self._person_detected_in_clear_time = current_time
+            logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR 检测到人，开始确认(需持续{self.confirm_intrusion_seconds}秒)")
+        else:
+            # 持续有人，检查是否达到确认时间
+            person_elapsed = (current_time - self._person_detected_in_clear_time).total_seconds()
+            if person_elapsed >= self.confirm_intrusion_seconds:
+                # 确认有人重新入侵，回到ALARMED
+                self.state = self.STATE_ALARMED
+                self.state_start_time = None
+                self._person_detected_in_clear_time = None
+                logger.info(f"ROI {roi_id}: CONFIRMING_CLEAR → ALARMED (确认有人重新入侵，持续{person_elapsed:.1f}秒)")
+
+    def _handle_clear_no_person(self, roi_id: str, current_time: datetime, elapsed: float, alerts: List[Dict]):
+        """CONFIRMING_CLEAR 状态下无人的处理"""
+        self._person_detected_in_clear_time = None  # 清除临时计时
+
+        # 检查是否达到消失确认时间
+        if elapsed >= self.confirm_clear_seconds:
+            # 消失确认成功，发送resolve事件
+            if self._last_alarm_id and self._intrusion_start_time:
+                duration_ms = int((current_time - self._intrusion_start_time).total_seconds() * 1000)
+                alerts.append({
+                    "alert_type": "alarm_resolve",
+                    "resolve_alarm_id": self._last_alarm_id,
+                    "duration_ms": duration_ms,
+                    "last_frame_time": current_time.strftime('%Y-%m-%d %H:%M:%S'),
+                    "resolve_type": "intrusion_cleared",
+                })
+
+                logger.info(f"ROI {roi_id}: 告警已解决（入侵消失，持续无人{elapsed:.1f}秒）")
+
+            # 重置状态
+            self.state = self.STATE_IDLE
+            self.state_start_time = None
+            self._last_alarm_id = None
+            self._intrusion_start_time = None
+            self._person_detected_in_clear_time = None
+            logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR → IDLE (消失确认成功)")
+
     def process(
         self,
         roi_id: str,
@@ -556,54 +630,14 @@ class IntrusionAlgorithm:
             else:
                 elapsed = (current_time - self.state_start_time).total_seconds()
 
+                # Step 7: 使用拆分后的方法
                 if roi_has_person:
-                    # 检测到有人
+                    self._handle_clear_person_detected(roi_id, current_time)
-                    if self._person_detected_in_clear_time is None:
-                        # 第一次检测到人，记录时间
-                        self._person_detected_in_clear_time = current_time
-                        logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR 检测到人，开始确认(需持续{self.confirm_intrusion_seconds}秒)")
                 else:
-                        # 持续有人，检查是否达到确认时间
+                    self._handle_clear_no_person(roi_id, current_time, elapsed, alerts)
-                        person_elapsed = (current_time - self._person_detected_in_clear_time).total_seconds()
-                        if person_elapsed >= self.confirm_intrusion_seconds:
-                            # 确认有人重新入侵，回到ALARMED
-                            self.state = self.STATE_ALARMED
-                            self.state_start_time = None
-                            self._person_detected_in_clear_time = None
-                            logger.info(f"ROI {roi_id}: CONFIRMING_CLEAR → ALARMED (确认有人重新入侵，持续{person_elapsed:.1f}秒)")
-                else:
-                    # 没有人
-                    self._person_detected_in_clear_time = None  # 清除临时计时
-
-                    # 检查是否达到消失确认时间
-                    if elapsed >= self.confirm_clear_seconds:
-                        # 消失确认成功，发送resolve事件
-                        if self._last_alarm_id and self._intrusion_start_time:
-                            duration_ms = int((current_time - self._intrusion_start_time).total_seconds() * 1000)
-                            alerts.append({
-                                "alert_type": "alarm_resolve",
-                                "resolve_alarm_id": self._last_alarm_id,
-                                "duration_ms": duration_ms,
-                                "last_frame_time": current_time.strftime('%Y-%m-%d %H:%M:%S'),
-                                "resolve_type": "intrusion_cleared",
-                            })
-
-                            logger.info(f"ROI {roi_id}: 告警已解决（入侵消失，持续无人{elapsed:.1f}秒）")
-
-                        # 重置状态
-                        self.state = self.STATE_IDLE
-                        self.state_start_time = None
-                        self._last_alarm_id = None
-                        self._intrusion_start_time = None
-                        self._person_detected_in_clear_time = None
-                        logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR → IDLE (消失确认成功)")
 
         return alerts
 
-    def set_last_alarm_id(self, alarm_id: str):
-        """由 main.py 在告警生成后回填 alarm_id"""
-        self._last_alarm_id = alarm_id
-
     def reset(self):
         """重置算法状态"""
         self.state = self.STATE_IDLE
@@ -713,7 +747,7 @@ class IntrusionAlgorithm:
 #         self.alert_triggered.clear()
 
 
-class IllegalParkingAlgorithm:
+class IllegalParkingAlgorithm(BaseAlgorithm):
     """
     车辆违停检测算法（状态机版本 v1.0）
 
@@ -721,7 +755,7 @@ class IllegalParkingAlgorithm:
     IDLE → CONFIRMING_VEHICLE → PARKED_COUNTDOWN → ALARMED → CONFIRMING_CLEAR → IDLE
 
     业务流程：
-    1. 检测到车辆进入禁停区 → 车辆确认期（confirm_vehicle_sec，默认15秒，ratio≥0.6）
+    1. 检测到车辆进入禁停区 → 车辆确认期（confirm_vehicle_sec，默认15秒，ratio>=0.6）
     2. 确认有车 → 违停倒计时（parking_countdown_sec，默认300秒/5分钟）
     3. 倒计时结束仍有车 → 触发告警（ALARMED状态）
     4. 车辆离开 → 消失确认期（confirm_clear_sec，默认30秒，ratio<0.2）
@@ -743,6 +777,14 @@ class IllegalParkingAlgorithm:
     # 滑动窗口参数
     WINDOW_SIZE_SEC = 10
 
+    # Step 2: 阈值常量
+    RATIO_CONFIRMING_DROP = 0.3       # 确认期内命中率低于此值则回到IDLE
+    RATIO_CONFIRM_VEHICLE = 0.6       # 确认有车的命中率阈值
+    RATIO_PARKED_LEAVE = 0.2          # 倒计时期间车辆离开的判定阈值
+    RATIO_ALARMED_CLEAR = 0.15        # 已告警状态下进入消失确认的阈值
+    RATIO_CLEAR_RETURN = 0.5          # 消失确认期间车辆再次出现的阈值
+    RATIO_CLEAR_CONFIRM = 0.2         # 消失确认完成的阈值
+
     def __init__(
         self,
         confirm_vehicle_sec: int = 15,
@@ -752,6 +794,8 @@ class IllegalParkingAlgorithm:
         target_classes: Optional[List[str]] = None,
         alarm_level: Optional[int] = None,
     ):
+        super().__init__()
+
         self.confirm_vehicle_sec = confirm_vehicle_sec
         self.parking_countdown_sec = parking_countdown_sec
         self.confirm_clear_sec = confirm_clear_sec
@@ -763,23 +807,15 @@ class IllegalParkingAlgorithm:
         self.state: str = self.STATE_IDLE
         self.state_start_time: Optional[datetime] = None
 
-        # 滑动窗口：存储 (timestamp, has_vehicle: bool)
+        # 滑动窗口：存储 (timestamp, has_vehicle: bool) — Step 3: maxlen=1000
-        self._detection_window: deque = deque()
+        self._detection_window: deque = deque(maxlen=1000)
 
         # 告警追踪
-        self._last_alarm_id: Optional[str] = None
         self._parking_start_time: Optional[datetime] = None
 
         # 冷却期管理
         self.alert_cooldowns: Dict[str, datetime] = {}
 
-    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
-        matched_rois = detection.get("matched_rois", [])
-        for roi in matched_rois:
-            if roi.get("roi_id") == roi_id:
-                return True
-        return False
-
     def _check_target_classes(self, detection: Dict) -> bool:
         """检查检测目标是否属于车辆类别"""
         det_class = detection.get("class", "")
@@ -799,6 +835,28 @@ class IllegalParkingAlgorithm:
         hits = sum(1 for _, has in self._detection_window if has)
         return hits / len(self._detection_window)
 
+    # Step 5: 合并遍历方法
+    def _scan_tracks(self, tracks: List[Dict], roi_id: str) -> Tuple[bool, int, List[float], float]:
+        """
+        一次遍历 tracks，返回 (has_target, count, latest_bbox, max_confidence)。
+        过滤 target_classes。
+        """
+        has_target = False
+        count = 0
+        latest_bbox: List[float] = []
+        max_confidence = 0.0
+        for det in tracks:
+            if self._check_detection_in_roi(det, roi_id) and self._check_target_classes(det):
+                has_target = True
+                count += 1
+                if not latest_bbox:
+                    latest_bbox = det.get("bbox", [])
+                conf = det.get("confidence", 0.0)
+                if conf > max_confidence:
+                    max_confidence = conf
+        return has_target, count, latest_bbox, max_confidence
+
+    # 保留旧方法以防外部调用
     def _get_latest_bbox(self, tracks: List[Dict], roi_id: str) -> List[float]:
         for det in tracks:
             if self._check_detection_in_roi(det, roi_id) and self._check_target_classes(det):
@@ -824,14 +882,13 @@ class IllegalParkingAlgorithm:
         current_time = current_time or datetime.now()
         alerts = []
 
-        # 检查ROI内是否有车辆
+        # Step 5: 一次遍历获取所有信息
-        roi_has_vehicle = any(
+        roi_has_vehicle, vehicle_count, scan_bbox, scan_confidence = self._scan_tracks(tracks, roi_id)
-            self._check_detection_in_roi(det, roi_id) and self._check_target_classes(det)
-            for det in tracks
-        )
 
         # 更新滑动窗口
         self._update_window(current_time, roi_has_vehicle)
+
+        # Step 4: 计算一次比率，后续分支复用
         ratio = self._get_window_ratio()
 
         # === 状态机处理 ===
@@ -849,12 +906,12 @@ class IllegalParkingAlgorithm:
 
             elapsed = (current_time - self.state_start_time).total_seconds()
 
-            if ratio < 0.3:
+            if ratio < self.RATIO_CONFIRMING_DROP:
                 # 命中率过低，车辆可能只是路过
                 self.state = self.STATE_IDLE
                 self.state_start_time = None
-                logger.debug(f"ROI {roi_id}: CONFIRMING_VEHICLE → IDLE (ratio={ratio:.2f}<0.3)")
+                logger.debug(f"ROI {roi_id}: CONFIRMING_VEHICLE → IDLE (ratio={ratio:.2f}<{self.RATIO_CONFIRMING_DROP})")
-            elif elapsed >= self.confirm_vehicle_sec and ratio >= 0.6:
+            elif elapsed >= self.confirm_vehicle_sec and ratio >= self.RATIO_CONFIRM_VEHICLE:
                 # 确认有车辆停留，进入倒计时
                 self._parking_start_time = self.state_start_time
                 self.state = self.STATE_PARKED_COUNTDOWN
@@ -868,7 +925,7 @@ class IllegalParkingAlgorithm:
 
             elapsed = (current_time - self.state_start_time).total_seconds()
 
-            if ratio < 0.2:
+            if ratio < self.RATIO_PARKED_LEAVE:
                 # 车辆已离开
                 self.state = self.STATE_IDLE
                 self.state_start_time = None
@@ -880,16 +937,13 @@ class IllegalParkingAlgorithm:
                 if cooldown_key not in self.alert_cooldowns or \
                    (current_time - self.alert_cooldowns[cooldown_key]).total_seconds() > self.cooldown_sec:
 
-                    bbox = self._get_latest_bbox(tracks, roi_id)
-                    confidence = self._get_max_confidence(tracks, roi_id)
-
                     alerts.append({
                         "roi_id": roi_id,
                         "camera_id": camera_id,
-                        "bbox": bbox,
+                        "bbox": scan_bbox,
                         "alert_type": "illegal_parking",
                         "alarm_level": self._alarm_level,
-                        "confidence": confidence,
+                        "confidence": scan_confidence,
                         "message": f"检测到车辆违停（已停留{int(elapsed / 60)}分钟）",
                         "first_frame_time": self._parking_start_time.strftime('%Y-%m-%d %H:%M:%S') if self._parking_start_time else None,
                         "duration_minutes": elapsed / 60,
@@ -905,10 +959,10 @@ class IllegalParkingAlgorithm:
                     logger.debug(f"ROI {roi_id}: PARKED_COUNTDOWN → IDLE (冷却期内)")
 
         elif self.state == self.STATE_ALARMED:
-            if ratio < 0.15:
+            if ratio < self.RATIO_ALARMED_CLEAR:
                 self.state = self.STATE_CONFIRMING_CLEAR
                 self.state_start_time = current_time
-                logger.debug(f"ROI {roi_id}: ALARMED → CONFIRMING_CLEAR (ratio={ratio:.2f}<0.15)")
+                logger.debug(f"ROI {roi_id}: ALARMED → CONFIRMING_CLEAR (ratio={ratio:.2f}<{self.RATIO_ALARMED_CLEAR})")
 
         elif self.state == self.STATE_CONFIRMING_CLEAR:
             if self.state_start_time is None:
@@ -917,12 +971,12 @@ class IllegalParkingAlgorithm:
 
             elapsed = (current_time - self.state_start_time).total_seconds()
 
-            if ratio >= 0.5:
+            if ratio >= self.RATIO_CLEAR_RETURN:
                 # 车辆又出现，回到ALARMED
                 self.state = self.STATE_ALARMED
                 self.state_start_time = None
                 logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR → ALARMED (车辆仍在)")
-            elif elapsed >= self.confirm_clear_sec and ratio < 0.2:
+            elif elapsed >= self.confirm_clear_sec and ratio < self.RATIO_CLEAR_CONFIRM:
                 # 确认车辆已离开
                 if self._last_alarm_id and self._parking_start_time:
                     duration_ms = int((current_time - self._parking_start_time).total_seconds() * 1000)
@@ -944,10 +998,6 @@ class IllegalParkingAlgorithm:
 
         return alerts
 
-    def set_last_alarm_id(self, alarm_id: str):
-        """由 main.py 在告警生成后回填 alarm_id"""
-        self._last_alarm_id = alarm_id
-
     def reset(self):
         """重置算法状态"""
         self.state = self.STATE_IDLE
@@ -960,10 +1010,12 @@ class IllegalParkingAlgorithm:
     def get_state(self, current_time: Optional[datetime] = None) -> Dict[str, Any]:
         """获取当前状态"""
         current_time = current_time or datetime.now()
+        # Step 4: 缓存窗口比率
+        window_ratio = self._get_window_ratio()
         state_info = {
             "state": self.state,
             "state_start_time": self.state_start_time.isoformat() if self.state_start_time else None,
-            "window_ratio": self._get_window_ratio(),
+            "window_ratio": window_ratio,
         }
         if self.state in (self.STATE_ALARMED, self.STATE_PARKED_COUNTDOWN) and self._parking_start_time:
             state_info["parking_duration_sec"] = (current_time - self._parking_start_time).total_seconds()
@@ -971,7 +1023,7 @@ class IllegalParkingAlgorithm:
         return state_info
 
 
-class VehicleCongestionAlgorithm:
+class VehicleCongestionAlgorithm(BaseAlgorithm):
     """
     车辆拥堵检测算法（状态机版本 v1.0）
 
@@ -979,8 +1031,8 @@ class VehicleCongestionAlgorithm:
     NORMAL → CONFIRMING_CONGESTION → CONGESTED → CONFIRMING_CLEAR → NORMAL
 
     业务流程：
-    1. 检测到车辆数量 ≥ count_threshold → 拥堵确认期（confirm_congestion_sec，默认60秒）
+    1. 检测到车辆数量 >= count_threshold → 拥堵确认期（confirm_congestion_sec，默认60秒）
-    2. 确认拥堵（窗口内平均车辆数 ≥ threshold）→ 触发告警
+    2. 确认拥堵（窗口内平均车辆数 >= threshold）→ 触发告警
     3. 车辆减少 → 消散确认期（confirm_clear_sec，默认120秒）
     4. 确认消散（平均数 < threshold）→ 发送resolve事件 → 回到正常
 
@@ -999,6 +1051,9 @@ class VehicleCongestionAlgorithm:
     # 滑动窗口参数
     WINDOW_SIZE_SEC = 10
 
+    # Step 2: 阈值常量 — Step 6: 默认消散比例
+    DISSIPATION_RATIO = 0.5
+
     def __init__(
         self,
         count_threshold: int = 5,
@@ -1007,35 +1062,31 @@ class VehicleCongestionAlgorithm:
         cooldown_sec: int = 1800,
         target_classes: Optional[List[str]] = None,
         alarm_level: Optional[int] = None,
+        dissipation_ratio: float = 0.5,  # Step 6: 消散比例可配置
     ):
+        super().__init__()
+
         self.count_threshold = count_threshold
         self.confirm_congestion_sec = confirm_congestion_sec
         self.confirm_clear_sec = confirm_clear_sec
         self.cooldown_sec = cooldown_sec
         self.target_classes = target_classes or ["car", "truck", "bus", "motorcycle"]
         self._alarm_level = alarm_level if alarm_level is not None else self.DEFAULT_ALARM_LEVEL
+        self.dissipation_ratio = dissipation_ratio  # Step 6
 
         # 状态变量
         self.state: str = self.STATE_NORMAL
         self.state_start_time: Optional[datetime] = None
 
-        # 滑动窗口：存储 (timestamp, vehicle_count: int)
+        # 滑动窗口：存储 (timestamp, vehicle_count: int) — Step 3: maxlen=1000
-        self._count_window: deque = deque()
+        self._count_window: deque = deque(maxlen=1000)
 
         # 告警追踪
-        self._last_alarm_id: Optional[str] = None
         self._congestion_start_time: Optional[datetime] = None
 
         # 冷却期管理
         self.alert_cooldowns: Dict[str, datetime] = {}
 
-    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
-        matched_rois = detection.get("matched_rois", [])
-        for roi in matched_rois:
-            if roi.get("roi_id") == roi_id:
-                return True
-        return False
-
     def _check_target_classes(self, detection: Dict) -> bool:
         det_class = detection.get("class", "")
         return det_class in self.target_classes
@@ -1061,6 +1112,28 @@ class VehicleCongestionAlgorithm:
         total = sum(c for _, c in self._count_window)
         return total / len(self._count_window)
 
+    # Step 5: 合并遍历方法
+    def _scan_tracks(self, tracks: List[Dict], roi_id: str) -> Tuple[bool, int, List[float], float]:
+        """
+        一次遍历 tracks，返回 (has_target, count, latest_bbox, max_confidence)。
+        过滤 target_classes。
+        """
+        has_target = False
+        count = 0
+        latest_bbox: List[float] = []
+        max_confidence = 0.0
+        for det in tracks:
+            if self._check_detection_in_roi(det, roi_id) and self._check_target_classes(det):
+                has_target = True
+                count += 1
+                if not latest_bbox:
+                    latest_bbox = det.get("bbox", [])
+                conf = det.get("confidence", 0.0)
+                if conf > max_confidence:
+                    max_confidence = conf
+        return has_target, count, latest_bbox, max_confidence
+
+    # 保留旧方法以防外部调用
     def _get_max_confidence(self, tracks: List[Dict], roi_id: str) -> float:
         max_conf = 0.0
         for det in tracks:
@@ -1085,18 +1158,24 @@ class VehicleCongestionAlgorithm:
         current_time = current_time or datetime.now()
         alerts = []
 
-        # 统计ROI内车辆数
+        # Step 5: 一次遍历获取所有信息
-        vehicle_count = self._count_vehicles_in_roi(tracks, roi_id)
+        _has_target, vehicle_count, scan_bbox, scan_confidence = self._scan_tracks(tracks, roi_id)
+
         self._update_count_window(current_time, vehicle_count)
+
+        # Step 4: 计算一次均值，后续分支复用
         avg_count = self._get_avg_count()
 
+        # Step 6: 消散阈值使用可配置比例
+        dissipation_threshold = self.count_threshold * self.dissipation_ratio
+
         # === 状态机处理 ===
 
         if self.state == self.STATE_NORMAL:
             if avg_count >= self.count_threshold:
                 self.state = self.STATE_CONFIRMING_CONGESTION
                 self.state_start_time = current_time
-                logger.debug(f"ROI {roi_id}: NORMAL → CONFIRMING_CONGESTION (avg={avg_count:.1f}≥{self.count_threshold})")
+                logger.debug(f"ROI {roi_id}: NORMAL → CONFIRMING_CONGESTION (avg={avg_count:.1f}>={self.count_threshold})")
 
         elif self.state == self.STATE_CONFIRMING_CONGESTION:
             if self.state_start_time is None:
@@ -1117,16 +1196,14 @@ class VehicleCongestionAlgorithm:
                    (current_time - self.alert_cooldowns[cooldown_key]).total_seconds() > self.cooldown_sec:
 
                     self._congestion_start_time = self.state_start_time
-                    bbox = self._get_latest_bbox(tracks, roi_id)
-                    confidence = self._get_max_confidence(tracks, roi_id)
 
                     alerts.append({
                         "roi_id": roi_id,
                         "camera_id": camera_id,
-                        "bbox": bbox,
+                        "bbox": scan_bbox,
                         "alert_type": "vehicle_congestion",
                         "alarm_level": self._alarm_level,
-                        "confidence": confidence,
+                        "confidence": scan_confidence,
                         "message": f"检测到车辆拥堵（平均{avg_count:.0f}辆，持续{int(elapsed)}秒）",
                         "first_frame_time": self._congestion_start_time.strftime('%Y-%m-%d %H:%M:%S') if self._congestion_start_time else None,
                         "vehicle_count": int(avg_count),
@@ -1141,11 +1218,11 @@ class VehicleCongestionAlgorithm:
                     logger.debug(f"ROI {roi_id}: CONFIRMING_CONGESTION → NORMAL (冷却期内)")
 
         elif self.state == self.STATE_CONGESTED:
-            # 车辆数降到阈值的一半以下才开始确认消散（避免抖动）
+            # Step 6: 使用可配置的消散比例
-            if avg_count < self.count_threshold * 0.5:
+            if avg_count < dissipation_threshold:
                 self.state = self.STATE_CONFIRMING_CLEAR
                 self.state_start_time = current_time
-                logger.debug(f"ROI {roi_id}: CONGESTED → CONFIRMING_CLEAR (avg={avg_count:.1f}<{self.count_threshold * 0.5:.1f})")
+                logger.debug(f"ROI {roi_id}: CONGESTED → CONFIRMING_CLEAR (avg={avg_count:.1f}<{dissipation_threshold:.1f})")
 
         elif self.state == self.STATE_CONFIRMING_CLEAR:
             if self.state_start_time is None:
@@ -1158,7 +1235,7 @@ class VehicleCongestionAlgorithm:
                 # 又拥堵了，回到CONGESTED
                 self.state = self.STATE_CONGESTED
                 self.state_start_time = None
-                logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR → CONGESTED (avg={avg_count:.1f}≥{self.count_threshold})")
+                logger.debug(f"ROI {roi_id}: CONFIRMING_CLEAR → CONGESTED (avg={avg_count:.1f}>={self.count_threshold})")
             elif elapsed >= self.confirm_clear_sec:
                 # 确认消散
                 if self._last_alarm_id and self._congestion_start_time:
@@ -1181,10 +1258,6 @@ class VehicleCongestionAlgorithm:
 
         return alerts
 
-    def set_last_alarm_id(self, alarm_id: str):
-        """由 main.py 在告警生成后回填 alarm_id"""
-        self._last_alarm_id = alarm_id
-
     def reset(self):
         """重置算法状态"""
         self.state = self.STATE_NORMAL
@@ -1197,10 +1270,12 @@ class VehicleCongestionAlgorithm:
     def get_state(self, current_time: Optional[datetime] = None) -> Dict[str, Any]:
         """获取当前状态"""
         current_time = current_time or datetime.now()
+        # Step 4: 缓存均值计算
+        avg_vehicle_count = self._get_avg_count()
         state_info = {
             "state": self.state,
             "state_start_time": self.state_start_time.isoformat() if self.state_start_time else None,
-            "avg_vehicle_count": self._get_avg_count(),
+            "avg_vehicle_count": avg_vehicle_count,
         }
         if self.state in (self.STATE_CONGESTED, self.STATE_CONFIRMING_CLEAR) and self._congestion_start_time:
             state_info["congestion_duration_sec"] = (current_time - self._congestion_start_time).total_seconds()
@@ -1215,6 +1290,7 @@ class AlgorithmManager:
         self._update_lock = threading.Lock()
         self._registered_keys: set = set()  # 已注册的 (roi_id, bind_id, algo_type) 缓存
 
+        # Bug fix: 默认参数与算法构造函数一致
         self.default_params = {
             "leave_post": {
                 "confirm_on_duty_sec": 10,
@@ -1225,20 +1301,21 @@ class AlgorithmManager:
             "intrusion": {
                 "cooldown_seconds": 300,
                 "confirm_seconds": 5,
+                "confirm_clear_seconds": 180,  # Bug fix: 添加与构造函数一致的默认值
                 "target_class": None,
             },
             "illegal_parking": {
                 "confirm_vehicle_sec": 15,
                 "parking_countdown_sec": 300,
-                "confirm_clear_sec": 30,
+                "confirm_clear_sec": 120,  # Bug fix: 与算法构造函数默认值一致（120，非30）
-                "cooldown_sec": 600,
+                "cooldown_sec": 1800,      # Bug fix: 与算法构造函数默认值一致（1800，非600）
                 "target_classes": ["car", "truck", "bus", "motorcycle"],
             },
             "vehicle_congestion": {
-                "count_threshold": 3,
+                "count_threshold": 5,       # Bug fix: 与算法构造函数默认值一致（5，非3）
                 "confirm_congestion_sec": 60,
-                "confirm_clear_sec": 120,
+                "confirm_clear_sec": 180,  # Bug fix: 与算法构造函数默认值一致（180，非120）
-                "cooldown_sec": 600,
+                "cooldown_sec": 1800,      # Bug fix: 与算法构造函数默认值一致（1800，非600）
                 "target_classes": ["car", "truck", "bus", "motorcycle"],
             },
         }
@@ -1303,7 +1380,7 @@ class AlgorithmManager:
                             else:
                                 self.reload_all_algorithms()
                         else:
-                            # type="full" / "camera" / unknown → 全量重载
+                            # type="full" / "camera" / unknown -> 全量重载
                             self.reload_all_algorithms()
                     except Exception as e:
                         logger.error(f"处理配置更新消息失败: {e}")
@@ -1396,6 +1473,70 @@ class AlgorithmManager:
                         alarm_level=configured_alarm_level,
                     )
                     logger.info(f"已从Redis加载算法: {key}")
+                # Bug fix: 热更新支持 illegal_parking 和 vehicle_congestion
+                elif algo_code == "illegal_parking":
+                    configured_alarm_level = params.get("alarm_level")
+                    algo_params = {
+                        "confirm_vehicle_sec": params.get("confirm_vehicle_sec", 15),
+                        "parking_countdown_sec": params.get("parking_countdown_sec", 300),
+                        "confirm_clear_sec": params.get("confirm_clear_sec", 120),
+                        "cooldown_sec": params.get("cooldown_sec", 1800),
+                        "target_classes": params.get("target_classes", ["car", "truck", "bus", "motorcycle"]),
+                    }
+                    if key in self.algorithms.get(roi_id, {}) and "illegal_parking" in self.algorithms[roi_id].get(key, {}):
+                        algo = self.algorithms[roi_id][key]["illegal_parking"]
+                        algo.confirm_vehicle_sec = algo_params["confirm_vehicle_sec"]
+                        algo.parking_countdown_sec = algo_params["parking_countdown_sec"]
+                        algo.confirm_clear_sec = algo_params["confirm_clear_sec"]
+                        algo.cooldown_sec = algo_params["cooldown_sec"]
+                        algo.target_classes = algo_params["target_classes"]
+                        if configured_alarm_level is not None:
+                            algo._alarm_level = configured_alarm_level
+                        logger.info(f"已热更新违停算法参数: {key}")
+                    else:
+                        self.algorithms[roi_id][key] = {}
+                        self.algorithms[roi_id][key]["illegal_parking"] = IllegalParkingAlgorithm(
+                            confirm_vehicle_sec=algo_params["confirm_vehicle_sec"],
+                            parking_countdown_sec=algo_params["parking_countdown_sec"],
+                            confirm_clear_sec=algo_params["confirm_clear_sec"],
+                            cooldown_sec=algo_params["cooldown_sec"],
+                            target_classes=algo_params["target_classes"],
+                            alarm_level=configured_alarm_level,
+                        )
+                        logger.info(f"已从Redis加载违停算法: {key}")
+                elif algo_code == "vehicle_congestion":
+                    configured_alarm_level = params.get("alarm_level")
+                    algo_params = {
+                        "count_threshold": params.get("count_threshold", 5),
+                        "confirm_congestion_sec": params.get("confirm_congestion_sec", 60),
+                        "confirm_clear_sec": params.get("confirm_clear_sec", 180),
+                        "cooldown_sec": params.get("cooldown_sec", 1800),
+                        "target_classes": params.get("target_classes", ["car", "truck", "bus", "motorcycle"]),
+                        "dissipation_ratio": params.get("dissipation_ratio", 0.5),
+                    }
+                    if key in self.algorithms.get(roi_id, {}) and "vehicle_congestion" in self.algorithms[roi_id].get(key, {}):
+                        algo = self.algorithms[roi_id][key]["vehicle_congestion"]
+                        algo.count_threshold = algo_params["count_threshold"]
+                        algo.confirm_congestion_sec = algo_params["confirm_congestion_sec"]
+                        algo.confirm_clear_sec = algo_params["confirm_clear_sec"]
+                        algo.cooldown_sec = algo_params["cooldown_sec"]
+                        algo.target_classes = algo_params["target_classes"]
+                        algo.dissipation_ratio = algo_params["dissipation_ratio"]
+                        if configured_alarm_level is not None:
+                            algo._alarm_level = configured_alarm_level
+                        logger.info(f"已热更新拥堵算法参数: {key}")
+                    else:
+                        self.algorithms[roi_id][key] = {}
+                        self.algorithms[roi_id][key]["vehicle_congestion"] = VehicleCongestionAlgorithm(
+                            count_threshold=algo_params["count_threshold"],
+                            confirm_congestion_sec=algo_params["confirm_congestion_sec"],
+                            confirm_clear_sec=algo_params["confirm_clear_sec"],
+                            cooldown_sec=algo_params["cooldown_sec"],
+                            target_classes=algo_params["target_classes"],
+                            alarm_level=configured_alarm_level,
+                            dissipation_ratio=algo_params["dissipation_ratio"],
+                        )
+                        logger.info(f"已从Redis加载拥堵算法: {key}")
 
             return True
         except Exception as e:
@@ -1491,7 +1632,34 @@ class AlgorithmManager:
 
                 logger.info(f"[{roi_id}_{bind_id}] 更新周界入侵参数: intrusion={confirm_intrusion_sec}s, clear={confirm_clear_sec}s")
 
-            # 其他算法类型可以在此添加
+            # Bug fix: 热更新支持 illegal_parking 和 vehicle_congestion
+            elif algo_code == "illegal_parking":
+                existing_algo.confirm_vehicle_sec = params.get("confirm_vehicle_sec", 15)
+                existing_algo.parking_countdown_sec = params.get("parking_countdown_sec", 300)
+                existing_algo.confirm_clear_sec = params.get("confirm_clear_sec", 120)
+                existing_algo.cooldown_sec = params.get("cooldown_sec", 1800)
+                if "target_classes" in params:
+                    existing_algo.target_classes = params["target_classes"]
+                alarm_level = params.get("alarm_level")
+                if alarm_level is not None:
+                    existing_algo._alarm_level = alarm_level
+
+                logger.info(f"[{roi_id}_{bind_id}] 更新违停检测参数")
+
+            elif algo_code == "vehicle_congestion":
+                existing_algo.count_threshold = params.get("count_threshold", 5)
+                existing_algo.confirm_congestion_sec = params.get("confirm_congestion_sec", 60)
+                existing_algo.confirm_clear_sec = params.get("confirm_clear_sec", 180)
+                existing_algo.cooldown_sec = params.get("cooldown_sec", 1800)
+                if "target_classes" in params:
+                    existing_algo.target_classes = params["target_classes"]
+                if "dissipation_ratio" in params:
+                    existing_algo.dissipation_ratio = params["dissipation_ratio"]
+                alarm_level = params.get("alarm_level")
+                if alarm_level is not None:
+                    existing_algo._alarm_level = alarm_level
+
+                logger.info(f"[{roi_id}_{bind_id}] 更新拥堵检测参数")
 
             return True
 
@@ -1617,19 +1785,20 @@ class AlgorithmManager:
             self.algorithms[roi_id][key]["illegal_parking"] = IllegalParkingAlgorithm(
                 confirm_vehicle_sec=algo_params.get("confirm_vehicle_sec", 15),
                 parking_countdown_sec=algo_params.get("parking_countdown_sec", 300),
-                confirm_clear_sec=algo_params.get("confirm_clear_sec", 30),
+                confirm_clear_sec=algo_params.get("confirm_clear_sec", 120),
-                cooldown_sec=algo_params.get("cooldown_sec", 600),
+                cooldown_sec=algo_params.get("cooldown_sec", 1800),
                 target_classes=algo_params.get("target_classes", ["car", "truck", "bus", "motorcycle"]),
                 alarm_level=configured_alarm_level,
             )
         elif algorithm_type == "vehicle_congestion":
             self.algorithms[roi_id][key]["vehicle_congestion"] = VehicleCongestionAlgorithm(
-                count_threshold=algo_params.get("count_threshold", 3),
+                count_threshold=algo_params.get("count_threshold", 5),
                 confirm_congestion_sec=algo_params.get("confirm_congestion_sec", 60),
-                confirm_clear_sec=algo_params.get("confirm_clear_sec", 120),
+                confirm_clear_sec=algo_params.get("confirm_clear_sec", 180),
-                cooldown_sec=algo_params.get("cooldown_sec", 600),
+                cooldown_sec=algo_params.get("cooldown_sec", 1800),
                 target_classes=algo_params.get("target_classes", ["car", "truck", "bus", "motorcycle"]),
                 alarm_level=configured_alarm_level,
+                dissipation_ratio=algo_params.get("dissipation_ratio", 0.5),
             )
 
         self._registered_keys.add(cache_key)

config/config_models.py

@@ -20,6 +20,8 @@ class AlgorithmType(str, Enum):
     """算法类型枚举"""
     LEAVE_POST = "leave_post"
     INTRUSION = "intrusion"
+    ILLEGAL_PARKING = "illegal_parking"
+    VEHICLE_CONGESTION = "vehicle_congestion"
     CROWD_DETECTION = "crowd_detection"
     FACE_RECOGNITION = "face_recognition"
 

config/database.py

@@ -885,6 +885,37 @@ class SQLiteManager:
             logger.error(f"获取摄像头算法绑定失败: {e}")
             return []
 
+    def get_all_bindings(self) -> List[Dict[str, Any]]:
+        """获取所有启用的算法绑定（一次查询，避免 N+1）"""
+        try:
+            cursor = self._conn.cursor()
+            cursor.execute("""
+                SELECT b.bind_id, b.roi_id, b.algo_code, b.params, b.priority,
+                       b.enabled, b.created_at, b.updated_at,
+                       a.algo_name, a.target_class
+                FROM roi_algo_bind b
+                LEFT JOIN algorithm_registry a ON b.algo_code = a.algo_code
+                WHERE b.enabled = 1
+                ORDER BY b.priority DESC
+            """)
+            results = []
+            for row in cursor.fetchall():
+                result = dict(zip(
+                    ['bind_id', 'roi_id', 'algo_code', 'params', 'priority',
+                     'enabled', 'created_at', 'updated_at', 'algo_name', 'target_class'],
+                    row
+                ))
+                if result.get('params') and isinstance(result['params'], str):
+                    try:
+                        result['params'] = json.loads(result['params'])
+                    except (json.JSONDecodeError, TypeError):
+                        pass
+                results.append(result)
+            return results
+        except Exception as e:
+            logger.error(f"获取所有算法绑定失败: {e}")
+            return []
+    
     def delete_roi_algo_bind(self, bind_id: str) -> bool:
         """删除ROI算法绑定"""
         try:

core/alarm_upload_worker.py

@@ -47,6 +47,7 @@ class AlarmUploadWorker:
         self._logger = logging.getLogger("alarm_upload_worker")
 
         self._redis: Optional[redis.Redis] = None
+        self._redis_binary: Optional[redis.Redis] = None  # 用于读取截图 bytes
         self._cos_client = None  # 懒初始化
 
         self._thread: Optional[threading.Thread] = None
@@ -80,6 +81,16 @@ class AlarmUploadWorker:
             )
             self._redis.ping()
             self._logger.info(f"Worker Redis 连接成功: {redis_cfg.host}:{redis_cfg.port}/{redis_cfg.db}")
+
+            # 二进制 Redis 连接（用于读取截图 bytes，不做 decode）
+            self._redis_binary = redis.Redis(
+                host=redis_cfg.host,
+                port=redis_cfg.port,
+                db=redis_cfg.db,
+                password=redis_cfg.password,
+                decode_responses=False,
+                socket_connect_timeout=5,
+            )
         except Exception as e:
             self._logger.error(f"Worker Redis 连接失败: {e}")
             return
@@ -136,6 +147,12 @@ class AlarmUploadWorker:
             except Exception:
                 pass
 
+        if self._redis_binary:
+            try:
+                self._redis_binary.close()
+            except Exception:
+                pass
+
         self._logger.info("AlarmUploadWorker 已停止")
 
     def _worker_loop(self):
@@ -184,21 +201,43 @@ class AlarmUploadWorker:
 
         self._logger.info(f"开始处理告警: {alarm_id} (retry={retry_count})")
 
-        # Step 1: 上传截图到 COS（从 base64 解码后直接上传字节流）
+        # Step 1: 上传截图到 COS
+        snapshot_key = (alarm_data.get("ext_data") or {}).get("_snapshot_key")
         snapshot_b64 = alarm_data.get("snapshot_b64")
         object_key = None
 
-        if snapshot_b64:
+        if snapshot_key:
+            # 新格式：从独立 Redis key 获取原始 bytes
+            try:
+                image_bytes = self._redis_binary.get(snapshot_key) if self._redis_binary else None
+                if image_bytes is None:
+                    self._logger.warning(f"截图 key 已过期: {snapshot_key}, 无截图继续上报")
+                else:
+                    object_key = self._upload_snapshot_to_cos(
+                        image_bytes, alarm_id, alarm_data.get("device_id", "unknown")
+                    )
+                    if object_key is None:
+                        self._handle_retry(alarm_json, "COS 上传失败")
+                        return
+                    # 上传成功后删除临时 key
+                    try:
+                        if self._redis_binary:
+                            self._redis_binary.delete(snapshot_key)
+                    except Exception:
+                        pass
+            except Exception as e:
+                self._logger.error(f"截图获取/上传失败: {e}")
+                self._handle_retry(alarm_json, f"截图处理失败: {e}")
+                return
+        elif snapshot_b64:
+            # 兼容旧格式 (Base64)
             try:
                 import base64
                 image_bytes = base64.b64decode(snapshot_b64)
                 object_key = self._upload_snapshot_to_cos(
-                    image_bytes,
+                    image_bytes, alarm_id, alarm_data.get("device_id", "unknown")
-                    alarm_id,
-                    alarm_data.get("device_id", "unknown"),
                 )
                 if object_key is None:
-                    # COS 上传失败，进入重试
                     self._handle_retry(alarm_json, "COS 上传失败")
                     return
             except Exception as e:

core/config_sync.py

@@ -215,6 +215,15 @@ class ConfigSyncManager:
             logger.error(f"本地 Redis 连接失败: {e}")
             self._local_redis = None
 
+    def _safe_close_cloud_redis(self):
+        """安全关闭云端 Redis 连接"""
+        if self._cloud_redis is not None:
+            try:
+                self._cloud_redis.close()
+            except Exception:
+                pass
+            self._cloud_redis = None
+
     def _init_cloud_redis(self):
         """初始化云端 Redis 连接"""
         try:
@@ -238,7 +247,7 @@ class ConfigSyncManager:
 
         except Exception as e:
             logger.warning(f"云端 Redis 连接失败（将使用本地缓存运行）: {e}")
-            self._cloud_redis = None
+            self._safe_close_cloud_redis()
 
     def _init_database(self):
         """初始化 SQLite 数据库连接"""
@@ -311,9 +320,7 @@ class ConfigSyncManager:
         try:
             cameras = self._db_manager.get_all_camera_configs()
             rois = self._db_manager.get_all_roi_configs()
-            binds = []
+            binds = self._db_manager.get_all_bindings()
-            for roi in rois:
-                binds.extend(self._db_manager.get_bindings_by_roi(roi["roi_id"]))
             logger.info(f"[EDGE] Loading config from local db ({source})...")
             logger.info(f"[EDGE] Camera count = {len(cameras)}")
             logger.info(f"[EDGE] ROI count = {len(rois)}")
@@ -378,7 +385,7 @@ class ConfigSyncManager:
                 if self._stop_event.is_set():
                     return
                 logger.warning(f"云端 Redis 连接断开: {e}, {backoff}s 后重连...")
-                self._cloud_redis = None
+                self._safe_close_cloud_redis()
                 self._stop_event.wait(backoff)
                 backoff = min(backoff * 2, max_backoff)
 
@@ -776,10 +783,7 @@ class ConfigSyncManager:
                 bindings_list = self._db_manager.get_bindings_by_camera(camera_id)
             else:
                 roi_configs = self._db_manager.get_all_roi_configs()
-                bindings_list = []
+                bindings_list = self._db_manager.get_all_bindings()
-                for roi in roi_configs:
-                    bindings = self._db_manager.get_bindings_by_roi(roi['roi_id'])
-                    bindings_list.extend(bindings)
 
             roi_dict = {r['roi_id']: r for r in roi_configs}
             bindings_dict: Dict[str, list] = {}
@@ -857,8 +861,7 @@ class ConfigSyncManager:
 
         binds: List[Dict[str, Any]] = []
         rois = self._db_manager.get_all_roi_configs()
-        for roi in rois:
+        binds = self._db_manager.get_all_bindings()
-            binds.extend(self._db_manager.get_bindings_by_roi(roi["roi_id"]))
         return binds
 
     def get_algo_bind_from_redis(self, bind_id: str) -> Optional[Dict[str, Any]]:

core/postprocessor.py

@@ -78,12 +78,14 @@ class NMSProcessor:
         max_output_size: int
     ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
         """GPU 加速 NMS"""
+        with torch.no_grad():
             boxes_t = torch.from_numpy(boxes).cuda()
             scores_t = torch.from_numpy(scores).cuda()
 
             keep = torch_nms(boxes_t, scores_t, iou_threshold=self.nms_threshold)
 
             keep_np = keep.cpu().numpy()
+            del boxes_t, scores_t, keep
 
             if len(keep_np) > max_output_size:
                 top_k = np.argsort(scores[keep_np])[::-1][:max_output_size]

core/result_reporter.py

@@ -112,9 +112,20 @@ class ResultReporter:
             self._logger.info(
                 f"Redis 连接成功: {redis_cfg.host}:{redis_cfg.port}/{redis_cfg.db}"
             )
+
+            # 二进制 Redis 连接（用于存储截图 bytes，不做 decode）
+            self._redis_binary = redis.Redis(
+                host=redis_cfg.host,
+                port=redis_cfg.port,
+                db=redis_cfg.db,
+                password=redis_cfg.password,
+                decode_responses=False,
+                socket_connect_timeout=5,
+            )
         except Exception as e:
             self._logger.error(f"Redis 连接失败: {e}")
             self._redis = None
+            self._redis_binary = None
 
     def report_alarm(self, alarm_info: AlarmInfo, screenshot: Optional[np.ndarray] = None) -> bool:
         """
@@ -129,13 +140,22 @@ class ResultReporter:
         """
         self._performance_stats["alerts_generated"] += 1
 
-        # 将截图编码为 JPEG base64，直接通过 Redis 传递给 Worker 上传 COS
+        # 将截图编码为 JPEG，直接存储 bytes 到独立 Redis key，避免 Base64 开销
         if screenshot is not None:
             try:
                 import cv2
-                import base64
                 success, buffer = cv2.imencode('.jpg', screenshot, [cv2.IMWRITE_JPEG_QUALITY, 85])
-                if success:
+                if success and self._redis_binary is not None:
+                    snapshot_key = f"local:alarm:snapshot:{alarm_info.alarm_id}"
+                    # 直接存储 JPEG bytes，避免 Base64 编解码开销
+                    self._redis_binary.set(snapshot_key, buffer.tobytes(), ex=3600)
+                    alarm_info.snapshot_b64 = None
+                    if alarm_info.ext_data is None:
+                        alarm_info.ext_data = {}
+                    alarm_info.ext_data["_snapshot_key"] = snapshot_key
+                elif success:
+                    # 降级：无二进制 Redis 连接时使用 Base64
+                    import base64
                     alarm_info.snapshot_b64 = base64.b64encode(buffer.tobytes()).decode('ascii')
                 else:
                     self._logger.warning("截图 JPEG 编码失败")
@@ -211,6 +231,12 @@ class ResultReporter:
             except Exception:
                 pass
 
+        if hasattr(self, '_redis_binary') and self._redis_binary:
+            try:
+                self._redis_binary.close()
+            except Exception:
+                pass
+
         self._logger.info("ResultReporter 清理完成")
 
     def cleanup(self):

core/screenshot_handler.py

@@ -59,6 +59,7 @@ class ScreenshotHandler:
 
         self._thread: Optional[threading.Thread] = None
         self._stop_event = threading.Event()
+        self._last_pending_check = 0.0
 
     # ==================== 生命周期 ====================
 
@@ -180,20 +181,26 @@ class ScreenshotHandler:
 
                 backoff = 5  # 重置退避
 
+                # 每 60 秒检查一次 pending 消息
+                import time as _time
+                if _time.time() - self._last_pending_check > 60:
+                    self._last_pending_check = _time.time()
+                    self._cleanup_pending_messages()
+
                 for stream_name, messages in results:
                     for msg_id, fields in messages:
                         try:
                             self._handle_request(fields)
-                        except Exception as e:
+                            # 处理成功才 ACK
-                            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:
+                            logger.error("[截图] 处理请求失败 (msg_id=%s): %s", msg_id, e)
+                            # 不 ACK，消息留在 pending list 等待重试
 
             except redis.ConnectionError as e:
                 if self._stop_event.is_set():
@@ -409,3 +416,38 @@ class ScreenshotHandler:
             logger.info("[截图] 降级写 Redis 成功: request_id=%s", request_id)
         except Exception as e:
             logger.error("[截图] 降级写 Redis 也失败: %s", e)
+
+    # ==================== Pending 消息清理 ====================
+
+    _MAX_RETRY_COUNT = 3
+    _PENDING_IDLE_MS = 30000  # 消息 pending 超过 30 秒才处理
+
+    def _cleanup_pending_messages(self):
+        """清理 pending list 中重试次数过多的消息"""
+        try:
+            pending = self._cloud_redis.xpending_range(
+                SNAP_REQUEST_STREAM, SNAP_CONSUMER_GROUP,
+                min="-", max="+", count=50,
+                consumername=self._consumer_name
+            )
+            for entry in pending:
+                msg_id = entry['message_id']
+                delivery_count = entry['times_delivered']
+                idle_ms = entry['time_since_delivered']
+
+                if idle_ms < self._PENDING_IDLE_MS:
+                    continue
+
+                if delivery_count > self._MAX_RETRY_COUNT:
+                    logger.warning(
+                        "[截图] 消息超过最大重试次数，丢弃: msg_id=%s, retries=%d",
+                        msg_id, delivery_count
+                    )
+                    try:
+                        self._cloud_redis.xack(
+                            SNAP_REQUEST_STREAM, SNAP_CONSUMER_GROUP, msg_id
+                        )
+                    except Exception:
+                        pass
+        except Exception as e:
+            logger.debug("[截图] 检查 pending list: %s", e)

docs/code_review_report.md

@@ -0,0 +1,646 @@
+# algorithms.py 代码审查报告
+
+> 审查日期: 2026-04-02
+> 审查文件: algorithms.py (1733行)
+> 审查范围: LeavePostAlgorithm, IntrusionAlgorithm, IllegalParkingAlgorithm, VehicleCongestionAlgorithm, AlgorithmManager
+
+---
+
+## 1. 功能基线清单
+
+### 1.1 LeavePostAlgorithm (离岗检测)
+
+**状态定义 (7个):**
+
+| 状态 | 常量 | 含义 |
+|------|------|------|
+| INIT | `STATE_INIT` | 初始化，等待检测到人 |
+| CONFIRMING_ON_DUTY | `STATE_CONFIRMING_ON_DUTY` | 上岗确认中（需持续检测到人） |
+| ON_DUTY | `STATE_ON_DUTY` | 已确认在岗 |
+| CONFIRMING_OFF_DUTY | `STATE_CONFIRMING_OFF_DUTY` | 离岗确认中（持续未检测到人） |
+| OFF_DUTY_COUNTDOWN | `STATE_OFF_DUTY_COUNTDOWN` | 离岗倒计时（确认离岗后等待告警） |
+| ALARMED | `STATE_ALARMED` | 已告警（等待回岗） |
+| NON_WORK_TIME | `STATE_NON_WORK_TIME` | 非工作时间 |
+
+**状态转换矩阵:**
+
+| 从 \ 到 | INIT | CONFIRMING_ON_DUTY | ON_DUTY | CONFIRMING_OFF_DUTY | OFF_DUTY_COUNTDOWN | ALARMED | NON_WORK_TIME |
+|---------|------|--------------------|---------|--------------------|-------------------|---------|--------------|
+| INIT | - | roi_has_person==True | - | - | - | - | not in_working_hours |
+| CONFIRMING_ON_DUTY | detection_ratio==0 | - | elapsed>=confirm_on_duty_sec AND ratio>=0.6 | - | - | - | not in_working_hours |
+| ON_DUTY | - | - | - | detection_ratio<0.2 | - | - | not in_working_hours |
+| CONFIRMING_OFF_DUTY | - | - | detection_ratio>=0.5 | - | elapsed>=confirm_off_duty_sec AND ratio<0.2 | - | not in_working_hours |
+| OFF_DUTY_COUNTDOWN | - | - | roi_has_person==True | - | - | elapsed>=leave_countdown_sec AND cooldown ok | not in_working_hours |
+| ALARMED | - | roi_has_person==True | - | - | - | - | not in_working_hours |
+| NON_WORK_TIME | in_working_hours | - | - | - | - | - | - |
+
+**关键行为:**
+- 使用滑动窗口（10秒）平滑检测结果，计算 detection_ratio
+- ALARMED -> CONFIRMING_ON_DUTY 复用上岗确认状态（不是独立的回岗确认状态）
+- 进入 ON_DUTY 状态时，若存在 `_last_alarm_id`，自动发送 `alarm_resolve` 事件
+- 进入 NON_WORK_TIME 时，若有未结束告警，发送 resolve_type="non_work_time" 的 resolve 事件
+- 冷却期检查使用 `cooldown_key = f"{camera_id}_{roi_id}"`
+- `_last_alarm_id` 由外部 main.py 通过 `set_last_alarm_id()` 回填
+- `_leave_start_time` 在进入 OFF_DUTY_COUNTDOWN 时记录（值等于 state_start_time）
+
+**构造函数参数:**
+- `confirm_on_duty_sec`: int = 10 (上岗确认窗口)
+- `confirm_off_duty_sec`: int = 30 (离岗确认窗口)
+- `confirm_return_sec`: int = 10 (回岗确认窗口 -- 注意: 代码中实际未使用此参数)
+- `leave_countdown_sec`: int = 300 (离岗倒计时)
+- `cooldown_sec`: int = 600 (告警冷却期)
+- `working_hours`: Optional[List[Dict]] = None
+- `target_class`: Optional[str] = "person"
+- `alarm_level`: Optional[int] = None (默认2)
+- `confirm_leave_sec`: Optional[int] = None (向后兼容旧参数名)
+
+### 1.2 IntrusionAlgorithm (周界入侵)
+
+**状态定义 (4个):**
+
+| 状态 | 常量 | 含义 |
+|------|------|------|
+| IDLE | `STATE_IDLE` | 空闲，无入侵 |
+| CONFIRMING_INTRUSION | `STATE_CONFIRMING_INTRUSION` | 入侵确认中 |
+| ALARMED | `STATE_ALARMED` | 已告警（等待入侵消失） |
+| CONFIRMING_CLEAR | `STATE_CONFIRMING_CLEAR` | 入侵消失确认中 |
+
+**状态转换矩阵:**
+
+| 从 \ 到 | IDLE | CONFIRMING_INTRUSION | ALARMED | CONFIRMING_CLEAR |
+|---------|------|---------------------|---------|-----------------|
+| IDLE | - | roi_has_person==True | - | - |
+| CONFIRMING_INTRUSION | not roi_has_person OR cooldown内 OR state_start_time==None | - | elapsed>=confirm_intrusion_seconds AND cooldown ok | - |
+| ALARMED | - | - | - | not roi_has_person |
+| CONFIRMING_CLEAR | elapsed>=confirm_clear_seconds AND no person | - | person_elapsed>=confirm_intrusion_seconds (持续有人) OR state_start_time==None | - |
+
+**关键行为:**
+- 不使用滑动窗口，直接使用当前帧的 `roi_has_person` 判断
+- CONFIRMING_CLEAR 有子状态追踪: `_person_detected_in_clear_time` 用于判断短暂有人 vs 持续有人
+- 冷却期内入侵确认直接回到 IDLE（不触发告警）
+- 包含防御性编程: state_start_time==None 时重置到 IDLE
+- `_check_target_class` 允许 target_class 为 None（匹配所有类别）
+
+**构造函数参数:**
+- `cooldown_seconds`: int = 300
+- `confirm_seconds`: int = 5 (向后兼容)
+- `confirm_intrusion_seconds`: Optional[int] = None (默认使用 confirm_seconds)
+- `confirm_clear_seconds`: Optional[int] = None (默认180)
+- `target_class`: Optional[str] = None
+- `alarm_level`: Optional[int] = None (默认1)
+
+### 1.3 IllegalParkingAlgorithm (车辆违停)
+
+**状态定义 (5个):**
+
+| 状态 | 常量 | 含义 |
+|------|------|------|
+| IDLE | `STATE_IDLE` | 空闲 |
+| CONFIRMING_VEHICLE | `STATE_CONFIRMING_VEHICLE` | 车辆确认中 |
+| PARKED_COUNTDOWN | `STATE_PARKED_COUNTDOWN` | 违停倒计时 |
+| ALARMED | `STATE_ALARMED` | 已告警 |
+| CONFIRMING_CLEAR | `STATE_CONFIRMING_CLEAR` | 消失确认中 |
+
+**状态转换矩阵:**
+
+| 从 \ 到 | IDLE | CONFIRMING_VEHICLE | PARKED_COUNTDOWN | ALARMED | CONFIRMING_CLEAR |
+|---------|------|-------------------|-----------------|---------|-----------------|
+| IDLE | - | roi_has_vehicle | - | - | - |
+| CONFIRMING_VEHICLE | ratio<0.3 OR state_start_time==None | - | elapsed>=confirm_vehicle_sec AND ratio>=0.6 | - | - |
+| PARKED_COUNTDOWN | ratio<0.2 (车离开) OR cooldown内 OR state_start_time==None | - | - | elapsed>=parking_countdown_sec AND cooldown ok | - |
+| ALARMED | - | - | - | - | ratio<0.15 |
+| CONFIRMING_CLEAR | elapsed>=confirm_clear_sec AND ratio<0.2 OR state_start_time==None | - | - | ratio>=0.5 | - |
+
+**关键行为:**
+- 使用滑动窗口（WINDOW_SIZE_SEC=10秒）
+- 支持多类车辆: target_classes 默认 ["car", "truck", "bus", "motorcycle"]
+- 告警字段包含 `confidence` 和 `duration_minutes`
+- CONFIRMING_CLEAR -> IDLE 时清除 alert_cooldowns（新车违停可正常告警）
+- ALARMED 进入 CONFIRMING_CLEAR 的阈值(0.15)比其他算法更严格
+
+**构造函数参数:**
+- `confirm_vehicle_sec`: int = 15
+- `parking_countdown_sec`: int = 300
+- `confirm_clear_sec`: int = 120
+- `cooldown_sec`: int = 1800
+- `target_classes`: Optional[List[str]] = None (默认 ["car", "truck", "bus", "motorcycle"])
+- `alarm_level`: Optional[int] = None (默认1)
+
+### 1.4 VehicleCongestionAlgorithm (车辆拥堵)
+
+**状态定义 (4个):**
+
+| 状态 | 常量 | 含义 |
+|------|------|------|
+| NORMAL | `STATE_NORMAL` | 正常 |
+| CONFIRMING_CONGESTION | `STATE_CONFIRMING_CONGESTION` | 拥堵确认中 |
+| CONGESTED | `STATE_CONGESTED` | 拥堵中 |
+| CONFIRMING_CLEAR | `STATE_CONFIRMING_CLEAR` | 消散确认中 |
+
+**状态转换矩阵:**
+
+| 从 \ 到 | NORMAL | CONFIRMING_CONGESTION | CONGESTED | CONFIRMING_CLEAR |
+|---------|--------|----------------------|-----------|-----------------|
+| NORMAL | - | avg_count >= count_threshold | - | - |
+| CONFIRMING_CONGESTION | avg_count < count_threshold OR cooldown内 OR state_start_time==None | - | elapsed >= confirm_congestion_sec AND cooldown ok | - |
+| CONGESTED | - | - | - | avg_count < count_threshold * 0.5 |
+| CONFIRMING_CLEAR | elapsed >= confirm_clear_sec OR state_start_time==None | - | avg_count >= count_threshold | - |
+
+**关键行为:**
+- 使用滑动窗口（WINDOW_SIZE_SEC=10秒）存储车辆计数，取平均值判断
+- 消散需车辆数降到阈值的 **50%** 以下才开始确认（避免抖动）
+- CONFIRMING_CLEAR -> NORMAL 时清除 alert_cooldowns
+- 告警字段包含 `vehicle_count` 和 `confidence`
+
+**构造函数参数:**
+- `count_threshold`: int = 5
+- `confirm_congestion_sec`: int = 60
+- `confirm_clear_sec`: int = 180
+- `cooldown_sec`: int = 1800
+- `target_classes`: Optional[List[str]] = None (默认 ["car", "truck", "bus", "motorcycle"])
+- `alarm_level`: Optional[int] = None (默认2)
+
+### 1.5 AlgorithmManager
+
+**数据结构:**
+```
+self.algorithms: Dict[str, Dict[str, Dict[str, Algorithm]]]
+  结构: { roi_id: { "{roi_id}_{bind_id}": { algo_type: algo_instance } } }
+```
+
+**公开方法:**
+- `start_config_subscription()` - 启动 Redis 配置订阅
+- `stop_config_subscription()` - 停止配置订阅
+- `load_bind_from_redis(bind_id)` - 从 Redis 加载单个绑定配置
+- `reload_bind_algorithm(bind_id)` - 重载单个绑定
+- `reload_algorithm(roi_id)` - 重载单个 ROI 的所有算法
+- `update_algorithm_params(roi_id, bind_id, bind_config)` - 仅更新参数，保留状态
+- `reload_all_algorithms(preserve_state=True)` - 重载全部算法
+- `register_algorithm(roi_id, bind_id, algorithm_type, params)` - 注册算法（带缓存）
+- `process(roi_id, bind_id, camera_id, algorithm_type, tracks, current_time)` - 处理检测结果
+- `update_roi_params(roi_id, bind_id, algorithm_type, params)` - 更新参数
+- `reset_algorithm(roi_id, bind_id=None)` - 重置算法状态
+- `reset_all()` - 重置所有算法
+- `remove_roi(roi_id)` - 移除 ROI
+- `remove_bind(roi_id, bind_id)` - 移除绑定
+- `get_status(roi_id)` - 获取状态
+
+---
+
+## 2. 接口契约清单
+
+### 2.1 process() 方法统一签名
+
+所有四个算法的 `process()` 方法具有相同签名:
+
+```python
+def process(
+    self,
+    roi_id: str,
+    camera_id: str,
+    tracks: List[Dict],
+    current_time: Optional[datetime] = None,
+) -> List[Dict]
+```
+
+**tracks 输入格式 (每个元素):**
+```python
+{
+    "track_id": str,           # 跟踪ID
+    "class": str,              # 检测类别 ("person", "car", "truck", ...)
+    "confidence": float,       # 置信度
+    "bbox": List[float],       # 边界框 [x1, y1, x2, y2]
+    "matched_rois": [          # 匹配的ROI列表
+        {"roi_id": str}
+    ],
+}
+```
+
+### 2.2 告警输出格式
+
+#### LeavePostAlgorithm 告警:
+```python
+{
+    "track_id": str,           # 等于 roi_id
+    "camera_id": str,
+    "bbox": List[float],       # 可能为空 []
+    "alert_type": "leave_post",
+    "alarm_level": int,        # 默认 2
+    "message": "人员离岗告警",
+    "first_frame_time": str,   # 格式: '%Y-%m-%d %H:%M:%S'
+}
+```
+注意: leave_post 告警使用 `track_id` 而非 `roi_id` 字段名（与其他算法不同）。
+
+#### IntrusionAlgorithm 告警:
+```python
+{
+    "roi_id": str,
+    "camera_id": str,
+    "bbox": List[float],
+    "alert_type": "intrusion",
+    "alarm_level": int,        # 默认 1
+    "message": "检测到周界入侵",
+    "first_frame_time": str,   # 格式: '%Y-%m-%d %H:%M:%S'
+}
+```
+
+#### IllegalParkingAlgorithm 告警:
+```python
+{
+    "roi_id": str,
+    "camera_id": str,
+    "bbox": List[float],
+    "alert_type": "illegal_parking",
+    "alarm_level": int,        # 默认 1
+    "confidence": float,
+    "message": str,            # 动态生成，包含停留分钟数
+    "first_frame_time": str,   # 格式: '%Y-%m-%d %H:%M:%S'，可能为 None
+    "duration_minutes": float,
+}
+```
+
+#### VehicleCongestionAlgorithm 告警:
+```python
+{
+    "roi_id": str,
+    "camera_id": str,
+    "bbox": List[float],
+    "alert_type": "vehicle_congestion",
+    "alarm_level": int,        # 默认 2
+    "confidence": float,
+    "message": str,            # 动态生成，包含平均车辆数和持续秒数
+    "first_frame_time": str,   # 格式: '%Y-%m-%d %H:%M:%S'，可能为 None
+    "vehicle_count": int,
+}
+```
+
+#### alarm_resolve 事件 (所有算法统一格式):
+```python
+{
+    "alert_type": "alarm_resolve",
+    "resolve_alarm_id": str,
+    "duration_ms": int,
+    "last_frame_time": str,    # 格式: '%Y-%m-%d %H:%M:%S'
+    "resolve_type": str,       # "person_returned" | "non_work_time" | "intrusion_cleared" | "vehicle_left" | "congestion_cleared"
+}
+```
+
+### 2.3 AlgorithmManager.process() 签名
+
+```python
+def process(
+    self,
+    roi_id: str,
+    bind_id: str,
+    camera_id: str,
+    algorithm_type: str,
+    tracks: List[Dict],
+    current_time: Optional[datetime] = None,
+) -> List[Dict]
+```
+
+### 2.4 AlgorithmManager.register_algorithm() 签名
+
+```python
+def register_algorithm(
+    self,
+    roi_id: str,
+    bind_id: str,
+    algorithm_type: str,                 # "leave_post" | "intrusion" | "illegal_parking" | "vehicle_congestion"
+    params: Optional[Dict[str, Any]] = None,
+)
+```
+
+---
+
+## 3. 已发现的潜在问题
+
+### 3.1 Critical (必须修复)
+
+**[C1] LeavePostAlgorithm: `confirm_return_sec` 参数声明但从未使用**
+- 位置: 第53行声明，但状态机中 ALARMED -> CONFIRMING_ON_DUTY -> ON_DUTY 的转换直接复用 `confirm_on_duty_sec`
+- 影响: 使用者设置 `confirm_return_sec` 以为可以独立控制回岗确认时长，但实际无效
+- 建议: 文档中声明此参数复用 `confirm_on_duty_sec`，或实现独立的回岗确认逻辑
+
+**[C2] LeavePostAlgorithm: resolve 事件的 duration_ms 计算依赖 `_leave_start_time`，但该值可能为 None**
+- 位置: 第325行 `duration_ms = int((current_time - self._leave_start_time).total_seconds() * 1000)`
+- 场景: 如果算法在 OFF_DUTY_COUNTDOWN 之前（即 _leave_start_time 赋值之前）因某种异常跳到 ON_DUTY 且 _last_alarm_id 非空，会抛出 TypeError
+- 风险: 低概率但会导致该帧整个 process 调用抛异常
+
+**[C3] LeavePostAlgorithm: `_leave_start_time` 赋值时机问题**
+- 位置: 第277行 `self._leave_start_time = self.state_start_time`
+- `state_start_time` 此时是 CONFIRMING_OFF_DUTY 的开始时间（非 OFF_DUTY_COUNTDOWN 的开始时间，因为 state_start_time 在下一行第276行才被更新为 current_time）
+- 实际效果: `_leave_start_time` 记录的是**离岗确认开始时间**，不是**倒计时开始时间**
+- 审查结论: 这是有意设计，离开时间应该从人离开被确认开始计算，但代码注释"记录离开时间"可能造成误解
+
+### 3.2 Important (应该修复)
+
+**[I1] LeavePostAlgorithm 告警字典使用 `track_id` 而非 `roi_id`**
+- 位置: 第299行 `"track_id": roi_id`
+- 其他三个算法统一使用 `"roi_id": roi_id`
+- 影响: main.py 中的 `_handle_detections` 不直接使用此字段（它有自己的 roi_id），所以不影响功能，但接口不一致
+
+**[I2] AlgorithmManager 缺乏线程安全**
+- `process()` 方法未加锁（第1637-1651行），而 `register_algorithm()` 也未加锁
+- `_update_lock` 仅在 `load_bind_from_redis` 和 `reload_all_algorithms` 中使用
+- 风险: 如果 config_update_worker 线程调用 `reload_all_algorithms` 同时主线程调用 `process`，可能读到不一致的 `self.algorithms` 字典
+- 缓解: Python GIL 在字典读操作上提供了一定程度的原子性保护，实际崩溃概率很低
+
+**[I3] AlgorithmManager.default_params 中 illegal_parking 的 confirm_clear_sec 默认值(30) 与 IllegalParkingAlgorithm 构造函数默认值(120) 不一致**
+- 位置: 第1233行 vs 第751行
+- 影响: 通过 AlgorithmManager 创建的 illegal_parking 算法 confirm_clear_sec 为 30，直接创建为 120
+
+**[I4] AlgorithmManager.default_params 中 vehicle_congestion 的 count_threshold 默认值(3) 与 VehicleCongestionAlgorithm 构造函数默认值(5) 不一致**
+- 位置: 第1237行 vs 第1004行
+- 影响: 通过 AlgorithmManager 创建的算法阈值为 3，直接创建为 5
+
+**[I5] `update_algorithm_params` 仅支持 leave_post 和 intrusion**
+- 位置: 第1461-1496行
+- 缺少 illegal_parking 和 vehicle_congestion 的参数更新逻辑（第1494行注释 "其他算法类型可以在此添加"）
+- 影响: `reload_all_algorithms(preserve_state=True)` 对 illegal_parking/vehicle_congestion 会回退到 `load_bind_from_redis`，会重置算法状态
+
+**[I6] `load_bind_from_redis` 仅支持 leave_post 和 intrusion**
+- 位置: 第1322-1403行
+- 缺少 illegal_parking 和 vehicle_congestion 的 Redis 加载逻辑
+- 影响: 从 Redis 热更新配置时，这两种算法无法被加载
+
+**[I7] IntrusionAlgorithm: `_get_latest_bbox` 不检查 target_class**
+- 位置: 第456-460行
+- 与 LeavePostAlgorithm 不同（第139-142行会检查 target_class）
+- 影响: 可能返回非目标类别的 bbox
+
+**[I8] `_is_in_working_hours` 不支持跨午夜时间段**
+- 位置: 第112行 `if start_minutes <= current_minutes < end_minutes`
+- 如果 working_hours 配置为 `{"start": "22:00", "end": "06:00"}`，则无法正确判断
+- 影响: 夜班场景可能不工作
+
+### 3.3 Suggestions (建议改进)
+
+**[S1] 滑动窗口的 window_size_sec 硬编码为 10 秒**
+- LeavePostAlgorithm 第80行: `self.window_size_sec = 10`
+- IllegalParkingAlgorithm 第744行: `WINDOW_SIZE_SEC = 10`
+- VehicleCongestionAlgorithm 第1000行: `WINDOW_SIZE_SEC = 10`
+- 建议: 提取为可配置参数
+
+**[S2] LeavePostAlgorithm._update_detection_window 与 IllegalParkingAlgorithm._update_window 实现逻辑相同但代码重复**
+- 可提取为基类方法或工具函数
+
+**[S3] LeavePostAlgorithm.get_state() 使用 datetime.now() 而非参数传入的 current_time**
+- 位置: 第367-372行
+- 在测试场景中会导致状态信息不准确（不影响核心逻辑，仅影响监控展示）
+
+**[S4] AlgorithmManager.get_status() 中 leave_post 分支访问 `alarm_sent` 属性**
+- 位置: 第1724行
+- LeavePostAlgorithm 实际上没有 `alarm_sent` 属性，getattr 返回 False
+- 这是旧版本残留代码
+
+**[S5] AlgorithmManager.remove_roi() 中 bind_id 解析逻辑脆弱**
+- 位置: 第1703行 `key.split("_")[-1]`
+- 如果 bind_id 本身包含下划线，解析会出错
+- key 格式为 `"{roi_id}_{bind_id}"`，应该用 `key[len(roi_id)+1:]` 提取 bind_id
+
+**[S6] `_is_in_working_hours` 中的 bare `except:` (第99行, 第124行)**
+- 应该至少 except Exception 或更具体的异常类型
+
+---
+
+## 4. 测试覆盖分析
+
+### 4.1 test_leave_post_full_workflow.py 覆盖的场景
+
+| 场景 | 状态路径 | 覆盖 |
+|------|----------|------|
+| 上岗确认成功 | INIT -> CONFIRMING_ON_DUTY -> ON_DUTY | YES |
+| 离岗确认 | ON_DUTY -> CONFIRMING_OFF_DUTY -> OFF_DUTY_COUNTDOWN | YES |
+| 倒计时触发告警 | OFF_DUTY_COUNTDOWN -> ALARMED | YES |
+| 回岗 resolve | ALARMED -> CONFIRMING_ON_DUTY -> ON_DUTY (+ resolve) | YES |
+| 告警字段验证 | 无 duration_minutes, 有 first_frame_time | YES |
+| resolve 字段验证 | duration_ms, resolve_alarm_id, resolve_type | YES |
+| set_last_alarm_id 回填 | - | YES |
+
+### 4.2 test_vehicle_algorithms.py 覆盖的场景
+
+**IllegalParkingAlgorithm:**
+| 场景 | 覆盖 |
+|------|------|
+| 完整生命周期 IDLE->CONFIRMING->COUNTDOWN->ALARMED->CLEAR->IDLE | YES |
+| 车辆短暂路过不触发 | YES |
+| 多类车辆检测 (truck, bus) | YES |
+| person 不触发违停 | YES |
+| 冷却期内不重复告警 | YES |
+| resolve 事件发送 | YES |
+
+**VehicleCongestionAlgorithm:**
+| 场景 | 覆盖 |
+|------|------|
+| 完整生命周期 NORMAL->CONFIRMING->CONGESTED->CLEAR->NORMAL | YES |
+| 少于阈值不触发 | YES |
+| 短暂拥堵不触发 | YES |
+| resolve 事件发送 | YES |
+
+**AlgorithmManager:**
+| 场景 | 覆盖 |
+|------|------|
+| 注册 illegal_parking | YES |
+| 注册 vehicle_congestion | YES |
+| process 调用 | YES |
+| get_status 调用 | YES |
+| 重复注册走缓存 | YES |
+| reset_algorithm | YES |
+
+### 4.3 测试覆盖缺口
+
+**LeavePostAlgorithm 未覆盖:**
+- [ ] CONFIRMING_ON_DUTY -> INIT (人消失)
+- [ ] CONFIRMING_OFF_DUTY -> ON_DUTY (人回来，ratio>=0.5)
+- [ ] OFF_DUTY_COUNTDOWN -> ON_DUTY (倒计时期间回来)
+- [ ] 非工作时间自动 resolve
+- [ ] NON_WORK_TIME -> INIT (工作时间恢复)
+- [ ] 冷却期内不重复告警
+- [ ] 空 tracks 输入
+- [ ] working_hours 配置解析（字符串格式）
+
+**IntrusionAlgorithm 完全未测试:**
+- [ ] 完整生命周期 IDLE->CONFIRMING->ALARMED->CLEAR->IDLE
+- [ ] 入侵确认中人消失
+- [ ] CONFIRMING_CLEAR 中短暂有人 vs 持续有人
+- [ ] 冷却期
+- [ ] resolve 事件
+- [ ] target_class=None 匹配所有类别
+- [ ] state_start_time==None 的防御性代码分支
+
+**IllegalParkingAlgorithm 未覆盖:**
+- [ ] state_start_time==None 的防御性代码分支 (CONFIRMING_VEHICLE, PARKED_COUNTDOWN, CONFIRMING_CLEAR)
+- [ ] CONFIRMING_CLEAR -> ALARMED (车辆又出现, ratio>=0.5)
+- [ ] ALARMED 状态下 ratio 在 0.15-0.5 之间（维持 ALARMED）
+
+**VehicleCongestionAlgorithm 未覆盖:**
+- [ ] state_start_time==None 的防御性代码分支
+- [ ] CONFIRMING_CLEAR -> CONGESTED (又拥堵了)
+- [ ] 消散阈值 0.5*count_threshold 的边界值
+- [ ] 冷却期测试
+
+**AlgorithmManager 未覆盖:**
+- [ ] start_config_subscription / stop_config_subscription
+- [ ] load_bind_from_redis
+- [ ] reload_all_algorithms (含孤立实例清理)
+- [ ] update_algorithm_params
+- [ ] remove_roi / remove_bind
+- [ ] 并发调用安全性
+- [ ] register_algorithm 的 leave_post 和 intrusion 类型
+
+---
+
+## 5. 优化安全边界
+
+### 5.1 不可修改区域 (功能合约)
+
+以下代码是外部依赖的契约，修改会破坏 main.py 或其他模块：
+
+1. **所有算法的 `process()` 方法签名** -- main.py 的 `_handle_detections` 直接调用
+2. **告警字典的字段名和类型** -- main.py 的 `_handle_detections` 依赖 `alert_type`, `alarm_level`, `confidence`, `bbox`, `message`, `first_frame_time`, `duration_minutes`, `vehicle_count`
+3. **`alarm_resolve` 事件格式** -- main.py 的 resolve 逻辑依赖 `resolve_alarm_id`, `duration_ms`, `last_frame_time`, `resolve_type`
+4. **`set_last_alarm_id(alarm_id)` 方法** -- main.py 回填 alarm_id
+5. **`reset()` 方法** -- AlgorithmManager 调用
+6. **`get_state()` 方法** -- AlgorithmManager.get_status() 调用
+7. **AlgorithmManager.process() 签名和返回值** -- main.py 直接调用
+8. **AlgorithmManager.register_algorithm() 签名** -- main.py 直接调用
+9. **AlgorithmManager.algorithms 的三层字典结构** -- main.py 直接访问内部实例来获取 `_leave_start_time` 等属性 (第905-911行)
+
+### 5.2 可安全优化区域
+
+以下代码修改不会影响外部行为：
+
+1. **滑动窗口实现** -- `_update_detection_window`, `_update_window`, `_update_count_window` 的内部实现可以优化，只要 `_get_detection_ratio()`, `_get_window_ratio()`, `_get_avg_count()` 的语义不变
+2. **`_check_detection_in_roi` / `_check_target_class` / `_check_target_classes`** -- 内部实现可优化，接口不变即可
+3. **`_get_latest_bbox` / `_get_max_confidence`** -- 辅助方法，内部实现可优化
+4. **`_is_in_working_hours` / `_parse_time_to_minutes`** -- 内部实现可优化（建议修复跨午夜问题）
+5. **AlgorithmManager 的 Redis 相关方法** -- `load_bind_from_redis`, `reload_*`, `_config_update_worker` 可以修改，不影响算法核心逻辑
+6. **日志输出** -- 所有 logger.* 调用可以调整
+7. **`default_params` 字典** -- 可以修正默认值不一致的问题
+
+### 5.3 高风险修改区域 (需要完整回归测试)
+
+1. **状态转换条件 (ratio 阈值)** -- 任何 detection_ratio, window_ratio 的阈值变更都可能影响告警灵敏度
+   - LeavePostAlgorithm: 0.6 (上岗), 0.2 (离岗开始), 0.5 (离岗恢复), 0.2 (离岗确认)
+   - IllegalParkingAlgorithm: 0.3 (放弃确认), 0.6 (确认有车), 0.2 (车离开), 0.15 (开始消失确认), 0.5 (车又来), 0.2 (消失确认)
+   - VehicleCongestionAlgorithm: count_threshold (开始确认), 0.5*count_threshold (开始消散)
+
+2. **时间比较逻辑** -- `elapsed >= xxx_sec` 的方向（大于等于 vs 大于）
+3. **冷却期检查** -- `cooldown_key` 的构造方式和比较逻辑
+4. **resolve 事件触发逻辑** -- LeavePostAlgorithm 的 "进入 ON_DUTY 且 _last_alarm_id 存在" 的检查
+
+### 5.4 重复代码可提取区域
+
+以下方法在多个算法中重复实现，可以提取为基类或 mixin：
+
+| 方法 | 出现在 | 可提取 |
+|------|--------|--------|
+| `_check_detection_in_roi` | 全部4个 | YES |
+| `_check_target_class` | LeavePost, Intrusion | YES |
+| `_check_target_classes` | IllegalParking, VehicleCongestion | YES |
+| `_get_latest_bbox` | 全部4个 | YES (注意 Intrusion 不检查 target_class) |
+| `_get_max_confidence` | IllegalParking, VehicleCongestion | YES |
+| `set_last_alarm_id` | 全部4个 | YES |
+| 滑动窗口逻辑 | LeavePost, IllegalParking, VehicleCongestion | YES |
+
+---
+
+## 6. config_models.py 与 algorithms.py 的一致性
+
+### 6.1 AlgorithmType 枚举缺失
+
+`config_models.py` 中的 `AlgorithmType` 枚举:
+```python
+LEAVE_POST = "leave_post"
+INTRUSION = "intrusion"
+CROWD_DETECTION = "crowd_detection"   # 已在 algorithms.py 中注释掉
+FACE_RECOGNITION = "face_recognition" # algorithms.py 中不存在
+```
+
+缺失:
+- `ILLEGAL_PARKING = "illegal_parking"` -- algorithms.py 已实现但枚举未添加
+- `VEHICLE_CONGESTION = "vehicle_congestion"` -- algorithms.py 已实现但枚举未添加
+
+### 6.2 ROIInfo 默认值
+
+`config_models.py` 的 `ROIInfo.confirm_leave_sec` 默认值为 **10**，而 `AlgorithmManager.default_params["leave_post"]["confirm_leave_sec"]` 为 **30**，`LeavePostAlgorithm.confirm_off_duty_sec` 默认为 **30**。
+
+---
+
+## 7. main.py 集成要点
+
+### 7.1 main.py 对 algorithms.py 的依赖
+
+1. **直接访问算法内部属性** (第905-911行):
+   ```python
+   for attr in ('_leave_start_time', '_parking_start_time', '_congestion_start_time', '_intrusion_start_time'):
+       val = getattr(algo, attr, None)
+   ```
+   这是紧耦合，如果内部属性名变更会导致 first_frame_time 丢失。
+
+2. **alarm_id 回填** (第943-945行):
+   ```python
+   algo.set_last_alarm_id(alarm_info.alarm_id)
+   ```
+
+3. **两层去重机制**:
+   - ROI级别: `_active_alarms[f"{roi_id}_{alert_type}"]`
+   - 摄像头级别: `_camera_alert_cooldown[f"{camera_id}_{alert_type}"]` (30秒冷却)
+
+4. **duration_ms 在 ext_data 中的计算** (第925行):
+   ```python
+   "duration_ms": int(alert.get("duration_minutes", 0) * 60 * 1000) if alert.get("duration_minutes") else None,
+   ```
+   仅 IllegalParkingAlgorithm 的告警包含 `duration_minutes`，其他算法的 ext_data.duration_ms 为 None。
+
+---
+
+## 附录: 状态机可视化
+
+### LeavePostAlgorithm
+```
+         +--> NON_WORK_TIME --+
+         |  (any state)       |  (in_working_hours)
+         |                    v
+  INIT --+--> CONFIRMING_ON_DUTY ---> ON_DUTY ---> CONFIRMING_OFF_DUTY
+    ^          |        ^                              |
+    +----------+        |                              v
+   (ratio==0)           |                    OFF_DUTY_COUNTDOWN
+                        |                       |          |
+                        |     (roi_has_person)  |          v
+                        |       +---------------+       ALARMED
+                        |       |                          |
+                        +-------+  (roi_has_person)        |
+                              ON_DUTY <-- (if _last_alarm_id, send resolve)
+```
+
+### IntrusionAlgorithm
+```
+  IDLE ---> CONFIRMING_INTRUSION ---> ALARMED ---> CONFIRMING_CLEAR ---> IDLE
+    ^            |                                     |        |
+    +------------+                                     +--------+
+   (person gone)                                  (person back >= confirm_intrusion_sec)
+                                                  -> ALARMED
+```
+
+### IllegalParkingAlgorithm
+```
+  IDLE --> CONFIRMING_VEHICLE --> PARKED_COUNTDOWN --> ALARMED --> CONFIRMING_CLEAR --> IDLE
+    ^          |                       |                              |
+    +----------+                       |                              v
+  (ratio<0.3)              (ratio<0.2) |                           ALARMED
+    ^                          |       |                          (ratio>=0.5)
+    +--------------------------+       v
+                                    ALARMED
+```
+
+### VehicleCongestionAlgorithm
+```
+  NORMAL --> CONFIRMING_CONGESTION --> CONGESTED --> CONFIRMING_CLEAR --> NORMAL
+    ^              |                                      |
+    +--------------+                                      v
+  (avg<threshold)                                      CONGESTED
+                                                    (avg>=threshold)
+```

docs/p0p1_review_report.md

@@ -0,0 +1,372 @@
+# P0+P1 修复涉及文件全面审查报告
+
+> 审查日期: 2026-04-02
+> 审查范围: main.py, config_sync.py, screenshot_handler.py, tensorrt_engine.py, result_reporter.py, alarm_upload_worker.py, postprocessor.py 及相关依赖
+
+---
+
+## 1. 功能基线清单
+
+### 1.1 main.py - EdgeInferenceService
+
+| 方法 | 行号 | 行为描述 |
+|------|------|----------|
+| `__init__` | 42-96 | 初始化两个去重字典: `_camera_alert_cooldown` (摄像头级, Dict[str, datetime]) 和 `_active_alarms` (ROI级, Dict[str, str])。冷却期默认30秒。 |
+| `_handle_detections` | 790-954 | 核心告警处理入口。接收检测结果后: (1) 调用算法管理器获取alerts; (2) 对 `alarm_resolve` 类型从 `_active_alarms` 中清除对应记录; (3) ROI级去重: 检查 `_active_alarms[f"{roi_id}_{alert_type}"]` 是否存在; (4) 摄像头级去重: 检查 `_camera_alert_cooldown[f"{camera_id}_{alert_type}"]` 时间间隔; (5) 构建 AlarmInfo 并调用 `report_alarm()`; (6) 写入 `_active_alarms` 并回填 alarm_id 到算法实例。 |
+| `_batch_process_rois` | 676-751 | 从队列取出 ROI 任务, 按 max_batch_size=8 分块, 调用 TensorRT 推理, 后处理后逐个调用 `_handle_detections`。 |
+| `_process_frame` | 613-651 | 获取 ROI 配置(含绑定), 预处理裁剪, 组装 roi_items 推入队列。 |
+| `_scheduler_worker` | 653-674 | 中心调度线程, 轮询所有视频流取最新帧, 丢弃超龄帧(>0.5s), 调用 `_process_frame`。 |
+| `_inference_worker` | 956-976 | 推理线程, 攒批窗口50ms, 调用 `_batch_process_rois`。 |
+
+**`_camera_alert_cooldown` 读写位置:**
+- 写入: 第900行 (`self._camera_alert_cooldown[dedup_key] = now`)
+- 读取: 第890行 (`self._camera_alert_cooldown.get(dedup_key)`)
+- 无其他模块引用此字典
+
+**`_active_alarms` 读写位置:**
+- 写入: 第940行 (`self._active_alarms[active_key] = alarm_info.alarm_id`)
+- 删除: 第865-868行 (resolve 事件清除)
+- 读取: 第880行 (`active_key in self._active_alarms`)
+- 无其他模块引用此字典
+
+### 1.2 config/config_sync.py - ConfigSyncManager
+
+| 方法 | 行为描述 |
+|------|----------|
+| `_init_cloud_redis` (219-241) | 创建云端Redis连接, 参数: socket_connect_timeout=10, socket_timeout=10, retry_on_timeout=True, socket_keepalive=True + TCP keepalive选项, health_check_interval=15。连接失败时将 `_cloud_redis` 设为 None 但不抛异常。 |
+| `_listen_config_stream` (326-390) | Stream 监听主循环。外层while循环: 若 `_cloud_redis` 为None则调用 `_init_cloud_redis()` 重连。内层while循环: XREAD BLOCK 5000ms, 无消息时 PING 保活。`redis.ConnectionError` except 块: 将 `_cloud_redis = None` (第381行)。通用 `Exception` except 块: 不置 None, 仅等待重试。 |
+| `get_roi_configs_with_bindings` (760-806) | 当传入 camera_id 时走优化路径 `get_bindings_by_camera(camera_id)` (单SQL JOIN查询)。当无 camera_id 时存在 N+1 问题: 先 `get_all_roi_configs()` 再逐个 `get_bindings_by_roi(roi_id)`。 |
+
+**`self._cloud_redis = None` 出现位置:**
+- 第381行: `_listen_config_stream` 中 `redis.ConnectionError` except 块
+- 第241行: `_init_cloud_redis` 中初始化失败
+
+### 1.3 core/screenshot_handler.py - ScreenshotHandler
+
+| 方法 | 行为描述 |
+|------|----------|
+| `_listen_loop` (159-220) | XREADGROUP 消费截图请求。finally 块中(第191-196行) ACK 消息, ACK 失败时静默 pass -- 意味着消息可能被重复消费。无消息时 PING 保活。ConnectionError 时重连(带指数退避)。当 `_cloud_redis is None` 时也会主动重连。 |
+| `_handle_request` (224-280) | 流程: 校验必填字段 -> 设备隔离检查 -> 抓帧 -> 上传COS(失败重试1次) -> HTTP回调(失败降级写Redis)。 |
+| `_capture_frame` (284-300) | 优先从 MultiStreamManager 获取已有流帧, 无流时降级临时 RTSP 连接抓帧。 |
+| `_send_result` (375-402) | 优先 HTTP 回调, 失败降级写 Redis key `snap:result:{request_id}` (TTL 60s)。 |
+
+**补偿机制:** 除 COS 上传有1次重试、HTTP 回调有 Redis 降级外, 无其他补偿。xack 失败意味着 Redis 会再次投递该消息(at-least-once 语义)。
+
+### 1.4 core/tensorrt_engine.py - TensorRTEngine
+
+| 方法 | 行为描述 |
+|------|----------|
+| `load_engine` (109-142) | 在 `_lock` 保护下: (1) 若已有 context 则先释放; (2) **创建新的 CUDA context** `cuda.Device(device_id).make_context()`; (3) 创建 CUDA Stream; (4) 反序列化 engine; (5) 创建 execution context; (6) 分配 buffers。 |
+| `infer` (184-253) | 在 `_lock` 保护下: (1) `_cuda_context.push()`; (2) 设置动态 input shape; (3) H2D async memcpy; (4) `execute_async_v2`; (5) D2H async memcpy; (6) `stream.synchronize()`; (7) finally 块中 `_cuda_context.pop()`。 |
+| `release` (317-325) | 在 `_lock` 保护下, 幂等释放: `_cuda_context.pop()` + `_cuda_context.detach()`。 |
+| `_release_resources` (294-315) | 内部释放: pop/detach CUDA context, synchronize stream, 清空 bindings。 |
+
+**CUDA context 模式:** 每个 TensorRTEngine 实例创建独立的 CUDA context。`pycuda.autoinit` 在 import 时创建一个默认 context, 而 `load_engine` 再创建一个新的。`infer` 使用 push/pop 模式切换到自己的 context。
+
+**`_lock` 使用范围:** 覆盖 `load_engine`, `infer`, `release` 三个公开方法, 保证单引擎实例的线程安全。
+
+**EngineManager:** 持有 `Dict[str, TensorRTEngine]`, 有自己的 `_lock`。当前代码只创建一个 "default" 引擎, 不存在多引擎共享 CUDA context 的问题。
+
+### 1.5 core/result_reporter.py - ResultReporter
+
+| 方法 | 行为描述 |
+|------|----------|
+| `report_alarm` (119-165) | 接收 AlarmInfo + numpy screenshot。将截图 cv2.imencode JPEG (quality=85) 后 base64 编码写入 `alarm_info.snapshot_b64`。然后 JSON 序列化后 LPUSH 到 `local:alarm:pending`。 |
+| `report_alarm_resolve` (167-180) | 将 resolve 数据附加 `_type: "resolve"` 标记后 LPUSH 到同一队列 `local:alarm:pending`。 |
+
+**AlarmInfo 字段:** alarm_id, alarm_type, device_id, scene_id, event_time(ISO8601), alarm_level(0-3), snapshot_b64(Optional[str]), algorithm_code, confidence_score, ext_data(Dict)。
+
+### 1.6 core/alarm_upload_worker.py - AlarmUploadWorker
+
+| 方法 | 行为描述 |
+|------|----------|
+| `_worker_loop` (141-166) | 主循环: 先处理重试队列, 再 BRPOP `local:alarm:pending` (timeout=2s)。 |
+| `_process_alarm` (169-230) | 流程: JSON解析 -> 检查 `_type=="resolve"` 分流 -> base64.b64decode 截图 -> 上传 COS -> HTTP POST 告警元数据。COS 上传失败或 HTTP 失败都进入 `_handle_retry`。 |
+| `_handle_retry` (386-425) | 递增 `_retry_count`, 超过 `retry_max` 则 LPUSH 到 `local:alarm:dead`。未超限则计算指数退避延迟, 附加 `_retry_at` 时间戳后 LPUSH 到 `local:alarm:retry`。 |
+| `_process_retry_queue` (427-465) | RPOP 逐条检查 `local:alarm:retry`, 到期的放回 pending, 未到期的放回 retry 头部。 |
+| `_upload_snapshot_to_cos` (263-316) | base64 解码后直接 `put_object` 上传, object_key 格式: `alarms/{device_id}/{yyyy-MM-dd}/{alarm_id}.jpg`。 |
+
+### 1.7 core/postprocessor.py - PostProcessor
+
+| 方法 | 行为描述 |
+|------|----------|
+| `_process_gpu` (73-96) | `torch.from_numpy(boxes).cuda()` 转为 GPU 张量, 调用 `torchvision.ops.nms`, 结果 `.cpu().numpy()` 回到 CPU。**未使用 `torch.no_grad()`**。 |
+| `batch_process_detections` (705-812) | 解析 TensorRT 输出, 按 batch 拆分, 逐个做 YOLO 输出解析 + NMS。每次调用创建新的 NMSProcessor 实例。 |
+
+**GPU 张量生命周期:** `_process_gpu` 中 `boxes_t` 和 `scores_t` 是临时变量, 函数返回后即可被 GC 回收。`keep` 张量在 `.cpu().numpy()` 后也成为临时变量。无显式释放, 依赖 Python GC + PyTorch 缓存分配器。
+
+---
+
+## 2. 接口契约
+
+### 2.1 方法签名与返回值
+
+```python
+# main.py
+def _handle_detections(
+    self, camera_id: str, roi, bind, frame: VideoFrame,
+    boxes: Any, scores: Any, class_ids: Any, scale_info: tuple
+) -> None
+
+# config_sync.py
+def get_roi_configs_with_bindings(
+    self, camera_id: Optional[str] = None, force_refresh: bool = False
+) -> List[ROIInfoNew]
+
+def _init_cloud_redis(self) -> None  # 失败时 self._cloud_redis = None
+
+# screenshot_handler.py
+def _handle_request(self, fields: dict) -> None
+# fields 预期字段: request_id, camera_code, cos_path, callback_url, device_id(可选), rtsp_url(可选)
+
+# tensorrt_engine.py
+def infer(self, input_batch: np.ndarray) -> Tuple[List[np.ndarray], float]
+# input_batch: shape=[batch, 3, 480, 480], dtype=float16
+# returns: (outputs_list, inference_time_ms)
+
+def load_engine(self, engine_path: Optional[str] = None) -> bool
+
+# result_reporter.py
+def report_alarm(self, alarm_info: AlarmInfo, screenshot: Optional[np.ndarray] = None) -> bool
+def report_alarm_resolve(self, resolve_data: dict) -> bool
+
+# alarm_upload_worker.py
+def _process_alarm(self, alarm_json: str) -> None
+def _handle_retry(self, alarm_json: str, error: str) -> None
+def _upload_snapshot_to_cos(self, image_bytes: bytes, alarm_id: str, device_id: str) -> Optional[str]
+
+# postprocessor.py
+def batch_process_detections(
+    self, batch_outputs: List[np.ndarray], batch_size: int,
+    conf_threshold: Optional[float] = None, nms_threshold: Optional[float] = None,
+    per_item_conf_thresholds: Optional[List[float]] = None
+) -> List[Tuple[np.ndarray, np.ndarray, np.ndarray]]
+```
+
+### 2.2 Redis Key 名与格式
+
+| Key | Redis实例 | 类型 | 格式 | 模块 |
+|-----|-----------|------|------|------|
+| `local:alarm:pending` | 本地 | List | JSON(AlarmInfo.to_dict() 或 resolve_data) | result_reporter / alarm_upload_worker |
+| `local:alarm:retry` | 本地 | List | JSON(带 _retry_count, _retry_at 字段) | alarm_upload_worker |
+| `local:alarm:dead` | 本地 | List | JSON(带 _dead_reason, _dead_at 字段) | alarm_upload_worker |
+| `device:{device_id}:config` | 云端 | String | JSON(完整配置) | config_sync |
+| `device:{device_id}:version` | 云端 | String | int 版本号 | config_sync |
+| `device_config_stream` | 云端 | Stream | {device_id, version, action} | config_sync |
+| `local:device:config:current` | 本地 | String | JSON(完整配置) | config_sync |
+| `local:device:config:backup` | 本地 | String | JSON(上一版本配置) | config_sync |
+| `local:device:config:version` | 本地 | String | int 版本号 | config_sync |
+| `local:device:config:stream_last_id` | 本地 | String | Stream message ID | config_sync |
+| `edge_snap_request` | 云端 | Stream | {request_id, camera_code, cos_path, callback_url, ...} | screenshot_handler |
+| `snap:result:{request_id}` | 云端 | String(TTL=60s) | JSON(降级结果) | screenshot_handler |
+
+### 2.3 AlarmInfo 数据结构 (完整字段)
+
+```python
+@dataclass
+class AlarmInfo:
+    alarm_id: str               # 格式: edge_{device_id}_{YYYYMMDDHHmmss}_{6hex}
+    alarm_type: str             # 算法返回的 alert_type
+    device_id: str              # 实际传入的是 camera_id (非 edge device_id)
+    scene_id: str               # ROI ID
+    event_time: str             # ISO8601 (frame.timestamp)
+    alarm_level: int            # 0=紧急 1=重要 2=普通 3=轻微
+    snapshot_b64: Optional[str] # JPEG base64, 由 report_alarm 填充
+    algorithm_code: Optional[str]
+    confidence_score: Optional[float]
+    ext_data: Optional[Dict]    # 包含: duration_ms, roi_id, bbox, target_class,
+                                #   bind_id, message, edge_node_id, first_frame_time,
+                                #   vehicle_count, area_id
+```
+
+---
+
+## 3. 依赖关系图
+
+```
+                    +-----------------+
+                    | main.py         |
+                    | EdgeInference   |
+                    |   Service       |
+                    +--------+--------+
+                             |
+          +------------------+------------------+
+          |         |        |        |         |
+          v         v        v        v         v
+    config_sync  stream   engine  postprocess  algorithm
+    (ConfigSync  manager  manager  (PostProc)   manager
+     Manager)                                     |
+          |                   |                   |
+          v                   v                   v
+      database.py      tensorrt_engine.py    algorithms/
+      (SQLiteManager)  (TensorRTEngine)
+
+                    +------------------+
+                    | result_reporter  |
+                    | (ResultReporter) |
+                    +--------+---------+
+                             |
+                             | LPUSH local:alarm:pending
+                             v
+                    +------------------+
+                    | alarm_upload     |
+                    | _worker          |
+                    +--------+---------+
+                             |
+                    +--------+---------+
+                    | COS Upload       |
+                    | HTTP POST cloud  |
+                    +------------------+
+
+    screenshot_handler (独立Redis连接) --> 云端 Redis Stream
+```
+
+**数据流转:**
+1. `_scheduler_worker` 轮询视频流 -> `_process_frame` 获取ROI+预处理 -> 推入 `_batch_roi_queue`
+2. `_inference_worker` 消费队列 -> `_batch_process_rois` 做 TensorRT 推理 + 后处理 -> `_handle_detections`
+3. `_handle_detections` -> 算法管理器判定 -> 去重过滤 -> `ResultReporter.report_alarm()` LPUSH Redis
+4. `AlarmUploadWorker` BRPOP Redis -> base64 decode -> COS upload -> HTTP POST 云端
+
+---
+
+## 4. 安全边界
+
+### 4.1 绝对不能动的代码 (Critical Path)
+
+| 文件 | 代码区域 | 原因 |
+|------|----------|------|
+| `tensorrt_engine.py` | `infer()` 的 push/pop/synchronize 顺序 | CUDA context 操作顺序错误会导致段错误或GPU挂死 |
+| `tensorrt_engine.py` | `_allocate_buffers()` 的 buffer 分配逻辑 | 改变 buffer 大小会导致推理崩溃 |
+| `result_reporter.py` | `AlarmInfo.to_dict()` 的字段名 | alarm_upload_worker 和云端 API 依赖这些字段名 |
+| `alarm_upload_worker.py` | Redis key 常量引用 | 必须与 result_reporter 一致 |
+| `postprocessor.py` | `_parse_yolo_output` 的输出格式解析 (84行 = 4+80) | 与模型输出格式强耦合 |
+
+### 4.2 可以安全修改的代码
+
+| 文件 | 代码区域 | 注意事项 |
+|------|----------|----------|
+| `main.py` | `_handle_detections` 的去重逻辑 | 只影响告警频率, 不影响推理管线。但要确保线程安全(当前单线程调用, 无锁)。 |
+| `main.py` | `_camera_cooldown_seconds` 默认值 | 可调整, 纯业务参数 |
+| `config_sync.py` | `_listen_config_stream` 的重连逻辑 | 注意 `_cloud_redis = None` 的时机 |
+| `config_sync.py` | `get_roi_configs_with_bindings` 的 N+1 查询 | 可优化为 JOIN 查询, 但要保持返回值格式不变 |
+| `screenshot_handler.py` | `_listen_loop` 的 xack 逻辑 | 可增加重试, 但要注意不能阻塞主循环 |
+| `postprocessor.py` | `_process_gpu` 添加 `torch.no_grad()` | 纯优化, 不影响功能 |
+
+### 4.3 修改时需要同步更新的代码对
+
+| 修改点 | 需要同步的位置 |
+|--------|----------------|
+| `AlarmInfo` 字段变更 | `to_dict()`, `alarm_upload_worker._process_alarm`, 云端API |
+| Redis key 名变更 | `result_reporter.py` 常量 + `alarm_upload_worker.py` import |
+| `_active_alarms` key 格式变更 | 第865行 resolve 清除逻辑 + 第880行查重逻辑 + 第940行写入 |
+| `_camera_alert_cooldown` key 格式变更 | 第888行构建 + 第890行读取 + 第900行写入 |
+
+---
+
+## 5. 已有测试
+
+| 测试文件 | 覆盖模块 | 状态 |
+|----------|----------|------|
+| `tests/test_config_sync.py` | config_sync.py | 存在 |
+| `tests/test_postprocessor.py` | postprocessor.py | 存在 |
+| `tests/test_result_reporter.py` | result_reporter.py | 存在 |
+| `tests/test_tensorrt.py` | tensorrt_engine.py | 存在 |
+| `tests/test_preprocessor.py` | preprocessor.py | 存在 |
+| `tests/test_video_stream.py` | video_stream.py | 存在 |
+| `tests/test_utils.py` | utils | 存在 |
+| `test_leave_post_full_workflow.py` | 离岗检测集成 | 存在(项目根目录) |
+| `test_vehicle_algorithms.py` | 车辆算法 | 存在(项目根目录) |
+
+**缺失测试:**
+- `main.py` (`_handle_detections`, 去重逻辑) -- **无单元测试**
+- `alarm_upload_worker.py` -- **无单元测试**
+- `screenshot_handler.py` -- **无单元测试**
+
+---
+
+## 6. 潜在风险
+
+### 6.1 Critical (必须关注)
+
+**[C1] `_handle_detections` 无线程安全保护**
+- `_camera_alert_cooldown` 和 `_active_alarms` 两个字典在 `_handle_detections` 中读写, 该方法被 `_inference_worker` 线程调用。当前架构下只有一个推理线程, 所以实际上是单线程安全的。但如果未来增加多推理线程, 将产生竞态条件。
+- 风险等级: 当前低, 架构变更时高
+
+**[C2] `_active_alarms` resolve 清除使用遍历+break**
+- 第865-868行: `for k, v in list(self._active_alarms.items())` 遍历查找匹配的 alarm_id 后 break。如果同一个 alarm_id 被错误地写入多个 key, 只会清除第一个。
+- 风险等级: 低(alarm_id 是 UUID 级唯一)
+
+**[C3] CUDA context 与 pycuda.autoinit 共存**
+- `import pycuda.autoinit` 在模块加载时创建一个默认 CUDA context。`load_engine` 又创建新的 context。两个 context 共存, 依赖 push/pop 正确切换。如果任何代码在 push/pop 之外使用了 CUDA 操作(如 PostProcessor 的 GPU NMS), 将使用 autoinit 的 context, 与 TensorRT 的 context 不同。
+- 风险等级: 中(当前 NMS 使用 PyTorch CUDA, 与 PyCUDA context 独立)
+
+### 6.2 Important (应当修复)
+
+**[I1] `config_sync._listen_config_stream` 通用 Exception 不置 None**
+- 第385-390行: 通用 Exception 分支不将 `_cloud_redis` 设为 None, 但 ConnectionError 分支会。如果出现非 ConnectionError 的 Redis 异常(如 ResponseError), 会一直使用同一个可能已损坏的连接重试, 而不是重建连接。
+- 建议: 在通用 Exception 中也加入 `self._cloud_redis = None` 触发重连, 或至少尝试 ping 验证连接健康。
+
+**[I2] `get_roi_configs_with_bindings` 的 N+1 查询**
+- 当 `camera_id` 为 None 时, 先查所有 ROI, 再逐个查 bindings。ROI 数量多时性能差。
+- 注意: 当 `camera_id` 非空时已经使用了 `get_bindings_by_camera` 优化查询, 这是正确的。
+- 建议: 添加一个 `get_all_bindings()` 方法或使用 JOIN 查询, 一次取出所有 bindings。
+
+**[I3] `screenshot_handler._listen_loop` 中 xack 失败静默 pass**
+- 第195行: xack 失败时 pass, 不记录日志。导致无法发现 ACK 累积问题, PEL (Pending Entries List) 会持续增长。
+- 建议: 至少记录 warning 日志。
+
+**[I4] `_process_gpu` 未使用 `torch.no_grad()`**
+- NMS 是纯推理操作, 不需要梯度计算。未包裹 `torch.no_grad()` 会导致不必要的计算图记录和额外内存占用。
+- 建议: 用 `with torch.no_grad():` 包裹 GPU NMS 调用。
+
+**[I5] `report_alarm` 中截图 base64 编码在调用线程中执行**
+- 第133-143行: JPEG 编码 + base64 编码在推理线程中同步执行。一张 1080p 截图约 100-300KB JPEG, base64 后约 130-400KB。对高频告警场景可能阻塞推理线程。
+- 当前影响: 有 `_camera_cooldown_seconds=30` 限频, 实际影响有限。
+
+### 6.3 Suggestions (改进建议)
+
+**[S1] `_handle_detections` 中 import 在函数内**
+- 第914行: `from core.result_reporter import AlarmInfo, generate_alarm_id` 在热路径函数内 import。虽然 Python 会缓存模块, 但每次调用仍有字典查找开销。
+- 建议: 移到文件顶部 import。
+
+**[S2] `batch_process_detections` 每次创建新 NMSProcessor**
+- 第753行: 每次调用都 `NMSProcessor(nms_threshold, use_gpu=True)`, 而 PostProcessor.__init__ 已经创建了 `self._nms`。
+- 建议: 复用 `self._nms` 或在需要不同阈值时参数化调用。
+
+**[S3] `alarm_upload_worker._process_retry_queue` 的 RPOP+LPUSH 非原子**
+- 重试队列的检查逻辑: RPOP -> 检查时间 -> LPUSH 回去。在高并发下可能有短暂数据丢失窗口(RPOP 后进程崩溃)。
+- 当前影响: 单线程消费, 实际风险极低。
+
+**[S4] `_camera_alert_cooldown` 字典无清理机制**
+- 随着时间推移, 已下线的摄像头 + 告警类型的 key 会一直驻留内存。
+- 当前影响: 每个 key 约 100 字节, 不太可能成为问题。长期运行建议定期清理。
+
+**[S5] AlarmInfo.device_id 语义歧义**
+- `_handle_detections` 第918行传入 `device_id=camera_id`, 但 AlarmInfo 字段名为 `device_id`。云端可能误认为这是边缘设备ID而非摄像头ID。实际的边缘设备ID在 `ext_data.edge_node_id` 中。
+- 建议: 确认云端 API 对此字段的预期, 考虑重命名或补充文档。
+
+---
+
+## 7. 修复点风险矩阵
+
+| 修复目标 | 涉及文件 | 变更范围 | 回归风险 | 测试覆盖 |
+|----------|----------|----------|----------|----------|
+| 告警去重逻辑优化 | main.py | _handle_detections 方法内部 | 低(不影响推理管线) | **无**(需新增) |
+| Redis 重连机制增强 | config_sync.py | _listen_config_stream except块 | 中(影响配置同步) | 有(test_config_sync) |
+| N+1 查询优化 | config_sync.py + database.py | get_roi_configs_with_bindings | 中(影响配置读取) | 有(test_config_sync) |
+| xack 失败处理 | screenshot_handler.py | _listen_loop finally块 | 低(仅日志增强) | **无**(需新增) |
+| GPU NMS 优化 | postprocessor.py | _process_gpu | 低(纯优化) | 有(test_postprocessor) |
+| CUDA context 安全 | tensorrt_engine.py | load_engine/infer | 高(GPU操作) | 有(test_tensorrt) |
+| 截图编码优化 | result_reporter.py | report_alarm | 中(涉及序列化格式) | 有(test_result_reporter) |
+
+---
+
+## 8. 审查总结
+
+**代码整体质量:** 项目架构清晰, 模块职责分明, 关键路径有合理的错误处理和日志输出。Redis 双层架构(云端分发+本地自治)设计合理, 支持离线运行。
+
+**主要关注点:**
+1. 告警去重字典 `_camera_alert_cooldown` 和 `_active_alarms` 是核心业务逻辑, 修改时务必保持 key 格式和读写位置的一致性。当前无测试覆盖, 是最大风险点。
+2. `config_sync._listen_config_stream` 的通用 Exception 处理策略需要与 ConnectionError 保持一致。
+3. TensorRT 的 CUDA context push/pop 模式是正确的, 但与 `pycuda.autoinit` 共存需要注意不要在 push/pop 之外的代码中使用 PyCUDA 操作。
+4. 截图通过 base64 经 Redis 传递的设计合理(避免文件IO), 但要注意大图场景下的内存和队列压力。

main.py

@@ -92,6 +92,10 @@ class EdgeInferenceService:
         # ROI级别告警去重：同ROI+同类型未resolve的告警不重复发送
         # key: f"{roi_id}_{alert_type}", value: alarm_id
         self._active_alarms: Dict[str, str] = {}
+        self._active_alarms_time: Dict[str, datetime] = {}  # 活跃告警创建时间
+        self._cleanup_counter = 0
+        self._cleanup_interval = 100  # 每 100 次 _handle_detections 清理一次
+        self._active_alarm_max_age_sec = 3600  # 活跃告警最大存活时间（1小时）
 
         self._logger.info("Edge_Inference_Service 初始化开始")
     
@@ -701,9 +705,11 @@ class EdgeInferenceService:
                 # 一次性推理整个 batch
                 outputs, inference_time_ms = engine.infer(batch_data)
                 self._performance_stats["inference_batches"] += 1
-                self._logger.log_inference_latency(
+                self._logger.performance(
+                    "inference_latency_ms",
                     inference_time_ms,
                     batch_size=len(chunk),
+                    throughput_fps=1000.0 / inference_time_ms if inference_time_ms > 0 else 0
                 )
 
                 # 诊断：输出原始推理结果形状（非告警诊断日志，使用 DEBUG 级别）
@@ -798,6 +804,12 @@ class EdgeInferenceService:
     ):
         """处理检测结果 - 算法接管判断权"""
         try:
+            # 惰性清理过期去重记录
+            self._cleanup_counter += 1
+            if self._cleanup_counter >= self._cleanup_interval:
+                self._cleanup_counter = 0
+                self._cleanup_dedup_dicts(frame.timestamp)
+
             if self._algorithm_manager is None:
                 self._logger.warning("算法管理器不可用，跳过算法处理")
                 return
@@ -863,6 +875,7 @@ class EdgeInferenceService:
                     for k, v in list(self._active_alarms.items()):
                         if v == resolve_alarm_id:
                             del self._active_alarms[k]
+                            self._active_alarms_time.pop(k, None)
                             self._logger.debug(f"[去重] 活跃告警已清除: {k} -> {resolve_alarm_id}")
                             break
 
@@ -936,6 +949,7 @@ class EdgeInferenceService:
 
                 # 记录活跃告警（用于 ROI 级去重）
                 self._active_alarms[active_key] = alarm_info.alarm_id
+                self._active_alarms_time[active_key] = frame.timestamp
 
                 # 回填 alarm_id 到算法实例（用于后续 resolve 追踪，泛化支持所有算法类型）
                 algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get(alert_type)
@@ -951,6 +965,31 @@ class EdgeInferenceService:
         except Exception as e:
             self._logger.error(f"处理检测结果失败: {e}")
 
+    def _cleanup_dedup_dicts(self, now: datetime):
+        """惰性清理过期的去重记录"""
+        # 清理 _camera_alert_cooldown 中已过冷却期的记录
+        expired_cooldown = [
+            k for k, v in self._camera_alert_cooldown.items()
+            if (now - v).total_seconds() > self._camera_cooldown_seconds * 2
+        ]
+        for k in expired_cooldown:
+            del self._camera_alert_cooldown[k]
+
+        # 清理 _active_alarms 中可能因 resolve 丢失而残留的记录
+        expired_active = [
+            k for k, t in self._active_alarms_time.items()
+            if (now - t).total_seconds() > self._active_alarm_max_age_sec
+        ]
+        for k in expired_active:
+            self._active_alarms.pop(k, None)
+            self._active_alarms_time.pop(k, None)
+            self._logger.warning(f"[去重] 活跃告警超时清除: {k}")
+
+        if expired_cooldown or expired_active:
+            self._logger.debug(
+                f"[去重] 清理完成: cooldown={len(expired_cooldown)}, active={len(expired_active)}"
+            )
+
     def _inference_worker(self):
         """推理线程：攒批窗口内收集 ROI 请求，批量推理"""
         while not self._stop_event.is_set():

utils/logger.py

@@ -223,15 +223,6 @@ class StructuredLogger:
         """记录性能指标"""
         self._performance_logger.record(metric_name, value, tags)
     
-        perf_data = {
-            "metric": metric_name,
-            "value": value,
-            "duration_ms": duration_ms,
-            "tags": tags
-        }
-        
-        self.info(f"性能指标: {metric_name} = {value}", **perf_data)
-    
     def log_inference_latency(self, latency_ms: float, batch_size: int = 1):
         """记录推理延迟"""
         self.performance(