292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""
|
||
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 Dict, List, Optional
|
||
|
||
from .logger import setup_logger
|
||
from .config_loader import load_config
|
||
from . import oracle_db
|
||
from .video_processor import VideoProcessor, validate_video
|
||
|
||
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.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', 10))
|
||
# 失败重试最小间隔（秒）：配额类瞬时故障（如 429）需等其恢复再重试，避免短时间重复打爆
|
||
self.retry_interval_sec = int(vp.get('retry_interval_sec', 3600))
|
||
# 文件校验：入队前用 OpenCV 确认可解码；mtime 稳定窗口防 rclone 写入中的半成品
|
||
self.file_validate = bool(vp.get('file_validate', True))
|
||
self.stable_window_sec = int(vp.get('stable_window_sec', 60))
|
||
|
||
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}
|
||
# 实时状态：当前正在处理的视频（服务状态界面用）
|
||
self._current = None
|
||
self._current_lock = threading.Lock()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 生产者
|
||
# ------------------------------------------------------------------
|
||
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:
|
||
# 新文件：先过 mtime 稳定窗口 + 可解码校验，通过才登记入队；失败标记 invalid
|
||
if self.file_validate:
|
||
if time.time() - os.path.getmtime(path) < self.stable_window_sec:
|
||
logger.info(f"文件仍在写入（mtime 未稳定），跳过本轮: {fn}")
|
||
continue
|
||
ok, verr, vmeta = validate_video(path)
|
||
if not ok:
|
||
vid = self.db.ensure_video(fn, path, camera_name=self.camera_name)
|
||
self.db.set_video_file_status(vid, False, verr)
|
||
self.db.mark_video_invalid(vid, verr)
|
||
logger.warning(f"文件校验失败，标记 invalid 不入队: {fn} ({verr})")
|
||
continue
|
||
if vmeta:
|
||
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.reset_video_to_pending(row['id'])
|
||
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 中未处理完的视频补入队（进程重启恢复）"""
|
||
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:
|
||
"""是否允许消费/重试该视频：pending 直接放行；failed 需未超重试上限且距上次失败够久；invalid 永不放行"""
|
||
status = row['status']
|
||
if status == 'pending':
|
||
return True
|
||
if status == 'invalid':
|
||
return False
|
||
# failed：未超重试上限，且距上次失败 >= retry_interval_sec（等配额类故障恢复）
|
||
if int(row['retry_count'] or 0) >= self.max_retries:
|
||
return False
|
||
last_fail = row['last_fail_at'] or row['updated_at'] or ''
|
||
if last_fail:
|
||
try:
|
||
from datetime import datetime
|
||
lt = datetime.strptime(last_fail[:19], '%Y-%m-%d %H:%M:%S')
|
||
elapsed = (datetime.now() - lt).total_seconds()
|
||
if elapsed < self.retry_interval_sec:
|
||
return False
|
||
except ValueError:
|
||
pass
|
||
return True
|
||
|
||
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):
|
||
# 每消费者独立 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, processor)
|
||
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, processor: VideoProcessor):
|
||
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
|
||
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, clip_ids = processor.process_video(
|
||
video_id, fn, row['local_path'],
|
||
timeout_multiplier=self.timeout_multiplier)
|
||
# 素材整段视频分割出的运动片段入队分析（片段不在监听目录，需手动入队）
|
||
for cid in (clip_ids or []):
|
||
self._enqueue(cid)
|
||
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()),
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# 生命周期
|
||
# ------------------------------------------------------------------
|
||
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))
|