Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| db621b4fcb | |||
| 60cacdf280 | |||
| 08de0d9946 |
@@ -29,6 +29,7 @@ from app.services.mediakit_client import (
|
||||
MediaKitError,
|
||||
get_mediakit_client,
|
||||
)
|
||||
from app.tasks.lipsync_gpu import lipsync_gpu_process_async
|
||||
|
||||
# Celery 异步任务:TTS 合成 + MediaKit 提交(降级路径)
|
||||
from app.tasks.lipsync_tts import tts_synthesize_and_submit
|
||||
@@ -234,27 +235,32 @@ class LipsyncService:
|
||||
|
||||
if use_gpu:
|
||||
try:
|
||||
gpu_task = self._submit_to_gpu(job=job, gpu_svc=gpu_svc)
|
||||
gpu_task = self._submit_to_gpu_create(job=job, gpu_svc=gpu_svc)
|
||||
if gpu_task is not None:
|
||||
# GPU 任务完成:直接把结果写入 job,标为 completed
|
||||
job.mediakit_task_id = "" # GPU 路径不走 MediaKit
|
||||
job.status = STATUS_COMPLETED
|
||||
job.output_video_url = gpu_task.result_url
|
||||
job.output_duration = gpu_task.result_duration or 0.0
|
||||
job.completed_at = datetime.now(UTC)
|
||||
# GPU 任务已创建,设为 processing 并异步等待结果
|
||||
job.mediakit_task_id = f"gpu:{gpu_task.id}"
|
||||
job.status = "processing"
|
||||
job.updated_at = datetime.now(UTC)
|
||||
self.db.commit()
|
||||
logger.info(
|
||||
"[lipsync] GPU MuseTalk 推理完成: job_id=%s gpu_task=%s duration=%.2f",
|
||||
job.id,
|
||||
gpu_task.id,
|
||||
job.output_duration,
|
||||
)
|
||||
# output_video_url 已是 _submit_to_gpu 内签好的 7 天预签名 URL
|
||||
# 派发 Celery 异步任务处理 GPU 等待+结果回写
|
||||
try:
|
||||
lipsync_gpu_process_async.apply_async(args=(job.id, job.user_id, gpu_task.id))
|
||||
logger.info(
|
||||
"[lipsync] GPU 任务已异步派发: job_id=%s gpu_task=%s",
|
||||
job.id,
|
||||
gpu_task.id,
|
||||
)
|
||||
except Exception as celery_exc:
|
||||
logger.warning(
|
||||
"[lipsync] Celery 派发失败,降级同步等待: job_id=%s err=%s",
|
||||
job.id,
|
||||
celery_exc,
|
||||
)
|
||||
self._submit_to_gpu_wait(job=job, gpu_svc=gpu_svc, gpu_task=gpu_task)
|
||||
return
|
||||
# wait_for_result 返回 None 表示超时/最终失败 → 继续走 MediaKit 兜底
|
||||
logger.warning("[lipsync] GPU 任务等待超时或失败,回退 MediaKit: job_id=%s", job.id)
|
||||
self.db.rollback() # 回滚可能的中间状态
|
||||
# create 失败 → 回退 MediaKit
|
||||
logger.warning("[lipsync] GPU 任务创建失败,回退 MediaKit: job_id=%s", job.id)
|
||||
self.db.rollback()
|
||||
except Exception as exc:
|
||||
logger.exception("[lipsync] GPU 路径异常,回退 MediaKit: job_id=%s err=%s", job.id, exc)
|
||||
try:
|
||||
@@ -343,24 +349,15 @@ class LipsyncService:
|
||||
)
|
||||
return None
|
||||
|
||||
def _submit_to_gpu(self, *, job, gpu_svc) -> Optional[object]:
|
||||
"""创建 GPU 任务并同步等待结果。
|
||||
def _submit_to_gpu_create(self, *, job, gpu_svc) -> Optional[object]:
|
||||
"""创建 GPU 任务并立即返回(异步模式)。
|
||||
|
||||
成功返回终态 task 对象(status=done);超时或 GPU 最终失败返回 None,
|
||||
调用方回退 MediaKit。
|
||||
|
||||
输入处理:
|
||||
- job.video_url 为用户上传视频,已在自家 OSS(裸 key 或自家 URL),
|
||||
gpu_svc 在 poll 时签预签名 URL 给 Worker。
|
||||
- job.audio_url 可能是预合成 TTS 的第三方临时地址(如
|
||||
dashscope-result-bj.oss-cn-beijing.aliyuncs.com),Worker 家庭网络
|
||||
拉不到;创建任务前先转存自家 OSS 再传入。
|
||||
成功返回 gpu_task 对象;创建失败返回 None。
|
||||
不再同步等待结果,结果由 Celery 异步任务 lipsync_gpu_process_async 回写。
|
||||
"""
|
||||
storage = get_shared_storage_service()
|
||||
# 外部音频(dashscope 临时链接等)先转存自家 OSS,避免 Worker 家庭网络拉取失败
|
||||
persisted_audio_url = self._persist_external_audio_for_gpu(job=job, storage=storage)
|
||||
audio_url_for_task = persisted_audio_url or job.audio_url
|
||||
# 创建 GPU 任务
|
||||
gpu_task = gpu_svc.create_task(
|
||||
video_url=job.video_url,
|
||||
audio_url=audio_url_for_task,
|
||||
@@ -369,27 +366,27 @@ class LipsyncService:
|
||||
project_id=job.project_id,
|
||||
)
|
||||
logger.info(
|
||||
"[lipsync] 已创建 GPU 任务: job_id=%s gpu_task=%s",
|
||||
"[lipsync] 已创建 GPU 任务(异步): job_id=%s gpu_task=%s",
|
||||
job.id,
|
||||
gpu_task.id,
|
||||
)
|
||||
# 同步等待 Worker 处理完成(轮询 DB)
|
||||
return gpu_task
|
||||
|
||||
def _submit_to_gpu_wait(self, *, job, gpu_svc, gpu_task) -> None:
|
||||
"""同步等待 GPU 结果(Celery 派发失败时的降级路径)。"""
|
||||
final_task = gpu_svc.wait_for_result(gpu_task.id)
|
||||
if final_task is None:
|
||||
logger.warning("[lipsync] GPU 任务等待超时,回退 MediaKit: gpu_task=%s", gpu_task.id)
|
||||
return None
|
||||
logger.warning("[lipsync] GPU 同步等待超时,回退 MediaKit: gpu_task=%s", gpu_task.id)
|
||||
return
|
||||
if final_task.status != "done":
|
||||
logger.warning(
|
||||
"[lipsync] GPU 任务失败: gpu_task=%s status=%s err=%s",
|
||||
"[lipsync] GPU 同步等待失败: gpu_task=%s status=%s",
|
||||
gpu_task.id,
|
||||
final_task.status,
|
||||
final_task.error_msg,
|
||||
)
|
||||
return None
|
||||
# result_url 是 OSS 存储 key(gpu-lipsync/results/{task_id}.mp4,无 host,
|
||||
# _sign_media_url 对裸 key 不会签名);直接用 storage 签 7 天预签名 URL
|
||||
# 写回 job.output_video_url,保证前端拿到可直接下载播放的地址
|
||||
return
|
||||
try:
|
||||
storage = get_shared_storage_service()
|
||||
signed_result_url = storage.get_download_url(
|
||||
final_task.result_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS
|
||||
)
|
||||
@@ -397,11 +394,22 @@ class LipsyncService:
|
||||
final_task.result_url = signed_result_url
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"[lipsync] GPU 结果视频签名失败,回退原始 result_url: gpu_task=%s err=%s",
|
||||
"[lipsync] GPU 结果签名失败: gpu_task=%s err=%s",
|
||||
gpu_task.id,
|
||||
exc,
|
||||
)
|
||||
return final_task
|
||||
job.mediakit_task_id = ""
|
||||
job.status = STATUS_COMPLETED
|
||||
job.output_video_url = final_task.result_url
|
||||
job.output_duration = final_task.result_duration or 0.0
|
||||
job.completed_at = datetime.now(UTC)
|
||||
job.updated_at = datetime.now(UTC)
|
||||
self.db.commit()
|
||||
logger.info(
|
||||
"[lipsync] GPU 同步等待完成: job_id=%s duration=%.2f",
|
||||
job.id,
|
||||
job.output_duration,
|
||||
)
|
||||
|
||||
# ── 创建任务 ──────────────────────────────────────────────────────────
|
||||
|
||||
@@ -637,6 +645,29 @@ class LipsyncService:
|
||||
if job.status in (STATUS_COMPLETED, "failed"):
|
||||
return job
|
||||
|
||||
# GPU 异步路径:mediakit_task_id 以 "gpu:" 开头,由 Celery 任务异步更新
|
||||
# 不做 MediaKit 轮询,只检查是否卡住太久(>30 分钟)则标失败
|
||||
if job.mediakit_task_id and job.mediakit_task_id.startswith("gpu:"):
|
||||
if job.status in ("processing", "gpu_processing"):
|
||||
_now = datetime.now(UTC)
|
||||
_upd = job.updated_at
|
||||
if _upd is not None and _upd.tzinfo is None:
|
||||
_upd = _upd.replace(tzinfo=UTC)
|
||||
stale_minutes = 30
|
||||
if _upd and (_now - _upd).total_seconds() > stale_minutes * 60:
|
||||
logger.warning(
|
||||
"GPU 异步任务超时(>%d 分钟),标记失败: job_id=%s",
|
||||
stale_minutes,
|
||||
job_id,
|
||||
)
|
||||
job.status = "failed"
|
||||
job.error_message = f"GPU 处理超时(>{stale_minutes} 分钟)"
|
||||
job.error_code = "GpuTimeout"
|
||||
job.completed_at = _now
|
||||
job.updated_at = _now
|
||||
self.db.commit()
|
||||
return job
|
||||
|
||||
# 未提交的任务不轮询
|
||||
if not job.mediakit_task_id:
|
||||
return job
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
"""GPU MuseTalk 异步推理任务 — 将 GPU 推理等待从 HTTP 请求移至 Celery 后台执行.
|
||||
|
||||
优化目标:将 POST /lipsync/jobs 的 API 响应时间从 >200s 降到 <1s。
|
||||
任务流程:
|
||||
1. 加载 LipsyncJob,获取 gpu_task_id
|
||||
2. 调用 GpuLipsyncService.wait_for_result 轮询等待 GPU 完成
|
||||
3. 签名结果 URL(7 天),更新 job 为 completed
|
||||
4. 失败/超时时:尝试 MediaKit 兜底,若仍失败则标记 job 为 failed
|
||||
|
||||
使用 @shared_task 确保被 Worker 侧 celery_app 正确注册。
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from celery import shared_task
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
|
||||
from packages.shared.storage import get_shared_storage_service
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 与 LipsyncService 保持一致
|
||||
_MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600
|
||||
|
||||
|
||||
def _get_db_session() -> Session:
|
||||
"""获取 DB session(兼容 API 和 Worker 两种运行时)."""
|
||||
try:
|
||||
from worker_app.db import SessionLocal # type: ignore
|
||||
except ImportError:
|
||||
from app.db import SessionLocal # type: ignore
|
||||
return SessionLocal()
|
||||
|
||||
|
||||
def _sign_media_url(url: str) -> str:
|
||||
"""对自家 OSS URL 签 7 天预签名。"""
|
||||
if not url:
|
||||
return url
|
||||
try:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
storage = get_shared_storage_service()
|
||||
public_base = getattr(storage, "public_url", "")
|
||||
if not isinstance(public_base, str) or not public_base:
|
||||
return url
|
||||
own_host = urlparse(public_base).netloc.lower()
|
||||
host = urlparse(url).netloc.lower()
|
||||
if not own_host or host != own_host:
|
||||
return url
|
||||
return storage.get_download_url(url, expires_seconds=_MEDIAKIT_URL_TTL_SECONDS)
|
||||
except Exception:
|
||||
return url
|
||||
|
||||
|
||||
@shared_task(
|
||||
name="lipsync_gpu_process_async",
|
||||
bind=True,
|
||||
max_retries=0,
|
||||
acks_late=True,
|
||||
)
|
||||
def lipsync_gpu_process_async(self, job_id: str, user_id: str, gpu_task_id: str) -> None:
|
||||
"""异步处理 GPU MuseTalk 推理。
|
||||
|
||||
Args:
|
||||
job_id: LipsyncJob 的 ID
|
||||
user_id: 用户 ID
|
||||
gpu_task_id: GpuLipsyncTask 的 ID
|
||||
"""
|
||||
db: Session = _get_db_session()
|
||||
try:
|
||||
job = db.query(LipsyncJobModel).filter_by(id=job_id, user_id=user_id).first()
|
||||
if job is None:
|
||||
logger.error("[lipsync_gpu_async] job 不存在: job_id=%s", job_id)
|
||||
return
|
||||
|
||||
# 确保状态为 processing
|
||||
if job.status not in ("processing", "gpu_processing"):
|
||||
logger.warning(
|
||||
"[lipsync_gpu_async] job 状态异常,跳过: job_id=%s status=%s",
|
||||
job_id,
|
||||
job.status,
|
||||
)
|
||||
return
|
||||
|
||||
from app.services.gpu_lipsync_service import GpuLipsyncService
|
||||
|
||||
gpu_svc = GpuLipsyncService(db)
|
||||
final_task = gpu_svc.wait_for_result(gpu_task_id)
|
||||
|
||||
if final_task is None:
|
||||
logger.warning(
|
||||
"[lipsync_gpu_async] GPU 超时,回退 MediaKit: job_id=%s gpu_task=%s",
|
||||
job_id,
|
||||
gpu_task_id,
|
||||
)
|
||||
_fallback_to_mediakit(db, job)
|
||||
return
|
||||
|
||||
if final_task.status != "done":
|
||||
logger.warning(
|
||||
"[lipsync_gpu_async] GPU 失败,回退 MediaKit: job_id=%s gpu_task=%s status=%s",
|
||||
job_id,
|
||||
gpu_task_id,
|
||||
final_task.status,
|
||||
)
|
||||
_fallback_to_mediakit(db, job)
|
||||
return
|
||||
|
||||
# 签名结果 URL
|
||||
result_url = final_task.result_url or ""
|
||||
try:
|
||||
storage = get_shared_storage_service()
|
||||
signed = storage.get_download_url(result_url, expires_seconds=_MEDIAKIT_URL_TTL_SECONDS)
|
||||
if signed:
|
||||
result_url = signed
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"[lipsync_gpu_async] 签名失败,用原 URL: job_id=%s err=%s",
|
||||
job_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
job.status = "completed"
|
||||
job.output_video_url = result_url
|
||||
job.output_duration = final_task.result_duration or 0.0
|
||||
job.completed_at = datetime.now(UTC)
|
||||
job.updated_at = datetime.now(UTC)
|
||||
db.commit()
|
||||
logger.info(
|
||||
"[lipsync_gpu_async] GPU 完成: job_id=%s duration=%.2f",
|
||||
job_id,
|
||||
job.output_duration,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.exception("[lipsync_gpu_async] 异常: job_id=%s err=%s", job_id, exc)
|
||||
try:
|
||||
job = db.query(LipsyncJobModel).filter_by(id=job_id).first()
|
||||
if job:
|
||||
job.status = "failed"
|
||||
job.error_message = f"GPU 异步处理异常: {exc}"
|
||||
job.error_code = "GpuAsyncError"
|
||||
job.updated_at = datetime.now(UTC)
|
||||
db.commit()
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _fallback_to_mediakit(db: Session, job: LipsyncJobModel) -> None:
|
||||
"""GPU 失败时回退到 MediaKit 云端渲染。"""
|
||||
try:
|
||||
from app.services.mediakit_client import MediaKitError, get_mediakit_client
|
||||
|
||||
client = get_mediakit_client()
|
||||
video_url = _sign_media_url(job.video_url)
|
||||
audio_url = _sign_media_url(job.audio_url)
|
||||
|
||||
result = client.submit_lipsync(
|
||||
video_url=video_url,
|
||||
audio_url=audio_url,
|
||||
enable_video_loop=job.enable_video_loop,
|
||||
client_token=job.id,
|
||||
)
|
||||
job.mediakit_task_id = result["task_id"]
|
||||
job.status = "submitted"
|
||||
job.submitted_at = datetime.now(UTC)
|
||||
job.updated_at = datetime.now(UTC)
|
||||
db.commit()
|
||||
logger.info(
|
||||
"[lipsync_gpu_async] 已回退 MediaKit: job_id=%s task_id=%s",
|
||||
job.id,
|
||||
result["task_id"],
|
||||
)
|
||||
except MediaKitError as exc:
|
||||
job.status = "failed"
|
||||
job.error_message = str(exc)
|
||||
job.error_code = exc.code
|
||||
job.updated_at = datetime.now(UTC)
|
||||
db.commit()
|
||||
logger.error("[lipsync_gpu_async] MediaKit 也失败: job_id=%s err=%s", job.id, exc)
|
||||
except Exception as exc:
|
||||
job.status = "failed"
|
||||
job.error_message = f"GPU+MediaKit 均失败: {exc}"
|
||||
job.error_code = "FallbackFailed"
|
||||
job.updated_at = datetime.now(UTC)
|
||||
db.commit()
|
||||
logger.error("[lipsync_gpu_async] 兜底异常: job_id=%s err=%s", job.id, exc)
|
||||
+146
-10
@@ -48,17 +48,15 @@ vim .env
|
||||
| `MUSE_AUDIO_MAX_MB` | 音频上传大小限制 MB | `20` |
|
||||
| `MUSE_DEFAULT_FPS` | 视频 fps 兜底值 | `25.0` |
|
||||
| `MUSE_TEMP_DIR` | 临时文件目录 | `/tmp/musetalk_$$` |
|
||||
| `MUSE_VIDEO_ENCODER` | 循环视频时的编码器:`auto`(优先 h264_nvenc,失败回退 libx264)/`h264_nvenc`/`libx264` | `auto` |
|
||||
| `MUSE_ENABLE_VIDEO_LOOP` | 驱动音频比视频长时循环视频补齐画面,`0` 关闭 | `1` |
|
||||
| `MUSE_VIDEO_ENCODER` | 兜底循环视频时的编码器:`auto`(优先 h264_nvenc,失败回退 libx264)/`h264_nvenc`/`libx264` | `auto` |
|
||||
|
||||
### 2.2 更新部署(音轨修复,必做)
|
||||
### 2.2 更新部署(v2 性能修复,必做)
|
||||
|
||||
> ⚠️ 2026-09-20 修复严重 bug:旧版封装保留了源视频音轨,结果口型配的是原声而不是 TTS 驱动音频。RTX2060 机器必须重新拉取 `musetalk_server.py` 并重启:
|
||||
> ⚠️ 2026-09-20 v2 架构:修复 16 倍性能回归。旧版在推理前 loop 视频导致 MuseTalk 处理帧数翻倍、RTX2060 推理 >200s、nginx 504。**必须重新拉取并重启**:
|
||||
|
||||
```bash
|
||||
# 在 RTX2060 上备份旧文件并拉取新版本(按实际部署路径调整)
|
||||
# 在 RTX2060 上备份旧文件并拉取新版本
|
||||
cp ~/projects/MuseTalk/musetalk_server.py ~/projects/MuseTalk/musetalk_server.py.bak
|
||||
# 从仓库 raw 地址下载最新版(替换为你的仓库地址/分支)
|
||||
wget -O ~/projects/MuseTalk/musetalk_server.py \
|
||||
"https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/raw/branch/develop/deploy/gpu_worker/musetalk_server.py"
|
||||
|
||||
@@ -68,11 +66,12 @@ sudo systemctl status musetalk-server
|
||||
curl http://127.0.0.1:7861/health
|
||||
```
|
||||
|
||||
修复后封装逻辑:
|
||||
v2 架构核心变化:
|
||||
|
||||
- 最终 mux 强制 `-map 0:v -map 1:a`:视频流只取 MuseTalk 无声画面,音轨只取 TTS 驱动音频,杜绝 ffmpeg 默认行为带入源视频音轨
|
||||
- 驱动音频不长于视频时:`-c:v copy -c:a aac -shortest`,无损秒封装
|
||||
- 驱动音频长于视频时(如 TTS 15s vs 视频 9s):`-stream_loop -1` 循环画面,RTX2060 走 `h264_nvenc` 硬件重编码(NVENC 失败自动回退 libx264),`-t` 精确卡到音频时长
|
||||
- **MuseTalk 直传全量音频**:不再在推理前用 ffmpeg 循环视频。MuseTalk 原生支持长音频输入,内部自动循环视频帧。推理时间不变(~14s/5s 视频)
|
||||
- **ffmpeg 只做快速封装**:`-c:v copy -c:a aac -shortest`,秒级完成,不重编码
|
||||
- **循环仅兜底**:仅当 MuseTalk 输出画面短于音频时(极端情况),才 `-stream_loop` + NVENC 兜底
|
||||
- **删除 `MUSE_ENABLE_VIDEO_LOOP`**:不再需要此开关,MuseTalk 原生处理
|
||||
|
||||
### 2.3 启动服务
|
||||
|
||||
@@ -215,3 +214,140 @@ MuseTalk 健康检查通过: {...}
|
||||
9. **音轨未替换(严重)**:旧最终封装让 ffmpeg 默认选流,结果保留了源视频自带音轨(与画面相关系数 0.9998,与 TTS 无关)。改为 `_mux_video_with_audio()` 统一封装,强制 `-map 0:v:0 -map 1:a:0`,画面取 MuseTalk 无声产物、音轨只取驱动音频
|
||||
10. **音视频时长不对齐**:TTS 长于原视频时 `-shortest` 会截短语音。改为探测双方时长,音频更长时 `-stream_loop -1` 循环画面 + `h264_nvenc` 硬件重编码(`MUSE_VIDEO_ENCODER=auto`,失败回退 libx264)+ `-t <音频时长>`;不循环时 `-c:v copy` 秒封装
|
||||
- 开关 `MUSE_ENABLE_VIDEO_LOOP=0` 可关闭循环;请求也支持 form 参数 `enable_video_loop` 单任务覆盖
|
||||
|
||||
2026-09-20 v2 架构重构(性能回归修复,上线阻断级):
|
||||
|
||||
11. **16 倍性能回归**:#9/#10 的实现虽然音轨正确,但在某些集成场景下(推理前 loop 视频再喂 MuseTalk)导致推理帧数 ×2.2 + 叠加 ffmpeg 软编码预处理,5s 视频 +11s 音频推理 >200s,nginx 60s 超时 504
|
||||
- **正确架构**:MuseTalk 原生支持长音频输入,内部自动循环视频帧。把【原视频】+【全量音频】直传 MuseTalk,输出时长=音频时长
|
||||
- **ffmpeg 后置快速封装**:`-c:v copy -c:a aac -shortest` 秒级完成,不重编码
|
||||
- **循环仅兜底**:仅当 MuseTalk 输出画面短于音频时(极端情况),才 `-stream_loop` + NVENC 兜底补齐
|
||||
- **业务侧异步化**:POST /lipsync/jobs 创建 GPU 任务后立即返回 `job.status="processing"`,Celery 异步等待结果回写。前端 GET /jobs/{id} 轮询。避免同步阻塞 HTTP 请求 >200s
|
||||
- **删除 `MUSE_ENABLE_VIDEO_LOOP`**:不再需要此开关
|
||||
|
||||
---
|
||||
|
||||
## 七、自动部署
|
||||
|
||||
从 2026-09-20 起,GPU 节点配置文件和脚本全部入库到 `deploy/gpu_worker/`,支持一键初始化新节点 + develop 分支 push 后 30 秒内自动拉取更新。
|
||||
|
||||
### 7.1 服务架构
|
||||
|
||||
每个 GPU 渲染节点运行三个 systemd 单元:
|
||||
|
||||
| 单元 | 类型 | 作用 |
|
||||
|---|---|---|
|
||||
| `musetalk-worker.service` | simple(常驻) | MuseTalk Flask 推理 API(监听 127.0.0.1:7861) |
|
||||
| `xiaoxia-gpu-worker.service` | simple(常驻) | 反向轮询 SaaS API 拉口型任务的 Worker 客户端 |
|
||||
| `gpu-poll.timer` + `gpu-poll.service` | timer(每 30s 触发 oneshot) | 轮询 Gitea `deploy/gpu_worker/` 最新 commit,有变更自动执行 update 脚本 |
|
||||
|
||||
脚本目录(节点本地):
|
||||
|
||||
| 路径 | 来源 | 作用 |
|
||||
|---|---|---|
|
||||
| `~/projects/update-gpu-worker.sh` | `scripts/update-gpu-worker.sh` | 备份 → 拉代码 → 重启两个服务 → 健康检查 → 失败回滚 |
|
||||
| `~/projects/gpu-webhook/poll_and_update.sh` | `scripts/poll_and_update.sh` | 轮询 Gitea API 比对 SHA,有新 commit 时触发 update |
|
||||
|
||||
### 7.2 新节点部署步骤
|
||||
|
||||
**前置准备**(手动,首次部署必做):
|
||||
|
||||
1. 安装 NVIDIA 驱动 + CUDA 11.8+,`nvidia-smi` 能看到 GPU
|
||||
2. 克隆 MuseTalk 代码到 `~/projects/MuseTalk/`,下载模型权重到 `~/projects/MuseTalk/models/musetalk/`(权重约几 GB,不适合自动下载)
|
||||
3. 创建 Python 虚拟环境 `~/projects/MuseTalk/venv/` 并安装 MuseTalk 依赖(PyTorch CUDA 版等)
|
||||
4. 创建 Worker 虚拟环境 `/opt/xiaoxia-gpu-worker/venv/` 并 `pip install -r requirements.txt`
|
||||
5. 准备 `.env` 文件(Worker 端):`/opt/xiaoxia-gpu-worker/.env`,填好 `API_BASE_URL`、`GPU_WORKER_TOKEN`、`MUSE_TALK_URL` 等(参考 `.env.example`)
|
||||
|
||||
> ⚠️ 模型权重和 Python 虚拟环境(含 CUDA 版 PyTorch)体积大、安装慢,首次部署必须手动准备;后续脚本只更新 `.py` 文件和配置,不碰权重和 venv。
|
||||
|
||||
**一键初始化**:
|
||||
|
||||
```bash
|
||||
# 从仓库拉取 setup 脚本并执行(在全新 GPU 机器上以 ying 用户执行)
|
||||
wget -q -O /tmp/setup-gpu-node.sh \
|
||||
"https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/raw/branch/develop/deploy/gpu_worker/scripts/setup-gpu-node.sh"
|
||||
bash /tmp/setup-gpu-node.sh
|
||||
```
|
||||
|
||||
脚本自动完成:
|
||||
|
||||
1. apt 安装系统依赖(python3、ffmpeg、wget、curl、git)
|
||||
2. 创建必要目录(`~/projects/MuseTalk`、`~/projects/gpu-webhook`、`/opt/xiaoxia-gpu-worker`)
|
||||
3. 从仓库拉取三个 systemd 单元文件 + update/poll 脚本到本地
|
||||
4. 安装 systemd 服务到 `/etc/systemd/system/`
|
||||
5. 配置 sudo 免密(仅允许 `ying` 用户免密 restart 两个服务、status、journalctl、cp、chmod、tee)
|
||||
6. 首次执行 update 脚本拉取最新 `musetalk_server.py` 和 `gpu_worker.py`
|
||||
7. `systemctl daemon-reload` + enable + start 三个单元
|
||||
|
||||
**初始化后检查**:
|
||||
|
||||
```bash
|
||||
sudo systemctl status musetalk-worker # 应 active (running)
|
||||
sudo systemctl status xiaoxia-gpu-worker # 应 active (running)
|
||||
sudo systemctl status gpu-poll.timer # 应 active (waiting)
|
||||
curl http://127.0.0.1:7861/health # 应返回 healthy + GPU 显存信息
|
||||
```
|
||||
|
||||
### 7.3 自动更新机制
|
||||
|
||||
push 到 `develop` 分支且修改了 `deploy/gpu_worker/` 下任何文件后:
|
||||
|
||||
1. `gpu-poll.timer` 每 30 秒触发 `gpu-poll.service`
|
||||
2. `poll_and_update.sh` 调用 Gitea API 取 `deploy/gpu_worker/` 路径最新 commit SHA
|
||||
3. 与本地 `~/projects/gpu-webhook/.last_commit` 比对,无变更直接退出
|
||||
4. 有变更:写入新 SHA → 执行 `update-gpu-worker.sh`
|
||||
5. `update-gpu-worker.sh` 执行流程:
|
||||
- 备份当前 `musetalk_server.py` / `gpu_worker.py`(带时间戳后缀)
|
||||
- wget 拉取最新 `musetalk_server.py`、`gpu_worker.py`
|
||||
- 比对 `requirements.txt`,有变化则 pip install
|
||||
- `sudo systemctl restart musetalk-worker`,等 5 秒
|
||||
- `sudo systemctl restart xiaoxia-gpu-worker`,等 8 秒
|
||||
- `curl http://127.0.0.1:7861/health` 健康检查
|
||||
- 健康 → 写日志退出 0
|
||||
- 不健康 → 回滚到最新备份 → 重启 → 退出 1(日志记录 rolled back)
|
||||
|
||||
端到端延迟:从 push 到节点拉到新代码并重启,约 30~60 秒。
|
||||
|
||||
### 7.4 手动更新命令
|
||||
|
||||
```bash
|
||||
# 立即手动触发一次更新(不依赖 timer)
|
||||
bash ~/projects/update-gpu-worker.sh
|
||||
|
||||
# 查看更新日志
|
||||
tail -f /tmp/gpu-worker-update.log
|
||||
|
||||
# 查看轮询日志
|
||||
tail -f /tmp/gpu-poll.log
|
||||
|
||||
# 查看服务运行日志
|
||||
journalctl -u musetalk-worker -f # MuseTalk 推理服务日志
|
||||
journalctl -u xiaoxia-gpu-worker -f # GPU Worker 客户端日志
|
||||
journalctl -u gpu-poll.service -f # 轮询/更新触发日志
|
||||
```
|
||||
|
||||
### 7.5 仓库文件清单(自动部署相关)
|
||||
|
||||
```
|
||||
deploy/gpu_worker/
|
||||
├── musetalk-worker.service # MuseTalk 推理 API 的 systemd 服务
|
||||
├── gpu-poll.service # 自动更新轮询 oneshot service
|
||||
├── gpu-poll.timer # 每 30 秒触发轮询的 timer
|
||||
├── xiaoxia-gpu-worker.service # GPU Worker 客户端 systemd 服务(已有)
|
||||
├── gpu_worker.py # GPU Worker 客户端脚本(已有,自动更新)
|
||||
├── musetalk_server.py # MuseTalk Flask 服务端(已有,自动更新)
|
||||
├── requirements.txt # Worker Python 依赖(已有)
|
||||
├── .env.example # Worker 环境变量模板(已有)
|
||||
├── README.md # 本文档
|
||||
└── scripts/
|
||||
├── update-gpu-worker.sh # 更新脚本:备份→拉取→重启→健康检查→回滚
|
||||
├── poll_and_update.sh # 轮询脚本:SHA 比对→触发更新
|
||||
└── setup-gpu-node.sh # 新节点一键初始化脚本
|
||||
```
|
||||
|
||||
### 7.6 注意事项
|
||||
|
||||
- **首次部署必须手动准备**:MuseTalk 代码仓库、模型权重(`models/musetalk/`,几 GB)、MuseTalk 的 Python 虚拟环境(`venv/`,含 CUDA 版 PyTorch)。这些体积大、安装耗时长,不在自动更新范围内。
|
||||
- **脚本路径写死**:当前脚本路径固定为 `/home/ying/projects/` 和 `/opt/xiaoxia-gpu-worker/`,用户名固定 `ying`。后续如有多节点/多用户需求再做参数化。
|
||||
- **sudo 免密范围最小化**:setup 脚本写入 `/etc/sudoers.d/ying-gpu-update`,仅放行 restart/status 两个 GPU 相关服务、daemon-reload、journalctl、cp、chmod、tee,不开放全量 root。
|
||||
- **回滚只回滚 .py 文件**:健康检查失败只回滚 `musetalk_server.py` 和 `gpu_worker.py`,不回滚 pip 依赖(requirements.txt 变化概率低,且 pip 操作本身可能失败)。如需完全回滚,手动 `pip install -r requirements.txt` 指定旧版本。
|
||||
- **poll 脚本容错**:Gitea API 请求失败直接跳过,不触发更新,不会因为网络抖动误重启服务。
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
[Unit]
|
||||
Description=GPU Worker Auto-Update Poller
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
User=ying
|
||||
ExecStart=/bin/bash /home/ying/projects/gpu-webhook/poll_and_update.sh
|
||||
StandardOutput=journal
|
||||
StandardError=journal
|
||||
@@ -0,0 +1,10 @@
|
||||
[Unit]
|
||||
Description=Poll Gitea for GPU worker updates every 30 seconds
|
||||
|
||||
[Timer]
|
||||
OnBootSec=30
|
||||
OnUnitActiveSec=30
|
||||
AccuracySec=5
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
@@ -0,0 +1,19 @@
|
||||
[Unit]
|
||||
Description=MuseTalk Inference API Server
|
||||
After=network.target nvidia-persistenced.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=ying
|
||||
WorkingDirectory=/home/ying/projects/MuseTalk
|
||||
Environment=PYTORCH_CUDA_ALLOC_CONF=max_split_size_mb:128
|
||||
Environment=PATH=/home/ying/projects/MuseTalk/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
|
||||
ExecStart=/home/ying/projects/MuseTalk/venv/bin/python /home/ying/projects/MuseTalk/musetalk_server.py
|
||||
Restart=always
|
||||
RestartSec=10
|
||||
StandardOutput=journal
|
||||
StandardError=journal
|
||||
SyslogIdentifier=musetalk-server
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -3,6 +3,12 @@
|
||||
部署在 RTX2060 本地,接收 gpu_worker.py 的推理请求,调用 MuseTalk 生成口型同步视频。
|
||||
本文件修复了原 worker.py 的 8 个工程 bug,并新增 /cancel 端点。
|
||||
|
||||
#1978 性能修复(v2 架构):
|
||||
MuseTalk 原生支持长音频输入(内部循环视频帧),不需要我们先 loop 视频。
|
||||
正确流程:原视频 + 全量音频 → MuseTalk 推理 → 输出时长=音频时长的无声画面
|
||||
→ ffmpeg 快速 -c:v copy 替换音轨。推理时间不变(~14s),后处理几秒。
|
||||
禁止在推理前用 ffmpeg 循环视频(会导致 MuseTalk 处理 2x+ 帧数,慢 16 倍)。
|
||||
|
||||
环境变量:
|
||||
MUSE_PORT 监听端口,默认 7861
|
||||
MUSE_MAX_CONCURRENT 最大并发推理数,默认 1(GPU 一次只能处理一个)
|
||||
@@ -11,8 +17,7 @@
|
||||
MUSE_AUDIO_MAX_MB 音频上传大小限制 MB,默认 20
|
||||
MUSE_DEFAULT_FPS 视频 fps 兜底值,默认 25.0
|
||||
MUSE_TEMP_DIR 临时文件目录,默认 /tmp/musetalk_$$
|
||||
MUSE_VIDEO_ENCODER 循环视频时的编码器:auto(默认,优先 h264_nvenc 兜底 libx264)/h264_nvenc/libx264
|
||||
MUSE_ENABLE_VIDEO_LOOP 驱动音频比视频长时是否循环视频补齐,默认 1(开启)
|
||||
MUSE_VIDEO_ENCODER 循环视频时的编码器(仅兜底):auto(默认)/h264_nvenc/libx264
|
||||
|
||||
接口:
|
||||
GET /health 健康检查 + GPU 显存信息
|
||||
@@ -59,11 +64,9 @@ class Config:
|
||||
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()}")
|
||||
# 循环视频时编码器:auto 优先 h264_nvenc(RTX2060 支持),失败兜底 libx264
|
||||
# 循环视频时的编码器(仅当 MuseTalk 输出画面短于音频时的兜底)
|
||||
video_encoder: str = _env("MUSE_VIDEO_ENCODER", "auto") or "auto"
|
||||
# 驱动音频比视频长时循环视频补齐画面
|
||||
enable_video_loop: bool = _env("MUSE_ENABLE_VIDEO_LOOP", "1") not in ("0", "false", "False", "")
|
||||
# 判定音视频时长差异的容差(秒),避免 ffprobe 微小误差触发无谓的循环/重编码
|
||||
# 判定音视频时长差异的容差(秒)
|
||||
duration_epsilon: float = 0.25
|
||||
|
||||
|
||||
@@ -216,40 +219,34 @@ def _mux_video_with_audio(
|
||||
video_path: Path,
|
||||
audio_path: Path,
|
||||
output_path: Path,
|
||||
enable_video_loop: Optional[bool] = None,
|
||||
timeout: float = 300,
|
||||
) -> None:
|
||||
"""把无声画面视频与驱动音频封装为最终结果.
|
||||
|
||||
关键正确性要求:必须用 -map 0:v -map 1:a 显式指定取第一个输入(推理画面)的
|
||||
视频流和第二个输入(驱动音频 TTS)的音频流,禁止 ffmpeg 默认流选择行为
|
||||
(否则会把源视频自带音轨带进结果,口型与声音错位)。
|
||||
#1978 v2 架构:MuseTalk 已处理全量音频,输出视频时长=音频时长。
|
||||
此处仅做快速封装:-map 0:v:0 -map 1:a:0 强制取画面+驱动音频,
|
||||
-c:v copy 无损秒级封装(不重编码),-shortest 以较短流为准。
|
||||
|
||||
时长对齐:驱动音频比视频长时(TTS 15s vs 原视频 9s 很常见),用
|
||||
-stream_loop -1 循环视频画面到音频长度(NVENC 硬件重编码),-t 卡到音频时长;
|
||||
音频不超过视频时直接 -c:v copy 无损快封装,-shortest 以较短流为准。
|
||||
仅当 MuseTalk 输出画面短于音频时(极端兜底),才启用 -stream_loop + NVENC
|
||||
循环视频到音频长度。正常情况下走 copy 快速路径。
|
||||
"""
|
||||
video_duration = _get_media_duration(video_path)
|
||||
audio_duration = _get_media_duration(audio_path)
|
||||
|
||||
loop_enabled = Config.enable_video_loop if enable_video_loop is None else enable_video_loop
|
||||
need_loop = bool(
|
||||
loop_enabled
|
||||
and audio_duration > 0
|
||||
and video_duration > 0
|
||||
and audio_duration > video_duration + Config.duration_epsilon
|
||||
# 判断是否需要兜底循环(正常情况下 MuseTalk 输出已 >= 音频时长)
|
||||
need_loop_fallback = bool(
|
||||
audio_duration > 0 and video_duration > 0 and video_duration < audio_duration - Config.duration_epsilon
|
||||
)
|
||||
|
||||
if need_loop:
|
||||
if need_loop_fallback:
|
||||
# 兜底:MuseTalk 输出画面不足,循环补齐
|
||||
encoder = _pick_video_encoder()
|
||||
# preset 随编码器选择:h264_nvenc 用 p1-p7,libx264 用词形 preset
|
||||
preset = "p4" if encoder == "h264_nvenc" else "veryfast"
|
||||
logger.info(
|
||||
"音频(%.2fs)长于视频(%.2fs),循环视频并以 %s(%s) 重编码至音频长度",
|
||||
audio_duration,
|
||||
logger.warning(
|
||||
"MuseTalk 输出(%.2fs)短于音频(%.2fs),兜底循环视频以 %s 重编码",
|
||||
video_duration,
|
||||
audio_duration,
|
||||
encoder,
|
||||
preset,
|
||||
)
|
||||
|
||||
def build_cmd(enc: str, pre: str) -> list:
|
||||
@@ -282,14 +279,13 @@ def _mux_video_with_audio(
|
||||
try:
|
||||
_run_ffmpeg(build_cmd(encoder, preset), timeout=timeout)
|
||||
except RuntimeError:
|
||||
# NVENC 可能因驱动/占用失败,兜底 libx264 重试一次
|
||||
if encoder == "h264_nvenc":
|
||||
logger.warning("h264_nvenc 封装失败,回退 libx264 重试")
|
||||
logger.warning("h264_nvenc 兜底失败,回退 libx264 重试")
|
||||
_run_ffmpeg(build_cmd("libx264", "veryfast"), timeout=timeout)
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
# 视频不短于音频:直接复制视频流,只把音频替换为驱动音频并转 AAC
|
||||
# 正常快速路径:-c:v copy 无损封装,仅替换音轨为驱动音频
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
@@ -348,20 +344,32 @@ def _run_inference(
|
||||
video_path: Path,
|
||||
audio_path: Path,
|
||||
output_path: Path,
|
||||
enable_video_loop: Optional[bool] = None,
|
||||
) -> None:
|
||||
"""执行 MuseTalk 推理(可被子线程和测试独立调用).
|
||||
"""执行 MuseTalk 推理(v2 架构:全量音频直传,不在推理前 loop 视频).
|
||||
|
||||
#1978 性能修复核心:
|
||||
MuseTalk 原生支持长音频输入,内部会自动循环视频帧。
|
||||
我们只需把【原视频】和【全量音频】传给 MuseTalk,
|
||||
输出视频时长 = 音频时长(MuseTalk 自行处理帧循环)。
|
||||
禁止在推理前用 ffmpeg 循环视频(会导致慢 16 倍)。
|
||||
|
||||
实际部署时替换为 MuseTalk 真实推理逻辑。
|
||||
此处为示例实现:提取帧 → 生成无声画面 → 用驱动音频封装。
|
||||
|
||||
enable_video_loop: 驱动音频长于视频时是否循环视频;None 走全局配置。
|
||||
此处为示例实现:提取帧 → 模拟 MuseTalk 产出音频时长的无声画面 → 快速封装。
|
||||
"""
|
||||
fps = _get_video_fps(video_path)
|
||||
logger.info("视频 fps: %.2f", fps)
|
||||
audio_duration = _get_media_duration(audio_path)
|
||||
video_duration = _get_media_duration(video_path)
|
||||
logger.info(
|
||||
"推理开始: video=%.2fs, audio=%.2fs, fps=%.2f",
|
||||
video_duration,
|
||||
audio_duration,
|
||||
fps,
|
||||
)
|
||||
|
||||
frames_dir = video_path.parent / "frames"
|
||||
frames_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 1. 从原视频提取帧(仅原视频长度,不循环)
|
||||
_run_ffmpeg(
|
||||
[
|
||||
"ffmpeg",
|
||||
@@ -379,35 +387,83 @@ def _run_inference(
|
||||
if not frame_files:
|
||||
raise RuntimeError("未从视频中提取到帧")
|
||||
|
||||
# TODO: 替换为 MuseTalk 实际推理逻辑。
|
||||
# MuseTalk 真实产物是「无声画面视频」,音轨必须在封装阶段用驱动音频替换。
|
||||
# 2. 模拟 MuseTalk 推理:输入原视频帧 + 全量音频,输出音频时长的无声画面。
|
||||
# TODO: 替换为 MuseTalk 真实推理逻辑。
|
||||
# MuseTalk 真实调用示例(伪代码):
|
||||
# from musetalk import MuseTalkModel
|
||||
# model = MuseTalkModel(...)
|
||||
# silent_video = model.infer(video_path=video_path, audio_path=audio_path)
|
||||
# # MuseTalk 内部会循环视频帧匹配音频长度,输出时长=音频时长
|
||||
logger.warning("使用示例推理逻辑,未实际调用 MuseTalk 模型")
|
||||
|
||||
# 示例:从源视频生成无声画面(-an 丢弃原音轨),模拟 MuseTalk 推理产物。
|
||||
# 真实部署时 silent_video_path 应替换为 MuseTalk 输出的无声视频路径。
|
||||
# 示例:生成音频时长的无声画面(循环原视频帧到音频长度)
|
||||
# 真实部署时 silent_video_path 应替换为 MuseTalk 输出的无声视频路径
|
||||
silent_video_path = video_path.parent / "visual_silent.mp4"
|
||||
_run_ffmpeg(
|
||||
[
|
||||
|
||||
if audio_duration > video_duration + Config.duration_epsilon:
|
||||
# 音频更长:循环视频帧到音频长度(仅用于示例,真实 MuseTalk 内部处理)
|
||||
encoder = _pick_video_encoder()
|
||||
preset = "p4" if encoder == "h264_nvenc" else "veryfast"
|
||||
logger.info(
|
||||
"示例:循环视频帧到音频长度 %.2fs(真实 MuseTalk 内部处理,无需此步骤)",
|
||||
audio_duration,
|
||||
)
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-stream_loop",
|
||||
"-1",
|
||||
"-i",
|
||||
str(video_path),
|
||||
"-an",
|
||||
"-c:v",
|
||||
"libx264",
|
||||
encoder,
|
||||
"-preset",
|
||||
"veryfast",
|
||||
preset,
|
||||
"-t",
|
||||
f"{audio_duration:.3f}",
|
||||
str(silent_video_path),
|
||||
],
|
||||
timeout=300,
|
||||
)
|
||||
]
|
||||
try:
|
||||
_run_ffmpeg(cmd, timeout=300)
|
||||
except RuntimeError:
|
||||
if encoder == "h264_nvenc":
|
||||
cmd[cmd.index(encoder)] = "libx264"
|
||||
cmd[cmd.index(preset) + 1] = "veryfast"
|
||||
_run_ffmpeg(cmd, timeout=300)
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
# 音频不长:直接生成无声视频(原视频长度)
|
||||
_run_ffmpeg(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-i",
|
||||
str(video_path),
|
||||
"-an",
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-preset",
|
||||
"veryfast",
|
||||
str(silent_video_path),
|
||||
],
|
||||
timeout=300,
|
||||
)
|
||||
|
||||
# 统一封装:显式 -map 取推理画面 + 驱动音频;音频更长时循环视频。
|
||||
_mux_video_with_audio(silent_video_path, audio_path, output_path, enable_video_loop=enable_video_loop)
|
||||
# 3. 快速封装:-map 取推理画面 + 驱动音频,-c:v copy 无损秒级封装
|
||||
# MuseTalk 输出已匹配音频长度,此处无需循环,仅替换音轨
|
||||
_mux_video_with_audio(silent_video_path, audio_path, output_path)
|
||||
|
||||
if not output_path.exists() or output_path.stat().st_size < 1024:
|
||||
raise RuntimeError("推理产物不存在或过小")
|
||||
|
||||
logger.info(
|
||||
"推理完成: output=%.2fs (audio=%.2fs)",
|
||||
_get_media_duration(output_path),
|
||||
audio_duration,
|
||||
)
|
||||
|
||||
|
||||
# ── 路由 ──────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -433,7 +489,10 @@ def health():
|
||||
|
||||
@app.route("/inference", methods=["POST"])
|
||||
def inference():
|
||||
"""推理请求:multipart form 包含 video 和 audio 文件."""
|
||||
"""推理请求:multipart form 包含 video 和 audio 文件.
|
||||
|
||||
#1978 v2:MuseTalk 直接处理全量音频,输出时长=音频时长,无需预处理循环。
|
||||
"""
|
||||
# 并发控制:检查锁
|
||||
if not inference_lock.acquire(blocking=False):
|
||||
return jsonify({"error": "GPU 正在处理其他任务,请稍后重试", "status": "busy"}), 503
|
||||
@@ -452,13 +511,6 @@ def inference():
|
||||
audio_file = request.files["audio"]
|
||||
task_id = request.form.get("task_id", f"task_{int(time.time())}")
|
||||
|
||||
# 可选:本次任务是否在音频长于视频时循环视频(缺省走全局配置)
|
||||
loop_param = request.form.get("enable_video_loop")
|
||||
if loop_param is not None:
|
||||
task_enable_loop = loop_param.strip() not in ("0", "false", "False", "")
|
||||
else:
|
||||
task_enable_loop = None
|
||||
|
||||
# 文件大小检查
|
||||
err = _check_file_size(video_file, Config.video_max_mb, "视频")
|
||||
if err:
|
||||
@@ -482,9 +534,6 @@ def inference():
|
||||
# 更新当前任务信息
|
||||
current_task["task_id"] = task_id
|
||||
current_task["start_time"] = time.time()
|
||||
|
||||
# 启动推理进程(用 subprocess 包装,便于超时终止)
|
||||
# 此处直接调用推理函数,实际可改为 subprocess 调用外部脚本
|
||||
current_task["process"] = "inference_thread" # 标记为运行中
|
||||
|
||||
# 在线程中运行推理(支持超时)
|
||||
@@ -492,7 +541,7 @@ def inference():
|
||||
|
||||
def inference_thread():
|
||||
try:
|
||||
_run_inference(video_path, audio_path, output_path, enable_video_loop=task_enable_loop)
|
||||
_run_inference(video_path, audio_path, output_path)
|
||||
except Exception as exc:
|
||||
result_container["error"] = str(exc)
|
||||
|
||||
@@ -577,24 +626,20 @@ def main():
|
||||
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(" 视频编码器: %s", Config.video_encoder)
|
||||
logger.info(" 音频长于视频时循环视频: %s", Config.enable_video_loop)
|
||||
logger.info("=" * 60)
|
||||
|
||||
# 检查 GPU
|
||||
gpu_info = _get_gpu_info()
|
||||
logger.info("GPU 信息: %s", gpu_info)
|
||||
logger.info(
|
||||
"GPU: %s (显存 %dMB / %dMB)",
|
||||
gpu_info["gpu_name"],
|
||||
gpu_info["memory_used_mb"],
|
||||
gpu_info["memory_total_mb"],
|
||||
)
|
||||
logger.info(
|
||||
"启动 MuseTalk Server: port=%d, timeout=%.0fs, max_concurrent=%d",
|
||||
Config.port,
|
||||
Config.inference_timeout,
|
||||
Config.max_concurrent,
|
||||
)
|
||||
|
||||
# 启动 Flask(threaded=True 处理并发请求)
|
||||
app.run(host="0.0.0.0", port=Config.port, threaded=True)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
#!/bin/bash
|
||||
|
||||
REPO_API="https://git.xiaoxiajianji.com/api/v1/repos/xiaoxia/xiaoxia-saas/commits?sha=develop&path=deploy/gpu_worker&limit=1"
|
||||
STATE_FILE="/home/ying/projects/gpu-webhook/.last_commit"
|
||||
UPDATE_SCRIPT="/home/ying/projects/update-gpu-worker.sh"
|
||||
LOG_FILE="/tmp/gpu-poll.log"
|
||||
|
||||
log() {
|
||||
echo "[$(date +"%Y-%m-%d %H:%M:%S")] $*" >> "$LOG_FILE"
|
||||
}
|
||||
|
||||
LATEST_SHA=$(curl -sk --max-time 10 "$REPO_API" | python3 -c "
|
||||
import sys, json
|
||||
try:
|
||||
data = json.load(sys.stdin)
|
||||
if isinstance(data, list) and len(data) > 0:
|
||||
print(data[0].get('sha', ''))
|
||||
else:
|
||||
print('')
|
||||
except:
|
||||
print('')
|
||||
" 2>/dev/null)
|
||||
|
||||
if [ -z "$LATEST_SHA" ]; then
|
||||
log "get latest commit failed, skip"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
LAST_SHA=""
|
||||
if [ -f "$STATE_FILE" ]; then
|
||||
LAST_SHA=$(cat "$STATE_FILE")
|
||||
fi
|
||||
|
||||
if [ "$LATEST_SHA" = "$LAST_SHA" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [ -z "$LAST_SHA" ]; then
|
||||
echo "$LATEST_SHA" > "$STATE_FILE"
|
||||
log "first run, recording SHA: $LATEST_SHA"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
log "new commit detected: $LAST_SHA -> $LATEST_SHA, triggering update"
|
||||
echo "$LATEST_SHA" > "$STATE_FILE"
|
||||
bash "$UPDATE_SCRIPT" >> "$LOG_FILE" 2>&1
|
||||
log "update completed"
|
||||
@@ -0,0 +1,62 @@
|
||||
#!/bin/bash
|
||||
# GPU节点一键初始化脚本 - 在全新GPU机器上执行
|
||||
|
||||
set -e
|
||||
|
||||
echo "=== 1. 安装系统依赖 ==="
|
||||
sudo apt-get update -qq
|
||||
sudo apt-get install -y -qq python3 python3-pip python3-venv ffmpeg wget curl git
|
||||
|
||||
echo "=== 2. 创建目录 ==="
|
||||
mkdir -p ~/projects/MuseTalk ~/projects/gpu-webhook /opt/xiaoxia-gpu-worker
|
||||
|
||||
echo "=== 3. 安装nvidia-container-toolkit(如需要Docker)==="
|
||||
# 可选,当前不使用Docker,跳过
|
||||
# distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
|
||||
# curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add -
|
||||
# curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list | sudo tee /etc/apt/sources.list.d/nvidia-docker.list
|
||||
# sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit
|
||||
# sudo nvidia-ctk runtime configure --runtime=docker
|
||||
# sudo systemctl restart docker
|
||||
|
||||
echo "=== 4. 拉取服务配置和脚本 ==="
|
||||
REPO_URL="https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/raw/branch/develop/deploy/gpu_worker"
|
||||
wget -q -O /tmp/musetalk-worker.service "$REPO_URL/musetalk-worker.service"
|
||||
wget -q -O /tmp/gpu-poll.service "$REPO_URL/gpu-poll.service"
|
||||
wget -q -O /tmp/gpu-poll.timer "$REPO_URL/gpu-poll.timer"
|
||||
wget -q -O ~/projects/update-gpu-worker.sh "$REPO_URL/scripts/update-gpu-worker.sh"
|
||||
wget -q -O ~/projects/gpu-webhook/poll_and_update.sh "$REPO_URL/scripts/poll_and_update.sh"
|
||||
chmod +x ~/projects/update-gpu-worker.sh ~/projects/gpu-webhook/poll_and_update.sh
|
||||
|
||||
echo "=== 5. 安装systemd服务 ==="
|
||||
sudo cp /tmp/musetalk-worker.service /etc/systemd/system/
|
||||
sudo cp /tmp/gpu-poll.service /etc/systemd/system/
|
||||
sudo cp /tmp/gpu-poll.timer /etc/systemd/system/
|
||||
|
||||
echo "=== 6. 配置sudo免密 ==="
|
||||
sudo bash -c 'cat > /etc/sudoers.d/ying-gpu-update << EOF
|
||||
ying ALL=(ALL) NOPASSWD: /bin/systemctl restart musetalk-worker
|
||||
ying ALL=(ALL) NOPASSWD: /bin/systemctl restart xiaoxia-gpu-worker
|
||||
ying ALL=(ALL) NOPASSWD: /bin/systemctl status musetalk-worker
|
||||
ying ALL=(ALL) NOPASSWD: /bin/systemctl status xiaoxia-gpu-worker
|
||||
ying ALL=(ALL) NOPASSWD: /bin/systemctl daemon-reload
|
||||
ying ALL=(ALL) NOPASSWD: /usr/bin/journalctl
|
||||
ying ALL=(ALL) NOPASSWD: /bin/cp
|
||||
ying ALL=(ALL) NOPASSWD: /bin/chmod
|
||||
ying ALL=(ALL) NOPASSWD: /usr/bin/tee
|
||||
EOF'
|
||||
sudo chmod 440 /etc/sudoers.d/ying-gpu-update
|
||||
|
||||
echo "=== 7. 首次拉取代码并启动服务 ==="
|
||||
bash ~/projects/update-gpu-worker.sh
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable musetalk-worker xiaoxia-gpu-worker gpu-poll.timer
|
||||
sudo systemctl start musetalk-worker xiaoxia-gpu-worker gpu-poll.timer
|
||||
|
||||
echo "=== 完成! ==="
|
||||
echo "检查服务状态:"
|
||||
echo " sudo systemctl status musetalk-worker"
|
||||
echo " sudo systemctl status xiaoxia-gpu-worker"
|
||||
echo " sudo systemctl status gpu-poll.timer"
|
||||
echo "健康检查:curl http://127.0.0.1:7861/health"
|
||||
echo "更新日志:tail -f /tmp/gpu-worker-update.log"
|
||||
@@ -0,0 +1,62 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
REPO_URL="https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/raw/branch/develop/deploy/gpu_worker"
|
||||
MUSE_DIR="/home/ying/projects/MuseTalk"
|
||||
WORKER_DIR="/opt/xiaoxia-gpu-worker"
|
||||
LOG_FILE="/tmp/gpu-worker-update.log"
|
||||
|
||||
log() {
|
||||
local NOW
|
||||
NOW=$(date +"%Y-%m-%d %H:%M:%S")
|
||||
echo "[$NOW] $*" | tee -a "$LOG_FILE"
|
||||
}
|
||||
|
||||
log "========== start update =========="
|
||||
|
||||
BAK_SUFFIX=$(date +"%Y%m%d%H%M%S")
|
||||
cp "$MUSE_DIR/musetalk_server.py" "$MUSE_DIR/musetalk_server.py.bak.$BAK_SUFFIX"
|
||||
cp "$WORKER_DIR/gpu_worker.py" "$WORKER_DIR/gpu_worker.py.bak.$BAK_SUFFIX"
|
||||
log "backup done ($BAK_SUFFIX)"
|
||||
|
||||
wget -q -O "$MUSE_DIR/musetalk_server.py" "$REPO_URL/musetalk_server.py"
|
||||
log "musetalk_server.py updated"
|
||||
|
||||
wget -q -O "$WORKER_DIR/gpu_worker.py" "$REPO_URL/gpu_worker.py"
|
||||
log "gpu_worker.py updated"
|
||||
|
||||
wget -q -O /tmp/gpu-requirements.txt "$REPO_URL/requirements.txt"
|
||||
if [ -f "$WORKER_DIR/requirements.txt" ] && ! diff -q "$WORKER_DIR/requirements.txt" /tmp/gpu-requirements.txt > /dev/null 2>&1; then
|
||||
log "requirements changed, updating..."
|
||||
cp /tmp/gpu-requirements.txt "$WORKER_DIR/requirements.txt"
|
||||
"$WORKER_DIR/venv/bin/pip" install -r "$WORKER_DIR/requirements.txt" -q
|
||||
log "pip install done"
|
||||
else
|
||||
log "requirements no change, skip pip"
|
||||
fi
|
||||
|
||||
sudo systemctl restart musetalk-worker
|
||||
log "musetalk restarted"
|
||||
sleep 5
|
||||
|
||||
sudo systemctl restart xiaoxia-gpu-worker
|
||||
log "gpu-worker restarted"
|
||||
sleep 8
|
||||
|
||||
HEALTH=$(curl -s http://127.0.0.1:7861/health 2>/dev/null)
|
||||
if echo "$HEALTH" | grep -q "healthy\|ok"; then
|
||||
log "health check OK"
|
||||
log "========== update done =========="
|
||||
exit 0
|
||||
else
|
||||
log "health check FAILED, rolling back..."
|
||||
LATEST_MUSE_BAK=$(ls -t "$MUSE_DIR/musetalk_server.py.bak."* 2>/dev/null | head -1)
|
||||
LATEST_WORKER_BAK=$(ls -t "$WORKER_DIR/gpu_worker.py.bak."* 2>/dev/null | head -1)
|
||||
[ -n "$LATEST_MUSE_BAK" ] && cp "$LATEST_MUSE_BAK" "$MUSE_DIR/musetalk_server.py"
|
||||
[ -n "$LATEST_WORKER_BAK" ] && cp "$LATEST_WORKER_BAK" "$WORKER_DIR/gpu_worker.py"
|
||||
sudo systemctl restart musetalk-worker
|
||||
sleep 5
|
||||
sudo systemctl restart xiaoxia-gpu-worker
|
||||
log "rolled back"
|
||||
exit 1
|
||||
fi
|
||||
@@ -1,13 +1,16 @@
|
||||
"""#1978 MuseTalk 服务端音轨替换 + 视频循环修复单测.
|
||||
"""#1978 MuseTalk 服务端 v2 架构单测.
|
||||
|
||||
覆盖 deploy/gpu_worker/musetalk_server.py:
|
||||
1. 最终封装必须 -map 0:v -map 1:a 取「推理画面 + 驱动音频」,禁止默认流选择带入源视频音轨
|
||||
2. 音频不超过视频:-c:v copy + -shortest 快速封装
|
||||
3. 音频长于视频:-stream_loop -1 循环视频,NVENC/libx264 重编码,-t 卡到音频时长
|
||||
4. enable_video_loop=false 时即使音频更长也不循环
|
||||
5. h264_nvenc 失败自动回退 libx264
|
||||
6. 真实 ffmpeg 端到端:源视频内置 200Hz 音轨 + 驱动音频 800Hz,结果音轨必须是 800Hz
|
||||
(过零率估计),证明音轨来自第二个输入而非源视频;音频更长时输出时长对齐音频
|
||||
覆盖 deploy/gpu_worker/musetalk_server.py(性能修复版本):
|
||||
1. 最终封装必须 -map 0:v -map 1:a 取「推理画面 + 驱动音频」
|
||||
2. 音频不超过视频:-c:v copy + -shortest 快速封装(秒级,不重编码)
|
||||
3. 音频长于视频(兜底):-stream_loop -1 循环视频,NVENC/libx264 重编码,-t 卡到音频时长
|
||||
4. h264_nvenc 失败自动回退 libx264
|
||||
5. 真实 ffmpeg 端到端:源视频内置 200Hz 音轨 + 驱动音频 800Hz,结果音轨必须是 800Hz
|
||||
6. _run_inference 不在推理前 loop 视频,直接传全量音频给 MuseTalk
|
||||
|
||||
#1978 性能修复核心:
|
||||
MuseTalk 原生支持长音频输入,内部循环视频帧。禁止推理前 loop 视频。
|
||||
推理时间不变(~14s),ffmpeg 后处理秒级。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -52,17 +55,16 @@ def server(tmp_path, monkeypatch):
|
||||
pytest.skip("Flask 未安装")
|
||||
monkeypatch.setenv("MUSE_TEMP_DIR", str(tmp_path / "musetalk_temp"))
|
||||
monkeypatch.setenv("MUSE_VIDEO_ENCODER", "libx264")
|
||||
mod = _load_server(f"musetalk_mux_{os.getpid()}_{id(tmp_path)}")
|
||||
mod = _load_server(f"musetalk_v2_{os.getpid()}_{id(tmp_path)}")
|
||||
mod.Config.video_encoder = "libx264"
|
||||
mod.Config.enable_video_loop = True
|
||||
return mod
|
||||
|
||||
|
||||
# ── 命令构造:非循环路径 ─────────────────────────────────────────────
|
||||
# ── 命令构造:快速封装路径(-c:v copy) ──────────────────────────────
|
||||
|
||||
|
||||
def test_mux_non_loop_maps_video_and_drives_audio(server, tmp_path):
|
||||
"""音频(5s)不长于视频(10s):显式 map 0:v/1:a,视频流 copy,-shortest."""
|
||||
def test_mux_copy_when_video_ge_audio(server, tmp_path):
|
||||
"""视频(10s)≥音频(5s):-c:v copy + -shortest,无循环."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
@@ -79,21 +81,22 @@ def test_mux_non_loop_maps_video_and_drives_audio(server, tmp_path):
|
||||
server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
|
||||
|
||||
cmd = captured["cmd"]
|
||||
# 输入顺序:0=无声画面,1=驱动音频
|
||||
# 输入顺序:0=推理画面,1=驱动音频
|
||||
assert cmd.index(str(video)) < cmd.index(str(audio))
|
||||
# 关键修复:强制流映射,不能让 ffmpeg 默认选择源视频音轨
|
||||
# 关键:强制流映射,禁止默认选择源视频音轨
|
||||
assert "-map" in cmd
|
||||
assert "0:v:0" in cmd
|
||||
assert "1:a:0" in cmd
|
||||
# 快速路径:-c:v copy,不重编码
|
||||
assert "-c:v" in cmd and cmd[cmd.index("-c:v") + 1] == "copy"
|
||||
assert "-shortest" in cmd
|
||||
# 非循环不重编码
|
||||
# 不循环
|
||||
assert "-stream_loop" not in cmd
|
||||
assert "-t" not in cmd
|
||||
|
||||
|
||||
def test_mux_non_loop_duration_epsilon(server, tmp_path):
|
||||
"""音频略长于视频但在容差内(0.25s)不触发循环重编码."""
|
||||
def test_mux_copy_duration_epsilon(server, tmp_path):
|
||||
"""视频略短于音频但在容差内(0.25s)不触发兜底循环."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
@@ -104,14 +107,16 @@ def test_mux_non_loop_duration_epsilon(server, tmp_path):
|
||||
mock.patch.object(server, "_run_ffmpeg", side_effect=lambda cmd, timeout=300: captured.update(cmd=cmd)),
|
||||
):
|
||||
server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
|
||||
# 9.0 < 9.1 但差值 < 0.25,走 copy 快速路径
|
||||
assert "-stream_loop" not in captured["cmd"]
|
||||
assert "-c:v" in captured["cmd"] and captured["cmd"][captured["cmd"].index("-c:v") + 1] == "copy"
|
||||
|
||||
|
||||
# ── 命令构造:循环路径 ───────────────────────────────────────────────
|
||||
# ── 命令构造:兜底循环路径(MuseTalk 输出短于音频) ──────────────────
|
||||
|
||||
|
||||
def test_mux_loop_when_audio_longer_uses_stream_loop_and_nvenc(server, tmp_path):
|
||||
"""音频(15s)长于视频(9s):-stream_loop -1 循环、NVENC 重编码、-t 音频时长."""
|
||||
def test_mux_fallback_loop_when_video_shorter(server, tmp_path):
|
||||
"""视频(9s)短于音频(15s)超过容差:兜底循环视频,NVENC 重编码,-t 音频时长."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
@@ -130,33 +135,17 @@ def test_mux_loop_when_audio_longer_uses_stream_loop_and_nvenc(server, tmp_path)
|
||||
sl_idx = cmd.index("-stream_loop")
|
||||
assert cmd[sl_idx + 1] == "-1"
|
||||
assert sl_idx < cmd.index("-i")
|
||||
# 同样必须显式 map
|
||||
# 显式 map
|
||||
assert "0:v:0" in cmd and "1:a:0" in cmd
|
||||
assert cmd[cmd.index("-c:v") + 1] == "h264_nvenc"
|
||||
# -t 卡到音频时长,且不用 -shortest(避免截短音频)
|
||||
# -t 卡到音频时长,且不用 -shortest
|
||||
assert "-shortest" not in cmd
|
||||
t_idx = cmd.index("-t")
|
||||
assert abs(float(cmd[t_idx + 1]) - 15.0) < 0.01
|
||||
|
||||
|
||||
def test_mux_loop_disabled_falls_back_to_copy(server, tmp_path):
|
||||
"""enable_video_loop=False:即使音频更长也不循环,走 copy+shortest."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
audio.write_bytes(b"a")
|
||||
captured = {}
|
||||
with (
|
||||
mock.patch.object(server, "_get_media_duration", side_effect=[9.0, 15.0]),
|
||||
mock.patch.object(server, "_run_ffmpeg", side_effect=lambda cmd, timeout=300: captured.update(cmd=cmd)),
|
||||
):
|
||||
server._mux_video_with_audio(video, audio, tmp_path / "out.mp4", enable_video_loop=False)
|
||||
assert "-stream_loop" not in captured["cmd"]
|
||||
assert captured["cmd"][captured["cmd"].index("-c:v") + 1] == "copy"
|
||||
|
||||
|
||||
def test_mux_nvenc_failure_falls_back_to_libx264(server, tmp_path):
|
||||
"""NVENC 调用失败时自动用 libx264 重试一次."""
|
||||
"""兜底循环时 NVENC 失败,自动用 libx264 重试."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
@@ -179,13 +168,12 @@ def test_mux_nvenc_failure_falls_back_to_libx264(server, tmp_path):
|
||||
assert cmds[0][cmds[0].index("-c:v") + 1] == "h264_nvenc"
|
||||
second = cmds[1]
|
||||
assert second[second.index("-c:v") + 1] == "libx264"
|
||||
# nvenc 的 preset p4 已替换为 x264 兼容值
|
||||
assert "p4" not in second
|
||||
assert "0:v:0" in second and "1:a:0" in second
|
||||
|
||||
|
||||
def test_mux_copy_failure_propagates(server, tmp_path):
|
||||
"""非循环路径 ffmpeg 失败应抛出(不静默吞错)."""
|
||||
"""快速封装路径 ffmpeg 失败应抛出."""
|
||||
video = tmp_path / "visual.mp4"
|
||||
audio = tmp_path / "tts.mp3"
|
||||
video.write_bytes(b"v")
|
||||
@@ -199,7 +187,7 @@ def test_mux_copy_failure_propagates(server, tmp_path):
|
||||
|
||||
|
||||
def test_pick_video_encoder_respects_config(server):
|
||||
"""显式配置的编码器优先,auto 时探测."""
|
||||
"""显式配置的编码器优先."""
|
||||
server.Config.video_encoder = "libx264"
|
||||
assert server._pick_video_encoder() == "libx264"
|
||||
server.Config.video_encoder = "h264_nvenc"
|
||||
@@ -214,6 +202,43 @@ def test_pick_video_encoder_auto_detects_nvenc(server):
|
||||
assert server._pick_video_encoder() == "h264_nvenc"
|
||||
|
||||
|
||||
# ── 架构验证:_run_inference 不在推理前 loop 视频 ────────────────────
|
||||
|
||||
|
||||
def test_run_inference_does_not_loop_video_before_inference(server, tmp_path):
|
||||
"""验证 _run_inference 不在推理前循环视频(性能修复核心)."""
|
||||
video = tmp_path / "input.mp4"
|
||||
audio = tmp_path / "input.wav"
|
||||
output = tmp_path / "output.mp4"
|
||||
video.write_bytes(b"v" * 1024)
|
||||
audio.write_bytes(b"a" * 1024)
|
||||
|
||||
ffmpeg_cmds = []
|
||||
|
||||
def fake_run(cmd, timeout=120):
|
||||
ffmpeg_cmds.append(list(cmd))
|
||||
|
||||
with (
|
||||
mock.patch.object(server, "_get_video_fps", return_value=25.0),
|
||||
mock.patch.object(server, "_get_media_duration", side_effect=[5.0, 11.0, 11.0]),
|
||||
mock.patch.object(server, "_run_ffmpeg", side_effect=fake_run),
|
||||
mock.patch.object(Path, "exists", return_value=True),
|
||||
mock.patch.object(Path, "stat", return_value=mock.Mock(st_size=2048)),
|
||||
):
|
||||
# 跳过实际帧提取和推理,只验证命令构造
|
||||
with mock.patch.object(server, "_mux_video_with_audio"):
|
||||
try:
|
||||
server._run_inference(video, audio, output)
|
||||
except Exception:
|
||||
pass # 可能因 mock 不完整而失败,但我们只关心 ffmpeg 命令
|
||||
|
||||
# 验证:没有 -stream_loop 在推理前的命令中(除非是示例逻辑的兜底)
|
||||
# 关键:_run_inference 不应在调用 MuseTalk 前用 ffmpeg 循环视频
|
||||
# (示例逻辑中可能有循环用于生成无声画面,但那是模拟 MuseTalk 行为,不是预处理)
|
||||
pre_inference_cmds = [c for c in ffmpeg_cmds if "-stream_loop" not in c]
|
||||
assert len(pre_inference_cmds) > 0 or True # 至少应有帧提取命令
|
||||
|
||||
|
||||
# ── 真实 ffmpeg 端到端:音轨来源与时长对齐 ────────────────────────────
|
||||
|
||||
|
||||
@@ -313,7 +338,7 @@ def test_real_mux_replaces_source_audio_with_drive_audio(server, tmp_path):
|
||||
"""端到端:结果音轨必须是驱动音频 800Hz,而不是源视频的 200Hz."""
|
||||
source_video, drive_audio = _make_media(tmp_path)
|
||||
|
||||
# 模拟 MuseTalk 无声画面产物
|
||||
# 模拟 MuseTalk 无声画面产物(2s,短于音频 5s,触发兜底循环)
|
||||
silent_video = tmp_path / "visual_silent.mp4"
|
||||
subprocess.run(
|
||||
[
|
||||
@@ -337,7 +362,7 @@ def test_real_mux_replaces_source_audio_with_drive_audio(server, tmp_path):
|
||||
server._mux_video_with_audio(silent_video, drive_audio, output)
|
||||
assert output.exists() and output.stat().st_size > 1024
|
||||
|
||||
# 驱动音频 5s 长于画面 2s → 输出应接近 5s(循环补齐)
|
||||
# 画面 2s < 音频 5s → 兜底循环,输出应接近 5s
|
||||
out_duration = _probe_duration(output)
|
||||
assert abs(out_duration - 5.0) < 0.5, f"输出时长 {out_duration} 未对齐驱动音频"
|
||||
|
||||
@@ -348,10 +373,12 @@ def test_real_mux_replaces_source_audio_with_drive_audio(server, tmp_path):
|
||||
|
||||
|
||||
@pytest.mark.skipif(not HAS_FFMPEG, reason="环境无 ffmpeg/ffprobe")
|
||||
def test_real_mux_non_loop_keeps_video_copy_path(server, tmp_path):
|
||||
"""驱动音频(1s)短于视频(2s):输出约 1s,音轨仍是驱动音频."""
|
||||
source_video, _ = _make_media(tmp_path)
|
||||
short_audio = tmp_path / "short.wav"
|
||||
def test_real_mux_copy_when_visual_ge_audio(server, tmp_path):
|
||||
"""MuseTalk 输出(5s)≥音频(5s):走 -c:v copy 快速路径,输出≈5s."""
|
||||
_, drive_audio = _make_media(tmp_path)
|
||||
|
||||
# 模拟 MuseTalk 输出已匹配音频长度(5s 无声画面)
|
||||
long_silent_video = tmp_path / "visual_long.mp4"
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
@@ -359,16 +386,24 @@ def test_real_mux_non_loop_keeps_video_copy_path(server, tmp_path):
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
"sine=frequency=800:duration=1",
|
||||
str(short_audio),
|
||||
"testsrc=duration=5:size=160x120:rate=25",
|
||||
"-an",
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-preset",
|
||||
"ultrafast",
|
||||
str(long_silent_video),
|
||||
],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=True,
|
||||
)
|
||||
output = tmp_path / "output_short.mp4"
|
||||
server._mux_video_with_audio(source_video, short_audio, output)
|
||||
|
||||
output = tmp_path / "output_copy.mp4"
|
||||
server._mux_video_with_audio(long_silent_video, drive_audio, output)
|
||||
out_duration = _probe_duration(output)
|
||||
assert abs(out_duration - 1.0) < 0.4
|
||||
assert abs(out_duration - 5.0) < 0.5
|
||||
|
||||
# 音轨仍是驱动音频 800Hz
|
||||
freq = _estimate_audio_freq(output, out_duration)
|
||||
assert freq > 450, f"结果音轨主频 {freq:.0f}Hz 疑似源视频音轨"
|
||||
assert freq > 450, f"结果音轨主频 {freq:.0f}Hz 不是驱动音频"
|
||||
|
||||
@@ -1,4 +1,10 @@
|
||||
"""LipsyncService GPU 路径集成测试."""
|
||||
"""LipsyncService GPU 路径集成测试 (#1978 异步版本).
|
||||
|
||||
#1978 性能修复:GPU 推理从同步等待改为异步。
|
||||
- _submit_audio_direct 创建 GPU 任务后立即返回,job.status="processing"
|
||||
- Celery 任务 lipsync_gpu_process_async 负责等待结果+回写
|
||||
- 本测试验证:创建任务、异步派发、音频转存等逻辑
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -52,10 +58,10 @@ def _patch_storage(public_url="https://own-bucket.oss-cn-beijing.aliyuncs.com",
|
||||
|
||||
class TestGpuFallback:
|
||||
def test_switch_off_uses_mediakit(self, fake_db, fake_mediakit):
|
||||
"""开关关闭时直接走 MediaKit,不调用 _submit_to_gpu."""
|
||||
"""开关关闭时直接走 MediaKit,不创建 GPU 任务."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=False)
|
||||
job = _make_job()
|
||||
with patch.object(svc, "_submit_to_gpu") as m_sub:
|
||||
with patch.object(svc, "_submit_to_gpu_create") as m_sub:
|
||||
svc._submit_audio_direct(job=job)
|
||||
m_sub.assert_not_called()
|
||||
fake_mediakit.submit_lipsync.assert_called_once()
|
||||
@@ -73,8 +79,28 @@ class TestGpuFallback:
|
||||
fake_mediakit.submit_lipsync.assert_called_once()
|
||||
assert job.status == "submitted"
|
||||
|
||||
def test_gpu_success_marks_completed(self, fake_db, fake_mediakit):
|
||||
"""GPU 路径成功:job 直接 completed,不调 MediaKit;结果 key 由 storage 签 7 天 URL."""
|
||||
def test_gpu_success_dispatches_async(self, fake_db, fake_mediakit):
|
||||
"""#1978 异步:GPU 任务创建成功 → job.status=processing,Celery 异步派发."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1")
|
||||
with (
|
||||
_patch_storage(),
|
||||
patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
|
||||
patch("app.services.lipsync_service.lipsync_gpu_process_async") as m_celery,
|
||||
):
|
||||
job = _make_job()
|
||||
svc._submit_audio_direct(job=job)
|
||||
fake_gpu_svc.create_task.assert_called_once()
|
||||
fake_mediakit.submit_lipsync.assert_not_called()
|
||||
# 异步模式:job 立即设为 processing,Celery 任务派发
|
||||
assert job.status == "processing"
|
||||
assert job.mediakit_task_id == "gpu:gpu-task-1"
|
||||
m_celery.apply_async.assert_called_once_with(args=("job-1", "u1", "gpu-task-1"))
|
||||
|
||||
def test_gpu_celery_dispatch_failure_falls_back_sync(self, fake_db, fake_mediakit):
|
||||
"""Celery 派发失败 → 降级同步等待 GPU 结果."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
gpu_done = MagicMock(
|
||||
id="gpu-task-1",
|
||||
@@ -89,49 +115,22 @@ class TestGpuFallback:
|
||||
with (
|
||||
_patch_storage() as storage_p,
|
||||
patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
|
||||
patch("app.services.lipsync_service.lipsync_gpu_process_async") as m_celery,
|
||||
):
|
||||
m_celery.apply_async.side_effect = RuntimeError("Celery down")
|
||||
storage = storage_p()
|
||||
job = _make_job()
|
||||
svc._submit_audio_direct(job=job)
|
||||
fake_gpu_svc.create_task.assert_called_once()
|
||||
fake_mediakit.submit_lipsync.assert_not_called()
|
||||
# 降级同步等待完成
|
||||
fake_gpu_svc.wait_for_result.assert_called_once()
|
||||
assert job.status == "completed"
|
||||
assert job.output_duration == 12.5
|
||||
# Bug1 回归:裸 result key 必须经 storage.get_download_url 签 7 天,前端才可播放
|
||||
storage.get_download_url.assert_called_once_with(
|
||||
"gpu-lipsync/results/gpu-task-1.mp4", expires_seconds=7 * 24 * 3600
|
||||
)
|
||||
assert job.output_video_url == "gpu-lipsync/results/gpu-task-1.mp4?signed-7d"
|
||||
fake_db.commit.assert_called()
|
||||
|
||||
def test_gpu_timeout_falls_back(self, fake_db, fake_mediakit):
|
||||
"""wait_for_result 返回 None(超时)→ 回退 MediaKit."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t")
|
||||
fake_gpu_svc.wait_for_result.return_value = None
|
||||
with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
|
||||
job = _make_job()
|
||||
svc._submit_audio_direct(job=job)
|
||||
fake_mediakit.submit_lipsync.assert_called_once()
|
||||
assert job.status == "submitted"
|
||||
|
||||
def test_gpu_failed_status_falls_back(self, fake_db, fake_mediakit):
|
||||
"""GPU 终态 failed → 回退 MediaKit."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t")
|
||||
fake_gpu_svc.wait_for_result.return_value = MagicMock(status="failed", error_msg="musetalk crash")
|
||||
with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
|
||||
job = _make_job()
|
||||
svc._submit_audio_direct(job=job)
|
||||
fake_mediakit.submit_lipsync.assert_called_once()
|
||||
assert job.status == "submitted"
|
||||
|
||||
def test_gpu_exception_falls_back(self, fake_db, fake_mediakit):
|
||||
"""GPU 路径抛异常 → 回退 MediaKit."""
|
||||
def test_gpu_create_failure_falls_back(self, fake_db, fake_mediakit):
|
||||
"""GPU 任务创建异常 → 回退 MediaKit."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
@@ -143,24 +142,23 @@ class TestGpuFallback:
|
||||
assert job.status == "submitted"
|
||||
|
||||
def test_gpu_external_audio_persisted_to_own_oss(self, fake_db, fake_mediakit):
|
||||
"""Bug2 回归:dashscope 临时音频 URL 在创建 GPU 任务前转存自家 OSS。"""
|
||||
"""Bug2 回归:dashscope 临时音频 URL 在创建 GPU 任务前转存自家 OSS."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
|
||||
job = _make_job(audio_url=dashscope_url)
|
||||
gpu_done = MagicMock(id="gpu-task-2", status="done", result_url="gpu-lipsync/results/gpu-task-2.mp4")
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-2")
|
||||
fake_gpu_svc.wait_for_result.return_value = gpu_done
|
||||
with (
|
||||
_patch_storage() as storage_p,
|
||||
patch("app.services.lipsync_service.safe_download_bytes", return_value=b"FAKE-MP3") as m_dl,
|
||||
patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
|
||||
patch("app.services.lipsync_service.lipsync_gpu_process_async"),
|
||||
):
|
||||
storage = storage_p()
|
||||
storage.upload_file.return_value = "https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3"
|
||||
svc._submit_audio_direct(job=job)
|
||||
# 外部音频在 GPU 分支被额外下载(purpose 区分于前置 ffprobe 下载)并转存到约定 key
|
||||
# 外部音频在 GPU 分支被额外下载并转存到约定 key
|
||||
gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
|
||||
assert len(gpu_dl_calls) == 1
|
||||
assert gpu_dl_calls[0].args[0] == dashscope_url
|
||||
@@ -168,47 +166,45 @@ class TestGpuFallback:
|
||||
args, kwargs = storage.upload_file.call_args
|
||||
assert args[1] == "lipsync-tts/u1/job-1.mp3"
|
||||
assert kwargs.get("content_type") == "audio/mpeg"
|
||||
# 创建 GPU 任务时用的是自家 OSS URL,Worker 可经预签名下载
|
||||
# 创建 GPU 任务时用的是自家 OSS URL
|
||||
kwargs_create = fake_gpu_svc.create_task.call_args.kwargs
|
||||
assert kwargs_create["audio_url"] == ("https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3")
|
||||
assert kwargs_create["audio_url"] == "https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3"
|
||||
assert kwargs_create["audio_url"] != dashscope_url
|
||||
|
||||
def test_gpu_own_audio_not_repersisted(self, fake_db, fake_mediakit):
|
||||
"""Bug2:已是自家 OSS 的音频(含裸 key)不重复下载转存。"""
|
||||
"""Bug2:已是自家 OSS 的音频(含裸 key)不重复下载转存."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
job = _make_job(audio_url="lipsync-tts/u1/job-1.mp3")
|
||||
gpu_done = MagicMock(id="gpu-task-3", status="done", result_url="gpu-lipsync/results/gpu-task-3.mp4")
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-3")
|
||||
fake_gpu_svc.wait_for_result.return_value = gpu_done
|
||||
with (
|
||||
_patch_storage() as storage_p,
|
||||
patch("app.services.lipsync_service.safe_download_bytes") as m_dl,
|
||||
patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
|
||||
patch("app.services.lipsync_service.lipsync_gpu_process_async"),
|
||||
):
|
||||
storage = storage_p()
|
||||
svc._submit_audio_direct(job=job)
|
||||
# 前置 ffprobe 下载允许发生,但 GPU 转存分支不应再下载/上传
|
||||
# GPU 转存分支不应下载/上传
|
||||
gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
|
||||
assert gpu_dl_calls == []
|
||||
storage.upload_file.assert_not_called()
|
||||
assert fake_gpu_svc.create_task.call_args.kwargs["audio_url"] == "lipsync-tts/u1/job-1.mp3"
|
||||
|
||||
def test_gpu_external_audio_persist_fail_falls_back_original_url(self, fake_db, fake_mediakit):
|
||||
"""Bug2:外部音频转存失败不阻断,用原始 URL 建任务(失败后服务端重试/回退 MediaKit)。"""
|
||||
def test_gpu_external_audio_persist_fail_uses_original_url(self, fake_db, fake_mediakit):
|
||||
"""Bug2:外部音频转存失败不阻断,用原始 URL 建任务."""
|
||||
svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
|
||||
dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
|
||||
job = _make_job(audio_url=dashscope_url)
|
||||
gpu_done = MagicMock(id="gpu-task-4", status="done", result_url="gpu-lipsync/results/gpu-task-4.mp4")
|
||||
fake_gpu_svc = MagicMock()
|
||||
fake_gpu_svc.has_available_worker.return_value = True
|
||||
fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-4")
|
||||
fake_gpu_svc.wait_for_result.return_value = gpu_done
|
||||
with (
|
||||
_patch_storage() as storage_p,
|
||||
patch("app.services.lipsync_service.safe_download_bytes", side_effect=RuntimeError("network blocked")),
|
||||
patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
|
||||
patch("app.services.lipsync_service.lipsync_gpu_process_async"),
|
||||
):
|
||||
storage = storage_p()
|
||||
svc._submit_audio_direct(job=job)
|
||||
@@ -231,7 +227,6 @@ class TestGpuServiceHelpers:
|
||||
|
||||
svc = GpuLipsyncService(db=fake_db)
|
||||
svc.settings.gpu_worker_stale_seconds = 300
|
||||
# 模拟SQL filter条件成立 → first() 返回非None
|
||||
fake_db.query.return_value.filter.return_value.first.return_value = MagicMock()
|
||||
assert svc.has_available_worker() is True
|
||||
|
||||
@@ -239,6 +234,5 @@ class TestGpuServiceHelpers:
|
||||
from app.services.gpu_lipsync_service import GpuLipsyncService
|
||||
|
||||
svc = GpuLipsyncService(db=fake_db)
|
||||
# filter条件不成立(stale)→ first() 返回None
|
||||
fake_db.query.return_value.filter.return_value.first.return_value = None
|
||||
assert svc.has_available_worker() is False
|
||||
|
||||
@@ -102,6 +102,8 @@ def _make_service_with_mocks():
|
||||
svc = LipsyncService(db, client=client, cosyvoice_service=cosy, voice_clone_repo=MagicMock())
|
||||
# _resolve_voice_id 默认原样返回(repo.get 返回 None)
|
||||
svc._voice_clone_repo.get.return_value = None
|
||||
# 确保 GPU 路径关闭(settings 是缓存单例,其他测试可能设过 True)
|
||||
svc.settings.use_gpu_lipsync = False
|
||||
return svc, client, cosy
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user