[重构] 运动事件驱动架构 - 整段素材按 ss_motion_events 分割运动片段再分析(不再整段送云端): oracle_db 加 motion_event_id/camera_id 列 + get_motion_events_in_range/has_unfinished_motion_in_range/get_video_by_motion_event_id; video_processor 素材→分割/片段→分析双分支(ffmpeg -c:v copy -c:a aac 保留音频); video_queue 片段入队; config 加 motion_segment 块

fam-edge/src/fam_edge/oracle_db.py

@@ -149,6 +149,14 @@ class OracleDB:
         ]:
             if col not in pe_cols:
                 c.execute(ddl)
+        # 兼容旧库：videos 表补运动片段字段（运动事件驱动架构加）
+        vid_cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()]
+        for col, ddl in [
+            ('motion_event_id', "ALTER TABLE videos ADD COLUMN motion_event_id INTEGER"),
+            ('camera_id', "ALTER TABLE videos ADD COLUMN camera_id INTEGER"),
+        ]:
+            if col not in vid_cols:
+                c.execute(ddl)
         self._conn.commit()
 
     # ------------------------------------------------------------------
@@ -244,19 +252,45 @@ class OracleDB:
             "ORDER BY start_time DESC LIMIT ?", (int(limit),)).fetchall()
         return [dict(r) for r in rows]
 
+    def record_motion_heartbeat(self):
+        """NAS 侧推送链路心跳（不管有没有真实事件，每隔几分钟都应该调一次）。
+
+        单独的心跳信号，跟"表里有没有历史数据"是两回事：表非空只能说明"曾经收到
+        过推送"，推送链路后来整个挂掉（NAS 服务崩溃/网络断开/DSM 侧 Webhook 规则
+        被误关）之后，表依然非空，has_motion_in_range_local() 原来的判断方式会
+        误以为"链路健康、这段时间确认无运动"从而错误跳过分析——心跳新鲜度检查就是
+        堵这个漏洞的。
+        """
+        self.set_cursor('motion_heartbeat_at', _now_iso())
+
+    def get_motion_heartbeat_age_sec(self) -> Optional[float]:
+        """距上次心跳过了多少秒；从未收到过心跳返回 None。"""
+        v = self.get_cursor('motion_heartbeat_at')
+        if not v:
+            return None
+        try:
+            hb = datetime.strptime(v, '%Y-%m-%d %H:%M:%S').replace(
+                tzinfo=timezone(timedelta(hours=8)))
+        except ValueError:
+            return None
+        return (datetime.now(timezone(timedelta(hours=8))) - hb).total_seconds()
+
     def has_motion_in_range_local(self, start_ts: int, end_ts: int,
-                                  camera_id: int = None) -> Optional[bool]:
+                                  camera_id: int = None,
+                                  max_heartbeat_age_sec: float = 900) -> Optional[bool]:
         """本地运动预过滤：判断 [start_ts, end_ts] 窗口内是否存在运动事件。
 
         替代原 dsm_motion_client 反向访问 NAS 的做法。返回:
-          - None : 本地运动事件表为空（冷启动，尚未收到 NAS 推送），调用方
-                   必须 fail-open（照常送云端分析），不能当作"无运动"跳过。
-          - True/False : 窗口内确有/确无运动事件。
+          - None : 推送链路心跳缺失或过期（从未收到过 / 距上次心跳超过
+                   max_heartbeat_age_sec），说明当前无法确认 NAS -> Oracle 这条
+                   推送链路是否存活——调用方必须 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:
+        age = self.get_motion_heartbeat_age_sec()
+        if age is None or age > max_heartbeat_age_sec:
             return None
         sql = ("SELECT COUNT(*) c FROM ss_motion_events "
                "WHERE event_type = 10 "
@@ -269,6 +303,53 @@ class OracleDB:
         cnt = self._conn.execute(sql, params).fetchone()['c']
         return cnt > 0
 
+    def get_motion_events_in_range(self, start_ts: int, end_ts: int,
+                                   camera_id: int = None,
+                                   finished_grace_sec: int = 10) -> List[Dict]:
+        """窗口内【已结束】的运动事件（运动片段分割用）。
+
+        返回按 start_time 升序的 [{event_id, camera_id, start_time, duration,
+        thumbnail_url}]。只返回已结束事件：SS 事件 duration 在动作进行中会显示 0、
+        结束才回填真实时长，因此只分割 start_time+duration 已落在当前时刻
+        （含 finished_grace_sec 秒容差）的事件，进行中的等下轮结束后再处理。
+        """
+        now_ts = int(datetime.now(timezone(timedelta(hours=8))).timestamp())
+        sql = ("SELECT event_id, camera_id, event_type, start_time, duration, thumbnail_url "
+               "FROM ss_motion_events "
+               "WHERE event_type = 10 "
+               "AND start_time <= ? "
+               "AND (start_time + COALESCE(duration,0)) >= ? "
+               "AND COALESCE(duration,0) > 0 "
+               "AND (start_time + COALESCE(duration,0)) <= ? + ?")
+        params = [end_ts, start_ts, now_ts, int(finished_grace_sec)]
+        if camera_id is not None:
+            sql += " AND camera_id = ?"
+            params.append(camera_id)
+        sql += " ORDER BY start_time ASC"
+        return [dict(r) for r in self._conn.execute(sql, params).fetchall()]
+
+    def has_unfinished_motion_in_range(self, start_ts: int, end_ts: int,
+                                       camera_id: int = None) -> bool:
+        """窗口内是否存在【未结束】的运动事件（duration 可能还在增长）。"""
+        now_ts = int(datetime.now(timezone(timedelta(hours=8))).timestamp())
+        sql = ("SELECT COUNT(*) c FROM ss_motion_events "
+               "WHERE event_type = 10 "
+               "AND start_time <= ? "
+               "AND (start_time + COALESCE(duration,0)) >= ? "
+               "AND (start_time + COALESCE(duration,0)) > ?")
+        params = [end_ts, start_ts, now_ts]
+        if camera_id is not None:
+            sql += " AND camera_id = ?"
+            params.append(camera_id)
+        return self._conn.execute(sql, params).fetchone()['c'] > 0
+
+    def get_video_by_motion_event_id(self, motion_event_id: int):
+        """按运动事件 id 查是否已生成对应运动片段（幂等去重）。"""
+        cur = self._conn.execute(
+            "SELECT * FROM videos WHERE motion_event_id=? LIMIT 1",
+            (int(motion_event_id),))
+        return cur.fetchone()
+
     def get_queue_status(self) -> Dict:
         """实时队列/处理状态（前端服务状态卡用）。"""
         total = self._conn.execute("SELECT COUNT(*) c FROM videos").fetchone()['c']
@@ -302,23 +383,29 @@ class OracleDB:
 
     def ensure_video(self, filename: str, local_path: str,
                      camera_name: str = '', event_start_time: str = '',
-                     duration_sec: float = 0.0, drive_file_id: str = '') -> int:
-        """视频进入监听目录时登记；已存在则更新路径。返回 video_id。"""
+                     duration_sec: float = 0.0, drive_file_id: str = '',
+                     motion_event_id: int = None, camera_id: int = None) -> int:
+        """视频进入监听目录时登记；已存在则更新路径。返回 video_id。
+
+        motion_event_id: 运动片段关联的 SS 运动事件 id（非空=运动片段，非素材）。
+        """
         now = _now_iso()
         row = self.get_video_by_filename(filename)
         if row:
             self._conn.execute(
                 "UPDATE videos SET local_path=?, camera_name=?, event_start_time=?, "
-                "duration_sec=?, updated_at=? WHERE id=?",
-                (local_path, camera_name, event_start_time, duration_sec, now, row['id']))
+                "duration_sec=?, motion_event_id=?, camera_id=?, updated_at=? WHERE id=?",
+                (local_path, camera_name, event_start_time, duration_sec,
+                 motion_event_id, camera_id, now, row['id']))
             self._conn.commit()
             return row['id']
         cur = self._conn.execute(
             "INSERT INTO videos (drive_file_id, filename, local_path, camera_name, "
-            "duration_sec, event_start_time, status, created_at, updated_at) "
-            "VALUES (?,?,?,?,?,?, 'pending', ?, ?)",
+            "duration_sec, event_start_time, motion_event_id, camera_id, "
+            "status, created_at, updated_at) "
+            "VALUES (?,?,?,?,?,?,?,?, 'pending', ?, ?)",
             (drive_file_id, filename, local_path, camera_name, duration_sec,
-             event_start_time, now, now))
+             event_start_time, motion_event_id, camera_id, now, now))
         self._conn.commit()
         return cur.lastrowid
 

fam-edge/src/fam_edge/video_processor.py

@@ -190,6 +190,16 @@ class VideoProcessor:
             'file_validate', True))
         self.parse_start = self.config.get('gdrive_sync', {}).get(
             'parse_start_from_filename', True)
+        # 运动预过滤"无运动"结论的可信心跳时限：NAS 推送心跳超过这么久没更新，
+        # 就认为推送链路可能已经挂了，fail-open（照常分析）
+        self.motion_max_heartbeat_age_sec = float(self.config.get(
+            'dsm_motion_prefilter', {}).get('max_heartbeat_age_sec', 900))
+        # 运动片段分割（运动事件驱动架构）：整段素材按 ss_motion_events 分割片段再分析
+        seg = self.config.get('motion_segment', {})
+        self.motion_clips_dir = seg.get('clips_dir', '/opt/fam-edge/motion_clips')
+        self.motion_keep_audio = bool(seg.get('keep_audio', True))
+        self.motion_min_duration = float(seg.get('min_duration_sec', 1))
+        self.motion_grace_sec = int(seg.get('unfinished_grace_sec', 10))
         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'}
@@ -206,16 +216,21 @@ class VideoProcessor:
         return ordered
 
     def process_video(self, video_id: int, filename: str, local_path: str,
-                      timeout_multiplier: float = 1.0) -> bool:
-        """处理一个视频记录，返回是否成功。
+                      timeout_multiplier: float = 1.0):
+        """处理一个视频记录。返回 (ok: bool, new_clip_ids: List[int])。
+
+        - 素材整段视频（无 motion_event_id）：不再整段分析，而是按 ss_motion_events
+          里【已结束】的运动事件分割成运动片段，返回片段 video_id 列表（调用方负责
+          入队分析）；素材记录在全部事件结束后标记 done，仍有未结束事件时保持
+          pending 由生产者下轮重试。
+        - 运动片段（有 motion_event_id）：本身即运动时段，直接云端分析（跳过预过滤）。
 
         timeout_multiplier: 云端模型消费的超时放大倍数（如 2 = 在配置 timeout 上 ×2）。
-        每次调用前临时放大对应 adapter.timeout，调用后恢复，避免影响其他调用方。
         """
         if not os.path.isfile(local_path):
             logger.error(f"[video_id={video_id}] 文件不存在，跳过: {local_path}")
             self.db.mark_video_failed(video_id, "file_missing")
-            return False
+            return False, []
 
         # 处理前二次确认文件有效性（防止登记后文件被破坏/截断；校验结果落库）
         if self.file_validate:
@@ -224,41 +239,29 @@ class VideoProcessor:
                 logger.error(f"[video_id={video_id}] 文件校验失败({verr})，标记 failed: {local_path}")
                 self.db.set_video_file_status(video_id, False, verr)
                 self.db.mark_video_failed(video_id, f"invalid_file:{verr}")
-                return False
+                return False, []
             if vmeta:
                 self.db.set_video_file_status(video_id, True, '', vmeta)
+        else:
+            vmeta = None
 
-        camera_name = self.db.get_video_by_filename(filename)['camera_name'] or ''
-        event_start = ''
-        if self.parse_start:
+        vrow = self.db.get_video_by_filename(filename)
+        is_motion_clip = bool(vrow and vrow.get('motion_event_id'))
+        camera_name = (vrow or {}).get('camera_name') or ''
+        event_start = (vrow or {}).get('event_start_time') or ''
+        if not event_start and self.parse_start:
             event_start = _parse_event_start_from_filename(filename)
-            # 回写解析到的开始时间
             if event_start:
                 self.db.set_event_start_time(video_id, event_start)
 
-        # 运动侦测预过滤（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:
-                # 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}] 运动预过滤：该时段无运动，跳过云端分析")
-                self.db.mark_video_processed(
-                    video_id, "（自动跳过：该时段未检测到运动）", [], [], 'skipped_no_motion')
-                return True
+        if not is_motion_clip:
+            # ---- 素材整段视频：分割成运动片段，不整段分析 ----
+            return self._segment_source_video(
+                video_id, local_path, event_start, camera_name, vmeta)
 
+        # ---- 运动片段：本身即运动时段，直接云端分析（跳过运动预过滤）----
         known = self.db.get_known_members_context()
-        logger.info(f"[video_id={video_id}] 开始整视频分析: {filename} "
+        logger.info(f"[video_id={video_id}] 开始运动片段分析: {filename} "
                     f"(event_start={event_start}, known_members={'有' if known else '无'})")
 
         last_err = "no_vision_adapter"
@@ -274,7 +277,7 @@ class VideoProcessor:
                 logger.info(f"[video_id={video_id}] {adapter.provider_name} 超时 "
                             f"{orig_timeout}s -> {adapter.timeout}s (×{timeout_multiplier})")
             try:
-                logger.info(f"[video_id={video_id}] 尝试 {adapter.provider_name} 整视频分析")
+                logger.info(f"[video_id={video_id}] 尝试 {adapter.provider_name} 运动片段分析")
                 result = adapter.analyze_video(local_path, known, event_start)
             except Exception as e:
                 logger.error(f"[video_id={video_id}] {adapter.provider_name} 异常: {e}")
@@ -284,14 +287,117 @@ class VideoProcessor:
                 adapter.timeout = orig_timeout
             if result:
                 self._store_result(video_id, result)
-                return True
+                return True, []
             else:
                 last_err = f"{adapter.provider_name}_failed"
                 logger.warning(f"[video_id={video_id}] {adapter.provider_name} 未返回结果，降级下一模型")
 
         logger.error(f"[video_id={video_id}] 所有视觉模型失败，标记 failed: {last_err}")
         self.db.mark_video_failed(video_id, last_err)
-        return False
+        return False, []
+
+    # ------------------------------------------------------------------
+    # 运动片段分割（运动事件驱动架构）
+    # ------------------------------------------------------------------
+    def _segment_source_video(self, video_id: int, local_path: str, event_start: str,
+                              camera_name: str, vmeta: dict):
+        """整段素材 -> 按已结束运动事件分割运动片段。返回 (ok, clip_ids)。
+
+        素材视频不再整段送云端分析；仅按运动事件窗口切出片段交给下游分析。
+        """
+        if not event_start:
+            logger.info(f"[video_id={video_id}] 素材无开始时间，无法对齐运动事件，标记 done")
+            self.db.mark_video_processed(
+                video_id, "（素材无开始时间，无法分割运动片段）", [], [], 'motion_segment')
+            return True, []
+        try:
+            start_dt = datetime.strptime(event_start, '%Y-%m-%d %H:%M:%S').replace(
+                tzinfo=timezone(timedelta(hours=8)))
+            start_ts = int(start_dt.timestamp())
+        except ValueError:
+            logger.warning(f"[video_id={video_id}] 素材开始时间解析失败: {event_start}")
+            self.db.mark_video_processed(
+                video_id, "（素材开始时间解析失败，无法分割）", [], [], 'motion_segment')
+            return True, []
+        if vmeta is None:
+            ok, verr, vmeta = validate_video(local_path)
+            if not ok:
+                self.db.set_video_file_status(video_id, False, verr)
+                self.db.mark_video_failed(video_id, f"invalid_file:{verr}")
+                return False, []
+        dur_sec = float((vmeta or {}).get('duration_sec') or 0)
+        end_ts = start_ts + int(dur_sec) if dur_sec > 0 else start_ts + 3600
+        clip_ids = self._segment_motion_clips(local_path, start_ts, end_ts, camera_name)
+        if self.db.has_unfinished_motion_in_range(start_ts, end_ts):
+            # 仍有进行中的事件（duration 未定型）：保持 pending，producer 下轮重试
+            logger.info(f"[video_id={video_id}] 素材仍有未结束运动事件，保持 pending 下轮再分割")
+            return True, clip_ids
+        self.db.mark_video_processed(
+            video_id, f"（整段素材已分割 {len(clip_ids)} 段运动片段）", [], [], 'motion_segment')
+        return True, clip_ids
+
+    def _segment_motion_clips(self, src_path: str, start_ts: int, end_ts: int,
+                              camera_name: str) -> List[int]:
+        """按窗口内已结束运动事件从整段素材分割运动片段。返回片段 video_id 列表。
+
+        幂等：已按 motion_event_id 生成过片段的跳过；片段文件已存在的跳过分割。
+        """
+        clips: List[int] = []
+        for me in self.db.get_motion_events_in_range(
+                start_ts, end_ts, finished_grace_sec=self.motion_grace_sec):
+            eid = me['event_id']
+            if self.db.get_video_by_motion_event_id(eid):
+                continue
+            dur = float(me.get('duration') or 0)
+            if dur < self.motion_min_duration:
+                continue
+            offset = float(me['start_time'] - start_ts)
+            if offset < -5 or offset > (end_ts - start_ts) + 5:
+                continue  # 事件窗口与素材窗口无重叠
+            offset = max(0.0, offset)
+            os.makedirs(self.motion_clips_dir, exist_ok=True)
+            clip_fn = f"motion_{eid}_{me['start_time']}.mp4"
+            clip_path = os.path.join(self.motion_clips_dir, clip_fn)
+            if not os.path.isfile(clip_path):
+                if not self._run_segment_ffmpeg(src_path, offset, dur, clip_path):
+                    logger.error(f"运动片段分割失败: {clip_fn}")
+                    continue
+            start_iso = datetime.fromtimestamp(
+                me['start_time'], tz=timezone(timedelta(hours=8))
+            ).strftime('%Y-%m-%d %H:%M:%S')
+            cid = self.db.ensure_video(
+                filename=clip_fn, local_path=clip_path, camera_name=camera_name,
+                event_start_time=start_iso, duration_sec=dur,
+                drive_file_id=f"motion_{eid}", motion_event_id=eid,
+                camera_id=me.get('camera_id'))
+            self.db.record_activity('segment', 'clip', f"{clip_fn} (event={eid})")
+            clips.append(cid)
+        return clips
+
+    def _run_segment_ffmpeg(self, src_path: str, offset: float, dur: float,
+                            out_path: str) -> bool:
+        """ffmpeg 从整段素材切运动片段：-ss/-t 定位，视频 copy 免转码，
+        音频按配置保留（pcm_alaw -> aac 转码）或丢弃（-an）。"""
+        import shutil
+        ffmpeg = shutil.which('ffmpeg') or 'ffmpeg'
+        args = ['-y', '-hide_banner', '-loglevel', 'error',
+                '-ss', f'{offset:.3f}', '-i', src_path, '-t', f'{dur:.3f}',
+                '-c:v', 'copy']
+        if self.motion_keep_audio:
+            args += ['-c:a', 'aac', '-b:a', '64k']
+        else:
+            args += ['-an']
+        args += ['-avoid_negative_ts', 'make_zero', out_path]
+        try:
+            proc = subprocess.run(args, capture_output=True, timeout=300)
+            if proc.returncode != 0:
+                logger.error(f"ffmpeg 分割失败: "
+                             f"{proc.stderr.decode(errors='ignore')[:200]}")
+                return False
+            return os.path.isfile(out_path) and os.path.getsize(out_path) > 0
+        except (subprocess.TimeoutExpired, OSError) as e:
+            logger.error(f"ffmpeg 分割异常: {e}")
+            return False
 
     def _store_result(self, video_id: int, result: Dict):
         events = result.get('events', [])

fam-edge/src/fam_edge/video_queue.py

@@ -214,9 +214,12 @@ class VideoQueue:
                              "started_at": None}
         self.db.record_activity('queue', 'process_start', f"video {video_id} {fn}")
         try:
-            ok = processor.process_video(
+            ok, clip_ids = processor.process_video(
                 video_id, fn, row['local_path'],
                 timeout_multiplier=self.timeout_multiplier)
+            # 素材整段视频分割出的运动片段入队分析（片段不在监听目录，需手动入队）
+            for cid in (clip_ids or []):
+                self._enqueue(cid)
             if ok:
                 self._stats["consumed_ok"] += 1
                 self.db.record_activity('queue', 'process_done', f"video {video_id} {fn}")