diff --git a/README.md b/README.md index 33154c4..27e33a7 100644 --- a/README.md +++ b/README.md @@ -149,7 +149,7 @@ Orchestrator 视觉阶段按 `fallback` 模式顺序降级：Gemini → NVIDIA N | 模块 | 文件 | 职责 | |------|------|------| | API-Gateway | `api_gateway/api_gateway.py` | `GET /api/oracle/sync`（增量拉取，since+token 校验）；`POST /api/oracle/people/correct`（命名校正）；`POST /api/edge/chat/ask`（问答编排）；`GET /health` | -| Watch-Processor | `watch_processor.py` | 30s 轮询 rclone 同步落地目录，登记新视频，串行触发 Video-Processor | +| Video-Queue | `video_queue.py` | 生产-消费队列：生产者 30s 轮询 rclone 同步落地目录登记新视频入队（含重启恢复）；消费者（`max_concurrent` 个线程）取队列调 Video-Processor；模型超时 = 原配置 ×`timeout_multiplier`；失败重试上限 `max_retries` | | Video-Processor | `video_processor.py` | 按 `vision_order` 调适配器 `analyze_video`（整视频）；首个成功即落库 Oracle `videos`+`events`+`people`；全失败标 `failed` | | Person-Service | `person_service.py` | 汇总全量人物 → LLM 合并为规范名 → `set_canonical`；生成 `known_members_context` 回灌视频提示；manual 命名优先不被覆盖 | | OracleDB | `oracle_db.py` | SQLite：videos / events / people / sync_cursor；`get_sync_delta(since)` 增量导出 | diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index ece3294..09f0e17 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -38,8 +38,10 @@ person_service: # 视频处理 video_processing: - max_concurrent: 1 - timeout: 900 # 单视频分析超时（整视频上云较慢） + max_concurrent: 1 # 消费者线程数（串行处理，避免云端并发超额） + timeout: 900 # 兜底单视频分析超时 + timeout_multiplier: 2 # 模型消费超时倍数：在 models[i].timeout 原值上 ×2（大视频上传+分析耗时） + max_retries: 2 # 单视频失败最大重试次数（failed 且 retry_count 超限不再消费） # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed vision_order: ["gemini", "nvidia"] diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py index fb58059..9359ac4 100644 --- a/fam-edge/src/fam_edge/app.py +++ b/fam-edge/src/fam_edge/app.py @@ -1,9 +1,9 @@ """ -FAM-Edge 主应用 - Flask 单进程（新架构 v2） +FAM-Edge 主应用 - Flask 单进程（新架构 v2.1） 承载: - API-Gateway（同步拉取 / 命名校正 / 智能问答） - - WatchProcessor（监听 Google 硬盘同步落地目录，整视频分析） + - VideoQueue（生产-消费队列：同步落地新视频入队，独立消费者云端分析，模型超时 ×2） - PersonService（人物汇总合并，定时） """ import os @@ -16,7 +16,7 @@ from .config_loader import load_config from .logger import setup_logger from .api_gateway.api_gateway import api_bp from . import state -from .watch_processor import WatchProcessor +from .video_queue import VideoQueue from .person_service import PersonService logger = setup_logger('fam-edge.app') @@ -27,18 +27,18 @@ app.register_blueprint(api_bp) @app.route('/', methods=['GET']) def root(): - return jsonify({"service": "fam-edge", "version": "2.0", - "mode": "drive-sync + whole-video analysis"}), 200 + return jsonify({"service": "fam-edge", "version": "2.1", + "mode": "drive-sync + whole-video analysis (producer-consumer queue)"}), 200 -# 启动监听处理器 + 人物服务 -_watch = None +# 启动生产-消费队列 + 人物服务 +_queue = None _person = None try: db = state.get_db() - _watch = WatchProcessor(db) - _watch.start() - logger.info("WatchProcessor 已启动") + _queue = VideoQueue(db) + _queue.start() + logger.info("VideoQueue 已启动") _person = PersonService(db) _person.start() diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index 1145fab..2be2dba 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -34,6 +34,7 @@ class OracleDB: self._conn = sqlite3.connect(db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA busy_timeout=10000") self._init_schema() # ------------------------------------------------------------------ @@ -49,6 +50,7 @@ class OracleDB: duration_sec REAL, event_start_time TEXT, status TEXT DEFAULT 'pending', + retry_count INTEGER DEFAULT 0, summary_json TEXT, events_json TEXT, people_json TEXT, @@ -82,6 +84,10 @@ class OracleDB: CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at); CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id); """) + # 兼容旧库：补 retry_count 列（生产-消费队列重试上限用） + cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] + if 'retry_count' not in cols: + c.execute("ALTER TABLE videos ADD COLUMN retry_count INTEGER DEFAULT 0") self._conn.commit() # ------------------------------------------------------------------ @@ -91,6 +97,10 @@ class OracleDB: cur = self._conn.execute("SELECT * FROM videos WHERE filename=?", (filename,)) return cur.fetchone() + def get_video_by_id(self, video_id: int) -> Optional[sqlite3.Row]: + cur = self._conn.execute("SELECT * FROM videos WHERE id=?", (video_id,)) + return cur.fetchone() + def ensure_video(self, filename: str, local_path: str, camera_name: str = '', event_start_time: str = '', duration_sec: float = 0.0, drive_file_id: str = '') -> int: @@ -141,7 +151,8 @@ class OracleDB: def mark_video_failed(self, video_id: int, error: str = ''): now = _now_iso() self._conn.execute( - "UPDATE videos SET status='failed', summary_json=?, updated_at=? WHERE id=?", + "UPDATE videos SET status='failed', retry_count=retry_count+1, " + "summary_json=?, updated_at=? WHERE id=?", (error, now, video_id)) self._conn.commit() diff --git a/fam-edge/src/fam_edge/state.py b/fam-edge/src/fam_edge/state.py index 61c0059..dee65a2 100644 --- a/fam-edge/src/fam_edge/state.py +++ b/fam-edge/src/fam_edge/state.py @@ -1,7 +1,7 @@ """ state - 进程内共享单例（OracleDB 实例） -watch_processor / person_service / api_gateway 都通过 get_db() 访问同一个 SQLite 连接， +video_queue / person_service / api_gateway 都通过 get_db() 访问同一个 SQLite 连接， 避免重复打开与循环 import。 """ from . import oracle_db diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index 2b72afd..6c3dc58 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/fam-edge/src/fam_edge/video_processor.py @@ -64,8 +64,13 @@ class VideoProcessor: ordered.append(a) return ordered - def process_video(self, video_id: int, filename: str, local_path: str) -> bool: - """处理一个视频记录，返回是否成功。""" + def process_video(self, video_id: int, filename: str, local_path: str, + timeout_multiplier: float = 1.0) -> bool: + """处理一个视频记录，返回是否成功。 + + timeout_multiplier: 云端模型消费的超时放大倍数（如 2 = 在配置 timeout 上 ×2）。 + 每次调用前临时放大对应 adapter.timeout，调用后恢复，避免影响其他调用方。 + """ if not os.path.isfile(local_path): logger.error(f"[video_id={video_id}] 文件不存在，跳过: {local_path}") self.db.mark_video_failed(video_id, "file_missing") @@ -88,6 +93,12 @@ class VideoProcessor: last_err = "no_vision_adapter" for adapter in self._ordered_vision_adapters(): + # 按模型原配置超时 × multiplier（默认 1x；队列消费默认 2x） + orig_timeout = adapter.get_timeout() + if timeout_multiplier != 1.0: + adapter.timeout = int(orig_timeout * timeout_multiplier) + logger.info(f"[video_id={video_id}] {adapter.provider_name} 超时 " + f"{orig_timeout}s -> {adapter.timeout}s (×{timeout_multiplier})") try: logger.info(f"[video_id={video_id}] 尝试 {adapter.provider_name} 整视频分析") result = adapter.analyze_video(local_path, known, event_start) @@ -95,6 +106,8 @@ class VideoProcessor: logger.error(f"[video_id={video_id}] {adapter.provider_name} 异常: {e}") last_err = str(e) continue + finally: + adapter.timeout = orig_timeout if result: self._store_result(video_id, result) return True diff --git a/fam-edge/src/fam_edge/video_queue.py b/fam-edge/src/fam_edge/video_queue.py new file mode 100644 index 0000000..9f60424 --- /dev/null +++ b/fam-edge/src/fam_edge/video_queue.py @@ -0,0 +1,199 @@ +""" +video_queue - 视频生产-消费队列（新架构 v2.1） + +职责拆分: + 生产者(Producer): 轮询 rclone 同步落地目录 + - 新文件 -> 登记 videos 表(status=pending) -> 放入内存队列 + - 启动时把 DB 中未处理完的(pending / failed 且未超重试上限)补入队 + 消费者(Consumer): 独立工作线程(max_concurrent 个) + - 从队列取 video_id -> 云端模型整视频分析（模型超时 = 原配置 × timeout_multiplier） + - 成功 -> done；全部模型失败 -> failed（retry_count+1，超上限不再重试） + +与旧 WatchProcessor 的区别: 生产/消费解耦，消费有独立线程与重试上限， +模型调用超时按各模型原配置 ×2（应对大视频上传+分析耗时）。 +""" +import os +import time +import queue +import threading +from typing import List, Optional + +from .logger import setup_logger +from .config_loader import load_config +from . import oracle_db +from .video_processor import VideoProcessor + +logger = setup_logger('fam-edge.video_queue') + +VIDEO_EXTS = ('.mp4', '.mkv', '.avi', '.mov', '.ts') + + +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( + 'watch_interval_sec', 30) + self.camera_name = self.config.get('gdrive_sync', {}).get( + 'camera_name', '摄像头') + vp = self.config.get('video_processing', {}) + self.max_concurrent = int(vp.get('max_concurrent', 1)) + # 模型超时放大倍数：消费时在模型原配置 timeout 上 ×N + self.timeout_multiplier = float(vp.get('timeout_multiplier', 2.0)) + # 单视频失败最大重试次数（failed 且 retry_count>=max_retries 不再消费） + self.max_retries = int(vp.get('max_retries', 2)) + + self._queue: "queue.Queue[int]" = queue.Queue() + self._queued: set = set() # 已在队中的 video_id（防重复入队） + self._lock = threading.Lock() + self._producer_stop = threading.Event() + self._consumer_stop = threading.Event() + self._producer = None + self._consumers: List[threading.Thread] = [] + self._stats = {"produced": 0, "consumed_ok": 0, "consumed_fail": 0} + + # ------------------------------------------------------------------ + # 生产者 + # ------------------------------------------------------------------ + def _scan_files(self) -> List[str]: + """递归扫描监听目录，返回所有视频文件完整路径""" + if not os.path.isdir(self.local_dir): + logger.warning(f"监听目录不存在: {self.local_dir}") + return [] + out = [] + for root, _dirs, files in os.walk(self.local_dir): + for fn in sorted(files): + if fn.lower().endswith(VIDEO_EXTS): + out.append(os.path.join(root, fn)) + return out + + def _produce_once(self): + for path in self._scan_files(): + fn = os.path.basename(path) + row = self.db.get_video_by_filename(fn) + if row is None: + vid = self.db.ensure_video(fn, path, camera_name=self.camera_name) + logger.info(f"登记新视频并入队: {fn} (id={vid})") + self._enqueue(vid) + elif row['status'] in ('pending', 'failed') and self._retry_allowed(row): + self._enqueue(row['id']) + + def _enqueue_existing(self): + """启动时把 DB 中未处理完的视频补入队（进程重启恢复）""" + rows = self.db.get_pending_videos(limit=10000) + for row in rows: + if self._retry_allowed(row): + self._enqueue(row['id']) + if rows: + logger.info(f"启动恢复入队 {len(rows)} 个待处理视频") + + def _retry_allowed(self, row) -> bool: + if row['status'] == 'pending': + return True + # failed：未超重试上限才允许再次消费 + return int(row['retry_count'] or 0) < self.max_retries + + def _enqueue(self, video_id: int): + with self._lock: + if video_id in self._queued: + return + self._queued.add(video_id) + self._queue.put(video_id) + self._stats["produced"] += 1 + + def _producer_loop(self): + logger.info(f"生产者启动，监听 {self.local_dir}，间隔 {self.watch_interval}s") + while not self._producer_stop.is_set(): + try: + self._produce_once() + except Exception as e: + logger.error(f"生产者扫描异常: {e}", exc_info=True) + for _ in range(self.watch_interval): + if self._producer_stop.is_set(): + break + time.sleep(1) + + # ------------------------------------------------------------------ + # 消费者 + # ------------------------------------------------------------------ + def _consume_loop(self): + 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) + except Exception as e: + logger.error(f"消费 video_id={video_id} 异常: {e}", exc_info=True) + finally: + with self._lock: + self._queued.discard(video_id) + self._queue.task_done() + + def _consume_one(self, video_id: int): + row = self.db.get_video_by_id(video_id) + if row is None: + logger.warning(f"消费到不存在的 video_id={video_id}，跳过") + return + if row['status'] == 'done': + return + if not self._retry_allowed(row): + logger.warning(f"[video_id={video_id}] 已达重试上限({row['retry_count']})，放弃") + return + ok = self.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 + + # ------------------------------------------------------------------ + # 生命周期 + # ------------------------------------------------------------------ + def start(self): + if self._producer is not None and self._producer.is_alive(): + return + # 启动恢复：先补入队 DB 中未处理完的 + try: + self._enqueue_existing() + except Exception as e: + logger.error(f"启动恢复入队失败: {e}", exc_info=True) + self._producer = threading.Thread( + target=self._producer_loop, daemon=True, name='video-producer') + self._producer.start() + for i in range(max(1, self.max_concurrent)): + t = threading.Thread( + target=self._consume_loop, daemon=True, + name=f'video-consumer-{i}') + t.start() + self._consumers.append(t) + logger.info(f"VideoQueue 已启动: {max(1, self.max_concurrent)} 个消费者, " + f"超时倍数 ×{self.timeout_multiplier}, 重试上限 {self.max_retries}") + + def stop(self): + self._producer_stop.set() + self._consumer_stop.set() + if self._producer: + self._producer.join(timeout=5) + for t in self._consumers: + t.join(timeout=5) + + def stats(self) -> dict: + return { + "queue_size": self._queue.qsize(), + "consumers": len(self._consumers), + "timeout_multiplier": self.timeout_multiplier, + "max_retries": self.max_retries, + "produced": self._stats["produced"], + "consumed_ok": self._stats["consumed_ok"], + "consumed_fail": self._stats["consumed_fail"], + } + + def is_alive(self) -> bool: + return (self._producer is not None and self._producer.is_alive() + and any(t.is_alive() for t in self._consumers)) diff --git a/fam-edge/src/fam_edge/watch_processor.py b/fam-edge/src/fam_edge/watch_processor.py deleted file mode 100644 index 6e4f023..0000000 --- a/fam-edge/src/fam_edge/watch_processor.py +++ /dev/null @@ -1,91 +0,0 @@ -""" -WatchProcessor - 监听 Google 硬盘同步落地目录，处理新视频 - -流程: -1. rclone 已把 Google 硬盘目录实时同步到 local_dir（视频文件） -2. 每 watch_interval_sec 轮询一次 local_dir -3. 发现未在 videos 表登记的文件 -> ensure_video 登记 -4. 取 pending/failed 的视频，逐个整视频分析（max_concurrent=1，串行避免过载） -""" -import os -import time -import threading -from typing import List - -from .logger import setup_logger -from .config_loader import load_config -from . import oracle_db -from .video_processor import VideoProcessor - -logger = setup_logger('fam-edge.watch_processor') - -VIDEO_EXTS = ('.mp4', '.mkv', '.avi', '.mov', '.ts') - - -class WatchProcessor: - def __init__(self, db: oracle_db.OracleDB): - self.config = load_config() - self.db = db - self.local_dir = self.config.get('gdrive_sync', {}).get('local_dir', '/opt/fam-edge/gdrive_videos') - self.interval = self.config.get('gdrive_sync', {}).get('watch_interval_sec', 30) - self.camera_name = self.config.get('gdrive_sync', {}).get('camera_name', '摄像头') - self.max_concurrent = self.config.get('video_processing', {}).get('max_concurrent', 1) - self.processor = VideoProcessor(db) - self._running = False - self._thread = None - - def _scan_files(self) -> List[str]: - """递归扫描监听目录，返回所有视频文件的完整路径""" - if not os.path.isdir(self.local_dir): - logger.warning(f"监听目录不存在: {self.local_dir}") - return [] - out = [] - for root, _dirs, files in os.walk(self.local_dir): - for fn in sorted(files): - if fn.lower().endswith(VIDEO_EXTS): - out.append(os.path.join(root, fn)) - return out - - def _register_new(self, files: List[str]): - for path in files: - fn = os.path.basename(path) - if self.db.get_video_by_filename(fn) is None: - self.db.ensure_video(fn, path, camera_name=self.camera_name) - logger.info(f"登记新视频: {fn}") - - def _process_pending(self): - pending = self.db.get_pending_videos(limit=self.max_concurrent) - for row in pending: - try: - self.processor.process_video(row['id'], row['filename'], row['local_path']) - except Exception as e: - logger.error(f"处理视频 {row['filename']} 异常: {e}", exc_info=True) - self.db.mark_video_failed(row['id'], f"watch_error: {e}") - - def _run(self): - logger.info(f"WatchProcessor 启动，监听 {self.local_dir}，间隔 {self.interval}s") - while self._running: - try: - files = self._scan_files() - self._register_new(files) - self._process_pending() - except Exception as e: - logger.error(f"WatchProcessor 轮询异常: {e}", exc_info=True) - # 处理完一小批后休眠 - for _ in range(self.interval): - if not self._running: - break - time.sleep(1) - - def start(self): - if self._running: - return - self._running = True - self._thread = threading.Thread(target=self._run, daemon=True, name='watch') - self._thread.start() - - def is_alive(self): - return self._thread is not None and self._thread.is_alive() - - def stop(self): - self._running = False