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

feat: 分块断点续传上传 — 20MB/块 + 分块级重试 + 断点查询

Browse Source 
Edge 端新增 3 个端点:
- POST /api/edge/video/chunk: 接收单块，保存到 task_{id}/chunk_{index:04d}
- GET /api/edge/video/chunks: 查询已上传分块（断点续传）
- POST /api/edge/video/assemble: 合并全部分块入队

NAS Dispatcher 重写:
- 大文件(>50MB)自动分块上传(20MB/块)
- 每块最多重试 3 次（分块级重试，非整文件级）
- 上传前查询已上传分块，跳过已有的（断点续传）
- 全部上传后调 /assemble 合并入队
- 小文件(<=50MB)走直接上传路径
- max_retries 3→5（文件级重试次数）
- scheduler 切回生产目录

解决: 360MB 视频跨公网单次上传超时/断连问题
This commit is contained in:
ericwyuan
2026-08-20 12:27:30 +08:00
parent fb4ccb64dc
commit b6c13a9047
3 changed files with 351 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
fam-core/config/config.yaml
View File

@@ -1,6 +1,8 @@
# FAM-Core 配置文件 (NAS 端) - 实际部署配置
# 注: Tailscale 防火墙待修复，当前 edge_url 使用 Oracle 公网 IP
# 异步队列模式: NAS 上传视频到 Edge /api/edge/video/enqueue 入队 → Edge 消费者异步处理
# 异步队列模式 + 分块断点续传:
# 小文件(<=50MB): 直接上传 /enqueue
# 大文件(>50MB): 分块(20MB/块)上传 /chunk → /assemble 合并入队
# → NAS Poller 定期从 /api/edge/results 拉取结果写库
# Ollama 未对外暴露，chat_handler 通过 FAM-Edge 代理
@@ -18,9 +20,8 @@ database:
scheduler:
scan_interval: 60
# E2E 测试期间指向独立测试目录（正式目录 /volume1/surveillance 有 285 个历史视频，
# 全量建任务会导致 100GB 跨公网上传，待与用户确认回补策略后再切回）
video_dir: "/volume1/web/sentinel-home-ai/e2e-test"
# 正式目录 /volume1/surveillance/Generic_ONVIF-001
video_dir: "/volume1/surveillance/Generic_ONVIF-001"
video_extensions: [".mp4", ".mkv", ".avi"]
file_stable_seconds: 60
camera_name: "客厅"
@@ -28,9 +29,8 @@ scheduler:
dispatcher:
poll_interval: 30
edge_url: "http://129.146.203.203:5000/api/edge/video/enqueue"
max_retries: 3
upload_timeout: 300 # 仅视频上传时间（不含 AI 处理）
stale_timeout: 600 # PROCESSING 超时回收（10分钟，Edge 处理 + 队列等待）
max_retries: 5 # 文件级重试次数（分块级重试另计，每块3次）
stale_timeout: 600 # PROCESSING 超时回收（10分钟）
poller:
poll_interval: 30

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

@@ -1,17 +1,21 @@
"""
Dispatcher - 30s 轮询 PENDING 任务，上传视频至 Edge 异步队列
流程（异步队列模式）:
流程（异步队列模式 + 分块断点续传）:
1. 读取任务对应的本地视频文件
2. multipart 上传至 Edge /api/edge/video/enqueue（附已知成员清单等元数据）
3. Edge 保存视频 + 入 SQLite 队列，立即返回 202
2a. 小文件 (<=50MB): 直接 multipart 上传至 /enqueue
2b. 大文件 (>50MB): 分块上传 (20MB/块) 至 /chunk，支持断点续传，最后调 /assemble 合并入队
3. Edge 保存视频 + 入 SQLite 队列，返回 202
4. Dispatcher 标记任务为 PROCESSING（已派发，等待 Poller 拉取结果）
5. Poller 线程定期从 Edge /api/edge/results 拉取结果，写库后标记 SUCCESS
退避重试: min(60 * (retry_count + 1) * 2, 600) 秒
分块级重试: 每块最多重试 3 次
"""
import os
import io
import time
import math
import threading
import requests
from datetime import datetime, timedelta
@@ -22,20 +26,27 @@ from .. import db_layer
logger = setup_logger('fam-core.dispatcher')
CHUNK_SIZE = 20 * 1024 * 1024 # 20MB per chunk
CHUNK_THRESHOLD = 50 * 1024 * 1024 # files > 50MB use chunked upload
MAX_CHUNK_RETRIES = 3
class Dispatcher:
"""任务下发器，30s 轮询（异步队列模式）"""
"""任务下发器，30s 轮询（异步队列 + 分块断点续传）"""
def __init__(self):
cfg = load_config()
self.poll_interval = cfg.get('dispatcher', {}).get('poll_interval', 30)
self.edge_url = cfg.get('dispatcher', {}).get('edge_url', 'http://localhost:5000/api/edge/video/enqueue')
self.edge_url = cfg.get('dispatcher', {}).get('edge_url',
'http://localhost:5000/api/edge/video/enqueue')
self.max_retries = cfg.get('dispatcher', {}).get('max_retries', 3)
# enqueue 模式只需上传时间，不含 AI 处理时间
self.upload_timeout = cfg.get('dispatcher', {}).get('upload_timeout', 300)
self.camera_name = cfg.get('scheduler', {}).get('camera_name', '默认摄像头')
# PROCESSING 超时回收：Edge 处理 + 队列等待可能很长
self.stale_timeout = cfg.get('dispatcher', {}).get('stale_timeout', 3600)
self.stale_timeout = cfg.get('dispatcher', {}).get('stale_timeout', 600)
# 推导 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"
self.chunks_query_url = f"{self.edge_base}/api/edge/video/chunks"
self.assemble_url = f"{self.edge_base}/api/edge/video/assemble"
self._running = False
self._thread = None
@@ -72,7 +83,7 @@ class Dispatcher:
return payload
def _dispatch_one(self, task):
"""上传视频至 Edge 异步队列，立即返回"""
"""上传视频至 Edge 异步队列（自动选择直接/分块模式）"""
task_id = task['task_id']
video_path = task['video_path']
@@ -80,24 +91,36 @@ class Dispatcher:
db_layer.update_task_status(
task_id, 'FAILED',
error_message=f"视频文件不存在: {video_path}",
failure_stage='upload'
failure_stage='callback'
)
logger.error(f"[task_id={task_id}] 视频文件不存在，标记 FAILED: {video_path}")
return
payload = self._build_payload(task)
size_mb = os.path.getsize(video_path) / (1024 * 1024)
file_size = os.path.getsize(video_path)
size_mb = file_size / (1024 * 1024)
db_layer.update_task_status(task_id, 'PROCESSING')
log_task(logger, task_id, 'dispatcher',
f'上传视频至 Edge 队列: {self.edge_url} ({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)
else:
log_task(logger, task_id, 'dispatcher',
f'直接上传: {self.edge_url} ({size_mb:.1f}MB)')
self._dispatch_direct(task, payload, video_path)
def _dispatch_direct(self, task, payload, video_path):
"""小文件直接上传至 /enqueue"""
task_id = task['task_id']
try:
with open(video_path, 'rb') as fh:
resp = requests.post(
self.edge_url,
data=payload,
files={'video': (os.path.basename(video_path), fh, 'video/mp4')},
timeout=(10, 60)
timeout=(10, 120)
)
except requests.RequestException as e:
logger.error(f"[task_id={task_id}] 上传失败: {e}")
@@ -121,6 +144,110 @@ class Dispatcher:
logger.error(f"[task_id={task_id}] Edge 返回异常状态码: {resp.status_code}")
self._schedule_retry(task)
def _dispatch_chunked(self, task, payload, video_path, file_size):
"""大文件分块上传 + 断点续传
1. 查询 Edge 端已上传分块（断点续传）
2. 上传缺失分块（每块最多重试 3 次）
3. 全部分块上传后调用 /assemble 合并入队
"""
task_id = task['task_id']
total_chunks = math.ceil(file_size / CHUNK_SIZE)
filename = os.path.basename(video_path)
# 1. 查询已上传分块（断点续传）
uploaded_set = set()
try:
resp = requests.get(
self.chunks_query_url,
params={'task_id': task_id},
timeout=(10, 15)
)
if resp.status_code == 200:
data = resp.json()
uploaded_set = set(data.get('uploaded_chunks', []))
if uploaded_set:
logger.info(f"[task_id={task_id}] 断点续传: 已有 {len(uploaded_set)}/{total_chunks} 块")
except requests.RequestException as e:
logger.warning(f"[task_id={task_id}] 查询已上传分块失败（将从头上传）: {e}")
# 2. 上传缺失分块
try:
with open(video_path, 'rb') as fh:
for idx in range(total_chunks):
if idx in uploaded_set:
continue
chunk_data = fh.read(CHUNK_SIZE)
if not chunk_data:
break
success = False
for attempt in range(MAX_CHUNK_RETRIES):
try:
cresp = requests.post(
self.chunk_url,
data={
'task_id': str(task_id),
'chunk_index': str(idx),
'total_chunks': str(total_chunks),
'filename': filename,
},
files={'chunk': (f'chunk_{idx}', io.BytesIO(chunk_data))},
timeout=(10, 120)
)
if cresp.status_code == 200:
success = True
break
logger.warning(f"[task_id={task_id}] 分块 {idx} 返回 {cresp.status_code}（尝试 {attempt+1}/{MAX_CHUNK_RETRIES}）")
except requests.RequestException as e:
logger.warning(f"[task_id={task_id}] 分块 {idx} 上传失败（尝试 {attempt+1}/{MAX_CHUNK_RETRIES}）: {e}")
if attempt < MAX_CHUNK_RETRIES - 1:
time.sleep(5 * (attempt + 1))
if not success:
logger.error(f"[task_id={task_id}] 分块 {idx} 超过最大重试次数，安排文件级重试")
self._schedule_retry(task)
return
if (idx + 1) % 5 == 0 or idx == total_chunks - 1:
logger.info(f"[task_id={task_id}] 分块进度: {idx + 1}/{total_chunks}")
except IOError as e:
logger.error(f"[task_id={task_id}] 读取视频文件失败: {e}")
self._schedule_retry(task)
return
# 3. 合并 + 入队
try:
aresp = requests.post(
self.assemble_url,
data={
'task_id': str(task_id),
'camera_name': payload.get('camera_name', ''),
'event_start_time': payload.get('event_start_time', ''),
'known_members_context': payload.get('known_members_context', ''),
},
timeout=(10, 60)
)
except requests.RequestException as e:
logger.error(f"[task_id={task_id}] 合并请求失败: {e}")
self._schedule_retry(task)
return
if aresp.status_code == 202:
try:
data = aresp.json()
queue_id = data.get('queue_id', '?')
asm_size = data.get('size_mb', '?')
logger.info(f"[task_id={task_id}] 分块合并入队成功 (queue_id={queue_id}, {asm_size}MB)，等待 Poller 拉取结果")
except ValueError:
logger.info(f"[task_id={task_id}] 分块合并入队成功，等待 Poller 拉取结果")
return
logger.error(f"[task_id={task_id}] 合并端点返回 {aresp.status_code}: {aresp.text[:200]}")
self._schedule_retry(task)
def _schedule_retry(self, task):
"""调度重试"""
task_id = task['task_id']
@@ -158,7 +285,7 @@ class Dispatcher:
def _run(self):
"""线程主循环"""
logger.info(f"Dispatcher 启动 (enqueue 模式)，轮询间隔 {self.poll_interval}s")
logger.info(f"Dispatcher 启动 (enqueue + 分块模式)，轮询间隔 {self.poll_interval}s")
while self._running:
try:
self._poll_once()

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

@@ -129,6 +129,205 @@ def enqueue_task():
return jsonify({"error": str(e)}), 500
# ========== 分块上传（断点续传）==========
CHUNK_SIZE = 20 * 1024 * 1024 # 20MB per chunk
def _chunk_dir(task_id: int) -> str:
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
d = os.path.join(upload_dir, f"task_{task_id}")
os.makedirs(d, exist_ok=True)
return d
@api_bp.route('/api/edge/video/chunk', methods=['POST'])
def upload_chunk():
"""接收单个分块，保存到 task_{id}/chunk_{index:04d}
断点续传：同一 task_id + chunk_index 重复上传会覆盖，
NAS 端可通过 /chunks 查询已上传分块，跳过已有的。
"""
task_id_raw = request.form.get('task_id')
chunk_index_raw = request.form.get('chunk_index')
total_chunks_raw = request.form.get('total_chunks')
filename = request.form.get('filename', 'video.mp4')
chunk_file = request.files.get('chunk')
if not task_id_raw or not chunk_index_raw or not chunk_file:
return jsonify({"error": "缺少必填字段: task_id, chunk_index, chunk"}), 400
try:
task_id = int(task_id_raw)
chunk_index = int(chunk_index_raw)
total_chunks = int(total_chunks_raw) if total_chunks_raw else 0
except ValueError:
return jsonify({"error": "task_id/chunk_index 必须是整数"}), 400
d = _chunk_dir(task_id)
chunk_path = os.path.join(d, f"chunk_{chunk_index:04d}")
try:
chunk_file.save(chunk_path)
size_kb = os.path.getsize(chunk_path) / 1024
# 写元数据（首次上传时）
meta_path = os.path.join(d, "meta.json")
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)
# 统计已上传分块
uploaded = sorted([
int(fn.split('_')[1]) for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
])
logger.info(f"[task_id={task_id}] 分块 {chunk_index}/{total_chunks} 上传成功 "
f"({size_kb:.0f}KB, 已上传 {len(uploaded)}/{total_chunks})")
return jsonify({
"status": "ok",
"task_id": task_id,
"chunk_index": chunk_index,
"uploaded_count": len(uploaded),
"total_chunks": total_chunks,
}), 200
except Exception as e:
logger.error(f"[task_id={task_id}] 分块上传失败: {e}", exc_info=True)
return jsonify({"error": str(e)}), 500
@api_bp.route('/api/edge/video/chunks', methods=['GET'])
def query_chunks():
"""查询已上传分块列表（断点续传：NAS 重启后查询跳过已有分块）"""
task_id_raw = request.args.get('task_id')
if not task_id_raw:
return jsonify({"error": "缺少 task_id"}), 400
try:
task_id = int(task_id_raw)
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
d = _chunk_dir(task_id)
uploaded = sorted([
int(fn.split('_')[1]) for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
]) if os.path.isdir(d) else []
total = 0
meta_path = os.path.join(d, "meta.json")
if os.path.exists(meta_path):
import json
try:
with open(meta_path) as f:
total = json.load(f).get('total_chunks', 0)
except (json.JSONDecodeError, IOError):
pass
return jsonify({
"task_id": task_id,
"uploaded_chunks": uploaded,
"uploaded_count": len(uploaded),
"total_chunks": total,
}), 200
@api_bp.route('/api/edge/video/assemble', methods=['POST'])
def assemble_chunks():
"""合并所有分块为完整视频文件，入 SQLite 队列
NAS 上传完全部分块后调用此端点触发合并 + 入队。
"""
task_id_raw = request.form.get('task_id')
if not task_id_raw:
return jsonify({"error": "缺少 task_id"}), 400
try:
task_id = int(task_id_raw)
except ValueError:
return jsonify({"error": "task_id 必须是整数"}), 400
camera_name = request.form.get('camera_name', '')
event_start_time = request.form.get('event_start_time', '')
known_members_context = request.form.get('known_members_context', '')
d = _chunk_dir(task_id)
# 读取元数据
import json
meta_path = os.path.join(d, "meta.json")
if not os.path.exists(meta_path):
return jsonify({"error": "元数据不存在，请先上传分块"}), 400
try:
with open(meta_path) as f:
meta = json.load(f)
except json.JSONDecodeError:
return jsonify({"error": "元数据损坏"}), 500
filename = meta.get('filename', 'video.mp4')
total_chunks = meta.get('total_chunks', 0)
# 检查分块完整性
chunk_files = sorted([
fn for fn in os.listdir(d)
if fn.startswith('chunk_') and len(fn.split('_')) == 2
])
if total_chunks and len(chunk_files) < total_chunks:
missing = total_chunks - len(chunk_files)
return jsonify({
"error": f"分块不完整: {len(chunk_files)}/{total_chunks}，缺 {missing} 块",
"uploaded_count": len(chunk_files),
"total_chunks": total_chunks,
}), 400
# 合并分块
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
video_filename = f"task_{task_id}_{filename}"
video_path = os.path.join(upload_dir, video_filename)
try:
with open(video_path, 'wb') as out:
for cf in chunk_files:
chunk_path = os.path.join(d, cf)
with open(chunk_path, 'rb') as chunk_f:
out.write(chunk_f.read())
size_mb = os.path.getsize(video_path) / 1024 / 1024
logger.info(f"[task_id={task_id}] 分块合并完成: {filename} ({size_mb:.1f}MB, {len(chunk_files)} 块)")
# 清理分块目录
import shutil
shutil.rmtree(d, ignore_errors=True)
# 入队
queue_id = queue_manager.enqueue(
nas_task_id=task_id,
video_filename=filename,
video_path=video_path,
camera_name=camera_name,
event_start_time=event_start_time,
known_members_context=known_members_context,
)
return jsonify({
"status": "queued",
"task_id": task_id,
"queue_id": queue_id,
"size_mb": round(size_mb, 1),
}), 202
except Exception as e:
logger.error(f"[task_id={task_id}] 合并失败: {e}", exc_info=True)
return jsonify({"error": str(e)}), 500
@api_bp.route('/api/edge/results', methods=['GET'])
def get_results():
"""返回已完成但未拉取的结果，标记为已交付"""