[新增] NAS 运动监测通知服务 MotionNotifier - 轮询/接收 SS 事件后主动 POST 推送到甲骨文 /api/ss/motion（单向 NAS->Oracle）：新增 motion_notifier 子包、/api/ss/webhook 实时转发蓝图、db_layer 运动游标、app.py 接线与 config 块

2026-08-22 10:26:03 +08:00
parent 38dea6ba2e
commit 01bbb39750
6 changed files with 377 additions and 0 deletions
--- a/fam-core/src/fam_core/app.py
+++ b/fam-core/src/fam_core/app.py
@@ -17,9 +17,11 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from .config_loader import load_config
 from .logger import setup_logger
 from .oracle_sync import get_sync
+from .motion_notifier.motion_notifier import get_motion_notifier
 from .chat_handler.chat_handler import chat_bp
 from .member_manager.member_manager import member_bp
 from .img_proxy import img_bp
+from .motion_bp import motion_bp
 from .ui_api import ui_bp
 from .static_app import static_bp

@@ -32,6 +34,7 @@ app = Flask(__name__)
 app.register_blueprint(chat_bp)
 app.register_blueprint(member_bp)
 app.register_blueprint(img_bp)
+app.register_blueprint(motion_bp)
 app.register_blueprint(ui_bp)
 app.register_blueprint(static_bp)

@@ -56,6 +59,15 @@ try:
 except Exception as e:
    logger.error(f"Oracle-Sync 启动失败: {e}")

+# 初始化运动监测通知服务（NAS 轮询 SS 事件 -> 推送甲骨文；enabled 才真正启动线程）
+_notifier = None
+try:
+    _notifier = get_motion_notifier()
+    _notifier.start()
+    logger.info("MotionNotifier 已初始化")
+except Exception as e:
+    logger.error(f"MotionNotifier 初始化失败: {e}")
+

@app.route('/api/status', methods=['GET'])
 def status():
@@ -63,6 +75,7 @@ def status():
    return jsonify({
        "service": "fam-core",
        "sync": _sync.status() if _sync else {"running": False, "error": "未初始化"},
+        "motion": _notifier.status() if _notifier else {"running": False, "error": "未初始化"},
    }), 200


--- a/fam-core/src/fam_core/db_layer.py
+++ b/fam-core/src/fam_core/db_layer.py
@@ -413,6 +413,35 @@ def set_sync_cursor(value: str):
        conn.close()


+# ============================================================
+# 运动通知游标（MotionNotifier 增量去重用，复用 sync_cursor 表）
+# ============================================================
+
+def get_motion_cursor() -> str:
+    """返回上次推送到甲骨文的最大 SS 事件 id（字符串），无则返回空。"""
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT `value` FROM sync_cursor WHERE `key`='motion_last_event_id'")
+        row = cur.fetchone()
+        return row[0] if row else ''
+    finally:
+        conn.close()
+
+
+def set_motion_cursor(value: str):
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute(
+            """INSERT INTO sync_cursor (`key`, `value`) VALUES ('motion_last_event_id', %s)
+               ON DUPLICATE KEY UPDATE `value`=VALUES(`value`)""",
+            (value,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
 # ============================================================
 # 统计
 # ============================================================
--- a/fam-core/src/fam_core/motion_bp.py
+++ b/fam-core/src/fam_core/motion_bp.py
@@ -0,0 +1,53 @@
+"""
+运动监测 Webhook 接收（fam-core）
+
+群晖 Surveillance Station 可配置 HTTP 推送（Webhook），将事件实时 POST 到本端点，
+本端点即时转发到甲骨文 FAM-Edge。与 MotionNotifier 轮询互补，提供更低的事件延迟。
+
+SS Webhook 的实际 payload 格式随套件版本而异，本路由做宽松解析:
+  - 接受 JSON 或表单；
+  - 兼容 {events:[...]} / {data:[...]} / 单事件对象 / 裸数组；
+  - 字段名兼容 id/event_id、thumbnail_url/thumbnail_dir。
+无法识别的字段会原样透传，甲骨文端按已知字段落库（未知字段忽略）。
+"""
+from flask import Blueprint, request, jsonify
+
+from .logger import setup_logger
+from .motion_notifier.motion_notifier import get_motion_notifier
+
+logger = setup_logger('fam-core.motion_bp')
+
+motion_bp = Blueprint('motion_bp', __name__)
+
+
+def _coerce_events(payload) -> list:
+    """从各种可能的 SS payload 形态中提取事件列表。"""
+    if isinstance(payload, list):
+        return payload
+    if isinstance(payload, dict):
+        for key in ('events', 'data', 'event', 'items'):
+            v = payload.get(key)
+            if isinstance(v, list):
+                return v
+        # 单事件对象
+        return [payload]
+    return []
+
+
+@motion_bp.route('/api/ss/webhook', methods=['POST'])
+def ss_webhook():
+    data = request.get_json(silent=True)
+    if not data:
+        data = request.form.to_dict() or None
+    if not data:
+        return jsonify({"error": "Invalid payload"}), 400
+    events = _coerce_events(data)
+    if not events:
+        return jsonify({"error": "no events found in payload"}), 400
+    pushed = get_motion_notifier().push_events_to_oracle(events)
+    return jsonify({"status": "ok", "received": len(events), "pushed": pushed}), 200
+
+
+@motion_bp.route('/api/ss/status', methods=['GET'])
+def ss_status():
+    return jsonify(get_motion_notifier().status()), 200
--- a/fam-core/src/fam_core/motion_notifier/init.py
+++ b/fam-core/src/fam_core/motion_notifier/init.py
@@ -0,0 +1,4 @@
+"""MotionNotifier 包：NAS 端运动监测通知服务。"""
+from .motion_notifier import MotionNotifier, get_motion_notifier
+
+__all__ = ["MotionNotifier", "get_motion_notifier"]
--- a/fam-core/src/fam_core/motion_notifier/motion_notifier.py
+++ b/fam-core/src/fam_core/motion_notifier/motion_notifier.py
@@ -0,0 +1,260 @@
+"""
+MotionNotifier - NAS 端运动监测通知服务（新架构，2026-08-22）
+
+职责:
+  在 NAS 本机轮询群晖 Surveillance Station 的运动侦测事件
+  （SYNO.SurveillanceStation.EventCenter.Event，参数名下划线风格 camera_ids/
+   start_time/end_time，event_type=10 即运动），把"需要的信息"主动 POST 推送到
+  甲骨文 FAM-Edge 的 /api/ss/motion 接口。
+
+数据方向（关键约束）: NAS -> Oracle，单向。甲骨文不再反向访问 NAS。
+  - 原 fam-edge 的 dsm_motion_client（甲骨文主动查 SS API）已停用；
+  - 改由本服务在 NAS 上拉取 SS 事件，推送后由甲骨文本地落库 ss_motion_events，
+    供 video_processor 本地预过滤使用。
+
+两种推送触发（都汇聚到 push_events_to_oracle）:
+  1. 轮询（默认开）: 每 poll_interval_sec 拉一次新事件，增量推送到 Oracle。
+  2. Webhook（可选）: fam-core 暴露 POST /api/ss/webhook，群晖 SS 配置 HTTP 推送后
+     可实时转发（见 fam_core.motion_bp）。
+"""
+import os
+import re
+import time
+import threading
+from datetime import datetime, timezone
+
+import requests
+
+from ..logger import setup_logger
+from ..config_loader import load_config
+from .. import db_layer
+
+logger = setup_logger('fam-core.motion_notifier')
+
+_MOTION_NOTIFIER = None
+
+
+def get_motion_notifier():
+    """模块级单例（app.py 启动时创建并 start，Webhook 路由经此获取）。"""
+    global _MOTION_NOTIFIER
+    if _MOTION_NOTIFIER is None:
+        _MOTION_NOTIFIER = MotionNotifier()
+    return _MOTION_NOTIFIER
+
+
+class MotionNotifier:
+    def __init__(self):
+        cfg = load_config().get('motion_notifier', {})
+        self.enabled = bool(cfg.get('enabled', False))
+        self.dsm_host = cfg.get('dsm_host', '192.168.50.64')
+        self.dsm_port = int(cfg.get('dsm_port', 5000))
+        self.dsm_account = self._resolve(cfg.get('dsm_account', ''))
+        self.dsm_password = self._resolve(cfg.get('dsm_password', ''))
+        self.camera_ids = cfg.get('camera_ids', [2])
+        self.oracle_base_url = cfg.get('oracle_base_url',
+                                       'http://129.146.203.203:5000').rstrip('/')
+        self.oracle_token = self._resolve(cfg.get('oracle_token', '${ORACLE_SYNC_TOKEN}'))
+        self.poll_interval_sec = int(cfg.get('poll_interval_sec', 60))
+        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))
+        self._base = f"http://{self.dsm_host}:{self.dsm_port}/webapi"
+        self._sid = None
+        self._running = False
+        self._thread = None
+        self._last_event_id = None
+        self._last_poll_at = None
+        self._last_error = None
+        self._pushed_total = 0
+
+    @staticmethod
+    def _resolve(v):
+        """解析 ${ENV} 引用；非字符串或不含 ${...} 原样返回。"""
+        if isinstance(v, str) and v.startswith('${') and v.endswith('}'):
+            return os.environ.get(v[2:-1], '')
+        return v
+
+    # ------------------------------------------------------------------
+    # Surveillance Station 登录 / 事件查询
+    # ------------------------------------------------------------------
+    def _login(self):
+        try:
+            resp = requests.get(
+                f"{self._base}/auth.cgi",
+                params={"api": "SYNO.API.Auth", "version": 6, "method": "login",
+                        "account": self.dsm_account, "passwd": self.dsm_password,
+                        "session": "SurveillanceStation", "format": "sid"},
+                timeout=self.timeout)
+            data = resp.json()
+            if data.get('success'):
+                return data['data']['sid']
+            logger.warning(f"SS 登录失败: {data.get('error')}")
+        except Exception as e:
+            logger.warning(f"SS 登录异常: {e}")
+        return None
+
+    def _fetch_events(self, start_ts: int, end_ts: int):
+        """查询 [start_ts, end_ts] 窗口内的 SS 事件。返回事件列表或 None（查询失败）。"""
+        if self._sid is None:
+            self._sid = self._login()
+            if self._sid is None:
+                return None
+        for attempt in range(2):
+            try:
+                resp = requests.get(
+                    f"{self._base}/entry.cgi",
+                    params={"api": "SYNO.SurveillanceStation.EventCenter.Event",
+                            "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},
+                    timeout=self.timeout)
+                data = resp.json()
+            except Exception as e:
+                logger.warning(f"SS 事件查询异常: {e}")
+                return None
+            if not data.get('success'):
+                code = (data.get('error') or {}).get('code')
+                if code in (106, 107, 119) and attempt == 0:
+                    # session 过期/被顶掉，重新登录重试一次
+                    self._sid = self._login()
+                    if self._sid is None:
+                        return None
+                    continue
+                logger.warning(f"SS 事件查询失败: {data.get('error')}")
+                return None
+            # 响应按 ds_id（CMS 多机场景的服务器 id，单机固定 "0"）分组，拉平
+            events = [e for grp in (data.get('data') or {}).values()
+                      for e in (grp or [])]
+            return events
+        return None
+
+    # ------------------------------------------------------------------
+    # 推送到甲骨文
+    # ------------------------------------------------------------------
+    def push_events_to_oracle(self, events) -> int:
+        """把标准化后的事件列表推送到 Oracle /api/ss/motion。返回成功推送条数。"""
+        if not events:
+            return 0
+        norm = []
+        for e in events:
+            eid = e.get('id') or e.get('event_id')
+            if eid is None:
+                continue
+            norm.append({
+                "event_id": int(eid),
+                "camera_id": e.get('camera_id'),
+                "event_type": e.get('event_type'),
+                "start_time": e.get('start_time'),
+                "duration": e.get('duration'),
+                "thumbnail_url": e.get('thumbnail_url') or e.get('thumbnail_dir'),
+            })
+        if not norm:
+            return 0
+        try:
+            resp = requests.post(
+                f"{self.oracle_base_url}/api/ss/motion",
+                json={"token": self.oracle_token, "events": norm},
+                timeout=(10, 30))
+        except requests.RequestException as e:
+            logger.error(f"推送运动事件到 Oracle 失败: {e}")
+            self._last_error = str(e)
+            return 0
+        if resp.status_code != 200:
+            logger.error(f"推送运动事件到 Oracle 返回 {resp.status_code}: {resp.text[:200]}")
+            self._last_error = f"HTTP {resp.status_code}"
+            return 0
+        try:
+            stored = resp.json().get('stored', 0)
+        except ValueError:
+            stored = len(norm)
+        self._pushed_total += stored
+        self._last_error = None
+        logger.info(f"运动事件推送成功: {len(norm)} 条 -> Oracle 存储 {stored}")
+        return stored
+
+    # ------------------------------------------------------------------
+    # 增量轮询主循环
+    # ------------------------------------------------------------------
+    def _init_cursor(self):
+        """启动时初始化 last_event_id: 取当前 SS 最大事件 id，只推增量（不回灌历史）。"""
+        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}")
+
+    def _poll_once(self):
+        now = int(datetime.now(timezone.utc).timestamp())
+        start_ts = now - int(self.poll_window_hours * 3600)
+        events = self._fetch_events(start_ts, now)
+        if events is None:
+            return  # 查询失败，下一轮重试
+        # 只保留 id 大于游标的新事件（SS 事件 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:
+            return
+        new.sort(key=lambda e: int(e.get('id', 0)))
+        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)
+        db_layer.set_motion_cursor(str(self._last_event_id))
+
+    def _run(self):
+        logger.info(f"MotionNotifier 启动，轮询间隔 {self.poll_interval_sec}s，"
+                    f"目标 SS {self.dsm_host}:{self.dsm_port}")
+        try:
+            self._init_cursor()
+        except Exception as e:
+            logger.warning(f"运动通知游标初始化失败（从 0 开始）: {e}")
+            self._last_event_id = 0
+        while self._running:
+            try:
+                self._poll_once()
+                self._last_poll_at = datetime.now()
+            except Exception as e:
+                self._last_error = str(e)
+                logger.error(f"运动通知轮询异常: {e}", exc_info=True)
+            # 分段休眠，便于 stop 快速唤醒
+            for _ in range(self.poll_interval_sec):
+                if not self._running:
+                    break
+                time.sleep(1)
+
+    # ------------------------------------------------------------------
+    def start(self):
+        if not self.enabled:
+            logger.info("MotionNotifier 未启用（motion_notifier.enabled=false）")
+            return
+        if self._running:
+            return
+        self._running = True
+        self._thread = threading.Thread(target=self._run, daemon=True,
+                                        name='motion-notifier')
+        self._thread.start()
+
+    def is_alive(self):
+        return self._thread is not None and self._thread.is_alive()
+
+    def stop(self):
+        self._running = False
+        if self._thread:
+            self._thread.join(timeout=5)
+
+    def status(self) -> dict:
+        return {
+            "running": self.is_alive(),
+            "enabled": self.enabled,
+            "last_event_id": self._last_event_id,
+            "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,
+            "oracle": self.oracle_base_url,
+            "dsm": f"{self.dsm_host}:{self.dsm_port}",
+        }