xiaoxia
d29a788753
- domain: ViralVideoJob 状态机（PENDING/RUNNING/WAIT_USER_CONFIRM/COMPLETED/FAILED/CANCELLED）、 11 阶段枚举、PromptType 7 值（含 v1.3 video_style_integration/style_constraint） - repository: 接口 + SQLAlchemy 实现（jobs/style_templates/prompt_templates） - API: 6 个 REST 端点（generate/history/detail/retry/confirm-intent/analyze-style/style-templates） - Celery: ViralVideoOrchestrator 10 步流水线，WS viral_video:progress 进度推送， Credits CREDITS_VIRAL_VIDEO_COST=50 扣点/失败自动回滚 - ai_service: 新增 call_llm/call_vision（复用现有豆包客户端） - 测试 40 个单测（domain/schema/repository/流水线/集成）
87 lines
4.4 KiB
Python
Executable File
87 lines
4.4 KiB
Python
Executable File
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.asset_quality_scoring_task",
|
||
"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.viral_video", # #2039 爆款视频编排器（10步流水线）
|
||
"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",
|
||
# #1998 GPU MuseTalk 异步推理：wait_for_result→签名 URL→回写 lipsync_jobs
|
||
# 必须在 Worker 侧注册，否则 apply_async 消息无人消费，job 永远卡在 processing
|
||
"app.tasks.lipsync_gpu",
|
||
)
|
||
|
||
# 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},
|
||
},
|
||
# 音色克隆卡死巡检：worker 重启/消息丢失后 processing 卡 10 分钟标 failed，用户可点重试
|
||
"cleanup-stale-voice-clones": {
|
||
"task": "worker.cleanup_stale_voice_clones",
|
||
"schedule": 300.0, # 每 5 分钟
|
||
"options": {"expires": 240},
|
||
},
|
||
}
|