deploy/gpu_worker/musetalk_server.py

@@ -2,6 +2,15 @@
 
 部署在 RTX2060 本地，接收 gpu_worker.py 的推理请求，调用 MuseTalk 生成口型同步视频。
 
+Bug 修复（2026-09-21）：
+  - Bug1: 60fps降帧逻辑 — 输入视频 >30fps 时先降帧至 25fps 推理，推理后用
+    minterpolate MCI 插帧回原帧率，并记录 input_fps/inference_fps
+  - Bug2: 超时终止机制 — 推理线程改为 daemon + abort_event 机制，超时时 set event
+    让推理循环检测退出，同时 kill 所有活跃 ffmpeg 子进程，等线程退出后再释放锁和清理目录
+  - Bug3: /health 接口增加 gfpgan_loaded 和 gfpgan_load_error 字段
+  - Bonus: 动态超时计算（基础120s + 帧数*0.15s，上限1800s）
+  - Bonus: GFPGAN 增强异常时 log warning 而非静默跳过
+
 #2000 关键修复：
   - 集成真实 MuseTalk 推理（替换原有 stub 代码）
   - 音频预处理：22050Hz MP3 → 16kHz mono 16bit WAV（MuseTalk 要求）
@@ -95,6 +104,15 @@ class Config:
 inference_lock = threading.Lock()
 current_task: dict = {"task_id": None, "process": None, "start_time": 0.0}
 shutdown_event = threading.Event()
+# 当前推理线程的终止信号和线程引用
+current_abort_event: Optional[threading.Event] = None
+current_thread: Optional[threading.Thread] = None
+# 全局追踪正在运行的 ffmpeg 子进程（用于超时终止）
+_active_ffmpeg_procs: list[subprocess.Popen] = []
+_active_ffmpeg_lock = threading.Lock()
+# GFPGAN 加载状态（供 /health 接口查询）
+_gfpgan_loaded = False
+_gfpgan_load_error: Optional[str] = None
 
 # ── MuseTalk 模型懒加载 ─────────────────────────────────────────────
 _muse_models = None
@@ -374,21 +392,33 @@ def _check_file_size(file, max_mb: int, label: str) -> Optional[str]:
 
 
 def _run_ffmpeg(cmd: list, timeout: float = 120) -> subprocess.CompletedProcess:
-    """运行 ffmpeg 命令，检查返回码和超时."""
+    """运行 ffmpeg 命令，检查返回码和超时。进程会被注册到全局列表以便外部终止."""
+    proc = None
     try:
-        result = subprocess.run(
+        proc = subprocess.Popen(
             cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            timeout=timeout,
-            check=True,
         )
-        return result
-    except subprocess.CalledProcessError as exc:
-        stderr = exc.stderr.decode(errors="ignore") if exc.stderr else ""
-        raise RuntimeError(f"ffmpeg 失败 (code={exc.returncode}): {stderr[:500]}") from exc
-    except subprocess.TimeoutExpired as exc:
-        raise RuntimeError(f"ffmpeg 超时（>{timeout}s）") from exc
+        with _active_ffmpeg_lock:
+            _active_ffmpeg_procs.append(proc)
+        try:
+            stdout, stderr = proc.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait(timeout=5)
+            raise RuntimeError(f"ffmpeg 超时（>{timeout}s）")
+        if proc.returncode != 0:
+            stderr_text = stderr.decode(errors="ignore") if stderr else ""
+            raise RuntimeError(f"ffmpeg 失败 (code={proc.returncode}): {stderr_text[:500]}")
+        return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr)
+    finally:
+        if proc is not None:
+            with _active_ffmpeg_lock:
+                try:
+                    _active_ffmpeg_procs.remove(proc)
+                except ValueError:
+                    pass
 
 
 # ── MuseTalk 模型加载 ────────────────────────────────────────────────
@@ -401,7 +431,7 @@ def _load_musetalk_models():
     加载到 GPU 后转为 FP16（如果配置开启）以节省显存。
     RTX2060 6G 显存，FP16 大约需要 3-4GB。
     """
-    global _muse_models, _muse_models_loaded, _muse_load_error
+    global _muse_models, _muse_models_loaded, _muse_load_error, _gfpgan_loaded, _gfpgan_load_error
 
     if _muse_models_loaded:
         return _muse_models
@@ -500,13 +530,21 @@ def _load_musetalk_models():
                             gfpgan_model = gfpgan_model.half()
                         gfpgan_model = gfpgan_model.to(device)
                         del gfpgan_ckpt
+                        _gfpgan_loaded = True
+                        _gfpgan_load_error = None
                         logger.info("GFPGAN 加载完成 (FP16=%s)", Config.use_float16)
                     else:
+                        _gfpgan_loaded = False
+                        _gfpgan_load_error = f"模型文件不存在: {gfpgan_path}"
                         logger.warning("GFPGAN 模型不存在: %s，跳过人脸增强", gfpgan_path)
                 except Exception as e:
+                    _gfpgan_loaded = False
+                    _gfpgan_load_error = str(e)
                     logger.warning("GFPGAN 加载失败，跳过人脸增强: %s", e)
                     gfpgan_model = None
             else:
+                _gfpgan_loaded = False
+                _gfpgan_load_error = "已通过环境变量禁用 (MUSE_USE_GFPGAN=0)"
                 logger.info("GFPGAN 已禁用 (MUSE_USE_GFPGAN=0)")
 
             _muse_models = {
@@ -562,6 +600,7 @@ def _run_inference(
     audio_path: Path,
     output_path: Path,
     bbox_shift: int = 0,
+    abort_event: threading.Event = None,
 ) -> None:
     """执行 MuseTalk 真实推理.
 
@@ -678,13 +717,39 @@ def _run_inference(
     audio_processor.feature2chunks = _types.MethodType(_feature2chunks, audio_processor)
 
     fps = _get_video_fps(video_path)
+    original_fps = fps
     audio_duration = _get_media_duration(audio_path)
     video_duration = _get_media_duration(video_path)
     logger.info(
-        "MuseTalk 推理开始: video=%.2fs, audio=%.2fs, fps=%.1f, bbox_shift=%d",
+        "MuseTalk 推理开始: video=%.2fs, audio=%.2fs, input_fps=%.1f, bbox_shift=%d",
         video_duration, audio_duration, fps, bbox_shift,
     )
 
+    # ── Step 0: 高帧率视频降帧（>30fps → 25fps 推理，推理后再插帧回原帧率）──
+    inference_fps = fps
+    video_downsampled = False
+    if fps > 30:
+        inference_fps = 25.0
+        downsampled_video = video_path.parent / "input_25fps.mp4"
+        logger.info("检测到高帧率视频 %.1f fps，降帧至 %.1f fps 进行推理", fps, inference_fps)
+        try:
+            _run_ffmpeg([
+                "ffmpeg", "-y", "-v", "warning",
+                "-i", str(video_path),
+                "-r", str(inference_fps),
+                "-c:v", "libx264", "-preset", "veryfast",
+                "-crf", "18", "-pix_fmt", "yuv420p",
+                str(downsampled_video),
+            ], timeout=120)
+            video_path = downsampled_video
+            video_downsampled = True
+            # 更新帧数和时长
+            total_input_frames_approx = int(audio_duration * inference_fps)
+            logger.info("降帧完成: input_fps=%.1f → inference_fps=%.1f", original_fps, inference_fps)
+        except Exception as e:
+            logger.warning("视频降帧失败，使用原始帧率 %.1f fps 推理: %s", fps, e)
+            inference_fps = fps
+
     # ── Step 1: 音频预处理（关键修复：22050Hz MP3 → 16kHz mono WAV）──
     audio_wav_path = video_path.parent / "audio_16k_mono.wav"
     _preprocess_audio(audio_path, audio_wav_path, target_sr=16000)
@@ -703,7 +768,7 @@ def _run_inference(
     logger.info("人脸检测完成，有效 bbox: %d/%d", sum(1 for c in coord_list if c is not coord_placeholder), total_frames)
 
     # 使用 mirror indexing 循环帧和坐标（避免硬切跳变）
-    num_output_frames = int(audio_duration * fps)
+    num_output_frames = int(audio_duration * inference_fps)
     if num_output_frames <= 0:
         num_output_frames = total_frames
 
@@ -713,7 +778,7 @@ def _run_inference(
     audio_array, _ = librosa.load(str(audio_wav_path), sr=16000, mono=True)
     whisper_features = audio_processor.feature2chunks(
         feature_array=audio_array,
-        fps=fps,
+        fps=inference_fps,
         weight_dtype=(torch.float16 if Config.use_float16 else torch.float32),
         batch_size=Config.batch_size,
     )
@@ -769,6 +834,10 @@ def _run_inference(
     total_batches = (video_num + bs - 1) // bs
 
     for bi in tqdm(range(total_batches), desc="MuseTalk 推理"):
+        # 检查是否收到终止信号
+        if abort_event is not None and abort_event.is_set():
+            logger.warning("收到终止信号，中止推理 (batch %d/%d)", bi, total_batches)
+            raise InterruptedError("推理被外部终止")
         whisper_batch = whisper_features[bi*bs:(bi+1)*bs]
         if len(whisper_batch) == 0:
             break
@@ -818,7 +887,7 @@ def _run_inference(
     _ff_cmd = [
         "ffmpeg", "-y", "-v", "warning",
         "-f", "rawvideo", "-pix_fmt", "bgr24",
-        "-s", f"{frame_w}x{frame_h}", "-r", str(fps),
+        "-s", f"{frame_w}x{frame_h}", "-r", str(inference_fps),
         "-i", "-",
         "-vcodec", "libx264", "-preset", "veryfast",
         "-vf", "format=yuv420p", "-crf", "18",
@@ -826,10 +895,17 @@ def _run_inference(
     ]
     import subprocess as _sp
     _ff_proc = _sp.Popen(_ff_cmd, stdin=_sp.PIPE, stdout=_sp.DEVNULL, stderr=_sp.PIPE)
+    # 注册 ffmpeg 进程到全局列表，以便超时终止
+    with _active_ffmpeg_lock:
+        _active_ffmpeg_procs.append(_ff_proc)
 
     n_out = min(len(res_frame_list), num_output_frames)
     try:
         for i in tqdm(range(n_out), desc="合成帧"):
+            # 检查终止信号
+            if abort_event is not None and abort_event.is_set():
+                logger.warning("收到终止信号，中止合成 (frame %d/%d)", i, n_out)
+                raise InterruptedError("推理被外部终止")
             cyc_i = i % len(coord_cycle)
             x1, y1, x2, y2 = coord_cycle[cyc_i]
             ori_frame = copy.deepcopy(frame_list_cycle[cyc_i])
@@ -863,8 +939,8 @@ def _run_inference(
                     res_frame_resized = cv2.resize(_out_bgr, (_fw, _fh),
                                                     interpolation=cv2.INTER_LANCZOS4)
                     del _face_t, _out, _out_bgr
-                except Exception:
-                    pass
+                except Exception as _gfpgan_err:
+                    logger.warning("GFPGAN 增强失败（帧 %d），使用原图: %s", i, _gfpgan_err)
 
             # face parsing 融合
             try:
@@ -882,27 +958,60 @@ def _run_inference(
 
         _ff_proc.stdin.close()
         _ff_ret = _ff_proc.wait(timeout=120)
+        # 从全局列表中移除已完成的 ffmpeg 进程
+        with _active_ffmpeg_lock:
+            try:
+                _active_ffmpeg_procs.remove(_ff_proc)
+            except ValueError:
+                pass
         if _ff_ret != 0:
             _ff_err = _ff_proc.stderr.read().decode(errors="ignore") if _ff_proc.stderr else ""
             raise RuntimeError(f"ffmpeg编码失败(exit={_ff_ret}): {_ff_err[-300:]}")
     except Exception:
         try: _ff_proc.kill()
         except Exception: pass
+        with _active_ffmpeg_lock:
+            try:
+                _active_ffmpeg_procs.remove(_ff_proc)
+            except ValueError:
+                pass
         raise
 
-    shutil.copy2(str(silent_video_path), str(output_path))
+    # ── Step 8: 高帧率视频插帧还原（如输入 >30fps，从 25fps 插帧回原帧率）──
+    final_video = silent_video_path
+    if video_downsampled:
+        upscaled_video = video_path.parent / "output_upscaled.mp4"
+        logger.info("将推理结果从 %.1f fps 插帧还原至 %.1f fps", inference_fps, original_fps)
+        try:
+            _run_ffmpeg([
+                "ffmpeg", "-y", "-v", "warning",
+                "-i", str(silent_video_path),
+                "-vf", f"minterpolate=mi_mode=mci:mc_mode=aobmc:me_mode=bidir:vsbmc=1:fps={int(original_fps)}",
+                "-c:v", "libx264", "-preset", "veryfast",
+                "-crf", "18", "-pix_fmt", "yuv420p",
+                str(upscaled_video),
+            ], timeout=max(300, int(original_fps * 10)))
+            final_video = upscaled_video
+            logger.info("插帧还原完成: %.1f fps", original_fps)
+        except Exception as e:
+            logger.warning("插帧还原失败，使用 %.1f fps 结果: %s", inference_fps, e)
+            final_video = silent_video_path
+
+    shutil.copy2(str(final_video), str(output_path))
 
     try:
         torch.cuda.empty_cache()
         if audio_wav_path.exists():
             audio_wav_path.unlink()
-        if silent_video_path.exists() and str(silent_video_path) != str(output_path):
-            silent_video_path.unlink()
+        # 清理中间文件
+        for _tmp in [silent_video_path] + ([video_path] if video_downsampled else []):
+            if _tmp.exists() and str(_tmp) != str(output_path):
+                _tmp.unlink()
     except Exception as e:
         logger.warning("清理中间文件失败: %s", e)
 
-    logger.info("MuseTalk 推理完成: output=%s, duration=%.2fs",
-                output_path.name, _get_media_duration(output_path))
+    logger.info("MuseTalk 推理完成: output=%s, duration=%.2fs, input_fps=%.1f, inference_fps=%.1f",
+                output_path.name, _get_media_duration(output_path), original_fps, inference_fps)
 
 
 # ── 路由 ──────────────────────────────────────────────────────────────
@@ -924,6 +1033,8 @@ def health():
             "current_task": task_info,
             "musetalk_loaded": _muse_models_loaded,
             "musetalk_load_error": str(_muse_load_error) if _muse_load_error else None,
+            "gfpgan_loaded": _gfpgan_loaded,
+            "gfpgan_load_error": _gfpgan_load_error,
             "timestamp": time.time(),
         }
     )
@@ -982,24 +1093,52 @@ def inference():
         current_task["start_time"] = time.time()
         current_task["process"] = "inference_thread"  # 标记为运行中
 
-        # 在线程中运行推理（支持超时）
+        # 在线程中运行推理（支持超时 + 终止信号）
         result_container = {"error": None}
+        abort_event = threading.Event()
+        global current_abort_event, current_thread
+        current_abort_event = abort_event
 
         def inference_thread():
             try:
-                _run_inference(video_path, audio_path, output_path, bbox_shift=bbox_shift)
+                _run_inference(video_path, audio_path, output_path, bbox_shift=bbox_shift, abort_event=abort_event)
             except Exception as exc:
                 logger.exception("推理异常: %s", exc)
                 result_container["error"] = str(exc)
 
-        thread = threading.Thread(target=inference_thread)
+        thread = threading.Thread(target=inference_thread, daemon=True)
+        current_thread = thread
         thread.start()
-        thread.join(timeout=Config.inference_timeout)
+
+        # 动态超时：基础 120s + 预估帧数×0.15s，上限 1800s
+        video_fps = _get_video_fps(video_path)
+        video_duration = _get_media_duration(video_path)
+        estimated_frames = int(video_duration * video_fps) if video_duration > 0 else 500
+        dynamic_timeout = min(max(Config.inference_timeout, 120 + estimated_frames * 0.15), 1800)
+
+        thread.join(timeout=dynamic_timeout)
 
         if thread.is_alive():
-            # 超时，终止
-            logger.error("推理超时 (>%ds)，终止任务 %s", Config.inference_timeout, task_id)
-            return jsonify({"error": f"推理超时（>{Config.inference_timeout}s）", "task_id": task_id}), 504
+            # 超时处理：发终止信号 + 杀 ffmpeg 子进程 + 等待线程退出
+            logger.error("推理超时 (>%ds)，终止任务 %s", dynamic_timeout, task_id)
+            abort_event.set()
+            # 终止所有正在运行的 ffmpeg 子进程
+            with _active_ffmpeg_lock:
+                for proc in _active_ffmpeg_procs[:]:
+                    try:
+                        proc.kill()
+                        proc.wait(timeout=5)
+                    except Exception:
+                        pass
+                _active_ffmpeg_procs.clear()
+            # 等待线程退出（daemon 线程会在主进程退出时自动终止，但这里给一定时间让它清理）
+            thread.join(timeout=10)
+            # 标记超时错误，由 finally 统一清理
+            result_container["error"] = f"推理超时（>{dynamic_timeout:.0f}s）"
+            result_container["timeout"] = True
+
+        if result_container.get("timeout"):
+            return jsonify({"error": result_container["error"], "task_id": task_id}), 504
 
         if result_container["error"]:
             logger.error("推理失败 task_id=%s: %s", task_id, result_container["error"])
@@ -1019,6 +1158,8 @@ def inference():
         current_task["task_id"] = None
         current_task["process"] = None
         current_task["start_time"] = 0.0
+        current_abort_event = None
+        current_thread = None
 
         # 清理临时文件
         if video_path and video_path.parent.exists():
@@ -1032,20 +1173,35 @@ def inference():
 @app.route("/cancel", methods=["POST"])
 def cancel():
     """终止当前正在进行的推理任务."""
+    global current_abort_event, current_thread
+
     if current_task["task_id"] is None:
         return jsonify({"message": "当前无正在运行的任务"})
 
     task_id = current_task["task_id"]
     logger.info("收到取消请求，终止任务 %s", task_id)
 
-    # 终止推理进程（如果是 subprocess）
-    if current_task["process"] and current_task["process"] != "inference_thread":
-        try:
-            current_task["process"].terminate()
-            current_task["process"].wait(timeout=5)
-            logger.info("已终止推理进程")
-        except Exception as exc:
-            logger.warning("终止进程失败: %s", exc)
+    # 发送终止信号给推理线程
+    if current_abort_event is not None:
+        current_abort_event.set()
+        logger.info("已发送终止信号给推理线程")
+
+    # 终止所有正在运行的 ffmpeg 子进程
+    with _active_ffmpeg_lock:
+        for proc in _active_ffmpeg_procs[:]:
+            try:
+                proc.kill()
+                proc.wait(timeout=5)
+                logger.info("已终止 ffmpeg 子进程")
+            except Exception as exc:
+                logger.warning("终止 ffmpeg 子进程失败: %s", exc)
+        _active_ffmpeg_procs.clear()
+
+    # 等待推理线程退出（daemon 线程会在主进程退出时自动终止）
+    if current_thread is not None and current_thread.is_alive():
+        current_thread.join(timeout=10)
+        if current_thread.is_alive():
+            logger.warning("推理线程未能在 10s 内退出，将作为 daemon 线程自动终止")
 
     # 清理临时文件
     task_dir = Path(Config.temp_dir) / task_id
@@ -1060,6 +1216,8 @@ def cancel():
     current_task["task_id"] = None
     current_task["process"] = None
     current_task["start_time"] = 0.0
+    current_abort_event = None
+    current_thread = None
 
     return jsonify({"message": f"已取消任务 {task_id}"})