feat: video analysis switched to push mode (upload whole video, sync response)

Rationale: Oracle cannot reach NAS (Tailscale userspace mode on NAS, no TUN),
the old pull+webhook design requires Edge to download video from NAS and
callback to NAS - both blocked. New design is one-way NAS -> Oracle:

- FAM-Edge: new POST /api/edge/video/push endpoint accepts multipart video
  upload, reuses existing OpenCV scene-change keyframe selection, analyzes
  synchronously and returns the result payload directly in the HTTP response
  (no webhook callback). Old /api/edge/video/analyze kept for compatibility.
- FAM-Edge: VideoPreprocessor.save_upload() saves the uploaded file
- FAM-Edge: AIOrchestrator.process_push_task() runs the full pipeline
  (health check -> extract -> select -> compress -> VLM -> fusion) and
  returns callback-style payload dict
- FAM-Core: Dispatcher rewritten to push mode - reads local video file,
  uploads with task metadata (camera_name, event_start_time from file mtime,
  known_members_context), applies the result to DB via shared
  event_receiver.apply_success_event()
- FAM-Core: event_receiver success logic extracted into reusable
  apply_success_event() (used by both webhook route and dispatcher)
- config: edge_url -> /api/edge/video/push, push_timeout 1800s, gunicorn
  Edge timeout raised to 1800s for long synchronous analysis

fam-core/config/config.yaml

@@ -1,6 +1,7 @@
 # FAM-Core 配置文件 (NAS 端) - 实际部署配置
-# Tailscale: NAS=100.70.234.39, Oracle=100.74.137.126
 # 注: Tailscale 防火墙待修复，当前 edge_url 使用 Oracle 公网 IP
+# 推送模式: NAS 上传视频到 Edge /api/edge/video/push，结果同步随响应返回，
+#          无需 Oracle 反向访问 NAS（webhook/Video-Server 拉取均不再使用）
 # Ollama 未对外暴露，chat_handler 通过 FAM-Edge 代理
 
 server:
@@ -24,12 +25,12 @@ scheduler:
 
 dispatcher:
   poll_interval: 30
-  edge_url: "http://129.146.203.203:5000/api/edge/video/analyze"
+  edge_url: "http://129.146.203.203:5000/api/edge/video/push"
-  webhook_url: "http://100.70.234.39:8000/api/core/callback/event"
   max_retries: 3
+  push_timeout: 1800
 
 video_server:
-  base_url: "http://100.70.234.39:8000/media"
+  base_url: "http://127.0.0.1:8000/media"
   token: "sentinel-media-2026"
   video_dir: "/volume1/surveillance"
 

fam-core/config/config.yaml.example

@@ -21,9 +21,9 @@ scheduler:
 
 dispatcher:
   poll_interval: 30           # 轮询间隔（秒）
-  edge_url: "http://100.x.x.20:5000/api/edge/video/analyze"
+  edge_url: "http://100.x.x.20:5000/api/edge/video/push"   # 推送模式端点（视频上传，同步返回结果）
-  webhook_url: "http://100.x.x.10:8000/api/core/callback/event"
   max_retries: 3
+  push_timeout: 1800          # 推送+分析同步超时（秒）
 
 video_server:
   base_url: "http://100.x.x.10:8000/media"

fam-core/src/fam_core/dispatcher/dispatcher.py

@@ -1,9 +1,15 @@
 """
-Dispatcher - 30s 轮询 PENDING 任务，下发至 Edge
+Dispatcher - 30s 轮询 PENDING 任务，推送视频至 Edge（推送模式）
+
+流程（纯单向通信，NAS → Oracle，无需 Oracle 反向访问 NAS）:
+1. 读取任务对应的本地视频文件
+2. multipart 上传至 Edge /api/edge/video/push（附已知成员清单等元数据）
+3. Edge 同步抽帧+分析，结果直接随 HTTP 响应返回
+4. Dispatcher 收到响应后直接写 monitor_events/event_details，任务标记 SUCCESS
 
 退避重试: min(60 * (retry_count + 1) * 2, 600) 秒
-payload 注入 family_members 表的已命名+未命名成员清单
 """
+import os
 import time
 import threading
 import requests
@@ -12,18 +18,22 @@ from datetime import datetime, timedelta
 from ..logger import setup_logger, log_task
 from ..config_loader import load_config
 from .. import db_layer
+from ..event_receiver.event_receiver import apply_success_event
 
 logger = setup_logger('fam-core.dispatcher')
 
 
 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/analyze')
+        self.edge_url = cfg.get('dispatcher', {}).get('edge_url', 'http://localhost:5000/api/edge/video/push')
         self.max_retries = cfg.get('dispatcher', {}).get('max_retries', 3)
+        # 推送+分析全程同步，耗时较长（大视频上传 + ARM 多帧分析）
+        self.push_timeout = cfg.get('dispatcher', {}).get('push_timeout', 1800)
+        self.camera_name = cfg.get('scheduler', {}).get('camera_name', '默认摄像头')
         self._running = False
         self._thread = None
 
@@ -42,41 +52,83 @@ class Dispatcher:
         return True
 
     def _build_payload(self, task):
-        """构建下发 payload，注入已知成员清单"""
+        """构建推送元数据，注入已知成员清单与事件时间"""
-        known_members = db_layer.get_known_members_context()
+        video_path = task['video_path']
-        return {
+        payload = {
-            "task_id": task['task_id'],
+            "task_id": str(task['task_id']),
-            "video_url": task['video_url'],
+            "camera_name": self.camera_name,
-            "webhook_url": load_config().get('dispatcher', {}).get(
+            "known_members_context": db_layer.get_known_members_context(),
-                'webhook_url', 'http://localhost:8000/api/core/callback/event'
+            "event_start_time": "",
-            ),
+            "event_end_time": "",
-            "known_members_context": known_members
         }
+        # 用文件 mtime 近似事件开始时间
+        try:
+            mtime = os.path.getmtime(video_path)
+            start_dt = datetime.fromtimestamp(mtime)
+            payload["event_start_time"] = start_dt.strftime('%Y-%m-%d %H:%M:%S')
+        except OSError:
+            pass
+        return payload
 
     def _dispatch_one(self, task):
-        """下发单个任务"""
+        """推送单个任务：上传视频 → 同步等结果 → 直接写库"""
         task_id = task['task_id']
+        video_path = task['video_path']
+
+        # 本地视频必须存在，否则直接失败（重试无意义）
+        if not video_path or not os.path.isfile(video_path):
+            db_layer.update_task_status(
+                task_id, 'FAILED',
+                error_message=f"视频文件不存在: {video_path}",
+                failure_stage='upload'
+            )
+            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)
+        db_layer.update_task_status(task_id, 'PROCESSING')
+        log_task(logger, task_id, 'dispatcher',
+                 f'推送视频至 Edge: {self.edge_url} ({size_mb:.1f}MB)')
 
         try:
-            log_task(logger, task_id, 'dispatcher', f'下发至 Edge: {self.edge_url}')
+            with open(video_path, 'rb') as fh:
-            resp = requests.post(self.edge_url, json=payload, timeout=30)
+                resp = requests.post(
-
+                    self.edge_url,
-            if resp.status_code == 202:
+                    data=payload,
-                db_layer.update_task_status(task_id, 'PROCESSING')
+                    files={'video': (os.path.basename(video_path), fh, 'video/mp4')},
-                log_task(logger, task_id, 'dispatcher', 'Edge 接受任务，状态切换为 PROCESSING')
+                    timeout=self.push_timeout
-            elif resp.status_code == 429:
+                )
-                logger.warning(f"[task_id={task_id}] Edge 队列已满 (429)，稍后重试")
+        except requests.RequestException as e:
-            elif resp.status_code == 503:
+            logger.error(f"[task_id={task_id}] 推送失败: {e}")
-                logger.warning(f"[task_id={task_id}] Edge Ollama 不可用 (503)，退避重试")
             self._schedule_retry(task)
-            else:
+            return
+
+        if resp.status_code == 429:
+            logger.warning(f"[task_id={task_id}] Edge 忙 (429)，回到 PENDING 稍后重试")
+            db_layer.update_task_status(task_id, 'PENDING')
+            return
+
+        if resp.status_code != 200:
             logger.error(f"[task_id={task_id}] Edge 返回异常状态码: {resp.status_code}")
             self._schedule_retry(task)
+            return
 
-        except requests.RequestException as e:
+        result = resp.json(silent=True) or {}
-            logger.error(f"[task_id={task_id}] 下发失败: {e}")
+        if result.get('status') == 'success':
-            self._schedule_retry(task)
+            try:
+                event_id = apply_success_event(task_id, result)
+                log_task(logger, task_id, 'dispatcher', f'推送任务完成: event_id={event_id}')
+            except Exception as e:
+                logger.error(f"[task_id={task_id}] 结果落库失败: {e}", exc_info=True)
+                db_layer.update_task_status(
+                    task_id, 'FAILED', error_message=str(e), failure_stage='callback')
+        else:
+            error_message = result.get('error_message', 'unknown')
+            failure_stage = result.get('failure_stage', 'edge')
+            logger.error(f"[task_id={task_id}] Edge 分析失败: stage={failure_stage}, error={error_message}")
+            db_layer.update_task_status(
+                task_id, 'FAILED', error_message=error_message, failure_stage=failure_stage)
 
     def _schedule_retry(self, task):
         """调度重试"""

fam-core/src/fam_core/event_receiver/event_receiver.py

@@ -45,19 +45,13 @@ def _upsert_abstract_members(frame_details: list):
         logger.info(f"upsert family_member: {label} (feature={feature})")
 
 
-@event_bp.route('/api/core/callback/event', methods=['POST'])
+def apply_success_event(task_id, data: dict) -> int:
-def receive_event():
+    """将成功结果写库，返回 event_id
-    """接收 Edge 回调"""
-    data = request.get_json(silent=True)
-    if not data:
-        return jsonify({"error": "Invalid JSON"}), 400
 
-    task_id = data.get('task_id')
+    供两条路径复用:
-    status = data.get('status')
+    - webhook 回调路由 (拉取模式)
-    logger.info(f"[task_id={task_id}] 收到回调: status={status}")
+    - Dispatcher 收到推送模式同步响应后直接落库
-
+    """
-    if status == 'success':
-        try:
     # 1. 插入 monitor_events
     event_id = db_layer.insert_event(
         task_id=task_id,
@@ -91,7 +85,23 @@ def receive_event():
     # 4. 更新任务状态
     db_layer.update_task_status(task_id, 'SUCCESS')
     logger.info(f"[task_id={task_id}] 事件处理完成: event_id={event_id}, frame_details={len(frame_details)}条")
+    return event_id
 
+
+@event_bp.route('/api/core/callback/event', methods=['POST'])
+def receive_event():
+    """接收 Edge 回调"""
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"error": "Invalid JSON"}), 400
+
+    task_id = data.get('task_id')
+    status = data.get('status')
+    logger.info(f"[task_id={task_id}] 收到回调: status={status}")
+
+    if status == 'success':
+        try:
+            event_id = apply_success_event(task_id, data)
             return jsonify({"status": "ok", "event_id": event_id}), 200
 
         except Exception as e:

fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py

@@ -246,7 +246,7 @@ class AIOrchestrator:
             logger.error(f"[task_id={task_id}] 失败回调也失败: {e}")
 
     def process_task(self, task_data: dict):
-        """端到端处理任务"""
+        """端到端处理任务（拉取模式，webhook 回调）"""
         task_id = task_data.get('task_id')
         video_url = task_data.get('video_url')
         webhook_url = task_data.get('webhook_url')
@@ -325,3 +325,82 @@ class AIOrchestrator:
                 preprocessor.cleanup()
 
         return 200
+
+    def process_push_task(self, task_data: dict, video_path: str,
+                          preprocessor: 'VideoPreprocessor') -> dict:
+        """推送模式：同步处理上传的视频，结果直接返回（无 webhook 回调）
+
+        返回 payload 结构与原 webhook 回调一致:
+        - 成功: {task_id, status: "success", event_start_time, ..., frame_details, ...}
+        - 失败: {task_id, status: "failed", failure_stage, error_message}
+        """
+        task_id = task_data.get('task_id')
+        known_members = task_data.get('known_members_context', '')
+        event_start_time = task_data.get('event_start_time', '')
+
+        logger.info(f"[task_id={task_id}] ====== 开始处理推送任务 ======")
+        start_time = time.time()
+
+        try:
+            # 1. 健康检查
+            healthy_adapters = self.health_check_all()
+            if not healthy_adapters:
+                logger.error(f"[task_id={task_id}] 所有模型不健康")
+                return {
+                    "task_id": task_id, "status": "failed",
+                    "failure_stage": "vlm_visual",
+                    "error_message": "All models unhealthy"
+                }
+
+            # 2. 抽帧（视频已由调用方保存到本地，无需下载）
+            candidate_frames = preprocessor.extract_candidate_frames(video_path)
+            if not candidate_frames:
+                raise Exception("抽帧失败，无候选帧")
+
+            key_frames = preprocessor.select_key_frames(candidate_frames)
+            compressed_frames = preprocessor.compress_frames(key_frames)
+            if not compressed_frames:
+                raise Exception("压缩后无可用帧")
+
+            frame_timestamps = preprocessor.compute_timestamps(
+                video_path, len(compressed_frames), event_start_time
+            )
+
+            # 3. 并行视觉分析
+            model_outputs = self.run_visual_analysis(
+                healthy_adapters, compressed_frames, frame_timestamps, known_members
+            )
+            if not model_outputs:
+                raise Exception('All models failed in visual analysis')
+
+            # 4. 文本融合
+            fusion_result = self.run_text_fusion(model_outputs, known_members, task_id)
+
+            total_ms = int((time.time() - start_time) * 1000)
+            log_task(logger, task_id, 'overall', '推送任务完成', duration_ms=total_ms)
+
+            return {
+                "task_id": task_id,
+                "status": "success",
+                "event_start_time": event_start_time,
+                "event_end_time": task_data.get('event_end_time', ''),
+                "camera_name": task_data.get('camera_name', ''),
+                "global_summary": fusion_result.get('global_summary', ''),
+                "entities_json": fusion_result.get('entities_json', []),
+                "frame_details": fusion_result.get('frame_details', []),
+                "compute_provider": fusion_result.get('compute_provider', []),
+                "error_message": None
+            }
+
+        except VLMOutputInvalidError as e:
+            logger.error(f"[task_id={task_id}] VLM 输出解析失败: {e}")
+            return {
+                "task_id": task_id, "status": "failed",
+                "failure_stage": "vlm_fusion", "error_message": str(e)
+            }
+        except Exception as e:
+            logger.error(f"[task_id={task_id}] 推送任务处理失败: {e}", exc_info=True)
+            return {
+                "task_id": task_id, "status": "failed",
+                "failure_stage": "process", "error_message": str(e)
+            }

fam-edge/src/fam_edge/api_gateway/api_gateway.py

@@ -9,6 +9,7 @@ from flask import Blueprint, request, jsonify
 
 from ..logger import setup_logger
 from ..ai_orchestrator.orchestrator import AIOrchestrator
+from ..video_preprocessor.preprocessor import VideoPreprocessor
 
 logger = setup_logger('fam-edge.api_gateway')
 
@@ -71,6 +72,63 @@ def receive_task():
     return jsonify({"status": "accepted", "task_id": task_id}), 202
 
 
+@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('/health', methods=['GET'])
 def health():
     """健康检查"""

fam-edge/src/fam_edge/video_preprocessor/preprocessor.py

@@ -68,6 +68,17 @@ class VideoPreprocessor:
         log_task(logger, self.task_id, 'download', f'下载完成: {size_mb:.1f}MB', duration_ms=duration_ms)
         return self.video_path
 
+    def save_upload(self, file_storage) -> str:
+        """保存推送模式上传的视频文件（multipart），替代 download_video"""
+        os.makedirs(self.work_dir, exist_ok=True)
+        start = time.time()
+        file_storage.save(self.video_path)
+        duration_ms = int((time.time() - start) * 1000)
+        size_mb = os.path.getsize(self.video_path) / (1024 * 1024)
+        log_task(logger, self.task_id, 'upload',
+                 f'保存上传视频: {size_mb:.1f}MB', duration_ms=duration_ms)
+        return self.video_path
+
     def _get_video_duration(self, video_path: str) -> float:
         """用 ffprobe 获取视频时长（秒）"""
         try: