43b3178a9d
1. fam-edge 新增 frame_marker.py: Edge 端人脸检测画红框+统计人脸数（NAS ARM 太弱只存图零计算），orchestrator 分析后回传前标记，新增 /api/edge/mark_frames 批量补标端点 2. fam-core 新增 tools/backfill_mark_frames.py: 存量关键帧批量补红框（幂等+备份 frames_orig） 3. db_layer.name_member 增强: 支持自动注册新标签/重命名回溯（旧真名一并替换）/多人组合字符串 REPLACE 4. fam-ui 成员命名页改为人物管理页: 所有人物照片墙+命名重命名+统计去重 5. event_receiver/member_manager 配套适配
531 lines
18 KiB
Python
531 lines
18 KiB
Python
"""
|
||
API-Gateway - Flask 蓝图，接收任务
|
||
|
||
模式:
|
||
1. enqueue (异步): NAS 上传视频 → Edge 入队 → 立即返回 → 消费者异步处理 → NAS 轮询拉取结果
|
||
2. push (同步, 兼容保留): NAS 上传 → Edge 同步处理 → 结果随响应返回
|
||
3. analyze (旧拉取模式, 兼容保留)
|
||
"""
|
||
import os
|
||
import base64
|
||
import threading
|
||
import requests
|
||
from flask import Blueprint, request, jsonify
|
||
|
||
from ..logger import setup_logger
|
||
from ..ai_orchestrator.orchestrator import AIOrchestrator
|
||
from ..video_preprocessor.preprocessor import VideoPreprocessor
|
||
from ..queue import queue_manager
|
||
|
||
logger = setup_logger('fam-edge.api_gateway')
|
||
|
||
api_bp = Blueprint('api_gateway', __name__)
|
||
|
||
_current_task_lock = threading.Lock()
|
||
_currently_processing = False
|
||
|
||
_orchestrator = None
|
||
|
||
|
||
def get_orchestrator():
|
||
global _orchestrator
|
||
if _orchestrator is None:
|
||
_orchestrator = AIOrchestrator()
|
||
return _orchestrator
|
||
|
||
|
||
@api_bp.route('/api/edge/video/analyze', methods=['POST'])
|
||
def receive_task():
|
||
"""接收分析任务"""
|
||
global _currently_processing
|
||
|
||
data = request.get_json(silent=True)
|
||
if not data:
|
||
return jsonify({"error": "Invalid JSON"}), 400
|
||
|
||
task_id = data.get('task_id')
|
||
video_url = data.get('video_url')
|
||
webhook_url = data.get('webhook_url')
|
||
|
||
if not task_id or not video_url or not webhook_url:
|
||
return jsonify({"error": "缺少必填字段: task_id, video_url, webhook_url"}), 400
|
||
|
||
logger.info(f"[task_id={task_id}] 收到任务: {video_url}")
|
||
|
||
# 并发控制
|
||
with _current_task_lock:
|
||
if _currently_processing:
|
||
logger.warning(f"[task_id={task_id}] 队列已满 (当前有任务处理中)，返回 429")
|
||
return jsonify({"error": "Queue full", "retry_after": 60}), 429
|
||
_currently_processing = True
|
||
|
||
# 异步处理
|
||
def _process():
|
||
global _currently_processing
|
||
try:
|
||
orch = get_orchestrator()
|
||
orch.process_task(data)
|
||
except Exception as e:
|
||
logger.error(f"[task_id={task_id}] 处理异常: {e}", exc_info=True)
|
||
finally:
|
||
with _current_task_lock:
|
||
_currently_processing = False
|
||
|
||
thread = threading.Thread(target=_process, daemon=True, name=f'task-{task_id}')
|
||
thread.start()
|
||
|
||
return jsonify({"status": "accepted", "task_id": task_id}), 202
|
||
|
||
|
||
@api_bp.route('/api/edge/video/enqueue', methods=['POST'])
|
||
def enqueue_task():
|
||
"""异步模式：接收 multipart 视频上传，入队后立即返回
|
||
|
||
NAS 上传视频 → Edge 保存到磁盘 + 入 SQLite 队列 → 返回 task_id
|
||
消费者线程异步处理，NAS 通过 /api/edge/results 拉取结果
|
||
"""
|
||
task_id_raw = request.form.get('task_id')
|
||
file = request.files.get('video')
|
||
if not task_id_raw or not file:
|
||
return jsonify({"error": "缺少必填字段: task_id, video"}), 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', '')
|
||
|
||
upload_dir = os.environ.get('FAM_UPLOAD_DIR', '/tmp/fam_uploads')
|
||
os.makedirs(upload_dir, exist_ok=True)
|
||
video_filename = f"task_{task_id}_{file.filename}"
|
||
video_path = os.path.join(upload_dir, video_filename)
|
||
|
||
try:
|
||
file.save(video_path)
|
||
size_mb = os.path.getsize(video_path) / 1024 / 1024
|
||
logger.info(f"[task_id={task_id}] 入队: {file.filename} ({size_mb:.1f}MB)")
|
||
|
||
queue_id = queue_manager.enqueue(
|
||
nas_task_id=task_id,
|
||
video_filename=file.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,
|
||
}), 202
|
||
|
||
except Exception as e:
|
||
logger.error(f"[task_id={task_id}] 入队失败: {e}", exc_info=True)
|
||
if os.path.exists(video_path):
|
||
os.remove(video_path)
|
||
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:
|
||
# 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块并更新元数据
|
||
import json
|
||
meta_path = os.path.join(d, "meta.json")
|
||
if total_chunks and os.path.exists(meta_path):
|
||
try:
|
||
with open(meta_path) as mf:
|
||
old_meta = json.load(mf)
|
||
if old_meta.get('total_chunks') and old_meta['total_chunks'] != total_chunks:
|
||
logger.warning(f"[task_id={task_id}] total_chunks 变更 "
|
||
f"({old_meta['total_chunks']}→{total_chunks})，清理旧分块")
|
||
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):
|
||
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():
|
||
"""返回已完成但未拉取的结果，标记为已交付"""
|
||
limit = int(request.args.get('limit', 10))
|
||
results = queue_manager.get_undelivered_results(limit=limit)
|
||
|
||
import json
|
||
payload = []
|
||
task_ids = []
|
||
for r in results:
|
||
try:
|
||
result_json = json.loads(r['result_json']) if r['result_json'] else None
|
||
except json.JSONDecodeError:
|
||
result_json = None
|
||
if r['status'] == 'FAILED':
|
||
result_json = {
|
||
"status": "failed",
|
||
"error_message": r['error_message'] or 'unknown',
|
||
"failure_stage": r['failure_stage'] or '',
|
||
}
|
||
payload.append({
|
||
"nas_task_id": r['nas_task_id'],
|
||
"result": result_json,
|
||
})
|
||
task_ids.append(r['id'])
|
||
|
||
if task_ids:
|
||
queue_manager.mark_delivered(task_ids)
|
||
|
||
return jsonify({"results": payload, "count": len(payload)}), 200
|
||
|
||
|
||
@api_bp.route('/api/edge/queue/stats', methods=['GET'])
|
||
def queue_stats():
|
||
"""队列状态统计"""
|
||
stats = queue_manager.get_queue_stats()
|
||
return jsonify(stats), 200
|
||
|
||
|
||
@api_bp.route('/api/edge/video/push', methods=['POST'])
|
||
def receive_push_task():
|
||
"""推送模式：接收 multipart 视频上传，同步分析，结果随 HTTP 响应返回
|
||
|
||
NAS 无法被 Oracle 反向访问（Tailscale 不通），因此改为 NAS 主动上传视频，
|
||
Edge 用 OpenCV 场景变化检测抽帧后分析，摘要直接放在响应里带回。
|
||
"""
|
||
global _currently_processing
|
||
|
||
task_id_raw = request.form.get('task_id')
|
||
file = request.files.get('video')
|
||
if not task_id_raw or not file:
|
||
return jsonify({"error": "缺少必填字段: task_id, video"}), 400
|
||
|
||
try:
|
||
task_id = int(task_id_raw)
|
||
except ValueError:
|
||
return jsonify({"error": "task_id 必须是整数"}), 400
|
||
|
||
logger.info(f"[task_id={task_id}] 收到推送任务: {file.filename}")
|
||
|
||
# 并发控制（同步处理，占用整个请求周期）
|
||
with _current_task_lock:
|
||
if _currently_processing:
|
||
logger.warning(f"[task_id={task_id}] 已有任务处理中，返回 429")
|
||
return jsonify({"error": "Queue full", "retry_after": 60}), 429
|
||
_currently_processing = True
|
||
|
||
preprocessor = None
|
||
try:
|
||
preprocessor = VideoPreprocessor(task_id)
|
||
video_path = preprocessor.save_upload(file)
|
||
|
||
task_data = {
|
||
"task_id": task_id,
|
||
"camera_name": request.form.get('camera_name', ''),
|
||
"event_start_time": request.form.get('event_start_time', ''),
|
||
"event_end_time": request.form.get('event_end_time', ''),
|
||
"known_members_context": request.form.get('known_members_context', ''),
|
||
}
|
||
|
||
result = get_orchestrator().process_push_task(task_data, video_path, preprocessor)
|
||
return jsonify(result), 200
|
||
|
||
except Exception as e:
|
||
logger.error(f"[task_id={task_id}] 推送任务异常: {e}", exc_info=True)
|
||
return jsonify({
|
||
"task_id": task_id, "status": "failed",
|
||
"failure_stage": "upload", "error_message": str(e)
|
||
}), 200
|
||
finally:
|
||
if preprocessor is not None:
|
||
preprocessor.cleanup()
|
||
with _current_task_lock:
|
||
_currently_processing = False
|
||
|
||
|
||
@api_bp.route('/api/edge/mark_frames', methods=['POST'])
|
||
def mark_frames():
|
||
"""NAS 存量关键帧批量补红框（检测计算在 Edge，NAS 只存图）"""
|
||
data = request.get_json(silent=True)
|
||
if not data:
|
||
return jsonify({"error": "Invalid JSON"}), 400
|
||
images = data.get('images')
|
||
if not isinstance(images, list) or not images or len(images) > 12:
|
||
return jsonify({"error": "images 需要 1-12 项 [{key, data}]"}), 400
|
||
|
||
from ..frame_marker import mark_jpeg
|
||
results = []
|
||
for item in images:
|
||
key = item.get('key', '')
|
||
b64 = item.get('data', '')
|
||
try:
|
||
marked, faces = mark_jpeg(base64.b64decode(b64))
|
||
results.append({
|
||
"key": key,
|
||
"data": base64.b64encode(marked).decode('ascii'),
|
||
"faces": faces
|
||
})
|
||
except Exception as e:
|
||
logger.warning(f"补标失败 {key}: {e}")
|
||
results.append({"key": key, "data": None, "faces": 0, "error": str(e)})
|
||
return jsonify({"results": results}), 200
|
||
|
||
|
||
@api_bp.route('/health', methods=['GET'])
|
||
def health():
|
||
"""健康检查"""
|
||
global _currently_processing
|
||
orch = get_orchestrator()
|
||
healthy = orch.health_check_all()
|
||
if not healthy:
|
||
return jsonify({
|
||
"status": "unavailable",
|
||
"healthy_models": [],
|
||
"processing": _currently_processing
|
||
}), 503
|
||
return jsonify({
|
||
"status": "ok",
|
||
"healthy_models": [a.provider_name for a in healthy],
|
||
"processing": _currently_processing
|
||
}), 200
|
||
|
||
|
||
@api_bp.route('/api/edge/chat', methods=['POST'])
|
||
def chat_proxy():
|
||
"""代理转发至本地 Ollama /api/generate（兼容旧调用，Ollama 未对外暴露）"""
|
||
data = request.get_json(silent=True)
|
||
if not data:
|
||
return jsonify({"error": "Invalid JSON"}), 400
|
||
|
||
try:
|
||
resp = requests.post(
|
||
'http://127.0.0.1:11434/api/generate',
|
||
json=data,
|
||
timeout=data.get('options', {}).get('timeout', 120)
|
||
)
|
||
return jsonify(resp.json()), resp.status_code
|
||
except requests.RequestException as e:
|
||
logger.error(f"Chat proxy error: {e}")
|
||
return jsonify({"error": f"Ollama unreachable: {e}"}), 502
|
||
|
||
|
||
@api_bp.route('/api/edge/chat/ask', methods=['POST'])
|
||
def chat_ask():
|
||
"""智能问答编排：Gemini → NVIDIA → 本地 Ollama（两云端都失败才用本地兜底）
|
||
|
||
请求: {"prompt": "..."}
|
||
响应: {"answer": "...", "provider": "gemini"|"nvidia"|"ollama"}
|
||
"""
|
||
data = request.get_json(silent=True)
|
||
if not data or 'prompt' not in data:
|
||
return jsonify({"error": "缺少必填字段: prompt"}), 400
|
||
|
||
prompt = data['prompt']
|
||
max_tokens = int(data.get('max_tokens', 512))
|
||
|
||
answer, provider = get_orchestrator().run_qa(prompt, max_tokens=max_tokens)
|
||
if answer is None:
|
||
return jsonify({
|
||
"error": "所有模型均不可用（Gemini / NVIDIA / Ollama 全部失败）"
|
||
}), 503
|
||
|
||
return jsonify({"answer": answer, "provider": provider}), 200
|