From c45a2ce9b1fbd35f24caea2a7d1a54de48189400 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 30 Sep 2026 00:30:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20viral=5Fvideo=20Celery=20?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E6=94=B9=E7=94=A8=20@shared=5Ftask=EF=BC=8C?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=8D=95=E6=B5=8B=E5=9C=A8=20CI=20=E4=B8=8B?= =?UTF-8?q?=20`resume=5Fviral=5Fvideo=5Fpipeline.run`=20=E6=89=BE=E4=B8=8D?= =?UTF-8?q?=E5=88=B0=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - @celery_app.task 在 conftest mock Task.delay 之后若模块重新加载,decorator 返回的是未绑定原函数而非 Task 实例,导致 .run 方法不存在 - @shared_task(和 lipsync_tts.py 保持一致)在任务被任意 Celery app 加载时自动绑定,测试和生产环境行为一致 - 保留 celery_app import 为 noqa F401,确保 worker_app.celery_app 模块加载时会注册 viral_video tasks --- apps/worker/worker_app/tasks/viral_video.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index e9fed1e2c..a904e0fd3 100755 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -19,9 +19,9 @@ from __future__ import annotations import logging import os -from celery import Task +from celery import Task, shared_task from celery.exceptions import Retry -from worker_app.celery_app import celery_app +from worker_app.celery_app import celery_app # noqa: F401 - 加载 app 以注册任务 from worker_app.db import SessionLocal from packages.adapters.sqlalchemy_impl.viral_video_repository import ( @@ -379,7 +379,7 @@ def _step_upload(job: ViralVideoJob, video_path: str) -> str: # ── 主编排器 ──────────────────────────────────────────────────────────── -@celery_app.task(bind=True, max_retries=2, name="worker.run_viral_video_pipeline") +@shared_task(bind=True, max_retries=2, name="worker.run_viral_video_pipeline") def run_viral_video_pipeline(self: Task, job_id: str) -> dict: """爆款视频 10 步流水线编排器。""" session = None @@ -452,7 +452,7 @@ def run_viral_video_pipeline(self: Task, job_id: str) -> dict: session.close() -@celery_app.task(bind=True, max_retries=2, name="worker.resume_viral_video_pipeline") +@shared_task(bind=True, max_retries=2, name="worker.resume_viral_video_pipeline") def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: """用户确认意图后,从断点恢复流水线(步骤 3-10)。""" session = None @@ -539,7 +539,7 @@ def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: session.close() -@celery_app.task(bind=True, max_retries=1, name="worker.run_video_style_analysis") +@shared_task(bind=True, max_retries=1, name="worker.run_video_style_analysis") def run_video_style_analysis(self: Task, job_id: str) -> dict: """独立的视频风格分析任务(v1.3)。""" session = None