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, }, ):