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

fix(锁与查询组): #12 熔断器状态转换加锁；#10 问答人名匹配改 JSON_CONTAINS 精确匹配(防 LIKE 子串/特殊字符误匹配)；#4 OracleDB 复合写加锁；#6 每消费者独立 VideoProcessor(消除 adapter.timeout 共享竞争)；#15 done 视频文件被覆盖(mtime>processed_at)自动重置重分析

Browse Source
This commit is contained in:
ericwyuan
2026-08-21 14:24:12 +08:00
parent 034dca92b2
commit 4e85a98944
4 changed files with 68 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
fam-core/src/fam_core/db_layer.py
View File

@@ -212,9 +212,11 @@ def query_sync_events_for_person_date(person: str, date_str: str) -> List[Dict]:
v.camera_name, v.filename, v.event_start_time, v.processed_at
FROM sync_events e
JOIN sync_videos v ON e.video_id = v.id
WHERE v.processed_at LIKE %s AND e.person_list_json LIKE %s
WHERE v.processed_at LIKE %s
AND e.person_list_json IS NOT NULL
AND JSON_CONTAINS(e.person_list_json, JSON_QUOTE(%s), '$')
ORDER BY v.processed_at ASC, e.ts ASC""",
(f'{date_str}%', f'%{person}%'))
(f'{date_str}%', person))
return cur.fetchall()
finally:
conn.close()

23
fam-edge/src/fam_edge/model_adapters/circuit_breaker.py
View File

@@ -8,6 +8,7 @@
"""
from collections import deque
import time
import threading
class CircuitBreaker:
@@ -21,27 +22,31 @@ class CircuitBreaker:
self.cooldown = cooldown
self.state = 'CLOSED'
self.last_failure = None
self._lock = threading.Lock() # 状态转换加锁，防多线程竞争
def record_failure(self):
if not self.enabled:
return
self.failures.append(time.time())
if len(self.failures) >= self.threshold:
self.state = 'OPEN'
self.last_failure = time.time()
with self._lock:
self.failures.append(time.time())
if len(self.failures) >= self.threshold:
self.state = 'OPEN'
self.last_failure = time.time()
def record_success(self):
if not self.enabled:
return
self.failures.clear()
self.state = 'CLOSED'
with self._lock:
self.failures.clear()
self.state = 'CLOSED'
def is_open(self):
if not self.enabled:
return False
if self.state == 'OPEN' and self.last_failure and time.time() - self.last_failure > self.cooldown:
self.state = 'HALF_OPEN'
return self.state == 'OPEN'
with self._lock:
if self.state == 'OPEN' and self.last_failure and time.time() - self.last_failure > self.cooldown:
self.state = 'HALF_OPEN'
return self.state == 'OPEN'
def __repr__(self):
return f"CircuitBreaker(state={self.state}, enabled={self.enabled})"

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

@@ -17,6 +17,7 @@ Oracle 本地库（SQLite） - 视频摘要 / 事件 / 人物 存储
import os
import json
import sqlite3
import threading
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Optional
@@ -35,6 +36,7 @@ class OracleDB:
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA busy_timeout=10000")
self._write_lock = threading.Lock() # 复合写（如 DELETE+INSERT+commit）串行化
self._init_schema()
# ------------------------------------------------------------------
@@ -182,25 +184,26 @@ class OracleDB:
def mark_video_processed(self, video_id: int, summary: str, events: List[dict],
people: List[str], compute_provider: str) -> List[int]:
"""落库视频结果；返回新插入事件的 id 列表（与 events 参数一一对应，供事件截图用）"""
now = _now_iso()
self._conn.execute(
"UPDATE videos SET status='done', summary_json=?, events_json=?, "
"people_json=?, compute_provider=?, updated_at=?, processed_at=? WHERE id=?",
(summary, json.dumps(events, ensure_ascii=False), json.dumps(people, ensure_ascii=False),
compute_provider, now, now, video_id))
# 事件落独立表，便于 NAS 拉取
self._conn.execute("DELETE FROM events WHERE video_id=?", (video_id,))
event_ids: List[int] = []
for ev in events:
cur = self._conn.execute(
"INSERT INTO events (video_id, ts, description, person_list_json, "
"is_attention_event) VALUES (?,?,?,?,?)",
(video_id, ev.get('timestamp', ''), ev.get('description', ''),
json.dumps(ev.get('people', []), ensure_ascii=False),
1 if ev.get('is_attention_event') else 0))
event_ids.append(cur.lastrowid)
self._conn.commit()
return event_ids
with self._write_lock:
now = _now_iso()
self._conn.execute(
"UPDATE videos SET status='done', summary_json=?, events_json=?, "
"people_json=?, compute_provider=?, updated_at=?, processed_at=? WHERE id=?",
(summary, json.dumps(events, ensure_ascii=False), json.dumps(people, ensure_ascii=False),
compute_provider, now, now, video_id))
# 事件落独立表，便于 NAS 拉取
self._conn.execute("DELETE FROM events WHERE video_id=?", (video_id,))
event_ids: List[int] = []
for ev in events:
cur = self._conn.execute(
"INSERT INTO events (video_id, ts, description, person_list_json, "
"is_attention_event) VALUES (?,?,?,?,?)",
(video_id, ev.get('timestamp', ''), ev.get('description', ''),
json.dumps(ev.get('people', []), ensure_ascii=False),
1 if ev.get('is_attention_event') else 0))
event_ids.append(cur.lastrowid)
self._conn.commit()
return event_ids
def mark_video_failed(self, video_id: int, error: str = ''):
now = _now_iso()

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

@@ -32,7 +32,6 @@ class VideoQueue:
def __init__(self, db: oracle_db.OracleDB):
self.config = load_config()
self.db = db
self.processor = VideoProcessor(db)
self.local_dir = self.config.get('gdrive_sync', {}).get(
'local_dir', '/opt/fam-edge/gdrive_videos')
self.watch_interval = self.config.get('gdrive_sync', {}).get(
@@ -105,6 +104,28 @@ class VideoQueue:
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._conn.execute(
"UPDATE videos SET status='pending', retry_count=0, summary_json=NULL, "
"events_json=NULL, people_json=NULL, compute_provider=NULL, "
"processed_at=NULL, file_valid=1 WHERE id=?",
(row['id'],))
self.db._conn.commit()
self._enqueue(row['id'])
def _file_changed(self, row, path: str) -> bool:
"""判断视频文件在处理后被覆盖（mtime 晚于 processed_at）。"""
try:
proc = row['processed_at'] or ''
if not proc:
return False
from datetime import datetime
pt = datetime.strptime(proc[:19], '%Y-%m-%d %H:%M:%S').timestamp()
return os.path.getmtime(path) > pt + 5 # 5s 容差
except Exception:
return False
def _enqueue_existing(self):
"""启动时把 DB 中未处理完的视频补入队（进程重启恢复）"""
@@ -161,13 +182,16 @@ class VideoQueue:
# 消费者
# ------------------------------------------------------------------
def _consume_loop(self):
# 每消费者独立 VideoProcessor（各自 build_adapters），
# 避免多消费者共享 adapter 实例导致 timeout 等实例属性改写竞争
processor = VideoProcessor(self.db)
while not self._consumer_stop.is_set():
try:
video_id = self._queue.get(timeout=1)
except queue.Empty:
continue
try:
self._consume_one(video_id)
self._consume_one(video_id, processor)
except Exception as e:
logger.error(f"消费 video_id={video_id} 异常: {e}", exc_info=True)
finally:
@@ -175,7 +199,7 @@ class VideoQueue:
self._queued.discard(video_id)
self._queue.task_done()
def _consume_one(self, video_id: int):
def _consume_one(self, video_id: int, processor: VideoProcessor):
row = self.db.get_video_by_id(video_id)
if row is None:
logger.warning(f"消费到不存在的 video_id={video_id}，跳过")
@@ -185,7 +209,7 @@ class VideoQueue:
if not self._retry_allowed(row):
logger.warning(f"[video_id={video_id}] 已达重试上限({row['retry_count']})，放弃")
return
ok = self.processor.process_video(
ok = processor.process_video(
video_id, row['filename'], row['local_path'],
timeout_multiplier=self.timeout_multiplier)
if ok: