diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index 19c768a..e83768a 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -47,18 +47,24 @@ video_processing: # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed vision_order: ["gemini", "nvidia"] -# 运动侦测预过滤（2026-08-22 重构，2026-08-22 补心跳）： -# - 甲骨文【不再反向访问 NAS】。原 dsm_motion_client（甲骨文主动查 SS API）已停用， -# 改由 NAS 端 fam-core 的 MotionNotifier 接收 SS Webhook 后，主动 POST 推送到 -# 甲骨文 /api/ss/motion，落库 ss_motion_events；NAS 侧还会定期用空 events 单纯 -# 调一次这个接口当心跳（不要求真的有事件），证明推送链路还活着。 -# - video_processor 分析前调用 db.has_motion_in_range_local()（本地运动事件表）做 -# 预过滤：窗口内无运动事件则跳过云端分析（compute_provider=skipped_no_motion）。 -# - 安全阀门: 距上次心跳超过 max_heartbeat_age_sec 没更新（NAS 服务挂了/网络断了/ -# DSM Webhook 规则被误关，历史数据表非空但推送链路已经死了），一律 fail-open -# （照常分析），不能仅凭"表里曾经有数据"就信任"这段时间确认无运动"的结论。 +# 运动事件（2026-08-22 运动事件驱动架构）： +# - NAS 端 fam-core MotionNotifier 轮询 SS 事件后，POST 推送到甲骨文 +# /api/ss/motion，落库 ss_motion_events（start_time/duration 为 Unix epoch）； +# 并定期用空 events 调接口当心跳，证明推送链路存活。 +# - video_processor 不再整段分析：整段素材按 ss_motion_events 中【已结束】的 +# 运动事件分割成运动片段（motion_segment 配置），只分析运动片段。 dsm_motion_prefilter: - max_heartbeat_age_sec: 900 # 心跳超过 15 分钟没更新 -> 判定推送链路可能已死，fail-open + max_heartbeat_age_sec: 900 # 心跳超过 15 分钟没更新 -> 判定推送链路可能已死 + +# 运动片段分割（运动事件驱动架构，2026-08-22 新增）： +# 整段素材视频（rclone 同步落地）按 ss_motion_events 分割成运动片段再分析。 +# 只分割已结束事件（start_time+duration 落在当前时刻附近），进行中的事件 +# 等结束后的下一轮再分割；素材保持 pending 直到窗口内事件全部结束。 +motion_segment: + clips_dir: "/opt/fam-edge/motion_clips" # 分割产物目录 + keep_audio: true # 保留音频（pcm_alaw -> aac 64k 转码）；false 则 -an 去音频 + min_duration_sec: 1 # 短于该时长的事件不分割 + unfinished_grace_sec: 10 # start_time+duration 距当前 ≤ 该秒视为"已结束"容差 # 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama models: diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index 589074d..59e7ac7 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/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 diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index d7cf601..0c8e9b0 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/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', []) diff --git a/fam-edge/src/fam_edge/video_queue.py b/fam-edge/src/fam_edge/video_queue.py index 1ced703..5a7eae3 100644 --- a/fam-edge/src/fam_edge/video_queue.py +++ b/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}") diff --git a/fam-edge/tests/test_oracle_db.py b/fam-edge/tests/test_oracle_db.py index 8fc0742..8d47046 100644 --- a/fam-edge/tests/test_oracle_db.py +++ b/fam-edge/tests/test_oracle_db.py @@ -7,6 +7,13 @@ def _db(tmp_path): return OracleDB(str(tmp_path / "oracle.db")) +def _set_heartbeat_age(db, age_sec): + """把心跳时间戳直接改写成"距现在 age_sec 秒前"，用于测试新鲜度阈值边界。""" + from datetime import datetime, timedelta, timezone + ts = (datetime.now(timezone(timedelta(hours=8))) - timedelta(seconds=age_sec)) + db.set_cursor('motion_heartbeat_at', ts.strftime('%Y-%m-%d %H:%M:%S')) + + def test_upsert_person_same_gender_merges_into_one_row(tmp_path): db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男", "hair": "短发黑色"}) @@ -70,3 +77,107 @@ def test_upsert_person_no_features_never_triggers_split(tmp_path): rows = db._conn.execute("SELECT * FROM people").fetchall() assert len(rows) == 1 assert rows[0]["appearances"] == 2 + + +# ---------------------------------------------------------------------- +# 运动侦测事件（NAS 推送） +# ---------------------------------------------------------------------- + +def test_record_motion_events_upserts_by_event_id(tmp_path): + db = _db(tmp_path) + n = db.record_motion_events([ + {"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 5}, + {"event_id": 2, "camera_id": 2, "event_type": 10, "start_time": 2000, "duration": 3}, + ]) + assert n == 2 + rows = db._conn.execute("SELECT * FROM ss_motion_events ORDER BY event_id").fetchall() + assert len(rows) == 2 + # 重复推送同一个 event_id（幂等）应该更新而不是新增一行 + db.record_motion_events( + [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 99}]) + rows = db._conn.execute("SELECT * FROM ss_motion_events").fetchall() + assert len(rows) == 2 + updated = db._conn.execute( + "SELECT duration FROM ss_motion_events WHERE event_id=1").fetchone() + assert updated['duration'] == 99 + + +def test_record_motion_events_skips_missing_event_id(tmp_path): + db = _db(tmp_path) + n = db.record_motion_events([{"camera_id": 2, "start_time": 1000}]) + assert n == 0 + + +def test_heartbeat_age_none_when_never_recorded(tmp_path): + db = _db(tmp_path) + assert db.get_motion_heartbeat_age_sec() is None + + +def test_heartbeat_age_near_zero_right_after_recording(tmp_path): + db = _db(tmp_path) + db.record_motion_heartbeat() + age = db.get_motion_heartbeat_age_sec() + assert age is not None and age < 5 + + +def test_has_motion_in_range_local_fails_open_without_heartbeat(tmp_path): + """核心诉求: 从未收到过心跳（冷启动，NAS 推送链路还没接上）必须 fail-open， + 不能因为本地表是空的就悄悄跳过分析。""" + db = _db(tmp_path) + assert db.has_motion_in_range_local(1000, 2000) is None + + +def test_has_motion_in_range_local_fails_open_when_heartbeat_stale(tmp_path): + """核心诉求: 表里有大量历史运动事件（曾经推送链路是健康的），但心跳已经 + 过期太久（NAS 服务挂了/网络断了/DSM Webhook 规则被误关）——这时候不能信任 + "查询结果是 0 条 = 确认无运动"，必须当作链路已死，fail-open。""" + db = _db(tmp_path) + db.record_motion_events( + [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 500, "duration": 10}]) + _set_heartbeat_age(db, 1000) # 超过默认阈值 900s + assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None + + +def test_has_motion_in_range_local_trusts_result_when_heartbeat_fresh(tmp_path): + db = _db(tmp_path) + db.record_motion_heartbeat() + assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is False + db.record_motion_events( + [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 2500, "duration": 5}]) + assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is True + + +def test_has_motion_in_range_local_respects_heartbeat_boundary(tmp_path): + db = _db(tmp_path) + _set_heartbeat_age(db, 899) + assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is not None + _set_heartbeat_age(db, 901) + assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None + + +def test_has_motion_in_range_local_overlap_semantics(tmp_path): + """事件区间 [start_time, start_time+duration] 只要和查询窗口有重叠就算命中， + 不要求事件完全落在窗口内部（也不要求窗口完全覆盖事件）。""" + db = _db(tmp_path) + db.record_motion_heartbeat() + # 事件在窗口开始之前就开始，但持续到窗口内 -> 应该命中 + db.record_motion_events( + [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1990, "duration": 20}]) + assert db.has_motion_in_range_local(2000, 3000) is True + + +def test_has_motion_in_range_local_ignores_non_motion_event_type(tmp_path): + db = _db(tmp_path) + db.record_motion_heartbeat() + db.record_motion_events( + [{"event_id": 1, "camera_id": 2, "event_type": 99, "start_time": 2500, "duration": 5}]) + assert db.has_motion_in_range_local(2000, 3000) is False + + +def test_has_motion_in_range_local_filters_by_camera_id(tmp_path): + db = _db(tmp_path) + db.record_motion_heartbeat() + db.record_motion_events( + [{"event_id": 1, "camera_id": 99, "event_type": 10, "start_time": 2500, "duration": 5}]) + assert db.has_motion_in_range_local(2000, 3000, camera_id=2) is False + assert db.has_motion_in_range_local(2000, 3000, camera_id=99) is True