[功能] 事件-片段一致性对账 - /api/oracle/activity 加 segment.consistency(事件总数/已结束/已分割/缺口+明细/素材覆盖范围), 服务状态页展示一致性数字

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

@@ -163,6 +163,8 @@ def activity():
         segment['clips_files'] = len(os.listdir(seg_dir)) if os.path.isdir(seg_dir) else 0
     except Exception:
         segment['clips_files'] = 0
+    # 事件↔片段一致性对账（数量可验证）
+    segment['consistency'] = db.get_segment_consistency()
     return jsonify({
         "queue": q_status,
         "db": db.get_queue_status(),

fam-edge/src/fam_edge/oracle_db.py

@@ -398,6 +398,50 @@ class OracleDB:
             "last": dict(last) if last else None,
         }
 
+    def get_segment_consistency(self, gap_limit: int = 20) -> Dict:
+        """事件↔片段一致性对账（数量可验证）。
+
+        - event_total : ss_motion_events 全部运动事件（SS 已回灌的）
+        - finished    : 已结束且 duration>0（具备分割条件）的事件数
+        - segmented   : 已生成片段的去重事件数（videos.motion_event_id）
+        - gap_count   : 已结束但未生成片段（缺口 = finished - segmented 的下界）
+        - gaps        : 缺口明细（event_id/start_time/duration，北京时间）
+        - material_range: 素材文件覆盖的 event_start_time 范围（判断缺口是否因素材缺失）
+        """
+        now_ts = int(datetime.now(timezone(timedelta(hours=8))).timestamp())
+        fin = ("event_type = 10 AND COALESCE(duration,0) > 0 "
+               "AND (start_time + COALESCE(duration,0)) <= ?")
+        event_total = self._conn.execute(
+            "SELECT COUNT(*) c FROM ss_motion_events WHERE event_type=10"
+        ).fetchone()['c']
+        finished = self._conn.execute(
+            f"SELECT COUNT(*) c FROM ss_motion_events WHERE {fin}", (now_ts,)
+        ).fetchone()['c']
+        segmented = self._conn.execute(
+            "SELECT COUNT(DISTINCT motion_event_id) c FROM videos "
+            "WHERE motion_event_id IS NOT NULL"
+        ).fetchone()['c']
+        gaps = self._conn.execute(
+            f"""SELECT me.event_id, me.camera_id, me.start_time, me.duration
+                FROM ss_motion_events me
+                WHERE {fin}
+                  AND NOT EXISTS (SELECT 1 FROM videos v
+                                  WHERE v.motion_event_id = me.event_id)
+                ORDER BY me.start_time DESC LIMIT ?""",
+            (now_ts, int(gap_limit))).fetchall()
+        mrange = self._conn.execute(
+            "SELECT MIN(event_start_time) mn, MAX(event_start_time) mx "
+            "FROM videos WHERE motion_event_id IS NULL AND event_start_time != ''"
+        ).fetchone()
+        return {
+            "event_total": event_total,
+            "finished": finished,
+            "segmented": segmented,
+            "gap_count": max(0, finished - segmented),
+            "gaps": [dict(r) for r in gaps],
+            "material_range": {"min": mrange['mn'], "max": mrange['mx']} if mrange and mrange['mn'] else None,
+        }
+
     def set_video_file_status(self, video_id: int, valid: bool,
                               error: str = '', media_meta: dict = None):
         """登记/更新文件校验结果：valid / file_error / media_meta_json"""