99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
from airshelf.celery import app
|
|
|
|
from apps.ai.models import AITask
|
|
from apps.ai.services import poll_video_segment
|
|
from apps.projects.models import ExportJob
|
|
from apps.projects.models import VideoSegment
|
|
from apps.projects.services.export import run_export_job
|
|
|
|
|
|
@app.task(bind=True, max_retries=120)
|
|
def poll_video_segment_task(self, video_segment_id: str) -> str:
|
|
segment = VideoSegment.objects.select_related("project", "project__created_by").get(id=video_segment_id)
|
|
ai_task = (
|
|
segment.project.ai_tasks.filter(
|
|
task_type=AITask.Type.VIDEO_SEGMENT,
|
|
request_payload__video_segment_id=str(segment.id),
|
|
status__in=[AITask.Status.SUBMITTED, AITask.Status.POLLING],
|
|
)
|
|
.select_related("created_by")
|
|
.order_by("-created_at")
|
|
.first()
|
|
)
|
|
if ai_task is None:
|
|
return video_segment_id
|
|
|
|
user = ai_task.created_by or segment.project.created_by
|
|
version = poll_video_segment(video_segment=segment, user=user)
|
|
if version is None and segment.status in [VideoSegment.Status.RUNNING, VideoSegment.Status.QUEUED]:
|
|
raise self.retry(countdown=30)
|
|
return video_segment_id
|
|
|
|
|
|
@app.task(bind=True, max_retries=2)
|
|
def run_export_job_task(self, export_job_id: str) -> str:
|
|
try:
|
|
run_export_job(export_job_id)
|
|
except Exception as exc:
|
|
export_job = ExportJob.objects.filter(id=export_job_id).first()
|
|
if export_job:
|
|
export_job.status = ExportJob.Status.FAILED
|
|
export_job.error_message = str(exc)
|
|
export_job.save(update_fields=["status", "error_message", "updated_at"])
|
|
raise
|
|
return export_job_id
|
|
|
|
|
|
QUICK_CREATE_QUEUE = "airshelf.quick"
|
|
|
|
|
|
# 豆包长思考脚本允许完整跑 30 分钟；硬上限额外留 60 秒让 soft timeout 的
|
|
# 收尾/落库完成，避免 15 分钟时仍在正常输出却被 worker 强制中断。
|
|
@app.task(bind=True, max_retries=0, soft_time_limit=1800, time_limit=1860, queue=QUICK_CREATE_QUEUE)
|
|
def run_quick_script_task(self, quick_job_id: str) -> str:
|
|
"""脚本生成单独跑，避免把整条极速成片编排堵在一次 SSE 消费里。
|
|
|
|
软超时必须长于豆包思考流（允许最长 30 分钟）。短于 HTTP 流超时会 SIGUSR1 掐连接，
|
|
任务监视器就记成 stream aborted (client disconnected)。
|
|
"""
|
|
from celery.exceptions import SoftTimeLimitExceeded
|
|
|
|
from apps.projects.models import QuickCreateJob
|
|
from apps.projects.services.quick_create import consume_quick_script, fail_quick_create
|
|
|
|
try:
|
|
consume_quick_script(quick_job_id)
|
|
except SoftTimeLimitExceeded as exc:
|
|
job = QuickCreateJob.objects.select_related("project").filter(id=quick_job_id).first()
|
|
if job is not None and job.status not in {
|
|
QuickCreateJob.Status.SUCCEEDED,
|
|
QuickCreateJob.Status.FAILED,
|
|
QuickCreateJob.Status.CANCELLED,
|
|
}:
|
|
fail_quick_create(
|
|
job,
|
|
"脚本生成时间较长，系统会自动重试",
|
|
internal_error=f"SoftTimeLimitExceeded: stream aborted (client disconnected); {exc}",
|
|
)
|
|
return quick_job_id
|
|
return quick_job_id
|
|
|
|
|
|
@app.task(bind=True, max_retries=0)
|
|
def advance_quick_create_task(self, quick_job_id: str) -> str:
|
|
"""一次只推进一个可重入状态，等待型阶段通过重新入队轮询，不占 worker 睡眠。
|
|
|
|
编排走默认 celery 队列：worker 即使没听 airshelf.quick，也不会停在「等待开始」。
|
|
"""
|
|
from apps.projects.models import QuickCreateJob
|
|
from apps.projects.services.quick_create import _claim_next_advance, _enqueue_advance, advance_quick_create
|
|
|
|
next_delay = advance_quick_create(quick_job_id)
|
|
if next_delay is not None:
|
|
if not _claim_next_advance(quick_job_id, int(next_delay)):
|
|
return quick_job_id
|
|
job = QuickCreateJob.objects.filter(id=quick_job_id).first()
|
|
if job is not None:
|
|
_enqueue_advance(job, countdown=max(1, int(next_delay)))
|
|
return quick_job_id
|