From 72d8e999913f3991bb87fecfa083e53922595637 Mon Sep 17 00:00:00 2001
From: xiaoxia <xiaoxia@xiaoxiajianji.com>
Date: Thu, 10 Sep 2026 20:48:24 +0800
Subject: [PATCH] =?UTF-8?q?fix(lipsync):=20=E4=BF=AE=E5=A4=8D=20lipsync=5F?=
 =?UTF-8?q?tts=20=E8=B7=A8=E5=AE=B9=E5=99=A8=20import=20=E9=94=99=E8=AF=AF?=
 =?UTF-8?q?=E5=AF=BC=E8=87=B4=20API/Worker=20=E5=90=AF=E5=8A=A8=E5=B4=A9?=
 =?UTF-8?q?=E6=BA=83?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

lipsync_tts.py 原从 apps.worker.worker_app 导入 celery_app，而 API 镜像中没有 apps.worker 目录，
导致 API 容器导入 lipsync_service 时 ModuleNotFoundError，健康检查失败触发自动回滚。

修复：
1. celery_app 改为从 app.core.celery_app 导入（API 侧 Celery app，API/Worker 镜像均存在）
2. _sign_media_url 改为模块内独立函数，去掉对 LipsyncService.__new__ 反射调用的依赖
   （Worker 镜像不复制 services 目录，反射导入会失败）
3. worker.Dockerfile 补充 apps/api/app/tasks/、mediakit_client.py 及必需的 __init__.py
   （Worker 进程需能 import lipsync_tts task 并注册 Celery 路由）
---
 apps/api/app/tasks/lipsync_tts.py | 60 ++++++++++++++++++++++++++-----
 infra/docker/worker.Dockerfile    |  7 ++++
 2 files changed, 58 insertions(+), 9 deletions(-)

diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py
index 45e1ed11e..59c97d395 100644
--- a/apps/api/app/tasks/lipsync_tts.py
+++ b/apps/api/app/tasks/lipsync_tts.py
@@ -14,11 +14,42 @@
 import io
 import logging
 from datetime import datetime, timezone
+from urllib.parse import urlparse
 
-from apps.worker.worker_app.celery_app import celery_app
+from app.core.celery_app import celery_app
 
 logger = logging.getLogger(__name__)
 
+# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService 保持一致
+_MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600
+
+
+def _sign_media_url(url: str) -> str:
+    """对自家 OSS 私有桶 URL 重签长有效期预签名（与 LipsyncService._sign_media_url 保持一致）.
+
+    - 自家 OSS URL → 重签 7 天有效期
+    - 外部临时 URL → 原样透传
+    - 任何异常降级原样返回，不阻断主流程
+    """
+    if not url:
+        return url
+    try:
+        from packages.shared.storage import get_shared_storage_service
+
+        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
+        signed = storage.get_download_url(url, expires_seconds=_MEDIAKIT_URL_TTL_SECONDS)
+        return signed or url
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("[lipsync_tts] URL 重签失败，原样返回: url_prefix=%s err=%s", url[:80], exc)
+        return url
+
 
 @celery_app.task(
     bind=True,
@@ -109,7 +140,13 @@ def tts_synthesize_and_submit(
             audio_data = safe_download_bytes(
                 temp_url,
                 purpose="lipsync_tts_audio",
-                allowed_mime_types=("audio/mpeg", "audio/mp3", "audio/wav", "audio/mp4", "audio/x-m4a"),
+                allowed_mime_types=(
+                    "audio/mpeg",
+                    "audio/mp3",
+                    "audio/wav",
+                    "audio/mp4",
+                    "audio/x-m4a",
+                ),
                 timeout=60.0,
             )
             storage = get_shared_storage_service()
@@ -118,17 +155,18 @@ def tts_synthesize_and_submit(
             logger.info("[lipsync_tts] TTS 音频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
             job.audio_url = permanent_url
         except Exception as exc:
-            logger.warning("[lipsync_tts] TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s", job_id, exc)
+            logger.warning(
+                "[lipsync_tts] TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s",
+                job_id,
+                exc,
+            )
             job.audio_url = temp_url
 
         db.commit()
 
         # 3. 签名 URL 并提交到 MediaKit
-        from app.services.lipsync_service import LipsyncService
-
-        temp_service = LipsyncService.__new__(LipsyncService)
-        audio_url = temp_service._sign_media_url(job.audio_url)
-        video_url = temp_service._sign_media_url(job.video_url)
+        audio_url = _sign_media_url(job.audio_url)
+        video_url = _sign_media_url(job.video_url)
 
         client = get_mediakit_client()
         try:
@@ -141,7 +179,11 @@ def tts_synthesize_and_submit(
             job.mediakit_task_id = mk_result["task_id"]
             job.status = "submitted"
             job.submitted_at = datetime.now(timezone.utc)
-            logger.info("[lipsync_tts] 已提交 MediaKit: job_id=%s task_id=%s", job_id, mk_result["task_id"])
+            logger.info(
+                "[lipsync_tts] 已提交 MediaKit: job_id=%s task_id=%s",
+                job_id,
+                mk_result["task_id"],
+            )
         except MediaKitError as exc:
             job.status = "failed"
             job.error_message = str(exc)
diff --git a/infra/docker/worker.Dockerfile b/infra/docker/worker.Dockerfile
index 9ca769af3..9721a69ff 100755
--- a/infra/docker/worker.Dockerfile
+++ b/infra/docker/worker.Dockerfile
@@ -30,6 +30,13 @@ COPY migrations/ /app/migrations/
 COPY packages/ /app/packages/
 COPY apps/api/app/config.py /app/apps/api/app/config.py
 COPY apps/api/app/core/ /app/apps/api/app/core/
+# API 侧 Celery 任务（lipsync_tts 等）在 worker 进程中执行，需复制任务文件、依赖及 __init__.py
+RUN mkdir -p /app/apps && touch /app/apps/__init__.py
+COPY apps/api/__init__.py /app/apps/api/__init__.py
+COPY apps/api/app/__init__.py /app/apps/api/app/__init__.py
+COPY apps/api/app/services/__init__.py /app/apps/api/app/services/__init__.py
+COPY apps/api/app/services/mediakit_client.py /app/apps/api/app/services/mediakit_client.py
+COPY apps/api/app/tasks/ /app/apps/api/app/tasks/
 
 # Worker 启动脚本
 COPY infra/docker/entrypoint-worker.sh /usr/local/bin/entrypoint-worker.sh