ericwyuan/sentinel-home-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[重构] 运动监测恢复轮询主路径(简单稳定) - poll_enabled 默认开启;修复3隐患: 推送失败批次不前进游标(下轮重试不丢事件)、重启优先续用DB游标(停机期间事件补推)、fetch limit 提到1000;Webhook降级为可选补充

Browse Source
This commit is contained in:
ericwyuan
2026-08-22 12:00:03 +08:00
parent b2b537dde5
commit 7adc32486a
3 changed files with 129 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
fam-core/src/fam_core/motion_bp.py
View File

@@ -1,9 +1,11 @@
"""
运动监测 Webhook 接收（fam-core）
运动监测 Webhook 接收（fam-core，可选补充，非主路径）
群晖 Surveillance Station 配置「行動規則 -> 事件=偵測到動作 -> 動作=Webhook」，
把运动事件实时 POST 到本端点，本端点即时映射字段并推送到甲骨文 FAM-Edge。
数据源完全来自 SS Webhook（不轮询 SS），数据方向 NAS -> Oracle 单向。
主数据源是 MotionNotifier 轮询 SS EventCenter.Event.List（真实
event_id/start_time/duration）。本端点是可选的低延迟补充：若在群晖
Surveillance Station 配置「行動規則 -> 事件=偵測到動作 -> 動作=Webhook」，
SS 会把运动事件实时 POST 到本端点，本端点映射字段后走同一个
push_events_to_oracle 推送到甲骨文 FAM-Edge（数据方向 NAS -> Oracle 单向）。
SS Webhook 仅提供模板变量（无 event_id / camera_id 数字字段）:
%EVENT_TIME% -> event_time (本地时间字符串)

121
fam-core/src/fam_core/motion_notifier/motion_notifier.py
View File

@@ -1,21 +1,27 @@
"""
MotionNotifier - NAS 端运动监测通知服务（Webhook 驱动，2026-08-22 重构）
MotionNotifier - NAS 端运动监测通知服务（轮询主路径，2026-08-22 定型）
职责:
作为 SS Webhook 的接收侧客户端，把群晖 Surveillance Station 推送来的运动侦测
事件（SYNO.SurveillanceStation.EventCenter.Event 的字段，event_type=10 即运动）
映射后主动 POST 推送到甲骨文 FAM-Edge 的 /api/ss/motion 接口。
在 NAS 本机**轮询**群晖 Surveillance Station 的运动侦测事件
（SYNO.SurveillanceStation.EventCenter.Event method=List，参数名下划线风格
camera_ids/start_time/end_time，event_type=10 即运动），增量推送到甲骨文
FAM-Edge 的 /api/ss/motion 接口。轮询能拿到真实 event_id/start_time/duration。
数据方向（关键约束）: NAS -> Oracle，单向。甲骨文不再反向访问 NAS。
- 原 fam-edge 的 dsm_motion_client（甲骨文主动查 SS API）已停用；
- SS Webhook 不提供 event_id / camera_id 数字字段，本服务负责:
* 合成稳定 event_id（device_name+event_time+thumbnail_url 哈希，幂等去重）
* 把 %DEVICE_NAME% 经摄像头名->id 映射解析为 camera_id
* 把 %EVENT_TIME% 解析为 Unix epoch
- 推送后由甲骨文本地落库 ss_motion_events，供 video_processor 本地预过滤使用。
轮询已禁用（poll_enabled=false，用户要求不轮询），仅保留轮询代码路径作为可选能力。
摄像头名->id 映射在启动时一次性从 SS 拉取补全（非轮询），并以配置 camera_name_to_id 兜底。
可靠性设计:
- 游标（MariaDB sync_cursor.motion_last_event_id）持久化；重启优先续用 DB 游标，
停机期间的事件由窗口回看补推；仅首次部署时才初始化为 SS 当前最大 id。
- 推送失败的批次不推进游标，下一轮窗口回看重试（不丢事件）。
- 心跳线程（enabled 即跑，与轮询无关）：定期空 POST /api/ss/motion，证明
NAS->Oracle 推送链路存活，供甲骨文侧判断"无运动"结论是否可信。
Webhook（可选补充，非主路径）:
fam-core 仍暴露 POST /api/ss/webhook，SS 行动规则若配置可实时低延迟推送，
经 build_event_from_webhook 映射后与轮询共用 push_events_to_oracle；
其合成 event_id 与轮询真实 event_id 不冲突（不同命名空间，幂等互不干扰）。
"""
import os
import re
@@ -59,8 +65,15 @@ class MotionNotifier:
self.poll_window_hours = int(cfg.get('poll_window_hours', 2))
self.batch_size = int(cfg.get('batch_size', 100))
self.timeout = int(cfg.get('timeout_sec', 10))
# 是否启用轮询：用户要求不轮询，改由 SS Webhook 推送（默认关闭）
self.poll_enabled = bool(cfg.get('poll_enabled', False))
# 是否启用轮询（默认开启）：轮询 EventCenter.Event.List 为主数据源，拿到
# 真实 event_id/start_time/duration；Webhook 仅作为可选的低延迟补充
# （SS 行动规则未配置时不会触发，不影响主路径）
self.poll_enabled = bool(cfg.get('poll_enabled', True))
# 心跳：跟"轮询 SS"是两回事——不查 SS，只是定期空 POST 一下甲骨文，证明
# NAS->Oracle 这条推送链路本身还活着。enabled=true 时始终跑（不受
# poll_enabled 影响），供甲骨文侧 has_motion_in_range_local() 判断
# "这段时间没收到运动事件"是真的没运动，还是推送链路已经挂了。
self.heartbeat_interval_sec = int(cfg.get('heartbeat_interval_sec', 300))
# 轮询路径（poll_enabled=true 时）关注的摄像头；Webhook 路径用摄像头名映射
self.camera_ids = cfg.get('camera_ids', [2])
# 摄像头名 -> camera_id 静态映射（优先），启动时再从 SS 拉一份补全
@@ -77,6 +90,10 @@ class MotionNotifier:
self._last_poll_at = None
self._last_error = None
self._pushed_total = 0
self._hb_running = False
self._hb_thread = None
self._last_heartbeat_at = None
self._last_heartbeat_error = None
@staticmethod
def _resolve(v):
@@ -118,7 +135,7 @@ class MotionNotifier:
"version": 1, "method": "List",
"camera_ids": ",".join(str(c) for c in self.camera_ids),
"start_time": start_ts, "end_time": end_ts,
"limit": 100, "_sid": self._sid},
"limit": 1000, "_sid": self._sid},
timeout=self.timeout)
data = resp.json()
except Exception as e:
@@ -312,20 +329,66 @@ class MotionNotifier:
logger.info(f"运动事件推送成功: {len(norm)} 条 -> Oracle 存储 {stored}")
return stored
def send_heartbeat(self) -> bool:
"""空 events 调一次 /api/ss/motion，只为证明 NAS->Oracle 推送链路还活着。
跟 push_events_to_oracle 分开一个方法，是因为那个方法 events 为空时直接
return 0（不发请求）——心跳恰恰就是要在没有真实事件时也发一次请求。
"""
try:
resp = requests.post(
f"{self.oracle_base_url}/api/ss/motion",
json={"token": self.oracle_token, "events": []},
timeout=(10, 30))
except requests.RequestException as e:
logger.warning(f"运动心跳推送失败: {e}")
self._last_heartbeat_error = str(e)
return False
if resp.status_code != 200:
logger.warning(f"运动心跳推送返回 {resp.status_code}: {resp.text[:200]}")
self._last_heartbeat_error = f"HTTP {resp.status_code}"
return False
self._last_heartbeat_at = datetime.now()
self._last_heartbeat_error = None
return True
def _heartbeat_run(self):
logger.info(f"运动心跳线程启动，间隔 {self.heartbeat_interval_sec}s")
while self._hb_running:
try:
self.send_heartbeat()
except Exception as e:
self._last_heartbeat_error = str(e)
logger.error(f"运动心跳异常: {e}", exc_info=True)
for _ in range(self.heartbeat_interval_sec):
if not self._hb_running:
break
time.sleep(1)
# ------------------------------------------------------------------
# 增量轮询主循环
# ------------------------------------------------------------------
def _init_cursor(self):
"""启动时初始化 last_event_id: 取当前 SS 最大事件 id，只推增量（不回灌历史）。"""
"""启动时初始化 last_event_id。
优先使用 DB 已保存的游标：服务停机/重启期间产生的新事件，会在重启后
被下一轮窗口回看捞到并补推（不丢事件）。
仅当 DB 无有效游标（首次部署）时，取 SS 当前最大事件 id 作为起点，
避免把历史事件全部回灌一遍。
"""
saved = db_layer.get_motion_cursor()
if saved and int(saved) > 0:
self._last_event_id = int(saved)
logger.info(f"运动通知游标初始化（DB）: last_event_id={self._last_event_id}")
return
now = int(datetime.now(timezone.utc).timestamp())
events = self._fetch_events(now - 3600, now) # 最近 1h 用于定位最大 id
if events:
self._last_event_id = max(
int(e.get('id', 0)) for e in events if e.get('id'))
else:
saved = db_layer.get_motion_cursor()
self._last_event_id = int(saved) if saved else 0
logger.info(f"运动通知游标初始化: last_event_id={self._last_event_id}")
self._last_event_id = 0
logger.info(f"运动通知游标初始化（SS 当前最大）: last_event_id={self._last_event_id}")
def _poll_once(self):
now = int(datetime.now(timezone.utc).timestamp())
@@ -339,10 +402,15 @@ class MotionNotifier:
if not new:
return
new.sort(key=lambda e: int(e.get('id', 0)))
cursor = self._last_event_id
for i in range(0, len(new), self.batch_size):
batch = new[i:i + self.batch_size]
self.push_events_to_oracle(batch)
self._last_event_id = max(int(e.get('id', 0)) for e in new)
stored = self.push_events_to_oracle(batch)
if stored > 0:
# 只有推送成功的批次才推进游标；失败批次保持原地，
# 下一轮窗口回看会重新捞到并重试（不丢事件）
cursor = max(int(e.get('id', 0)) for e in batch)
self._last_event_id = cursor
db_layer.set_motion_cursor(str(self._last_event_id))
def _run(self):
@@ -371,9 +439,16 @@ class MotionNotifier:
if not self.enabled:
logger.info("MotionNotifier 未启用（motion_notifier.enabled=false）")
return
# 心跳线程跟轮询是否开启无关：只要 MotionNotifier 整体 enabled，就该
# 持续证明推送链路活着，哪怕当前正好没有真实运动事件可推送。
if not self._hb_running:
self._hb_running = True
self._hb_thread = threading.Thread(
target=self._heartbeat_run, daemon=True, name='motion-heartbeat')
self._hb_thread.start()
if not self.poll_enabled:
logger.info("MotionNotifier 轮询已禁用（poll_enabled=false），仅作为 "
"Webhook 推送客户端 + 摄像头名映射使用")
"Webhook 推送客户端 + 心跳 + 摄像头名映射使用")
return
if self._running:
return
@@ -389,9 +464,15 @@ class MotionNotifier:
self._running = False
if self._thread:
self._thread.join(timeout=5)
self._hb_running = False
if self._hb_thread:
self._hb_thread.join(timeout=5)
def status(self) -> dict:
return {
"heartbeat_running": self._hb_thread is not None and self._hb_thread.is_alive(),
"last_heartbeat_at": self._last_heartbeat_at.isoformat() if self._last_heartbeat_at else None,
"last_heartbeat_error": self._last_heartbeat_error,
"running": self.is_alive(),
"enabled": self.enabled,
"poll_enabled": self.poll_enabled,