feat(fam-edge): DSM 运动侦测预过滤 - 空转时段跳过云端视频分析

群晖 Surveillance Station 用 SYNO.SurveillanceStation.EventCenter.Event（未公开
文档的内部接口，参数是下划线风格 camera_ids/start_time/end_time，公开文档里的
cameraIds/fromTime/toTime 是旧版 Event API 的参数名，两者不通用）记录了摄像头
真实的运动侦测事件（event_type=10=运动，start_time+duration 精确到秒），比自己
本地跑 ffmpeg 帧差分更准、不需要额外算力，之前一直在这台 NAS 上验证可行性。

新增 DsmMotionClient：process_video() 分析每段视频前，先用视频的
[event_start, event_start+duration] 时间窗查一次这个接口，窗口内一条运动事件都
没有就跳过云端分析（标记 done，compute_provider=skipped_no_motion），省掉长期
无人时段白白消耗的 Gemini/NVIDIA 配额。

安全设计：查询本身失败/未配置/账号密码没填一律 fail-open（当作"有运动"，照常
分析），不会因为这层可选优化漏检真实事件——这是一个纯粹的省配额优化，不能反过来
影响监控系统的可靠性。

新增 12 个单测覆盖：禁用/缺凭证时的 fail-open、登录失败、网络异常、无事件、有
事件、按 camera_id 过滤（响应是按 ds_id 分组不是按 camera_id，容易搞混）、
最短运动时长阈值、session 过期重登录重试、env 变量解析。

账号密码走 .env 的 DSM_ACCOUNT/DSM_PASSWORD，不明文入库。

fam-edge/config/config.yaml

@@ -47,6 +47,23 @@ video_processing:
   # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed
   vision_order: ["gemini", "nvidia"]
 
+# DSM 运动侦测预过滤（2026-08-22 新增）：分析前先查一下群晖 Surveillance Station
+# 自己记录的运动侦测事件（SYNO.SurveillanceStation.EventCenter.Event，未公开文档的
+# 内部接口，参数名是下划线风格 camera_ids/start_time/end_time），这段时间窗口一条
+# 运动事件都没有就跳过云端分析（标记 done，compute_provider=skipped_no_motion），
+# 省掉长期无人时段白白消耗的 Gemini/NVIDIA 配额。
+# 账号密码走 .env（DSM_ACCOUNT/DSM_PASSWORD），不明文入库；查询失败/未配置一律
+# fail-open（照常送云端分析），不会因为这层可选优化漏检真实事件。
+dsm_motion_prefilter:
+  enabled: true
+  host: "192.168.50.64"
+  port: 5000
+  account: "${DSM_ACCOUNT}"
+  password: "${DSM_PASSWORD}"
+  camera_id: 2                  # Surveillance Station 里 Generic_ONVIF-001 的 camera_id
+  min_motion_seconds: 0         # 窗口内运动事件总时长需 ≥ 此值才算"有运动"（0=有事件就算）
+  timeout_sec: 10
+
 # 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama
 models:
   - provider: "gemini"

fam-edge/src/fam_edge/dsm_motion_client.py

@@ -0,0 +1,125 @@
+"""
+DsmMotionClient - 查询群晖 Surveillance Station 的运动侦测事件（预过滤用）
+
+背景: fam-edge 原来不管这段 30 分钟录像里有没有人走动，一律整段送云端 VLM 分析，
+配额/耗时都花在长期无人的空转时段上。DSM 的 Surveillance Station 用
+SYNO.SurveillanceStation.EventCenter.Event 这个未公开文档的内部 API 记录了摄像
+头真实的运动侦测窗口（start_time + duration，event_type=10 即运动），比自己在本地
+用 ffmpeg 帧差分更准，也不需要额外算力。
+
+用法: process_video() 分析前，用视频的 [event_start, event_start+duration] 时间窗
+查一次这个接口——窗口内一条运动事件都没有，就跳过云端分析（标记 done，
+compute_provider='skipped_no_motion'），有任何一条就正常送去分析。
+
+失败即放行（fail-open）: 网络错误/认证失败/账号未授权(401)等任何异常都视为
+"无法判断"，返回 None，调用方必须当作"照常分析"处理，而不是当作"跳过"处理——
+宁可多花一次配额，也不能因为这个可选的省钱层漏检真实事件。
+
+未公开文档的内部 API，字段名/权限模型可能随群晖套件升级变化，出问题时表现为
+每次都 fail-open（不再省钱），不会导致漏检。
+"""
+import time
+from datetime import datetime
+from typing import Optional
+
+import requests
+
+from .logger import setup_logger
+
+logger = setup_logger('fam-edge.dsm_motion_client')
+
+
+class DsmMotionClient:
+    def __init__(self, config: dict):
+        self.enabled = bool(config.get('enabled', False))
+        self.host = config.get('host', '')
+        self.port = config.get('port', 5000)
+        self.account = self._resolve(config.get('account', ''))
+        self.password = self._resolve(config.get('password', ''))
+        self.camera_id = config.get('camera_id')
+        self.min_motion_seconds = float(config.get('min_motion_seconds', 0))
+        self.timeout = config.get('timeout_sec', 10)
+        self._sid: Optional[str] = None
+        self._base = f"http://{self.host}:{self.port}/webapi"
+
+    @staticmethod
+    def _resolve(raw: str) -> str:
+        import os
+        if raw.startswith('${') and raw.endswith('}'):
+            return os.environ.get(raw[2:-1], '')
+        return raw
+
+    def _login(self) -> Optional[str]:
+        try:
+            resp = requests.get(
+                f"{self._base}/auth.cgi",
+                params={
+                    "api": "SYNO.API.Auth", "version": 6, "method": "login",
+                    "account": self.account, "passwd": self.password,
+                    "session": "SurveillanceStation", "format": "sid",
+                },
+                timeout=self.timeout,
+            )
+            data = resp.json()
+            if data.get('success'):
+                return data['data']['sid']
+            logger.warning(f"DSM 登录失败: {data.get('error')}")
+        except Exception as e:
+            logger.warning(f"DSM 登录异常: {e}")
+        return None
+
+    def has_motion_in_range(self, start_dt: datetime, duration_sec: float) -> Optional[bool]:
+        """查询 [start_dt, start_dt+duration_sec] 窗口内是否有运动事件。
+
+        返回 True/False 为确定结果；返回 None 表示查询本身失败，调用方必须
+        fail-open（当作"有运动"处理，照常送云端分析），不能当作"无运动"跳过。
+        """
+        if not self.enabled or not self.host or not self.account or not self.password \
+                or self.camera_id is None:
+            return None
+        if self._sid is None:
+            self._sid = self._login()
+            if self._sid is None:
+                return None
+        start_ts = int(start_dt.timestamp())
+        end_ts = int(start_dt.timestamp() + max(duration_sec, 0))
+        for attempt in range(2):  # 第 2 次仅用于 session 过期后重新登录重试一次
+            try:
+                resp = requests.get(
+                    f"{self._base}/entry.cgi",
+                    params={
+                        "api": "SYNO.SurveillanceStation.EventCenter.Event",
+                        "version": 1, "method": "List",
+                        "camera_ids": str(self.camera_id),
+                        "start_time": start_ts, "end_time": end_ts,
+                        "limit": 50, "_sid": self._sid,
+                    },
+                    timeout=self.timeout,
+                )
+                data = resp.json()
+            except Exception as e:
+                logger.warning(f"DSM 运动事件查询异常: {e}")
+                return None
+            if not data.get('success'):
+                err_code = (data.get('error') or {}).get('code')
+                if err_code in (106, 107, 119) and attempt == 0:
+                    # session 过期/被顶掉，重新登录重试一次
+                    self._sid = self._login()
+                    if self._sid is None:
+                        return None
+                    continue
+                logger.warning(f"DSM 运动事件查询失败: {data.get('error')}")
+                return None
+            # data 按 ds_id（群晖多主机 CMS 场景的服务器 id，单机固定是 "0"）分组，
+            # 不是按 camera_id 分组——这里把所有 ds_id 下的事件拉平，
+            # 用事件自带的 camera_id 字段再确认一次（双保险，服务端已按 camera_ids 过滤）。
+            events = [
+                e for group in (data.get('data') or {}).values()
+                for e in (group or [])
+                if e.get('camera_id') == self.camera_id
+            ]
+            if not events:
+                return False
+            total_motion = sum(float(e.get('duration', 0) or 0) for e in events)
+            return total_motion >= self.min_motion_seconds
+        return None

fam-edge/src/fam_edge/video_processor.py

@@ -22,6 +22,7 @@ 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')
@@ -193,6 +194,7 @@ 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 = []
@@ -236,6 +238,22 @@ class VideoProcessor:
             if event_start:
                 self.db.set_event_start_time(video_id, event_start)
 
+        # DSM 运动侦测预过滤：这段时间窗口里群晖自己记录的运动事件一条都没有，
+        # 就跳过云端分析（省配额）。查询本身失败/未配置一律 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)
+            except ValueError:
+                has_motion = None
+            if has_motion is False:
+                logger.info(f"[video_id={video_id}] DSM 运动预过滤：该时段无运动，跳过云端分析")
+                self.db.mark_video_processed(
+                    video_id, "（自动跳过：该时段未检测到运动）", [], [], 'skipped_no_motion')
+                return True
+
         known = self.db.get_known_members_context()
         logger.info(f"[video_id={video_id}] 开始整视频分析: {filename} "
                     f"(event_start={event_start}, known_members={'有' if known else '无'})")

fam-edge/tests/test_dsm_motion_client.py

@@ -0,0 +1,157 @@
+from datetime import datetime
+
+from fam_edge.dsm_motion_client import DsmMotionClient
+
+
+def _cfg(**overrides):
+    base = {
+        "enabled": True,
+        "host": "192.168.50.64",
+        "port": 5000,
+        "account": "ericwyuan",
+        "password": "secret",
+        "camera_id": 2,
+        "min_motion_seconds": 0,
+        "timeout_sec": 5,
+    }
+    base.update(overrides)
+    return base
+
+
+class _FakeResp:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+
+def _events_payload(events, ds_id="0"):
+    return {"success": True, "data": {ds_id: events}}
+
+
+def test_disabled_config_always_returns_none():
+    c = DsmMotionClient(_cfg(enabled=False))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is None
+
+
+def test_missing_credentials_returns_none():
+    c = DsmMotionClient(_cfg(account="", password=""))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is None
+
+
+def test_login_failure_returns_none(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        return _FakeResp({"success": False, "error": {"code": 400}})
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg())
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is None
+
+
+def test_login_network_exception_returns_none(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        raise ConnectionError("boom")
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg())
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is None
+
+
+def test_no_motion_events_returns_false(monkeypatch):
+    calls = {"n": 0}
+
+    def fake_get(url, params=None, timeout=None):
+        calls["n"] += 1
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload([]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg())
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is False
+
+
+def test_motion_event_present_returns_true(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 12}]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg())
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is True
+
+
+def test_events_grouped_by_ds_id_not_camera_id(monkeypatch):
+    """核心场景: 响应按 ds_id 分组（单机固定 "0"），不是按 camera_id 分组——
+    错把 camera_id 当 dict key 去取会永远拿到空列表。"""
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 5}], ds_id="0"))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg(camera_id=2))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is True
+
+
+def test_events_for_other_camera_ignored(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 99, "event_type": 10, "start_time": 1000, "duration": 999}]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg(camera_id=2))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is False
+
+
+def test_total_motion_below_threshold_is_treated_as_no_motion(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 1}]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg(min_motion_seconds=3))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is False
+
+
+def test_total_motion_meeting_threshold_counts_as_motion(monkeypatch):
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            return _FakeResp({"success": True, "data": {"sid": "abc"}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 2},
+             {"camera_id": 2, "event_type": 10, "start_time": 1100, "duration": 2}]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg(min_motion_seconds=3))
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is True
+
+
+def test_expired_session_relogins_and_retries_once(monkeypatch):
+    """session 过期(错误码 106)时应该重新登录一次再查，而不是直接放弃。"""
+    state = {"logins": 0, "queries": 0}
+
+    def fake_get(url, params=None, timeout=None):
+        if params.get("api") == "SYNO.API.Auth":
+            state["logins"] += 1
+            return _FakeResp({"success": True, "data": {"sid": f"sid{state['logins']}"}})
+        state["queries"] += 1
+        if state["queries"] == 1:
+            return _FakeResp({"success": False, "error": {"code": 106}})
+        return _FakeResp(_events_payload(
+            [{"camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 5}]))
+    monkeypatch.setattr("fam_edge.dsm_motion_client.requests.get", fake_get)
+    c = DsmMotionClient(_cfg())
+    assert c.has_motion_in_range(datetime(2026, 8, 22), 1800) is True
+    assert state["logins"] == 2
+    assert state["queries"] == 2
+
+
+def test_env_var_credentials_resolved():
+    import os
+    os.environ["TEST_DSM_ACCOUNT_XYZ"] = "realaccount"
+    try:
+        c = DsmMotionClient(_cfg(account="${TEST_DSM_ACCOUNT_XYZ}"))
+        assert c.account == "realaccount"
+    finally:
+        del os.environ["TEST_DSM_ACCOUNT_XYZ"]