From 01bbb39750c8dca793bc16e003a032fa411b2348 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Sat, 22 Aug 2026 10:26:03 +0800 Subject: [PATCH] =?UTF-8?q?[=E6=96=B0=E5=A2=9E]=20NAS=20=E8=BF=90=E5=8A=A8?= =?UTF-8?q?=E7=9B=91=E6=B5=8B=E9=80=9A=E7=9F=A5=E6=9C=8D=E5=8A=A1=20Motion?= =?UTF-8?q?Notifier=20-=20=E8=BD=AE=E8=AF=A2/=E6=8E=A5=E6=94=B6=20SS=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E5=90=8E=E4=B8=BB=E5=8A=A8=20POST=20?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E5=88=B0=E7=94=B2=E9=AA=A8=E6=96=87=20/api/s?= =?UTF-8?q?s/motion=EF=BC=88=E5=8D=95=E5=90=91=20NAS->Oracle=EF=BC=89?= =?UTF-8?q?=EF=BC=9A=E6=96=B0=E5=A2=9E=20motion=5Fnotifier=20=E5=AD=90?= =?UTF-8?q?=E5=8C=85=E3=80=81/api/ss/webhook=20=E5=AE=9E=E6=97=B6=E8=BD=AC?= =?UTF-8?q?=E5=8F=91=E8=93=9D=E5=9B=BE=E3=80=81db=5Flayer=20=E8=BF=90?= =?UTF-8?q?=E5=8A=A8=E6=B8=B8=E6=A0=87=E3=80=81app.py=20=E6=8E=A5=E7=BA=BF?= =?UTF-8?q?=E4=B8=8E=20config=20=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fam-core/config/config.yaml | 18 ++ fam-core/src/fam_core/app.py | 13 + fam-core/src/fam_core/db_layer.py | 29 ++ fam-core/src/fam_core/motion_bp.py | 53 ++++ .../src/fam_core/motion_notifier/__init__.py | 4 + .../motion_notifier/motion_notifier.py | 260 ++++++++++++++++++ 6 files changed, 377 insertions(+) create mode 100644 fam-core/src/fam_core/motion_bp.py create mode 100644 fam-core/src/fam_core/motion_notifier/__init__.py create mode 100644 fam-core/src/fam_core/motion_notifier/motion_notifier.py diff --git a/fam-core/config/config.yaml b/fam-core/config/config.yaml index 9509185..bb61c2b 100644 --- a/fam-core/config/config.yaml +++ b/fam-core/config/config.yaml @@ -32,3 +32,21 @@ chat_handler: # 智能问答统一走 FAM-Edge 编排端点（Gemini → NVIDIA → 本地 Ollama 兜底） qa_url: "http://129.146.203.203:5000/api/edge/chat/ask" timeout: 120 + +# 运动监测通知服务（2026-08-22 新增） +# 在 NAS 本机轮询群晖 Surveillance Station 的运动侦测事件，把所需信息主动 POST +# 推送到甲骨文 FAM-Edge（单向 NAS -> Oracle，甲骨文不再反向访问 NAS）。 +# SS 凭据走环境变量（${DSM_ACCOUNT}/${DSM_PASSWORD}），由启动脚本 source 的 .env 提供。 +motion_notifier: + enabled: true + dsm_host: "192.168.50.64" # Surveillance Station 所在地址（NAS 本机） + dsm_port: 5000 + dsm_account: "${DSM_ACCOUNT}" + dsm_password: "${DSM_PASSWORD}" + camera_ids: [2] # 关注的摄像头（Generic_ONVIF-001 的 camera_id） + oracle_base_url: "http://129.146.203.203:5000" # 与 oracle_sync.base_url 一致 + oracle_token: "${ORACLE_SYNC_TOKEN}" # 与 oracle_sync.token 一致 + poll_interval_sec: 60 # 轮询间隔 + poll_window_hours: 2 # 每轮回看窗口（小时），覆盖轮询间隔内的新事件 + batch_size: 100 # 单批推送上限 + timeout_sec: 10 # 单次 SS 请求超时 diff --git a/fam-core/src/fam_core/app.py b/fam-core/src/fam_core/app.py index 5b310e4..43c2b08 100644 --- 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 diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py index 2c301e2..9974777 100644 --- 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() + + # ============================================================ # 统计 # ============================================================ diff --git a/fam-core/src/fam_core/motion_bp.py b/fam-core/src/fam_core/motion_bp.py new file mode 100644 index 0000000..c26dc57 --- /dev/null +++ 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 diff --git a/fam-core/src/fam_core/motion_notifier/__init__.py b/fam-core/src/fam_core/motion_notifier/__init__.py new file mode 100644 index 0000000..5afb0ae --- /dev/null +++ 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"] diff --git a/fam-core/src/fam_core/motion_notifier/motion_notifier.py b/fam-core/src/fam_core/motion_notifier/motion_notifier.py new file mode 100644 index 0000000..b936964 --- /dev/null +++ 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}", + }