diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py index b4c4e9f..fdd680d 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/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(): """健康检查""" diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py index 9359ac4..3d4136f 100644 --- a/fam-edge/src/fam_edge/app.py +++ b/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) diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index 0de774a..dc48ddf 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/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""" diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index 4ed8675..0acead0 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/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]]: diff --git a/fam-edge/src/fam_edge/state.py b/fam-edge/src/fam_edge/state.py index dee65a2..8597cb2 100644 --- a/fam-edge/src/fam_edge/state.py +++ b/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 diff --git a/fam-edge/src/fam_edge/video_queue.py b/fam-edge/src/fam_edge/video_queue.py index 3fa432f..bd52519 100644 --- a/fam-edge/src/fam_edge/video_queue.py +++ b/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()), + } # ------------------------------------------------------------------ # 生命周期 diff --git a/fam-ui/src/app.py b/fam-ui/src/app.py index 3181370..4008e0f 100644 --- a/fam-ui/src/app.py +++ b/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('
暂时无法获取服务状态
', unsafe_allow_html=True) + else: + # ---- 服务状态卡 ---- + st.markdown('
各服务当前状态
', 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'
' + f'
{"运行中" if q.get("running") else "停止"}
' + f'
FAM-Edge 队列
' + f'
{q.get("queued", 0)}
' + f'
排队中
' + f'
{by_status.get("done", 0)}
' + f'
已完成
' + f'
{by_status.get("pending", 0)}
' + f'
待处理
' + f'
{by_status.get("failed", 0)}
' + f'
失败
' + f'
', unsafe_allow_html=True) + + # 详细状态行 + qs = q.get('stats') or {} + st.markdown( + f'
' + f'▶ 正在处理：{cur_txt}' + f'生产者已入队 {qs.get("produced", 0)} · ' + f'成功 {qs.get("consumed_ok", 0)} · 失败 {qs.get("consumed_fail", 0)}
', + unsafe_allow_html=True) + + # 服务卡：rclone / 人物 / NAS 同步 / 模型 + def _svc_card(icon, name, val, sub, color='#cbd5e1'): + return ( + f'
' + f'
{icon} {esc(name)}
' + f'
{val}
' + f'
{esc(sub)}
' + f'
') + + 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'
{"".join(cards)}
', + unsafe_allow_html=True) + + # ---- 活动时间流 ---- + acts = (act_data or {}).get('activities') or [] + st.markdown('
最近活动
', unsafe_allow_html=True) + if not acts: + st.markdown('
暂无活动记录（服务刚启动？）
', + unsafe_allow_html=True) + else: + _svc_badge = { + 'queue': ('队列'), + 'rclone': ('同步'), + 'person': ('人物'), + } + items = [] + for a in acts[:50]: + badge = _svc_badge.get(a.get('service'), f"{esc(a.get('service'))}") + ts = str(a.get('ts') or '')[:19] + items.append( + f'
' + f'
{esc(ts)}
' + f'
{badge}' + f'{esc(a.get("action"))}' + f'{esc(a.get("detail") or "")}' + f'
') + st.markdown( + f'
{"".join(items)}
', unsafe_allow_html=True) + st.caption(f'共 {len(acts)} 条记录（活动日志仅保留最近 7 天）')