feat(监控): 实时服务状态界面 - Oracle 新增 service_activity 活动表(只保留7天，写入时清理)与 /api/oracle/activity 接口(队列实时状态/当前处理视频/模型调用/rclone/人物合并/最近50条活动)；VideoQueue 打点+当前处理跟踪+status()；PersonService 打点；rclone_sync.sh 同步结果写活动表；fam-ui 新增'🖥 服务状态'页(状态卡+活动时间流，直连 Oracle 实时拉取)

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

@@ -52,6 +52,11 @@ def _sync_token():
     return _SYNC_TOK or ''
 
 
+def _now_iso_str() -> str:
+    from datetime import datetime, timezone, timedelta
+    return datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S')
+
+
 @api_bp.route('/api/oracle/sync', methods=['GET'])
 def sync_pull():
     """NAS 拉取增量数据。since=ISO 时间字符串（默认 '' 拉全量）。
@@ -145,6 +150,45 @@ def event_thumb(event_id):
     return send_file(path, mimetype='image/jpeg')
 
 
+@api_bp.route('/api/oracle/activity', methods=['GET'])
+def activity():
+    """实时服务状态 + 最近活动流（token 校验）。
+
+    返回各服务当前状态（队列/当前处理视频/rclone 最近同步/人物合并/模型调用）
+    与最近 50 条活动（service_activity，只保留 7 天）。
+    """
+    if not _check_token():
+        return jsonify({"error": "unauthorized"}), 401
+    db = state.get_db()
+    queue = state.get_queue()
+    # 队列实时状态
+    q_status = None
+    if queue is not None:
+        try:
+            q_status = queue.status()
+        except Exception as e:
+            logger.warning(f"queue.status 异常: {e}")
+    # 各服务最近活动（rclone / person / model 切换）
+    def _last_activity(service):
+        row = db._conn.execute(
+            "SELECT service, action, detail, ts FROM service_activity "
+            "WHERE service=? ORDER BY id DESC LIMIT 1", (service,)).fetchone()
+        return dict(row) if row else None
+    # 最近模型调用（实时模型卡）
+    model_calls = db._conn.execute(
+        "SELECT id, provider, model, filename, started_at, duration_sec, "
+        "success, error FROM model_calls ORDER BY id DESC LIMIT 5").fetchall()
+    return jsonify({
+        "queue": q_status,
+        "db": db.get_queue_status(),
+        "rclone": _last_activity('rclone'),
+        "person": _last_activity('person'),
+        "model_calls": [dict(m) for m in model_calls],
+        "activities": db.get_recent_activities(50),
+        "ts": _now_iso_str(),
+    }), 200
+
+
 @api_bp.route('/health', methods=['GET'])
 def health():
     """健康检查"""

fam-edge/src/fam_edge/app.py

@@ -38,6 +38,7 @@ try:
     db = state.get_db()
     _queue = VideoQueue(db)
     _queue.start()
+    state.set_queue(_queue)
     logger.info("VideoQueue 已启动")
 
     _person = PersonService(db)

fam-edge/src/fam_edge/oracle_db.py

@@ -99,9 +99,17 @@ class OracleDB:
             error TEXT,
             created_at TEXT
         );
+        CREATE TABLE IF NOT EXISTS service_activity (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            service TEXT,
+            action TEXT,
+            detail TEXT,
+            ts 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);
         """)
         # 兼容旧库：补 retry_count / file_valid / media 等列（生产-消费队列用）
         cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()]
@@ -141,6 +149,53 @@ class OracleDB:
              duration_sec, 1 if success else 0, error or '', now))
         self._conn.commit()
 
+    # ------------------------------------------------------------------
+    # 服务活动日志（实时状态界面用；只保留最近 7 天）
+    # ------------------------------------------------------------------
+    def record_activity(self, service: str, action: str, detail: str = ''):
+        """记录一条服务活动（queue/rclone/person/model...）。
+
+        写入时顺带清理 7 天前的旧记录（用户要求只保留最近七天）。
+        """
+        now = _now_iso()
+        with self._write_lock:
+            self._conn.execute(
+                "INSERT INTO service_activity (service, action, detail, ts) "
+                "VALUES (?,?,?,?)",
+                (service, action, str(detail or '')[:500], now))
+            # 只保留最近 7 天
+            self._conn.execute(
+                "DELETE FROM service_activity WHERE ts < ?",
+                ((datetime.now(timezone(timedelta(hours=8))) - timedelta(days=7))
+                 .strftime('%Y-%m-%d %H:%M:%S'),))
+            self._conn.commit()
+
+    def get_recent_activities(self, limit: int = 50) -> List[Dict]:
+        """最近活动（时间倒序）。"""
+        rows = self._conn.execute(
+            "SELECT id, service, action, detail, ts FROM service_activity "
+            "ORDER BY id DESC LIMIT ?", (int(limit),)).fetchall()
+        return [dict(r) for r in rows]
+
+    def get_queue_status(self) -> Dict:
+        """实时队列/处理状态（前端服务状态卡用）。"""
+        total = self._conn.execute("SELECT COUNT(*) c FROM videos").fetchone()['c']
+        by_status = {}
+        for r in self._conn.execute(
+                "SELECT status, COUNT(*) c FROM videos GROUP BY status").fetchall():
+            by_status[r['status']] = r['c']
+        # 最近处理的视频（done/failed，按 updated_at 倒序）
+        recent = self._conn.execute(
+            "SELECT id, filename, status, compute_provider, updated_at, "
+            "processed_at, event_start_time FROM videos "
+            "ORDER BY COALESCE(updated_at, created_at) DESC LIMIT 5"
+        ).fetchall()
+        return {
+            "total": total,
+            "by_status": by_status,
+            "recent": [dict(r) for r in recent],
+        }
+
     def set_video_file_status(self, video_id: int, valid: bool,
                               error: str = '', media_meta: dict = None):
         """登记/更新文件校验结果：valid / file_error / media_meta_json"""

fam-edge/src/fam_edge/person_service.py

@@ -63,6 +63,7 @@ class PersonService:
         unnamed = [r for r in rows if not r['canonical_name'] or r['canonical_name'] == r['label']]
         if not unnamed:
             logger.info("PersonService: 无待合并人物，跳过 LLM 合并")
+            self.db.record_activity('person', 'merge_skip', f"已校准 {len(label_videos)} 个标签，无待合并")
             return
 
         samples = self._collect_descriptions([r['label'] for r in unnamed])
@@ -80,6 +81,9 @@ class PersonService:
                 resolved = label_to_canonical.get(canonical, canonical)
                 self.db.set_canonical(label, resolved, source='llm')
                 updated += 1
+        self.db.record_activity(
+            'person', 'merge_done',
+            f"校准 {len(label_videos)} 标签，LLM 合并更新 {updated} 条（{', '.join(list(mapping)[:6])}）")
         logger.info(f"PersonService: LLM 合并完成，更新 {updated} 条")
 
     def _collect_descriptions(self, labels: List[str]) -> Dict[str, List[str]]:

fam-edge/src/fam_edge/state.py

@@ -1,5 +1,5 @@
 """
-state - 进程内共享单例（OracleDB 实例）
+state - 进程内共享单例（OracleDB / VideoQueue 实例）
 
 video_queue / person_service / api_gateway 都通过 get_db() 访问同一个 SQLite 连接，
 避免重复打开与循环 import。
@@ -8,6 +8,7 @@ from . import oracle_db
 from .config_loader import load_config
 
 _db = None
+_queue = None
 
 
 def get_db() -> oracle_db.OracleDB:
@@ -17,3 +18,13 @@ def get_db() -> oracle_db.OracleDB:
         path = cfg.get('oracle_db', {}).get('path', '/opt/fam-edge/data/oracle.db')
         _db = oracle_db.OracleDB(path)
     return _db
+
+
+def set_queue(q):
+    """注册 VideoQueue 实例（app 启动时调用，api_gateway 读取实时状态）。"""
+    global _queue
+    _queue = q
+
+
+def get_queue():
+    return _queue

fam-edge/src/fam_edge/video_queue.py

@@ -58,6 +58,9 @@ class VideoQueue:
         self._producer = None
         self._consumers: List[threading.Thread] = []
         self._stats = {"produced": 0, "consumed_ok": 0, "consumed_fail": 0}
+        # 实时状态：当前正在处理的视频（服务状态界面用）
+        self._current = None
+        self._current_lock = threading.Lock()
 
     # ------------------------------------------------------------------
     # 生产者
@@ -97,16 +100,19 @@ class VideoQueue:
                         vid = self.db.ensure_video(fn, path, camera_name=self.camera_name)
                         self.db.set_video_file_status(vid, True, '', vmeta)
                         logger.info(f"登记新视频并入队: {fn} (id={vid}, meta={vmeta})")
+                        self.db.record_activity('queue', 'register', f"{fn} (id={vid})")
                         self._enqueue(vid)
                         continue
                 vid = self.db.ensure_video(fn, path, camera_name=self.camera_name)
                 logger.info(f"登记新视频并入队: {fn} (id={vid})")
+                self.db.record_activity('queue', 'register', f"{fn} (id={vid})")
                 self._enqueue(vid)
             elif row['status'] in ('pending', 'failed') and self._retry_allowed(row):
                 self._enqueue(row['id'])
             elif row['status'] == 'done' and self._file_changed(row, path):
                 # 文件被覆盖（rclone 重新同步/更新）：重置 pending 重新分析
                 logger.info(f"文件内容变更，重置重新分析: {fn} (id={row['id']})")
+                self.db.record_activity('queue', 'reanalyze', f"{fn} (id={row['id']})")
                 self.db._conn.execute(
                     "UPDATE videos SET status='pending', retry_count=0, summary_json=NULL, "
                     "events_json=NULL, people_json=NULL, compute_provider=NULL, "
@@ -209,13 +215,38 @@ class VideoQueue:
         if not self._retry_allowed(row):
             logger.warning(f"[video_id={video_id}] 已达重试上限({row['retry_count']})，放弃")
             return
-        ok = processor.process_video(
-            video_id, row['filename'], row['local_path'],
-            timeout_multiplier=self.timeout_multiplier)
-        if ok:
-            self._stats["consumed_ok"] += 1
-        else:
-            self._stats["consumed_fail"] += 1
+        fn = row['filename'] or ''
+        with self._current_lock:
+            self._current = {"video_id": video_id, "filename": fn,
+                             "started_at": None}
+        self.db.record_activity('queue', 'process_start', f"video {video_id} {fn}")
+        try:
+            ok = processor.process_video(
+                video_id, fn, row['local_path'],
+                timeout_multiplier=self.timeout_multiplier)
+            if ok:
+                self._stats["consumed_ok"] += 1
+                self.db.record_activity('queue', 'process_done', f"video {video_id} {fn}")
+            else:
+                self._stats["consumed_fail"] += 1
+                self.db.record_activity('queue', 'process_fail', f"video {video_id} {fn}")
+        finally:
+            with self._current_lock:
+                self._current = None
+
+    # ------------------------------------------------------------------
+    # 实时状态（服务状态界面用）
+    # ------------------------------------------------------------------
+    def status(self) -> Dict:
+        with self._current_lock:
+            cur = dict(self._current) if self._current else None
+        return {
+            "queued": self._queue.qsize(),
+            "current": cur,
+            "stats": dict(self._stats),
+            "max_concurrent": self.max_concurrent,
+            "running": (self._producer is not None and self._producer.is_alive()),
+        }
 
     # ------------------------------------------------------------------
     # 生命周期

fam-ui/src/app.py

@@ -329,7 +329,7 @@ st.sidebar.markdown(
 # 顶部横向导航
 page = st.segmented_control(
     "功能页面",
-    ["🕒 事件时间轴", "💬 AI 对话", "📝 对话历史", "👤 人物管理", "📈 统计图表", "🤖 模型统计"],
+    ["🕒 事件时间轴", "💬 AI 对话", "📝 对话历史", "👤 人物管理", "📈 统计图表", "🤖 模型统计", "🖥 服务状态"],
     default="🕒 事件时间轴",
     label_visibility="collapsed"
 ) or "🕒 事件时间轴"
@@ -1032,3 +1032,159 @@ elif page == "🤖 模型统计":
     finally:
         if conn:
             conn.close()
+
+
+# ============================================================
+# 服务状态页（实时查看各服务在做什么：队列/模型/rclone/人物/NAS 同步）
+# ============================================================
+elif page == "🖥 服务状态":
+    page_header('🖥', '服务状态', '各服务实时动态 · 最近 7 天活动记录')
+
+    c1, c2, c3 = st.columns([1, 1, 5])
+    with c1:
+        if st.button("🔄 刷新", use_container_width=True):
+            st.rerun()
+    with c2:
+        st.caption("自动每 5 分钟更新页面（点击刷新立即更新）")
+
+    # ---- 拉取 Oracle 实时状态 + NAS 同步状态 ----
+    act_data = None
+    act_err = None
+    if _oracle_url and _oracle_token:
+        try:
+            r = requests.get(
+                f"{_oracle_url}/api/oracle/activity?token={_oracle_token}",
+                timeout=15)
+            if r.status_code == 200:
+                act_data = r.json()
+            else:
+                act_err = f"Oracle activity HTTP {r.status_code}"
+        except Exception as e:
+            act_err = f"连接 Oracle 失败: {e}"
+    else:
+        act_err = "未配置 oracle_url / oracle_token"
+
+    nas_sync = None
+    try:
+        rr = requests.get(f"{_core_url}/api/status", timeout=10)
+        if rr.status_code == 200:
+            nas_sync = rr.json().get('sync', {})
+    except Exception:
+        pass
+
+    if act_err:
+        st.warning(act_err)
+    if act_data is None and nas_sync is None:
+        st.markdown('<div class="fam-empty">暂时无法获取服务状态</div>', unsafe_allow_html=True)
+    else:
+        # ---- 服务状态卡 ----
+        st.markdown('<div style="font-size:15px;font-weight:700;color:#f1f5f9;'
+                    'margin:6px 0 12px 0;">各服务当前状态</div>', unsafe_allow_html=True)
+
+        q = (act_data or {}).get('queue') or {}
+        dbinfo = (act_data or {}).get('db') or {}
+        cur_vid = q.get('current')
+        cur_txt = '—'
+        if cur_vid:
+            cur_txt = f"#{cur_vid.get('video_id')} {esc((cur_vid.get('filename') or '')[-42:])}"
+        by_status = dbinfo.get('by_status') or {}
+        st.markdown(
+            f'<div class="stat-row">'
+            f'<div class="stat-card"><div class="s-val">{"运行中" if q.get("running") else "停止"}</div>'
+            f'<div class="s-label">FAM-Edge 队列</div></div>'
+            f'<div class="stat-card"><div class="s-val">{q.get("queued", 0)}</div>'
+            f'<div class="s-label">排队中</div></div>'
+            f'<div class="stat-card ok"><div class="s-val">{by_status.get("done", 0)}</div>'
+            f'<div class="s-label">已完成</div></div>'
+            f'<div class="stat-card warn"><div class="s-val">{by_status.get("pending", 0)}</div>'
+            f'<div class="s-label">待处理</div></div>'
+            f'<div class="stat-card err"><div class="s-val">{by_status.get("failed", 0)}</div>'
+            f'<div class="s-label">失败</div></div>'
+            f'</div>', unsafe_allow_html=True)
+
+        # 详细状态行
+        qs = q.get('stats') or {}
+        st.markdown(
+            f'<div style="font-size:12px;color:#8b93a7;margin:8px 0 4px 0;">'
+            f'▶ 正在处理：<b style="color:#7dd3fc;">{cur_txt}</b>'
+            f'<span style="margin-left:14px;">生产者已入队 {qs.get("produced", 0)} · '
+            f'成功 {qs.get("consumed_ok", 0)} · 失败 {qs.get("consumed_fail", 0)}</span></div>',
+            unsafe_allow_html=True)
+
+        # 服务卡：rclone / 人物 / NAS 同步 / 模型
+        def _svc_card(icon, name, val, sub, color='#cbd5e1'):
+            return (
+                f'<div style="flex:1;min-width:180px;background:#0f172a;border:1px solid #1e293b;'
+                f'border-radius:12px;padding:12px 14px;margin:6px 6px 6px 0;">'
+                f'<div style="font-size:12px;color:#64748b;">{icon} {esc(name)}</div>'
+                f'<div style="font-size:13px;color:{color};margin-top:6px;line-height:1.5;">{val}</div>'
+                f'<div style="font-size:11px;color:#475569;margin-top:4px;">{esc(sub)}</div>'
+                f'</div>')
+
+        rclone = (act_data or {}).get('rclone') or {}
+        person = (act_data or {}).get('person') or {}
+        cards = []
+        # rclone
+        if rclone:
+            cards.append(_svc_card('🔄', 'rclone 同步',
+                                   f"{esc((rclone.get('detail') or '')[:70])}",
+                                   f"最近 {(rclone.get('ts') or '')[:19]}", '#4ade80'))
+        else:
+            cards.append(_svc_card('🔄', 'rclone 同步', '暂无记录', '—'))
+        # 人物合并
+        if person:
+            cards.append(_svc_card('👤', '人物合并',
+                                   f"{esc((person.get('action') or ''))}",
+                                   f"{(person.get('ts') or '')[:19]} · {esc((person.get('detail') or '')[:46])}", '#c084fc'))
+        else:
+            cards.append(_svc_card('👤', '人物合并', '暂无记录', '—'))
+        # NAS 同步
+        if nas_sync:
+            cnt = nas_sync.get('last_count') or [0, 0, 0, 0]
+            cards.append(_svc_card('📡', 'NAS 同步',
+                                   f"游标 {esc(str(nas_sync.get('cursor') or '')[:19])}",
+                                   f"最近 {(str(nas_sync.get('last_sync_at') or ''))[:19]} · 增量 V{cnt[0]} E{cnt[1]} P{cnt[2]} M{cnt[3]}",
+                                   '#fbbf24'))
+        else:
+            cards.append(_svc_card('📡', 'NAS 同步', '不可达', '—'))
+        # 模型
+        mcs = (act_data or {}).get('model_calls') or []
+        if mcs:
+            m0 = mcs[0]
+            cards.append(_svc_card('🧠', '云端模型',
+                                   f"最近：{esc(m0.get('model') or '')} {'✅' if m0.get('success') else '❌ ' + esc((m0.get('error') or '')[:30])}",
+                                   f"{(m0.get('started_at') or '')[:19]} · 耗时 {round(m0.get('duration_sec') or 0, 1)}s · 近5次 成功{sum(1 for m in mcs if m.get('success'))}/{len(mcs)}",
+                                   '#4ade80'))
+        else:
+            cards.append(_svc_card('🧠', '云端模型', '暂无调用', '—'))
+        st.markdown(
+            f'<div style="display:flex;flex-wrap:wrap;gap:2px;">{"".join(cards)}</div>',
+            unsafe_allow_html=True)
+
+        # ---- 活动时间流 ----
+        acts = (act_data or {}).get('activities') or []
+        st.markdown('<div style="font-size:15px;font-weight:700;color:#f1f5f9;'
+                    'margin:24px 0 12px 0;">最近活动</div>', unsafe_allow_html=True)
+        if not acts:
+            st.markdown('<div class="fam-empty">暂无活动记录（服务刚启动？）</div>',
+                        unsafe_allow_html=True)
+        else:
+            _svc_badge = {
+                'queue': ('<span class="bdg" style="background:#0e2233;border:1px solid #164e63;color:#7dd3fc;">队列</span>'),
+                'rclone': ('<span class="bdg" style="background:#0c1f17;border:1px solid #14532d;color:#4ade80;">同步</span>'),
+                'person': ('<span class="bdg" style="background:#1f122e;border:1px solid #4a1d96;color:#c084fc;">人物</span>'),
+            }
+            items = []
+            for a in acts[:50]:
+                badge = _svc_badge.get(a.get('service'), f"<span class='bdg'>{esc(a.get('service'))}</span>")
+                ts = str(a.get('ts') or '')[:19]
+                items.append(
+                    f'<div class="ev-item" style="margin-bottom:5px;">'
+                    f'<div class="ev-time" style="min-width:110px;">{esc(ts)}</div>'
+                    f'<div class="ev-card" style="padding:6px 10px;">{badge}'
+                    f'<span style="font-weight:600;color:#e2e8f0;margin:0 6px;">{esc(a.get("action"))}</span>'
+                    f'<span style="font-size:12px;color:#94a3b8;">{esc(a.get("detail") or "")}</span>'
+                    f'</div></div>')
+            st.markdown(
+                f'<div class="ev-list">{"".join(items)}</div>', unsafe_allow_html=True)
+            st.caption(f'共 {len(acts)} 条记录（活动日志仅保留最近 7 天）')