xiaoxia-saas/apps/worker/worker_app/celery_app.py

from celery import Celery
from worker_app.core.config import get_settings

settings = get_settings()
celery_app = Celery(settings.worker_name)
celery_app.conf.broker_url = settings.broker_url
celery_app.conf.result_backend = settings.result_backend
celery_app.conf.broker_connection_retry_on_startup = True

# #1714 队列隔离：generation（高优，独占 worker）/ transcode（素材转码）/ celery（默认）
from packages.shared.celery_queues import (  # noqa: E402
    GENERATION_WORKER_PREFETCH_MULTIPLIER,
    apply_queue_settings,
)

apply_queue_settings(celery_app)
# 长渲染任务预取 1，避免任务被预取占住导致调度不均
celery_app.conf.worker_prefetch_multiplier = GENERATION_WORKER_PREFETCH_MULTIPLIER
celery_app.conf.task_acks_late = True  # worker 崩溃时未完成任务重回队列，由执行前守卫丢弃作废消息
# worker 进程被 OOM/容器硬杀时拒绝 ack，消息留在队列由其他 worker 接手
celery_app.conf.task_reject_on_worker_lost = True
# Redis broker 消息可见性超时（#1714）：acks_late 下，消息被预取后 visibility_timeout
# 内未 ack 才会重投。长任务（ingest HEVC 转码 20-30 分钟、生成硬超时 11 分钟）
# 必须远大于最长执行时间，否则正常任务会在执行中被误重投；4 小时覆盖最长转码 + 余量。
celery_app.conf.broker_transport_options = {"visibility_timeout": 4 * 60 * 60}

celery_app.conf.imports = (
    "worker_app.tasks.health",
    "worker_app.tasks.ingest",
    "worker_app.tasks.atom_clips",
    # #1970 片段级 AI 标签：必须显式 import 注册，否则 worker 报
    # "Received unregistered task of type 'worker.tag_atom_clip'"
    "worker_app.tasks.atom_clip_tagging",
    "worker_app.tasks.backfill_atom_clip_tags",
    "worker_app.tasks.classification",
    "worker_app.tasks.generation",
    "worker_app.tasks.voice_extraction",
    "worker_app.tasks.voice_clone",
    "worker_app.tasks.tts_synthesis",
    "worker_app.tasks.batch_download",
    "worker_app.tasks.duplication_check",
    # #1798 AI 数字人渲染：必须在 Worker 实例上注册同名任务，否则消息无人消费（渲染卡 0%）
    "worker_app.tasks.ai_avatar_render",
    "worker_app.tasks._startup",
    "apps.worker.video_processing.dedup",
    "worker_app.tasks.cleanup",
    # 注意：必须用 app.* 路径，不能用 apps.api.app.* 路径！
    # PYTHONPATH=/app/apps/api 下，app.tasks.lipsync_tts 可直接导入且不触发 apps/api/__init__.py
    # （apps/api/__init__.py 会 from .main import app，级联加载整个 FastAPI 栈，Worker 中不需要且会导致注册失败）
    "app.tasks.lipsync_tts",
)

# Celery Beat 定时任务调度
# 注：worker 单实例内嵌 beat（entrypoint-worker.sh -B），定时任务不会重复执行
celery_app.conf.beat_schedule = {
    # pending 任务超时清理：worker 停止消费后，卡 pending 的任务 15 分钟内释放限流名额
    "cleanup-stale-pending-tasks": {
        "task": "worker.cleanup_stale_pending_tasks",
        "schedule": 300.0,  # 每 5 分钟（秒）
        "options": {"expires": 240},  # 4 分钟过期，避免堆积
    },
    # running 孤儿任务巡检：容器重启/进程被杀后卡 running 的任务，20 分钟无更新则判失败
    "cleanup-stale-running-tasks": {
        "task": "worker.cleanup_stale_running_tasks",
        "schedule": 300.0,  # 每 5 分钟（秒）
        "options": {"expires": 240},
    },
    # 上传/转码链路孤儿巡检：worker 重启丢 prefetch 消息后，卡 pending/processing
    # 的 ingest_job + asset 占位超时标终态（#1714）。转码任务较长，10 分钟一轮
    "cleanup-stale-ingest-jobs": {
        "task": "worker.cleanup_stale_ingest_jobs",
        "schedule": 600.0,  # 每 10 分钟（秒）
        "options": {"expires": 540},
    },
}