diff --git a/fam-core/src/fam_core/motion_notifier/motion_notifier.py b/fam-core/src/fam_core/motion_notifier/motion_notifier.py index 468d74c..0b24cbd 100644 --- a/fam-core/src/fam_core/motion_notifier/motion_notifier.py +++ b/fam-core/src/fam_core/motion_notifier/motion_notifier.py @@ -87,6 +87,9 @@ class MotionNotifier: self._running = False self._thread = None self._last_event_id = None + # 推送过但 duration<=0(动作进行中,SS 尚未回填真实时长)的事件 id。 + # 下轮窗口回查时若已结束(duration>0)补推覆盖 Oracle,避免分割永远跳过。 + self._zero_dur_ids = set() self._last_poll_at = None self._last_error = None self._pushed_total = 0 @@ -396,7 +399,23 @@ class MotionNotifier: events = self._fetch_events(start_ts, now) if events is None: return # 查询失败,下一轮重试 - # 只保留 id 大于游标的新事件(SS 事件 id 单调递增) + by_id = {int(e.get('id')): e for e in events if e.get('id')} + + # 1) 补推:先前推送时 duration<=0(动作进行中)的事件,现在若已结束 + # (SS 回填真实 duration>0),重推覆盖 Oracle(ON CONFLICT UPDATE)。 + # 重启兜底:游标 DB 续用 → 窗口回看会把历史事件整体重推,届时同样覆盖。 + if self._zero_dur_ids: + recheck = [] + for eid in list(self._zero_dur_ids): + e = by_id.get(eid) + if e and int(e.get('duration') or 0) > 0: + recheck.append(e) + self._zero_dur_ids.discard(eid) + if recheck: + self.push_events_to_oracle(recheck) + logger.info(f"补推 {len(recheck)} 条已结束事件的真实 duration") + + # 2) 增量推送新事件(id 大于游标) new = [e for e in events if e.get('id') and int(e.get('id')) > (self._last_event_id or 0)] if not new: @@ -410,6 +429,10 @@ class MotionNotifier: # 只有推送成功的批次才推进游标;失败批次保持原地, # 下一轮窗口回看会重新捞到并重试(不丢事件) cursor = max(int(e.get('id', 0)) for e in batch) + # 记录推送时仍在进行中的事件(duration<=0),待下轮补推真实时长 + for e in batch: + if int(e.get('duration') or 0) <= 0: + self._zero_dur_ids.add(int(e.get('id'))) self._last_event_id = cursor db_layer.set_motion_cursor(str(self._last_event_id)) @@ -479,6 +502,7 @@ class MotionNotifier: "camera_map": {**self.camera_name_to_id, **self._ss_name_to_id}, "camera_loaded": self._camera_loaded, "last_event_id": self._last_event_id, + "zero_dur_pending": len(self._zero_dur_ids), "last_poll_at": self._last_poll_at.isoformat() if self._last_poll_at else None, "last_error": self._last_error, "pushed_total": self._pushed_total,