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

feat: NAS端FFmpeg预压缩 — 根治360MB视频跨境上传超时

Browse Source 
根因诊断（网络问题，非代码逻辑）:
- NAS(中国移动) → Oracle(美国Phoenix) 跨境上行带宽实测仅 0.5-0.9MB/s 且波动大
- 下载方向 3.0MB/s（非对称，典型国际出口拥塞）
- Tailscale P2P: ping 可通(224ms)但持续数据流被阻断(0 B/s, 127s超时)
- 360MB 原始视频上传需 10-20min，任何大分块都会超时

NAS dispatcher:
- 新增 _compress_video: 480p/CRF28/veryfast，静态监控场景实测 ~36x 压缩比
  (360MB → ~12MB，上传时间 20min → ~30s)
- ffmpeg 自动探测: CodecPack ffmpeg41(带libx264) > /usr/local/bin > PATH
  (Synology 系统 ffmpeg 被裁剪，无 h264 编解码)
- scale 两级滤镜: force_original_aspect_ratio=decrease + trunc(iw/2)*2
  (h264 要求偶数尺寸，853x480 会报错)
- 压缩缓存 /tmp/fam_compressed/task_{id}/，源文件未变则重试复用，TTL 24h
- 压缩完成后重置 PROCESSING 状态（重置 stale 回收计时基准）
- 压缩失败回退原始文件分块上传
- _query_uploaded_chunks 失败时记录日志（原静默失败导致全量重传无感知）

Edge api_gateway:
- fix: total_chunks 变更清理旧分块后同步重写 meta.json
  (原 bug: meta.json 不更新导致每块上传都触发清理，删除同批新分块死循环)

config:
- stale_timeout 600→1800（覆盖压缩+上传+Edge队列积压总时长）
- 新增 ffmpeg_path / compress_timeout 配置项
This commit is contained in:
ericwyuan
2026-08-20 14:59:50 +08:00
parent a9a1b1dda5
commit 203b076783
3 changed files with 139 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
fam-core/config/config.yaml.example
View File

@@ -21,9 +21,11 @@ scheduler:
dispatcher:
poll_interval: 30 # 轮询间隔（秒）
edge_url: "http://100.x.x.20:5000/api/edge/video/push" # 推送模式端点（视频上传，同步返回结果）
max_retries: 3
push_timeout: 1800 # 推送+分析同步超时（秒）
edge_url: "http://100.x.x.20:5000/api/edge/video/enqueue" # 异步队列端点（上传入队，Poller 拉取结果）
max_retries: 5 # 文件级重试次数（分块级重试另计，每块3次）
stale_timeout: 1800 # PROCESSING 僵尸回收（秒），需大于压缩+上传+Edge队列积压总时长
# ffmpeg_path: "/var/packages/CodecPack/target/bin/ffmpeg41" # 可选，默认自动探测（Synology 需 CodecPack 版，系统版无 h264 编码）
compress_timeout: 3600 # 单个视频预压缩超时（秒）
video_server:
base_url: "http://100.x.x.10:8000/media"

140
fam-core/src/fam_core/dispatcher/dispatcher.py
View File

@@ -1,13 +1,15 @@
"""
Dispatcher - 30s 轮询 PENDING 任务，上传视频至 Edge 异步队列
流程（异步队列模式 + 分块断点续传）:
流程（异步队列模式 + NAS 预压缩 + 分块断点续传）:
1. 读取任务对应的本地视频文件
2a. 小文件 (<=50MB): 直接 multipart 上传至 /enqueue
2b. 大文件 (>50MB): 分块上传 (20MB/块) 至 /chunk，支持断点续传，最后调 /assemble 合并入队
3. Edge 保存视频 + 入 SQLite 队列，返回 202
4. Dispatcher 标记任务为 PROCESSING（已派发，等待 Poller 拉取结果）
5. Poller 线程定期从 Edge /api/edge/results 拉取结果，写库后标记 SUCCESS
2. 大文件 (>20MB): NAS 端 FFmpeg 预压缩 (480p/CRF28, ~36x 压缩比)
（根因: NAS→Oracle 跨境上行带宽仅 ~0.5-0.9MB/s，360MB 原始上传需 10-20min 且频繁超时）
3a. 压缩后小文件 (<=20MB): 直接 multipart 上传至 /enqueue
3b. 仍超阈值: 分块上传 (5MB/块) 至 /chunk，支持断点续传，最后调 /assemble 合并入队
4. Edge 保存视频 + 入 SQLite 队列，返回 202
5. Dispatcher 标记任务为 PROCESSING（已派发，等待 Poller 拉取结果）
6. Poller 线程定期从 Edge /api/edge/results 拉取结果，写库后标记 SUCCESS
退避重试: min(30 * (retry_count + 1), 300) 秒
分块级重试: 每块最多重试 3 次
@@ -17,6 +19,8 @@ import os
import io
import time
import math
import shutil
import subprocess
import threading
import requests
from datetime import datetime, timedelta
@@ -29,8 +33,17 @@ logger = setup_logger('fam-core.dispatcher')
CHUNK_SIZE = 5 * 1024 * 1024 # 5MB per chunk (reliable at ~1Mbps upload)
CHUNK_THRESHOLD = 20 * 1024 * 1024 # files > 20MB use chunked upload
COMPRESS_THRESHOLD = 20 * 1024 * 1024 # files > 20MB get pre-compressed before upload
COMPRESS_DIR = '/tmp/fam_compressed'
COMPRESS_CACHE_TTL = 24 * 3600 # 压缩缓存保留 24h（供上传失败重试复用）
MAX_CHUNK_RETRIES = 3
# Synology 系统 ffmpeg 被裁剪（无 h264 编解码），CodecPack 的 ffmpeg41 带 libx264
FFMPEG_CANDIDATES = [
'/var/packages/CodecPack/target/bin/ffmpeg41',
'/usr/local/bin/ffmpeg',
]
class Dispatcher:
"""任务下发器，30s 轮询（异步队列 + 分块断点续传）"""
@@ -43,6 +56,8 @@ class Dispatcher:
self.max_retries = cfg.get('dispatcher', {}).get('max_retries', 3)
self.camera_name = cfg.get('scheduler', {}).get('camera_name', '默认摄像头')
self.stale_timeout = cfg.get('dispatcher', {}).get('stale_timeout', 600)
self.compress_timeout = cfg.get('dispatcher', {}).get('compress_timeout', 3600)
self.ffmpeg = self._find_ffmpeg(cfg)
# 推导 Edge base URL
self.edge_base = self.edge_url.rsplit('/api/edge/video/enqueue', 1)[0]
self.chunk_url = f"{self.edge_base}/api/edge/video/chunk"
@@ -51,6 +66,16 @@ class Dispatcher:
self._running = False
self._thread = None
@staticmethod
def _find_ffmpeg(cfg):
"""查找可用 ffmpeg：配置优先，其次 CodecPack（带 libx264），最后 PATH"""
configured = cfg.get('dispatcher', {}).get('ffmpeg_path')
candidates = ([configured] if configured else []) + FFMPEG_CANDIDATES
for path in candidates:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
return shutil.which('ffmpeg')
def _calculate_backoff(self, retry_count):
"""退避策略: min(30 * (retry_count + 1), 300)"""
return min(30 * (retry_count + 1), 300)
@@ -83,8 +108,87 @@ class Dispatcher:
pass
return payload
def _compress_video(self, task_id, video_path):
"""NAS 端预压缩: 480p/CRF28/veryfast，静态监控场景实测 ~36x 压缩比
成功返回压缩文件路径；失败返回 None（回退原始文件分块上传）。
压缩产物缓存在 /tmp/fam_compressed/task_{id}/，重试时源文件未变则复用。
"""
if not self.ffmpeg:
logger.warning(f"[task_id={task_id}] ffmpeg 不可用，跳过预压缩")
return None
self._cleanup_compress_cache()
out_dir = os.path.join(COMPRESS_DIR, f'task_{task_id}')
out_path = os.path.join(out_dir, os.path.basename(video_path))
# 缓存复用：源文件未变且压缩产物有效
try:
if (os.path.isfile(out_path)
and os.path.getsize(out_path) > 0
and os.path.getmtime(out_path) >= os.path.getmtime(video_path)):
logger.info(f"[task_id={task_id}] 复用压缩缓存: {out_path}")
return out_path
except OSError:
pass
os.makedirs(out_dir, exist_ok=True)
src_size = os.path.getsize(video_path)
start = time.time()
cmd = [
self.ffmpeg, '-y',
'-i', video_path,
# 第二级 scale 向下取偶（h264 要求偶数尺寸；force_divisible_by 需 ffmpeg>=4.3）
'-vf', ('scale=854:480:force_original_aspect_ratio=decrease,'
'scale=trunc(iw/2)*2:trunc(ih/2)*2'),
'-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28',
'-an',
out_path,
]
try:
result = subprocess.run(
cmd, capture_output=True, text=True,
timeout=self.compress_timeout,
)
except subprocess.TimeoutExpired:
logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s)，回退原始上传")
return None
except OSError as e:
logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}")
return None
if result.returncode != 0 or not os.path.isfile(out_path) or os.path.getsize(out_path) == 0:
stderr_tail = (result.stderr or '')[-300:]
logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}")
return None
dst_size = os.path.getsize(out_path)
elapsed = time.time() - start
logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → "
f"{dst_size/1048576:.1f}MB ({src_size/max(dst_size,1):.1f}x)，耗时 {elapsed:.0f}s")
return out_path
@staticmethod
def _cleanup_compress_cache():
"""清理超过 TTL 的压缩缓存目录（tmpfs 空间有限）"""
try:
if not os.path.isdir(COMPRESS_DIR):
return
cutoff = time.time() - COMPRESS_CACHE_TTL
for entry in os.listdir(COMPRESS_DIR):
path = os.path.join(COMPRESS_DIR, entry)
try:
if os.path.isdir(path) and os.path.getmtime(path) < cutoff:
shutil.rmtree(path, ignore_errors=True)
except OSError:
continue
except OSError:
pass
def _dispatch_one(self, task):
"""上传视频至 Edge 异步队列（自动选择直接/分块模式）"""
"""上传视频至 Edge 异步队列（大文件先预压缩，自动选择直接/分块模式）"""
task_id = task['task_id']
video_path = task['video_path']
@@ -101,16 +205,29 @@ class Dispatcher:
size_mb = file_size / (1024 * 1024)
db_layer.update_task_status(task_id, 'PROCESSING')
# 跨境上行带宽受限（实测 ~0.5-0.9MB/s），大文件先预压缩再上传
upload_path = video_path
if file_size > COMPRESS_THRESHOLD:
compressed = self._compress_video(task_id, video_path)
if compressed:
upload_path = compressed
file_size = os.path.getsize(upload_path)
size_mb = file_size / (1024 * 1024)
# 压缩耗时较长，重置 stale 计时基准（reclaim 按 updated_at 判断）
db_layer.update_task_status(task_id, 'PROCESSING')
else:
logger.warning(f"[task_id={task_id}] 压缩失败，回退原始文件上传 ({size_mb:.1f}MB)")
payload = self._build_payload(task)
if file_size > CHUNK_THRESHOLD:
logger.info(f"[task_id={task_id}] 大文件分块上传: {size_mb:.1f}MB, "
f"{math.ceil(file_size / CHUNK_SIZE)} 块")
self._dispatch_chunked(task, payload, video_path, file_size)
self._dispatch_chunked(task, payload, upload_path, file_size)
else:
log_task(logger, task_id, 'dispatcher',
f'直接上传: {self.edge_url} ({size_mb:.1f}MB)')
self._dispatch_direct(task, payload, video_path)
self._dispatch_direct(task, payload, upload_path)
def _dispatch_direct(self, task, payload, video_path):
"""小文件直接上传至 /enqueue"""
@@ -167,8 +284,9 @@ class Dispatcher:
f"≠ expected={expected_total}（chunk_size 已变更），从头上传")
return set(), edge_total
return uploaded, edge_total
except requests.RequestException:
pass
logger.warning(f"[task_id={task_id}] 查询已上传分块返回 {resp.status_code}，将全量重传")
except requests.RequestException as e:
logger.warning(f"[task_id={task_id}] 查询已上传分块失败（将全量重传）: {e}")
return set(), 0
def _dispatch_chunked(self, task, payload, video_path, file_size):

9
fam-edge/src/fam_edge/api_gateway/api_gateway.py
View File

@@ -168,10 +168,10 @@ def upload_chunk():
chunk_path = os.path.join(d, f"chunk_{chunk_index:04d}")
try:
# 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块
# 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块并更新元数据
import json
meta_path = os.path.join(d, "meta.json")
if total_chunks and os.path.exists(meta_path):
import json
try:
with open(meta_path) as mf:
old_meta = json.load(mf)
@@ -181,15 +181,16 @@ def upload_chunk():
for fn in os.listdir(d):
if fn.startswith('chunk_'):
os.remove(os.path.join(d, fn))
with open(meta_path, 'w') as f:
json.dump({"filename": filename, "total_chunks": total_chunks}, f)
except (json.JSONDecodeError, IOError):
pass
chunk_file.save(chunk_path)
size_kb = os.path.getsize(chunk_path) / 1024
# 写/更新元数据
# 写元数据（首次上传时）
if not os.path.exists(meta_path):
import json
meta = {"filename": filename, "total_chunks": total_chunks}
with open(meta_path, 'w') as f:
json.dump(meta, f)