diff --git a/apps/api/app/api/routes/lipsync.py b/apps/api/app/api/routes/lipsync.py index 7fc23a56e..09670e226 100644 --- a/apps/api/app/api/routes/lipsync.py +++ b/apps/api/app/api/routes/lipsync.py @@ -144,7 +144,22 @@ def get_lipsync_job( raise HTTPException(status_code=404, detail="任务不存在") if job.status not in ("completed", "failed"): - background.add_task(svc.refresh_job_status, job_id, current_user.user.id) + # 如果距上次更新超过 30 秒，同步刷新一次（避免 background task 静默失败导致永久卡 running）； + # 否则挂后台异步刷新（避免阻塞前端轮询）。 + from datetime import datetime, timezone + now = datetime.now(timezone.utc) + stale = ( + job.updated_at is None + or (now - job.updated_at).total_seconds() > 30 + ) + if stale: + try: + job = svc.refresh_job_status(job_id, current_user.user.id) or job + except Exception as exc: # noqa: BLE001 + logger.error("同步刷新对口型状态失败 job_id=%s err=%s", job_id, exc, exc_info=True) + background.add_task(svc.refresh_job_status, job_id, current_user.user.id) + else: + background.add_task(svc.refresh_job_status, job_id, current_user.user.id) return job diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index d76d09ec4..f19e6124a 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -310,26 +310,40 @@ class LipsyncService: mk_status = status_data.get("status", STATUS_RUNNING) logger.info("MediaKit 对口型状态 [%s]: %s", job_id, mk_status) - if mk_status == STATUS_COMPLETED: - result = status_data.get("result", {}) - job.status = STATUS_COMPLETED - output_url = result.get("video_url", "") - # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL） - job.output_video_url = self._persist_output_video(output_url, job_id, user_id) - job.output_duration = result.get("duration", 0.0) - job.completed_at = datetime.now(timezone.utc) - elif mk_status == STATUS_FAILED: - error = status_data.get("error", {}) - job.status = "failed" - job.error_message = error.get("message", "任务执行失败") - job.error_code = error.get("code", "TaskFailed") - job.completed_at = datetime.now(timezone.utc) - else: - # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted - if isinstance(mk_status, str) and mk_status: - job.status = mk_status - job.updated_at = datetime.now(timezone.utc) - self.db.commit() + try: + if mk_status == STATUS_COMPLETED: + result = status_data.get("result", {}) + job.status = STATUS_COMPLETED + output_url = result.get("video_url", "") + # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL） + job.output_video_url = self._persist_output_video(output_url, job_id, user_id) + job.output_duration = result.get("duration", 0.0) + job.completed_at = datetime.now(timezone.utc) + elif mk_status == STATUS_FAILED: + error = status_data.get("error", {}) + job.status = "failed" + job.error_message = error.get("message", "任务执行失败") + job.error_code = error.get("code", "TaskFailed") + job.completed_at = datetime.now(timezone.utc) + else: + # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted + if isinstance(mk_status, str) and mk_status: + job.status = mk_status + job.updated_at = datetime.now(timezone.utc) + self.db.commit() + except Exception as exc: # noqa: BLE001 - DB 提交失败必须记录日志并重试，否则后台任务静默失败 + logger.error( + "refresh_job_status 提交 DB 失败 job_id=%s mk_status=%s err=%s", + job_id, + mk_status, + exc, + exc_info=True, + ) + try: + self.db.rollback() + except Exception: + pass + # DB commit 失败不 raise，返回当前 job 对象让下次轮询再试 self.db.refresh(job) return job diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index d241c0b89..1383547ce 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -207,6 +207,14 @@ def tts_synthesize_and_submit( db.commit() + # 4. 链式触发 Celery 兜底轮询：MediaKit 提交成功后由 worker 主动拉取状态到终态， + # 不依赖前端轮询触发的 FastAPI background task（后台任务可能静默失败导致永久卡 running） + if job.status == "submitted" and job.mediakit_task_id: + poll_mediakit_status.apply_async( + kwargs={"job_id": job_id, "user_id": user_id}, + countdown=10, # 10 秒后开始轮询（给 MediaKit 一点处理时间） + ) + except Exception: logger.exception("[lipsync_tts] 未预期的异常: job_id=%s", job_id) try: @@ -221,3 +229,88 @@ def tts_synthesize_and_submit( logger.exception("[lipsync_tts] 回写失败状态时异常: job_id=%s", job_id) finally: db.close() + + +@shared_task( + bind=True, + name="lipsync_tts.poll_mediakit_status", + max_retries=60, # 最多轮询 60 次 + default_retry_delay=10, # 每次间隔 10 秒（总兜底时长 10 分钟） +) +def poll_mediakit_status(self, job_id: str, user_id: str): + """Celery 兜底轮询：TTS 提交 MediaKit 后，由 worker 主动拉取状态直到终态。 + + 不依赖前端轮询，避免 background task 静默失败导致任务永久卡 running/submitted。 + """ + from sqlalchemy.orm import Session as DBSession + + from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel + + 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 = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id, LipsyncJobModel.user_id == user_id).first() + if job is None: + logger.warning("[lipsync_poll] Job not found: job_id=%s", job_id) + return + + # 已终态，不需要再轮询 + if job.status in ("completed", "failed", "cancelled"): + return + + if not job.mediakit_task_id: + logger.warning("[lipsync_poll] Job has no mediakit_task_id: job_id=%s status=%s", job_id, job.status) + return + + from app.services.mediakit_client import MediaKitError, get_mediakit_client + + client = get_mediakit_client() + try: + status_data = client.get_task_status(job.mediakit_task_id) + except MediaKitError as exc: + logger.warning("[lipsync_poll] 拉取 MediaKit 状态失败，将重试: job_id=%s err=%s", job_id, exc) + raise self.retry(exc=exc) + + mk_status = status_data.get("status", "running") + + if mk_status == "succeeded": + # 复用 LipsyncService 的持久化逻辑 + from app.services.lipsync_service import LipsyncService + + svc = LipsyncService(db) + result = status_data.get("result", {}) + job.status = "completed" + output_url = result.get("video_url", "") + job.output_video_url = svc._persist_output_video(output_url, job_id, user_id) + job.output_duration = result.get("duration", 0.0) + job.completed_at = datetime.now(timezone.utc) + job.updated_at = datetime.now(timezone.utc) + db.commit() + logger.info("[lipsync_poll] 任务完成: job_id=%s", job_id) + elif mk_status in ("failed", "error"): + error = status_data.get("error", {}) + job.status = "failed" + job.error_message = error.get("message", "任务执行失败") + job.error_code = error.get("code", "TaskFailed") + job.completed_at = datetime.now(timezone.utc) + job.updated_at = datetime.now(timezone.utc) + db.commit() + logger.info("[lipsync_poll] 任务失败: job_id=%s err=%s", job_id, job.error_message) + else: + # 中间状态，更新时间戳，继续重试 + job.updated_at = datetime.now(timezone.utc) + if isinstance(mk_status, str) and mk_status: + job.status = mk_status + db.commit() + logger.debug("[lipsync_poll] 任务仍在 %s，继续轮询: job_id=%s", mk_status, job_id) + raise self.retry() + except Exception as exc: + logger.exception("[lipsync_poll] 未预期异常: job_id=%s", job_id) + db.rollback() + raise self.retry(exc=exc) + finally: + db.close()