From 177c2a83b76d86638e288fb4863b0293b227ba76 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Sat, 22 Aug 2026 13:25:20 +0800 Subject: [PATCH] =?UTF-8?q?[=E4=BF=AE=E5=A4=8D]=20=E8=BF=9B=E8=A1=8C?= =?UTF-8?q?=E4=B8=AD=E4=BA=8B=E4=BB=B6=20duration=3D0=20=E5=88=86=E5=89=B2?= =?UTF-8?q?=E4=B8=A2=E5=A4=B1=20-=20=E6=8E=A8=E9=80=81=E8=BF=87=20duration?= =?UTF-8?q?<=3D0=20=E7=9A=84=E4=BA=8B=E4=BB=B6=E8=AE=B0=E5=85=A5=20=5Fzero?= =?UTF-8?q?=5Fdur=5Fids,=20=E4=B8=8B=E8=BD=AE=E7=AA=97=E5=8F=A3=E5=9B=9E?= =?UTF-8?q?=E6=9F=A5=E5=B7=B2=E7=BB=93=E6=9D=9F(duration>0)=E8=A1=A5?= =?UTF-8?q?=E6=8E=A8=E8=A6=86=E7=9B=96=20Oracle;=20=E4=BF=9D=E8=AF=81=20ss?= =?UTF-8?q?=5Fmotion=5Fevents=20=E6=9C=80=E7=BB=88=E6=8C=81=E6=9C=89?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=20duration,=20=E5=88=86=E5=89=B2=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E8=B7=B3=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../motion_notifier/motion_notifier.py | 26 ++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) 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,