feat(监控): 实时服务状态界面 - Oracle 新增 service_activity 活动表(只保留7天,写入时清理)与 /api/oracle/activity 接口(队列实时状态/当前处理视频/模型调用/rclone/人物合并/最近50条活动);VideoQueue 打点+当前处理跟踪+status();PersonService 打点;rclone_sync.sh 同步结果写活动表;fam-ui 新增'🖥 服务状态'页(状态卡+活动时间流,直连 Oracle 实时拉取)
This commit is contained in:
@@ -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():
|
||||
"""健康检查"""
|
||||
|
||||
@@ -38,6 +38,7 @@ try:
|
||||
db = state.get_db()
|
||||
_queue = VideoQueue(db)
|
||||
_queue.start()
|
||||
state.set_queue(_queue)
|
||||
logger.info("VideoQueue 已启动")
|
||||
|
||||
_person = PersonService(db)
|
||||
|
||||
@@ -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"""
|
||||
|
||||
@@ -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]]:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()),
|
||||
}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 生命周期
|
||||
|
||||
@@ -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 天)')
|
||||
|
||||
Reference in New Issue
Block a user