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

feat(fam-edge): 视频生产-消费队列 - 新增 video_queue.py（生产者扫描新文件入队+重启恢复，独立消费者线程云端分析）；模型消费超时按原配置×timeout_multiplier(2)；failed 重试上限 max_retries(2)；删除旧 watch_processor

Browse Source
This commit is contained in:
ericwyuan
2026-08-21 12:01:25 +08:00
parent ab5b706eeb
commit ce27e4ab0c
8 changed files with 242 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

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

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

17
fam-edge/src/fam_edge/video_processor.py
View File

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

199
fam-edge/src/fam_edge/video_queue.py Normal file
View File

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

91
fam-edge/src/fam_edge/watch_processor.py
View File

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