diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py index fe90442..c73b7b0 100644 --- a/fam-core/src/fam_core/db_layer.py +++ b/fam-core/src/fam_core/db_layer.py @@ -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() diff --git a/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py b/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py index 85fee74..b494bce 100644 --- a/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py +++ b/fam-edge/src/fam_edge/model_adapters/circuit_breaker.py @@ -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})" diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index e99c75e..0de774a 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -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() diff --git a/fam-edge/src/fam_edge/video_queue.py b/fam-edge/src/fam_edge/video_queue.py index 88458ee..3fa432f 100644 --- a/fam-edge/src/fam_edge/video_queue.py +++ b/fam-edge/src/fam_edge/video_queue.py @@ -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: