From c015769a254f88dd6a9261dd02608d56898210c9 Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Fri, 11 Sep 2026 09:44:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(lipsync,=20P0):=20Worker=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=97=B6ModuleNotFoundError=E5=AF=BC=E8=87=B4=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=8D=A1=E5=9C=A8tts=5Fprocessing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因（经staging运行时日志确认）： 1. lipsync_tts.py 中 SessionLocal 导入路径错误： 从 packages.adapters.sqlalchemy_impl.database 导入，但该模块不存在， 实际 SessionLocal 在 worker_app.db / app.db 模块级通过 build_session_factory 创建。 修复：改为懒探测，Worker环境优先worker_app.db，回退到app.db。 2. packages/ 和 packages/adapters/ 缺少 __init__.py，在某些运行时环境下 可能影响子包导入稳定性（显式包声明优于隐式命名空间包）。 3. lipsync_service.py apply_async 异常被 try/except 吞掉只打 warning， 前端永远看不到失败原因，job 一直卡在 tts_processing。 修复：投递失败立即标failed，写error_message+error_code=AsyncDispatchFailed。 本地验证： - Worker PYTHONPATH 下 import app.tasks.lipsync_tts → Task正确绑定 xiaoxia-saas-worker - API PYTHONPATH 下 import app.tasks.lipsync_tts → Task正确绑定 xiaoxia-saas-api - 61/61 单元测试通过 --- apps/api/app/services/lipsync_service.py | 14 ++++++++++---- apps/api/app/tasks/lipsync_tts.py | 12 +++++++++++- packages/__init__.py | 1 + packages/adapters/__init__.py | 1 + tests/unit/test_lipsync_speed_optimization.py | 10 ++++++---- tests/unit/test_lipsync_tts.py | 19 +++++++++++++------ 6 files changed, 42 insertions(+), 15 deletions(-) create mode 100644 packages/__init__.py create mode 100644 packages/adapters/__init__.py diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index 385675980..d76d09ec4 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -211,12 +211,18 @@ class LipsyncService: normalize_emotion(emotion), ) ) - except Exception: - logger.warning( - "Celery 任务提交失败，TTS 任务已创建但未触发执行: %s", + except Exception as exc: + # 投递失败时立即把 job 标成 failed 并写入 error_message， + # 前端轮询时能直接看到失败原因，不会无限卡在 tts_processing。 + logger.exception( + "Celery 任务提交失败，TTS 任务已创建但未触发执行: job_id=%s err=%s", job_id, - exc_info=True, + exc, ) + job.status = "failed" + job.error_message = f"Celery 任务投递失败: {exc}" + job.error_code = "AsyncDispatchFailed" + job.updated_at = datetime.now(timezone.utc) else: # 2b. 直接音频模式：同步签名并提交 MediaKit video_url = self._sign_media_url(video_url) diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index 85a478828..434d325ea 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -76,11 +76,21 @@ def tts_synthesize_and_submit( from app.services.mediakit_client import MediaKitError, get_mediakit_client from sqlalchemy.orm import Session as DBSession - from packages.adapters.sqlalchemy_impl.database import SessionLocal from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService from packages.shared.url_security import safe_download_bytes + # SessionLocal 获取： + # - API 容器：app.db.SessionLocal（环境变量完整，导入即建引擎） + # - Worker 容器：worker_app.db.SessionLocal（Worker 自己的 settings 初始化引擎） + # API 侧没有 worker_app 模块 → ImportError 直接回退； + # Worker 侧 app.db 会因缺少 API 专有环境变量抛 pydantic ValidationError， + # 此时也要回退到 worker_app.db。 + try: + from worker_app.db import SessionLocal # type: ignore + except Exception: # noqa: BLE001 + from app.db import SessionLocal # type: ignore + db: DBSession = SessionLocal() try: job = ( diff --git a/packages/__init__.py b/packages/__init__.py new file mode 100644 index 000000000..100e2316b --- /dev/null +++ b/packages/__init__.py @@ -0,0 +1 @@ +# xiaoxia-saas shared packages namespace diff --git a/packages/adapters/__init__.py b/packages/adapters/__init__.py new file mode 100644 index 000000000..c4a601e0c --- /dev/null +++ b/packages/adapters/__init__.py @@ -0,0 +1 @@ +# adapter implementations namespace diff --git a/tests/unit/test_lipsync_speed_optimization.py b/tests/unit/test_lipsync_speed_optimization.py index d53bc153b..7c4a39eb3 100644 --- a/tests/unit/test_lipsync_speed_optimization.py +++ b/tests/unit/test_lipsync_speed_optimization.py @@ -143,8 +143,8 @@ class TestCreateJobAsyncTTS: assert args[2] == "v-1" # voice_id assert args[3] == "测试文本" # script_text - def test_tts_mode_celery_dispatch_failure_still_creates_job(self): - """Celery dispatch 失败时，job 记录已创建，状态保持 tts_processing.""" + def test_tts_mode_celery_dispatch_failure_marks_job_failed(self): + """Celery dispatch 失败时，job 标为 failed 并写入 error_message，前端轮询能直接看到错误.""" svc, client, cosy = _make_service_with_mocks() with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: @@ -157,9 +157,11 @@ class TestCreateJobAsyncTTS: script_text="测试文本", ) - # job 已创建 + # job 已创建且状态标为 failed assert job is not None - assert job.status == "tts_processing" + assert job.status == "failed" + assert "Celery 任务投递失败" in job.error_message + assert job.error_code == "AsyncDispatchFailed" # MediaKit 未被调用 client.submit_lipsync.assert_not_called() diff --git a/tests/unit/test_lipsync_tts.py b/tests/unit/test_lipsync_tts.py index 21355c1a0..ce6e6334e 100644 --- a/tests/unit/test_lipsync_tts.py +++ b/tests/unit/test_lipsync_tts.py @@ -73,12 +73,16 @@ def _apply_all_patches( lipsync_tts.run() 在函数体内部懒 import 多个模块，通过 sys.modules 注入 伪造包路径避免真实导入；对存在的模块用 patch() 替换返回值/side_effect。 """ - fake_db_mod = ModuleType("packages.adapters.sqlalchemy_impl.database") + # SessionLocal 通过懒探测获取（Worker 用 worker_app.db，API 用 app.db）， + # 测试环境里两个模块都能被真实导入，必须同时 mock 保证用的是 fake session。 + fake_app_db = ModuleType("app.db") + fake_worker_db = ModuleType("worker_app.db") session, factory = _build_session(job) - fake_db_mod.SessionLocal = factory + fake_app_db.SessionLocal = factory + fake_worker_db.SessionLocal = factory patches = [ - patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.database": fake_db_mod}), + patch.dict(sys.modules, {"app.db": fake_app_db, "worker_app.db": fake_worker_db}), patch( "app.tasks.lipsync_tts._sign_media_url", side_effect=lambda url: url + "?signed" if url else url, @@ -289,9 +293,11 @@ class TestTtsSynthesizeAndSubmit: from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() - fake_db_mod = ModuleType("packages.adapters.sqlalchemy_impl.database") + fake_app_db = ModuleType("app.db") + fake_worker_db = ModuleType("worker_app.db") session, factory = _build_session(job) - fake_db_mod.SessionLocal = factory + fake_app_db.SessionLocal = factory + fake_worker_db.SessionLocal = factory # CosyVoiceService 在 __init__ 抛 RuntimeError（非 CosyVoiceError/ValueError） fake_cosy_mod = ModuleType("packages.application.cosyvoice_service") @@ -309,7 +315,8 @@ class TestTtsSynthesizeAndSubmit: with patch.dict( sys.modules, { - "packages.adapters.sqlalchemy_impl.database": fake_db_mod, + "app.db": fake_app_db, + "worker_app.db": fake_worker_db, "packages.application.cosyvoice_service": fake_cosy_mod, }, ): -- 2.54.0