[重构] 运动预过滤改为本地事件(NAS推送) - 甲骨文不再反向访问NAS：新增 ss_motion_events 表/record_motion_events/has_motion_in_range_local，/api/ss/motion 接收端点，video_processor 改用本地运动事件预过滤，停用 dsm_motion_client 反向查询

fam-edge/src/fam_edge/api_gateway/api_gateway.py

@@ -197,6 +197,36 @@ def oracle_avatar():
     return Response(data, mimetype='image/jpeg')
 
 
+@api_bp.route('/api/ss/motion', methods=['POST'])
+def ss_motion():
+    """接收 NAS 推送的运动侦测事件（单向：NAS -> Oracle）。
+
+    请求体: {"token": "...", "events": [
+        {"event_id": 25349, "camera_id": 2, "event_type": 10,
+         "start_time": 1787318023, "duration": 149, "thumbnail_url": "107471,12075"}
+    ]}
+    start_time/duration 为 Unix epoch（与 SS 同源，时区无关）。
+    落库 ss_motion_events（按 event_id 幂等），供 video_processor 本地预过滤使用。
+    """
+    if not _check_token():
+        return jsonify({"error": "unauthorized"}), 401
+    data = request.get_json(silent=True)
+    if not data or 'events' not in data:
+        return jsonify({"error": "缺少 events"}), 400
+    events = data.get('events') or []
+    if not isinstance(events, list):
+        return jsonify({"error": "events 必须是数组"}), 400
+    try:
+        stored = state.get_db().record_motion_events(events)
+    except Exception as e:
+        logger.error(f"ss_motion 落库异常: {e}")
+        return jsonify({"error": str(e)}), 500
+    if stored:
+        state.get_db().record_activity(
+            'motion', 'push', f"接收 NAS 运动事件 {stored} 条")
+    return jsonify({"status": "ok", "received": len(events), "stored": stored}), 200
+
+
 @api_bp.route('/health', methods=['GET'])
 def health():
     """健康检查：DB 连通性 + producer/consumer 线程存活状态。

fam-edge/src/fam_edge/dsm_motion_client.py

@@ -1,5 +1,15 @@
 """
-DsmMotionClient - 查询群晖 Surveillance Station 的运动侦测事件（预过滤用）
+[已弃用 / DEPRECATED] 本模块自 2026-08-22 起不再被调用。
+
+架构约束：甲骨文 FAM-Edge 不得反向访问 NAS。原 video_processor 在此处主动查询
+NAS 的 Surveillance Station（甲骨文 -> NAS），违反该约束，已停用。
+
+替代方案：由 NAS 端 fam-core 的 MotionNotifier 轮询/接收 SS 事件后，主动 POST
+推送到甲骨文的 /api/ss/motion，落库 ss_motion_events；video_processor 改用
+db.has_motion_in_range_local() 做本地预过滤。本文件保留仅作参考，请勿再实例化。
+
+---
+DsmMotionClient（旧实现，仅作历史参考） - 查询群晖 Surveillance Station 的运动侦测事件（预过滤用）
 
 背景: fam-edge 原来不管这段 30 分钟录像里有没有人走动，一律整段送云端 VLM 分析，
 配额/耗时都花在长期无人的空转时段上。DSM 的 Surveillance Station 用

fam-edge/src/fam_edge/oracle_db.py

@@ -110,10 +110,21 @@ class OracleDB:
             detail TEXT,
             ts TEXT
         );
+        CREATE TABLE IF NOT EXISTS ss_motion_events (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            event_id INTEGER UNIQUE,
+            camera_id INTEGER,
+            event_type INTEGER,
+            start_time INTEGER,
+            duration INTEGER,
+            thumbnail_url TEXT,
+            received_at TEXT
+        );
         CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at);
         CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id);
         CREATE INDEX IF NOT EXISTS idx_model_calls_created ON model_calls(created_at);
         CREATE INDEX IF NOT EXISTS idx_activity_ts ON service_activity(ts);
+        CREATE INDEX IF NOT EXISTS idx_motion_window ON ss_motion_events(start_time, event_type);
         """)
         # 兼容旧库：补 retry_count / file_valid / media 等列（生产-消费队列用）
         cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()]
@@ -193,6 +204,71 @@ class OracleDB:
             "ORDER BY id DESC LIMIT ?", (int(limit),)).fetchall()
         return [dict(r) for r in rows]
 
+    # ------------------------------------------------------------------
+    # 运动侦测事件（NAS 推送，单向：NAS -> Oracle，不再反向访问 NAS）
+    # 由 fam-core 的 MotionNotifier 轮询/接收 SS 事件后 POST 到 /api/ss/motion。
+    # ------------------------------------------------------------------
+    def record_motion_events(self, events: List[Dict]) -> int:
+        """批量 upsert NAS 推送来的运动事件（按 event_id 幂等）。返回成功条数。"""
+        n = 0
+        now = _now_iso()
+        for e in (events or []):
+            eid = e.get('event_id')
+            if eid is None:
+                continue
+            self._conn.execute(
+                """INSERT INTO ss_motion_events
+                       (event_id, camera_id, event_type, start_time, duration,
+                        thumbnail_url, received_at)
+                   VALUES (?,?,?,?,?,?,?)
+                   ON CONFLICT(event_id) DO UPDATE SET
+                       camera_id=excluded.camera_id,
+                       event_type=excluded.event_type,
+                       start_time=excluded.start_time,
+                       duration=excluded.duration,
+                       thumbnail_url=excluded.thumbnail_url,
+                       received_at=excluded.received_at""",
+                (int(eid), e.get('camera_id'), e.get('event_type'),
+                 e.get('start_time'), e.get('duration'),
+                 e.get('thumbnail_url'), now))
+            n += 1
+        if n:
+            self._conn.commit()
+        return n
+
+    def get_recent_motion_events(self, limit: int = 50) -> List[Dict]:
+        """最近运动事件（按 start_time 倒序）。"""
+        rows = self._conn.execute(
+            "SELECT id, event_id, camera_id, event_type, start_time, duration, "
+            "thumbnail_url, received_at FROM ss_motion_events "
+            "ORDER BY start_time DESC LIMIT ?", (int(limit),)).fetchall()
+        return [dict(r) for r in rows]
+
+    def has_motion_in_range_local(self, start_ts: int, end_ts: int,
+                                  camera_id: int = None) -> Optional[bool]:
+        """本地运动预过滤：判断 [start_ts, end_ts] 窗口内是否存在运动事件。
+
+        替代原 dsm_motion_client 反向访问 NAS 的做法。返回:
+          - None : 本地运动事件表为空（冷启动，尚未收到 NAS 推送），调用方
+                   必须 fail-open（照常送云端分析），不能当作"无运动"跳过。
+          - True/False : 窗口内确有/确无运动事件。
+        start_ts/end_ts 为 Unix epoch（与 SS 事件 start_time 同源，时区无关）。
+        """
+        total = self._conn.execute(
+            "SELECT COUNT(*) c FROM ss_motion_events").fetchone()['c']
+        if total == 0:
+            return None
+        sql = ("SELECT COUNT(*) c FROM ss_motion_events "
+               "WHERE event_type = 10 "
+               "AND start_time <= ? "
+               "AND (start_time + COALESCE(duration,0)) >= ?")
+        params = [end_ts, start_ts]
+        if camera_id is not None:
+            sql += " AND camera_id = ?"
+            params.append(camera_id)
+        cnt = self._conn.execute(sql, params).fetchone()['c']
+        return cnt > 0
+
     def get_queue_status(self) -> Dict:
         """实时队列/处理状态（前端服务状态卡用）。"""
         total = self._conn.execute("SELECT COUNT(*) c FROM videos").fetchone()['c']

fam-edge/src/fam_edge/video_processor.py

@@ -22,7 +22,6 @@ from .logger import setup_logger
 from .config_loader import load_config
 from .model_adapters.adapter_factory import build_adapters
 from .model_adapters.base_adapter import BaseModelAdapter
-from .dsm_motion_client import DsmMotionClient
 from . import oracle_db
 
 logger = setup_logger('fam-edge.video_processor')
@@ -194,7 +193,6 @@ class VideoProcessor:
         adapters = build_adapters(self.config.get('models', []))
         self.vision_adapters: Dict[str, BaseModelAdapter] = {
             a.provider_name: a for a in adapters if a.get_role() == 'vision'}
-        self.dsm_motion = DsmMotionClient(self.config.get('dsm_motion_prefilter', {}))
 
     def _ordered_vision_adapters(self) -> List[BaseModelAdapter]:
         ordered = []
@@ -238,18 +236,23 @@ class VideoProcessor:
             if event_start:
                 self.db.set_event_start_time(video_id, event_start)
 
-        # DSM 运动侦测预过滤：这段时间窗口里群晖自己记录的运动事件一条都没有，
-        # 就跳过云端分析（省配额）。查询本身失败/未配置一律 fail-open（照常分析），
+        # 运动侦测预过滤（v2: 改用 NAS 推送来的本地运动事件，不再反向访问 NAS）：
+        # 这段时间窗口里没有任何运动事件，就跳过云端分析（省配额）。
+        # 本地运动事件表为空（冷启动，尚未收到 NAS 推送）一律 fail-open（照常分析），
         # 绝不能因为这层可选优化漏检真实事件。
         duration_sec = (vmeta or {}).get('duration_sec') if self.file_validate else None
         if event_start and duration_sec:
             try:
-                start_dt = datetime.strptime(event_start, '%Y-%m-%d %H:%M:%S')
-                has_motion = self.dsm_motion.has_motion_in_range(start_dt, duration_sec)
+                # event_start 是北京时间墙钟，转成与 SS 事件同源的 Unix epoch
+                start_dt = datetime.strptime(event_start, '%Y-%m-%d %H:%M:%S').replace(
+                    tzinfo=timezone(timedelta(hours=8)))
+                start_ts = int(start_dt.timestamp())
+                end_ts = start_ts + int(duration_sec)
+                has_motion = self.db.has_motion_in_range_local(start_ts, end_ts)
             except ValueError:
                 has_motion = None
             if has_motion is False:
-                logger.info(f"[video_id={video_id}] DSM 运动预过滤：该时段无运动，跳过云端分析")
+                logger.info(f"[video_id={video_id}] 运动预过滤：该时段无运动，跳过云端分析")
                 self.db.mark_video_processed(
                     video_id, "（自动跳过：该时段未检测到运动）", [], [], 'skipped_no_motion')
                 return True