ericwyuan/sentinel-home-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
ericwyuan
2026-08-21 15:52:36 +08:00
parent da7631976d
commit 0d0a7f6ef8
7 changed files with 311 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
fam-edge/src/fam_edge/api_gateway/api_gateway.py
View File

@@ -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():
"""健康检查"""

1
fam-edge/src/fam_edge/app.py
View File

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

55
fam-edge/src/fam_edge/oracle_db.py
View File

@@ -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"""

4
fam-edge/src/fam_edge/person_service.py
View File

@@ -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]]:

13
fam-edge/src/fam_edge/state.py
View File

@@ -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

45
fam-edge/src/fam_edge/video_queue.py
View File

@@ -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()),
}
# ------------------------------------------------------------------
# 生命周期