feat(gpu): #1978 AI数字人口型同步接入MuseTalk GPU Worker

业务侧（LipsyncService）集成 GpuLipsyncService： - 新增开关 USE_GPU_LIPSYNC（env，默认false；staging默认true方便联调，生产默认false待稳定后开） - _submit_audio_direct 在签名提交 MediaKit 前判断： * 开关关闭 → 直接走现有 MediaKit 云端 lipsync * 开关开但无可用 Worker（last_heartbeat_at 超5分钟窗口） → 回退 MediaKit * 开关开+Worker可用 → 创建GPU任务→同步轮询等待结果 - GPU路径成功：直接标记 job.status=completed、output_video_url=签名后的结果URL - GPU超时（默认1200s）/终态failed/异常 → 全部回退 MediaKit，对用户透明 - 轮询间隔默认5s，可配 GPU_LIPSYNC_POLL_INTERVAL - Worker心跳新鲜度窗口可配 GPU_WORKER_STALE_SECONDS（默认300s） GpuLipsyncService新增方法： - has_available_worker()：判断5分钟内有心跳的Worker存在 - wait_for_result(task_id, timeout, poll_interval)：同步轮询DB等待终态，期间顺手 _recover_timed_out_tasks；超时返回None让调用方兜底 env/CI配置： - .env.example 新增4个配置项注释 - deploy/configs/.env.staging USE_GPU_LIPSYNC=true（联调） - deploy/configs/.env.production USE_GPU_LIPSYNC=false（暂不开）单元测试：tests/unit/test_lipsync_gpu_integration.py 9个用例覆盖： - 开关关不调用GPU、无Worker回退、GPU成功completed、超时回退、 failed回退、异常回退、has_available_worker三种场景 - 全部 30个GPU相关测试全绿（含此前11+10个）
2026-09-19 16:00:58 +08:00
4 changed files with 63 additions and 913 deletions
@@ -1,148 +1,87 @@
-# MuseTalk GPU Worker 部署指南
+# MuseTalk GPU Worker — 部署指南
-本目录包含两个组件：
+本目录包含 RTX2060 本地电脑上运行的 GPU Worker 脚本。
 Worker 采用 **反向轮询模式**：主动向 SaaS API 拉取待处理的口型同步任务 → 调用本地 MuseTalk 推理 → 把结果视频回传到 SaaS。不需要内网穿透。
-1. **gpu_worker.py**：反向轮询客户端，部署在 RTX2060 本地，轮询 SaaS API 拉取口型任务，调用本地 MuseTalk 服务推理，上传结果回 SaaS。
+## 目录文件
 2. **musetalk_server.py**：MuseTalk Flask HTTP 服务端，接收 gpu_worker.py 的推理请求，调用 MuseTalk 模型生成口型同步视频。
---
+| 文件 | 作用 |
 |---|---|
 | `gpu_worker.py` | Worker 主程序（单文件，零项目代码依赖，仅依赖 `requests`） |
 | `requirements.txt` | Python 依赖（只有 `requests`） |
 | `xiaoxia-gpu-worker.service` | systemd 服务单元（开机自启、异常自动重启） |
 | `.env.example` | 环境变量样例，复制为 `.env` 后填入真实值 |
 ## 一、环境准备
-### 1.1 硬件要求
+1. **Python 3.10+**（Windows 建议从 python.org 安装；Linux 自带）
 2. **本地 MuseTalk 服务** 已启动在 `http://127.0.0.1:7861`，health 接口返回 `{"status":"ok","free_vram_mb":...}`
 3. **ffmpeg**（可选，用于读取输出视频时长；未装则 duration 报 0，不影响功能）
 4. 网络能访问 staging / 生产 API（`curl https://staging-api.xiaoxiajianji.com/health` 应返回 `{"status":"healthy"}`）
- GPU: NVIDIA RTX 2060 或更高（显存 ≥ 6GB）
+## 二、部署步骤（Linux，推荐 systemd）
 - CUDA: 11.8+
 - Python: 3.10+
 - ffmpeg: 需安装并加入 PATH
 ### 1.2 安装依赖
 ```bash
-cd deploy/gpu_worker
+# 1. 创建部署目录
 sudo mkdir -p /opt/xiaoxia-gpu-worker
 sudo chown $USER:$USER /opt/xiaoxia-gpu-worker
 cd /opt/xiaoxia-gpu-worker
 # 2. 拷贝脚本和依赖
 cp /path/to/deploy/gpu_worker/{gpu_worker.py,requirements.txt,xiaoxia-gpu-worker.service,.env.example} .
 cp .env.example .env
 #    编辑 .env，填入 API_BASE_URL 和 GPU_WORKER_TOKEN
 # 3. 创建虚拟环境并安装依赖
 python3 -m venv venv
-source venv/bin/activate
+./venv/bin/pip install -r requirements.txt
-pip install -r requirements.txt
+
 # 4. 前台先跑一次，确认日志正常
 ./venv/bin/python gpu_worker.py
 #    看到 "MuseTalk 健康检查通过" 和 "注册/心跳" 成功即可 Ctrl+C 退出
 # 5. 安装 systemd 服务
 sudo cp xiaoxia-gpu-worker.service /etc/systemd/system/
 sudo systemctl daemon-reload
 sudo systemctl enable --now xiaoxia-gpu-worker
 # 6. 查看日志
 sudo journalctl -u xiaoxia-gpu-worker -f
 ```
---
+## 三、部署步骤（Windows，快速测试）
-## 二、MuseTalk 服务端部署（musetalk_server.py）
+```bat
 :: 创建虚拟环境
 python -m venv venv
 venv\Scripts\pip install -r requirements.txt
-### 2.1 配置环境变量
+:: 复制并编辑 .env
 copy .env.example .env
 notepad .env
-复制 `.env.example` 为 `.env`，修改配置：
+:: 运行
-
+venv\Scripts\python gpu_worker.py
 ```bash
 cp .env.example .env
 vim .env
 ```
-关键配置：
+可在任务计划程序中添加开机启动项：程序选 `venv\Scripts\python.exe`，参数填 `gpu_worker.py`，起始目录填脚本所在目录。
-| 变量 | 说明 | 默认值 |
+## 四、SaaS 侧配套配置
 |------|------|--------|
 | `MUSE_PORT` | 监听端口 | `7861` |
 | `MUSE_INFERENCE_TIMEOUT` | 推理超时秒数 | `600` |
 | `MUSE_VIDEO_MAX_MB` | 视频上传大小限制 MB | `100` |
 | `MUSE_AUDIO_MAX_MB` | 音频上传大小限制 MB | `20` |
 | `MUSE_DEFAULT_FPS` | 视频 fps 兜底值 | `25.0` |
 | `MUSE_TEMP_DIR` | 临时文件目录 | `/tmp/musetalk_$$` |
-### 2.2 启动服务
+SaaS 后端部署完成后需配置：
-```bash
+1. 服务端环境变量 `GPU_WORKER_TOKEN` 设为一个随机强 Token（和 Worker `.env` 中一致）
-# 前台运行（调试用）
+2. 数据库已跑迁移 `081_add_gpu_lipsync_tasks`（自动随 API 启动的 alembic upgrade head 完成）
-python musetalk_server.py
+3. OSS bucket 中 `gpu-lipsync/results/` 路径可写（默认 bucket 已配）
-# 后台运行（生产用 systemd）
+## 五、验证联调
 sudo systemctl start musetalk-server
 sudo systemctl enable musetalk-server
 ```
-### 2.3 验证健康检查
+1. Worker 启动后日志看到 `注册/心跳` 成功
 2. 后端调用 `GpuLipsyncService.create_task(video_url=..., audio_url=...)` 放入一条测试任务
 3. Worker 在 5 秒内拉到任务，下载 → 推理 → 上传 → 上报
 4. 后端 `GET /api/v1/gpu/lipsync/status/{task_id}` 返回 `status=done`，`result_url` 非空
-```bash
+## 六、故障排查
 curl http://127.0.0.1:7861/health
 ```
 应返回：
 ```json
 {
  "status": "healthy",
  "gpu": {
    "gpu_name": "NVIDIA GeForce RTX 2060",
    "memory_total_mb": 6144,
    "memory_used_mb": 1024,
    "memory_free_mb": 5120
  },
  "current_task": {
    "task_id": null,
    "running": false,
    "elapsed_seconds": 0.0
  },
  "timestamp": 1700000000.0
 }
 ```
 ---
 ## 三、GPU Worker 客户端部署（gpu_worker.py）
 ### 3.1 配置环境变量
 复制 `.env.example` 为 `.env`，修改配置：
 ```bash
 cp .env.example .env
 vim .env
 ```
 关键配置：
 | 变量 | 说明 | 默认值 |
 |------|------|--------|
 | `API_BASE_URL` | SaaS API 基础 URL | `https://staging-api.xiaoxiajianji.com` |
 | `GPU_WORKER_TOKEN` | 长期 API Token（与服务端一致） | - |
 | `MUSE_TALK_URL` | 本地 MuseTalk 服务地址 | `http://127.0.0.1:7861` |
 | `POLL_INTERVAL` | 轮询间隔秒 | `5` |
 | `HEARTBEAT_INTERVAL` | 空闲心跳间隔秒 | `15` |
 | `REQUEST_TIMEOUT` | HTTP 请求超时秒 | `900` |
 | `TASK_MAX_RETRY` | 本地最大重试次数 | `1` |
 | `TASK_HEARTBEAT_INTERVAL` | 推理期间任务心跳间隔秒 | `30` |
 | `MIN_VIDEO_DURATION_SECONDS` | 最短输入视频时长秒 | `3` |
 ### 3.2 启动 Worker
 ```bash
 # 前台运行（调试用）
 python gpu_worker.py
 # 后台运行（生产用 systemd）
 sudo systemctl start xiaoxia-gpu-worker
 sudo systemctl enable xiaoxia-gpu-worker
 ```
 ### 3.3 验证启动日志
 应看到：
 ```
 ============================================================
 MuseTalk GPU Worker 启动
  worker_id   = rtx2060-xxxx
  api_base    = https://staging-api.xiaoxiajianji.com
  muse_talk   = http://127.0.0.1:7861
  poll        = 5.0s / heartbeat = 15.0s
 ============================================================
 MuseTalk 健康检查通过: {...}
 注册/心跳成功
 ```
 ---
 ## 四、常见问题排查
 | 现象 | 可能原因 / 排查 |
 |---|---|
@@ -153,34 +92,9 @@ MuseTalk 健康检查通过: {...}
 | 服务端看到任务回退到 pending 重试 | 任务心跳真正超时（默认 900s）：Worker 进程崩溃/断网，或推理彻底卡死；正常长推理期间心跳线程每 30s 续期，不会回退 |
 | 日志 `MuseTalk 推理超时或连接失败` | 视频太长或显存不足；可临时调大 REQUEST_TIMEOUT（服务端 GPU_TASK_TIMEOUT_SECONDS 需同步调大），或限制输入视频时长 |
 | 日志 `视频过短（x.xxs < 3s）` | 输入视频不足 3s，MuseTalk 对短视频会 division by zero，已在本地直接上报失败；可用 MIN_VIDEO_DURATION_SECONDS 调整阈值 |
 | MuseTalk 服务端 503 `GPU 正在处理其他任务` | 并发请求被锁拒绝，等当前推理完成即可 |
 | MuseTalk 服务端 504 `推理超时` | 推理超过 MUSE_INFERENCE_TIMEOUT，客户端会调 /cancel 终止服务端任务 |
---
+## 七、安全注意事项
 ## 五、安全注意事项
 - `.env` 包含长期 Token，文件权限设为 600（`chmod 600 .env`）
 - Token 泄露要立即在服务端更换 `GPU_WORKER_TOKEN` 并重启 Worker
 - Worker 只需要出站访问 SaaS API 和 OSS，不需要开放任何入站端口
 - MuseTalk 服务端只监听本地 127.0.0.1（或 0.0.0.0 但通过防火墙限制），不暴露到公网
 - 临时文件自动清理（推理完成/失败后），无需手动维护
 ---
 ## 六、工程改进记录（musetalk_server.py）
 相比原 `worker.py`，修复了以下 8 个 bug：
 1. **Flask 单线程阻塞**：`app.run(threaded=True)`，推理时 `/health` 仍可响应
 2. **fps=0 除零崩溃**：`_get_video_fps()` 兜底 `MUSE_DEFAULT_FPS`
 3. **ffmpeg 不检查返回码**：`subprocess.run(check=True)` + 超时检查，失败立即报错
 4. **无并发锁**：`threading.Lock` 控制并发，第二请求立即 503
 5. **无推理超时**：线程 join timeout，超时返回 504 并调 `/cancel`
 6. **结果文件不清理**：推理完成/失败后自动删除临时目录
 7. **无人脸检测兜底**：MuseTalk 推理内部处理（TODO: 可在 `_run_inference` 前置检查）
 8. **上传无大小限制**：`_check_file_size()` 校验，超限返回 413
 新增：
 - `/cancel` 端点：终止当前推理任务，清理临时文件
 - `/health` 端点：返回 GPU 显存信息和当前任务状态
@@ -229,26 +229,12 @@ def _call_musetalk(video_path: Path, audio_path: Path, out_path: Path) -> tuple[
        duration = _probe_duration(out_path)
        return True, duration, "", False
    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
-        # 瞬时网络/超时错误，允许本地重试 1 次；同时调 /cancel 让服务端终止僵尸推理
+        # 瞬时网络/超时错误，允许本地重试 1 次
        _cancel_musetalk()
        return False, 0.0, f"MuseTalk 推理超时或连接失败（>{Config.request_timeout}s）", True
    except Exception as exc:
        return False, 0.0, f"MuseTalk 调用异常: {exc}", False
 def _cancel_musetalk() -> None:
    """调 MuseTalk /cancel 端点终止服务端僵尸推理进程，避免超时后任务还在跑占显存."""
    try:
        r = requests.post(f"{Config.muse_talk_url}/cancel", timeout=10)
        if r.status_code == 200:
            logger.info("已调 MuseTalk /cancel，服务端终止推理")
        else:
            logger.warning("MuseTalk /cancel 返回 %d: %s", r.status_code, r.text[:200])
    except Exception as exc:
        # /cancel 失败不应影响主流程上报
        logger.warning("调 MuseTalk /cancel 异常（忽略）: %s", exc)
 def _probe_duration(path: Path) -> float:
    """用 ffprobe 读视频时长（若系统装了 ffmpeg）；否则返回 0."""
    try:
@@ -1,427 +0,0 @@
 """MuseTalk Flask HTTP 服务 — 反向轮询架构的服务端部分.
 部署在 RTX2060 本地，接收 gpu_worker.py 的推理请求，调用 MuseTalk 生成口型同步视频。
 本文件修复了原 worker.py 的 8 个工程 bug，并新增 /cancel 端点。
 环境变量：
  MUSE_PORT            监听端口，默认 7861
  MUSE_MAX_CONCURRENT  最大并发推理数，默认 1（GPU 一次只能处理一个）
  MUSE_INFERENCE_TIMEOUT 推理超时秒数，默认 600
  MUSE_VIDEO_MAX_MB    视频上传大小限制 MB，默认 100
  MUSE_AUDIO_MAX_MB    音频上传大小限制 MB，默认 20
  MUSE_DEFAULT_FPS     视频 fps 兜底值，默认 25.0
  MUSE_TEMP_DIR        临时文件目录，默认 /tmp/musetalk_$$
 接口：
  GET  /health          健康检查 + GPU 显存信息
  POST /inference       推理请求（multipart: video + audio）
  POST /cancel          终止当前推理任务
 """
 from __future__ import annotations
 import atexit
 import logging
 import os
 import shutil
 import signal
 import subprocess
 import threading
 import time
 from pathlib import Path
 from typing import Optional
 from flask import Flask, jsonify, request, send_file
 # ── 日志 ──────────────────────────────────────────────────────────────
 logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger("musetalk-server")
 # ── 配置 ──────────────────────────────────────────────────────────────
 def _env(name: str, default: str = "") -> str:
    v = os.environ.get(name, default)
    return v.strip() if isinstance(v, str) else default
 class Config:
    port: int = int(_env("MUSE_PORT", "7861"))
    max_concurrent: int = int(_env("MUSE_MAX_CONCURRENT", "1"))
    inference_timeout: float = float(_env("MUSE_INFERENCE_TIMEOUT", "600"))
    video_max_mb: int = int(_env("MUSE_VIDEO_MAX_MB", "100"))
    audio_max_mb: int = int(_env("MUSE_AUDIO_MAX_MB", "20"))
    default_fps: float = float(_env("MUSE_DEFAULT_FPS", "25.0"))
    temp_dir: str = _env("MUSE_TEMP_DIR", f"/tmp/musetalk_{os.getpid()}")
 # ── 全局状态 ──────────────────────────────────────────────────────────
 inference_lock = threading.Lock()
 current_task: dict = {"task_id": None, "process": None, "start_time": 0.0}
 shutdown_event = threading.Event()
 # ── Flask App ─────────────────────────────────────────────────────────
 app = Flask(__name__)
 def _cleanup_temp_dir():
    """退出时清理临时目录."""
    if os.path.exists(Config.temp_dir):
        try:
            shutil.rmtree(Config.temp_dir)
            logger.info("已清理临时目录: %s", Config.temp_dir)
        except Exception as exc:
            logger.warning("清理临时目录失败: %s", exc)
 atexit.register(_cleanup_temp_dir)
 def _signal_handler(signum, frame):
    """优雅退出."""
    logger.info("收到信号 %s，准备退出...", signum)
    shutdown_event.set()
    if current_task["process"]:
        logger.info("终止正在进行的推理进程...")
        try:
            current_task["process"].terminate()
            current_task["process"].wait(timeout=5)
        except Exception:
            pass
    _cleanup_temp_dir()
    exit(0)
 signal.signal(signal.SIGTERM, _signal_handler)
 signal.signal(signal.SIGINT, _signal_handler)
 # ── 工具函数 ──────────────────────────────────────────────────────────
 def _get_gpu_info() -> dict:
    """获取 GPU 显存信息（通过 nvidia-smi）."""
    try:
        out = subprocess.check_output(
            [
                "nvidia-smi",
                "--query-gpu=name,memory.total,memory.used,memory.free",
                "--format=csv,noheader,nounits",
            ],
            stderr=subprocess.DEVNULL,
            timeout=5,
        )
        parts = out.decode().strip().split(",")
        if len(parts) >= 4:
            return {
                "gpu_name": parts[0].strip(),
                "memory_total_mb": int(parts[1].strip()),
                "memory_used_mb": int(parts[2].strip()),
                "memory_free_mb": int(parts[3].strip()),
            }
    except Exception as exc:
        logger.warning("nvidia-smi 失败: %s", exc)
    return {"gpu_name": "unknown", "memory_total_mb": 0, "memory_used_mb": 0, "memory_free_mb": 0}
 def _get_video_fps(video_path: Path) -> float:
    """用 ffprobe 读视频帧率，失败或为 0 时返回 default_fps."""
    try:
        out = subprocess.check_output(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=r_frame_rate",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(video_path),
            ],
            stderr=subprocess.DEVNULL,
            timeout=10,
        )
        fps_str = out.decode().strip()
        if "/" in fps_str:
            num, den = fps_str.split("/")
            fps = float(num) / float(den) if float(den) != 0 else 0.0
        else:
            fps = float(fps_str) if fps_str else 0.0
        return fps if fps > 0 else Config.default_fps
    except Exception as exc:
        logger.warning("ffprobe 读 fps 失败: %s，使用默认 %.1f", exc, Config.default_fps)
        return Config.default_fps
 def _check_file_size(file, max_mb: int, label: str) -> Optional[str]:
    """检查文件大小，超限返回错误信息，否则返回 None."""
    file.seek(0, 2)
    size = file.tell()
    file.seek(0)
    max_bytes = max_mb * 1024 * 1024
    if size > max_bytes:
        return f"{label} 文件大小 {size / (1024*1024):.1f}MB 超过限制 {max_mb}MB"
    if size == 0:
        return f"{label} 文件为空"
    return None
 def _run_ffmpeg(cmd: list, timeout: float = 120) -> subprocess.CompletedProcess:
    """运行 ffmpeg 命令，检查返回码和超时."""
    try:
        result = subprocess.run(
            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
 def _run_inference(video_path: Path, audio_path: Path, output_path: Path) -> None:
    """执行 MuseTalk 推理（可被子线程和测试独立调用）.
    实际部署时替换为 MuseTalk 真实推理逻辑。
    此处为示例实现：提取帧 → 合并音视频。
    """
    fps = _get_video_fps(video_path)
    logger.info("视频 fps: %.2f", fps)
    frames_dir = video_path.parent / "frames"
    frames_dir.mkdir(parents=True, exist_ok=True)
    _run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-r",
            str(fps),
            str(frames_dir / "frame_%05d.png"),
        ],
        timeout=120,
    )
    frame_files = sorted(frames_dir.glob("*.png"))
    if not frame_files:
        raise RuntimeError("未从视频中提取到帧")
    # TODO: 替换为 MuseTalk 实际推理逻辑
    logger.warning("使用示例推理逻辑，未实际调用 MuseTalk 模型")
    _run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-i",
            str(audio_path),
            "-c:v",
            "libx264",
            "-c:a",
            "aac",
            "-shortest",
            str(output_path),
        ],
        timeout=300,
    )
    if not output_path.exists() or output_path.stat().st_size < 1024:
        raise RuntimeError("推理产物不存在或过小")
 # ── 路由 ──────────────────────────────────────────────────────────────
@app.route("/health", methods=["GET"])
 def health():
    """健康检查 + GPU 显存信息."""
    gpu_info = _get_gpu_info()
    task_info = {
        "task_id": current_task["task_id"],
        "running": current_task["process"] is not None,
        "elapsed_seconds": time.time() - current_task["start_time"] if current_task["start_time"] else 0.0,
    }
    return jsonify(
        {
            "status": "healthy",
            "gpu": gpu_info,
            "current_task": task_info,
            "timestamp": time.time(),
        }
    )
@app.route("/inference", methods=["POST"])
 def inference():
    """推理请求：multipart form 包含 video 和 audio 文件."""
    # 并发控制：检查锁
    if not inference_lock.acquire(blocking=False):
        return jsonify({"error": "GPU 正在处理其他任务，请稍后重试", "status": "busy"}), 503
    task_id = None
    video_path = None
    audio_path = None
    output_path = None
    try:
        # 解析参数
        if "video" not in request.files or "audio" not in request.files:
            return jsonify({"error": "缺少 video 或 audio 文件"}), 400
        video_file = request.files["video"]
        audio_file = request.files["audio"]
        task_id = request.form.get("task_id", f"task_{int(time.time())}")
        # 文件大小检查
        err = _check_file_size(video_file, Config.video_max_mb, "视频")
        if err:
            return jsonify({"error": err}), 413
        err = _check_file_size(audio_file, Config.audio_max_mb, "音频")
        if err:
            return jsonify({"error": err}), 413
        # 保存到临时目录
        task_dir = Path(Config.temp_dir) / task_id
        task_dir.mkdir(parents=True, exist_ok=True)
        video_path = task_dir / "input.mp4"
        audio_path = task_dir / "input_audio.wav"
        output_path = task_dir / "output.mp4"
        video_file.save(str(video_path))
        audio_file.save(str(audio_path))
        logger.info("开始推理 task_id=%s, video=%s, audio=%s", task_id, video_path.name, audio_path.name)
        # 更新当前任务信息
        current_task["task_id"] = task_id
        current_task["start_time"] = time.time()
        # 启动推理进程（用 subprocess 包装，便于超时终止）
        # 此处直接调用推理函数，实际可改为 subprocess 调用外部脚本
        current_task["process"] = "inference_thread"  # 标记为运行中
        # 在线程中运行推理（支持超时）
        result_container = {"error": None}
        def inference_thread():
            try:
                _run_inference(video_path, audio_path, output_path)
            except Exception as exc:
                result_container["error"] = str(exc)
        thread = threading.Thread(target=inference_thread)
        thread.start()
        thread.join(timeout=Config.inference_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
        if result_container["error"]:
            logger.error("推理失败 task_id=%s: %s", task_id, result_container["error"])
            return jsonify({"error": result_container["error"], "task_id": task_id}), 500
        # 返回结果文件
        logger.info("推理完成 task_id=%s, output=%s", task_id, output_path)
        return send_file(str(output_path), mimetype="video/mp4", as_attachment=True, download_name=f"{task_id}.mp4")
    except Exception as exc:
        logger.exception("推理异常: %s", exc)
        return jsonify({"error": str(exc)}), 500
    finally:
        # 释放锁，清理当前任务信息
        inference_lock.release()
        current_task["task_id"] = None
        current_task["process"] = None
        current_task["start_time"] = 0.0
        # 清理临时文件
        if video_path and video_path.parent.exists():
            try:
                shutil.rmtree(video_path.parent)
                logger.info("已清理临时目录: %s", video_path.parent)
            except Exception as exc:
                logger.warning("清理临时目录失败: %s", exc)
@app.route("/cancel", methods=["POST"])
 def cancel():
    """终止当前正在进行的推理任务."""
    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)
    # 清理临时文件
    task_dir = Path(Config.temp_dir) / task_id
    if task_dir.exists():
        try:
            shutil.rmtree(task_dir)
            logger.info("已清理临时目录: %s", task_dir)
        except Exception as exc:
            logger.warning("清理临时目录失败: %s", exc)
    # 重置当前任务
    current_task["task_id"] = None
    current_task["process"] = None
    current_task["start_time"] = 0.0
    return jsonify({"message": f"已取消任务 {task_id}"})
 # ── 主入口 ────────────────────────────────────────────────────────────
 def main():
    """启动 Flask 服务."""
    # 创建临时目录
    Path(Config.temp_dir).mkdir(parents=True, exist_ok=True)
    logger.info("临时目录: %s", Config.temp_dir)
    # 打印配置
    logger.info("=" * 60)
    logger.info("MuseTalk Flask Server 启动")
    logger.info("  端口: %d", Config.port)
    logger.info("  最大并发: %d", Config.max_concurrent)
    logger.info("  推理超时: %.0fs", Config.inference_timeout)
    logger.info("  视频大小限制: %dMB", Config.video_max_mb)
    logger.info("  音频大小限制: %dMB", Config.audio_max_mb)
    logger.info("  默认 fps: %.1f", Config.default_fps)
    logger.info("=" * 60)
    # 检查 GPU
    gpu_info = _get_gpu_info()
    logger.info("GPU 信息: %s", gpu_info)
    # 启动 Flask（threaded=True 处理并发请求）
    app.run(host="0.0.0.0", port=Config.port, threaded=True)
 if __name__ == "__main__":
    main()
@@ -1,323 +0,0 @@
 """#1970 MuseTalk Flask 服务端 8 项工程 bug 修复单测.
 覆盖 deploy/gpu_worker/musetalk_server.py（独立部署脚本，按文件路径动态加载）：
 1. threaded=True 启动，/health 在推理阻塞时仍可达
 2. fps 兜底：ffprobe 返回 0 或失败时使用 default_fps
 3. ffmpeg 走 subprocess.run(check=True)，失败抛 RuntimeError
 4. 并发锁：推理期间第二请求立即 503
 5. 推理超时：超过 MUSE_INFERENCE_TIMEOUT 返回 504
 6. 结果文件清理：临时目录在请求结束（成功/失败）后删除
 7. 文件大小限制：超过限制返回 413，空文件返回 400
 8. /cancel 端点：终止当前推理，清理临时文件
 """
 from __future__ import annotations
 import importlib.util
 import io
 import os
 import shutil
 import sys
 import threading
 import time
 from pathlib import Path
 from unittest import mock
 import pytest
 # 检查 Flask 是否可用（CI 环境可能没装）
 try:
    import flask  # noqa: F401
    HAS_FLASK = True
 except ImportError:
    HAS_FLASK = False
 pytestmark = pytest.mark.skipif(not HAS_FLASK, reason="Flask 未安装（gpu_worker 独立部署依赖）")
 ROOT = Path(__file__).resolve().parents[2]
 SERVER_PATH = ROOT / "deploy" / "gpu_worker" / "musetalk_server.py"
 def _load_server_module(name: str = "musetalk_server_test"):
    """加载 musetalk_server.py 为独立模块."""
    # 避免重复注册
    if name in sys.modules:
        del sys.modules[name]
    spec = importlib.util.spec_from_file_location(name, SERVER_PATH)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod
@pytest.fixture
 def server(tmp_path, monkeypatch):
    """加载一个干净的 musetalk_server 模块，使用独立临时目录和端口."""
    if not HAS_FLASK:
        pytest.skip("Flask 未安装（gpu_worker 独立部署依赖）")
    monkeypatch.setenv("MUSE_TEMP_DIR", str(tmp_path / "musetalk_temp"))
    monkeypatch.setenv("MUSE_PORT", "0")
    monkeypatch.setenv("MUSE_INFERENCE_TIMEOUT", "2")
    monkeypatch.setenv("MUSE_VIDEO_MAX_MB", "1")
    monkeypatch.setenv("MUSE_AUDIO_MAX_MB", "1")
    monkeypatch.setenv("MUSE_DEFAULT_FPS", "25.0")
    mod_name = f"musetalk_server_test_{os.getpid()}_{id(tmp_path)}"
    mod = _load_server_module(mod_name)
    # 确保配置已更新
    mod.Config.temp_dir = str(tmp_path / "musetalk_temp")
    mod.Config.inference_timeout = 2.0
    mod.Config.video_max_mb = 1
    mod.Config.audio_max_mb = 1
    mod.Config.default_fps = 25.0
    Path(mod.Config.temp_dir).mkdir(parents=True, exist_ok=True)
    # 重置全局状态
    mod.inference_lock = threading.Lock()
    mod.current_task = {"task_id": None, "process": None, "start_time": 0.0}
    return mod
 # ── 1. Flask threaded=True ──────────────────────────────────────────
 def test_flask_run_uses_threaded(server):
    """验证 app.run 调用时 threaded=True."""
    with mock.patch.object(server.app, "run") as mock_run:
        server.main()
        mock_run.assert_called_once()
        call_kwargs = mock_run.call_args
        assert call_kwargs.kwargs.get("threaded") is True
 # ── 2. fps=0 兜底 ───────────────────────────────────────────────────
 def test_get_video_fps_fallback_on_zero(server, tmp_path):
    """ffprobe 返回 0/1 时兜底为 default_fps."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", return_value=b"0/1"):
        fps = server._get_video_fps(fake_video)
        assert fps == 25.0
 def test_get_video_fps_normal(server, tmp_path):
    """正常 fps 解析."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", return_value=b"30/1"):
        fps = server._get_video_fps(fake_video)
        assert abs(fps - 30.0) < 0.01
 def test_get_video_fps_exception_fallback(server, tmp_path):
    """ffprobe 异常时兜底 default_fps."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", side_effect=Exception("no ffprobe")):
        fps = server._get_video_fps(fake_video)
        assert fps == 25.0
 # ── 3. ffmpeg 错误检查 ──────────────────────────────────────────────
 def test_run_ffmpeg_raises_on_nonzero_exit(server):
    """ffmpeg 返回非零应抛 RuntimeError."""
    import subprocess
    with mock.patch(
        "subprocess.run",
        side_effect=subprocess.CalledProcessError(1, "ffmpeg", stderr=b"decode error"),
    ):
        with pytest.raises(RuntimeError, match="ffmpeg 失败"):
            server._run_ffmpeg(["ffmpeg", "-i", "in", "out"])
 def test_run_ffmpeg_raises_on_timeout(server):
    """ffmpeg 超时应抛 RuntimeError."""
    import subprocess
    with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("ffmpeg", 10)):
        with pytest.raises(RuntimeError, match="ffmpeg 超时"):
            server._run_ffmpeg(["ffmpeg", "-i", "in", "out"], timeout=10)
 # ── 4. 并发锁 503 ──────────────────────────────────────────────────
 def test_inference_returns_503_when_busy(server):
    """推理期间第二请求立即 503."""
    server.inference_lock.acquire()
    server.current_task["task_id"] = "task-busy"
    server.current_task["start_time"] = time.time()
    try:
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 503
            assert resp.get_json()["status"] == "busy"
    finally:
        server.inference_lock.release()
        server.current_task = {"task_id": None, "process": None, "start_time": 0.0}
 # ── 5. 推理超时 504 ─────────────────────────────────────────────────
 def test_inference_timeout_returns_504(server):
    """推理超时返回 504."""
    def slow_inference(*args, **kwargs):
        time.sleep(10)  # 远超 2s 超时
    with mock.patch.object(server, "_run_inference", side_effect=slow_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 504
            assert "超时" in resp.get_json()["error"]
 # ── 6. 临时文件清理 ──────────────────────────────────────────────────
 def test_temp_files_cleaned_after_success(server, tmp_path):
    """推理成功后临时目录被清理."""
    def fake_inference(video_path, audio_path, output_path):
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_bytes(b"v" * 2048)
    with mock.patch.object(server, "_run_inference", side_effect=fake_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                    "task_id": "task-cleanup-ok",
                },
                content_type="multipart/form-data",
            )
            # send_file 返回 200 或推理异常 500
            assert resp.status_code in (200, 500)
            task_dir = Path(server.Config.temp_dir) / "task-cleanup-ok"
            assert not task_dir.exists(), f"临时目录 {task_dir} 应被清理"
 def test_temp_files_cleaned_after_failure(server, tmp_path):
    """推理失败后临时目录也被清理."""
    def failing_inference(*args, **kwargs):
        raise RuntimeError("MuseTalk crash")
    with mock.patch.object(server, "_run_inference", side_effect=failing_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                    "task_id": "task-cleanup-fail",
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 500
            task_dir = Path(server.Config.temp_dir) / "task-cleanup-fail"
            assert not task_dir.exists()
 # ── 7. 文件大小限制 ─────────────────────────────────────────────────
 def test_oversize_video_returns_413(server):
    """视频超过大小限制返回 413."""
    big_video = b"v" * (2 * 1024 * 1024)  # 2MB > 1MB limit
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={
                "video": (io.BytesIO(big_video), "v.mp4"),
                "audio": (io.BytesIO(b"a" * 100), "a.wav"),
            },
            content_type="multipart/form-data",
        )
        assert resp.status_code == 413
        assert "超过限制" in resp.get_json()["error"]
 def test_empty_file_returns_400(server):
    """空文件返回 400."""
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={
                "video": (io.BytesIO(b""), "v.mp4"),
                "audio": (io.BytesIO(b"a" * 100), "a.wav"),
            },
            content_type="multipart/form-data",
        )
        assert resp.status_code in (400, 413)
        assert "为空" in resp.get_json().get("error", "") or "超过限制" in resp.get_json().get("error", "")
 def test_missing_file_returns_400(server):
    """缺少必要文件返回 400."""
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={"video": (io.BytesIO(b"v" * 100), "v.mp4")},
            content_type="multipart/form-data",
        )
        assert resp.status_code == 400
 # ── 8. /cancel 端点 ─────────────────────────────────────────────────
 def test_cancel_no_running_task(server):
    """无任务时 /cancel 返回提示."""
    with server.app.test_client() as c:
        resp = c.post("/cancel")
        assert resp.status_code == 200
        assert "无正在运行" in resp.get_json()["message"]
 def test_cancel_terminates_running_task(server, tmp_path):
    """有任务时 /cancel 清理临时目录并重置状态."""
    task_dir = Path(server.Config.temp_dir) / "task-cancel"
    task_dir.mkdir(parents=True, exist_ok=True)
    (task_dir / "some_file.txt").write_text("temp")
    server.current_task["task_id"] = "task-cancel"
    server.current_task["start_time"] = time.time()
    server.current_task["process"] = "inference_thread"
    with server.app.test_client() as c:
        resp = c.post("/cancel")
        assert resp.status_code == 200
        assert "已取消" in resp.get_json()["message"]
        assert not task_dir.exists()
        assert server.current_task["task_id"] is None
        assert server.current_task["process"] is None
        assert server.current_task["start_time"] == 0.0