from __future__ import annotations import json import math import os import uuid from datetime import datetime, timedelta, timezone from collections.abc import Awaitable, Callable from types import SimpleNamespace from typing import Any from urllib.parse import urlsplit, urlunsplit from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.enums.celery_queue import CeleryQueue from app.enums.generation_task import ( ChatGenerationPipelineStage, ChatGenerationTaskStatus, GenerationType, ) from app.enums.video_upscale import ( LOCAL_PROCESSOR_KEYS, REMOTE_PROCESSOR_KEYS, VIDEO_UPSCALE_SOURCE_RETAINED_MARKER, VideoUpscaleInputSourceType, VideoUpscaleProbeStatus, VideoUpscaleStage, VideoUpscaleTaskStatus, ) from app.models.chat_generation_task import ChatGenerationTask from app.models.generation_record import GenerationRecord from app.models.video_upscale_task import VideoUpscaleTask from app.services.generation.pipeline.db_lock_service import apply_short_lock_timeout from app.services.generation.pipeline.lifecycle_service import notify_owner_finished from app.services.redis_registry_service import RedisExecutionLockError, RedisExecutionLockLost from app.services.media_token_usage_snapshot_service import ( sync_chat_generation_task_media_token_snapshot, sync_generation_record_media_token_snapshot, ) from app.services.video_upscale.log_service import log_video_upscale_event from app.services.resource_accounting_service import ( record_chat_task_generated_resource, record_generation_record_generated_resource, safe_file_size, ) from app.services.video_cover_service import async_create_video_cover_for_local_video from app.services.video_upscale.local_ffmpeg_service import execute_local_ffmpeg_crop from app.services.video_upscale.owner_service import ( VideoUpscaleOwner, load_upscale_owner, mark_owner_upscale_failed, owner_is_completed, owner_is_generating, restore_owner_for_upscale_retry, set_owner_stage, upscale_stage_value, ) from app.services.video_upscale.media_service import ( build_local_source_signed_url, download_video_to_path, is_valid_file, parse_tos_signed_url_expiry, probe_remote_url, probe_video, safe_remove, ) from app.services.video_upscale.snapshot_service import parse_video_upscale_snapshot from app.services.video_upscale.volc_service import ( VolcMediaKitError, is_remote_input_access_error, query_task, submit_video_enhance, ) from app.utils.id_gen import generate_id ExecutionGuard = Callable[[], Awaitable[None]] async def _ensure_guard(execution_guard: ExecutionGuard | None) -> None: if execution_guard is not None: await execution_guard() def _now() -> datetime: return datetime.now(timezone.utc) def _aware(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _date_dir(owner: VideoUpscaleOwner) -> str: fixed = str(getattr(owner, "download_storage_date_dir", None) or "").strip().strip("/") if fixed: return fixed created = _aware(owner.created_at) or _now() return created.strftime("%Y/%m/%d") def _final_video_path(owner: VideoUpscaleOwner) -> tuple[str, str]: date_dir = _date_dir(owner) path = os.path.join(settings.STORAGE_LOCAL_PATH, date_dir, f"{owner.id}.mp4") url = f"/generate/videos/{date_dir}/{owner.id}.mp4" return path, url def _snapshot(owner: VideoUpscaleOwner) -> dict[str, Any]: data = parse_video_upscale_snapshot(owner.video_upscale_snapshot_json) if not data: raise RuntimeError("任务缺少视频超分快照") return data def _sanitize_provider_payload(payload: dict[str, Any]) -> dict[str, Any]: data = dict(payload or {}) value = data.get("video_url") if isinstance(value, str) and value.startswith(("http://", "https://")): parts = urlsplit(value) data["video_url"] = urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) return data def _processor(snapshot: dict[str, Any]) -> dict[str, Any]: value = snapshot.get("processor") if not isinstance(value, dict): raise RuntimeError("超分快照缺少处理器参数") return value def _max_failures(snapshot: dict[str, Any]) -> int: return max(1, int(_processor(snapshot).get("max_attempts") or 3)) def _retry_at(failure_count: int) -> datetime: base = max(1, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)) multiplier = min(max(1, failure_count), 10) return _now() + timedelta(seconds=base * multiplier) def _countdown(value: datetime | None) -> int: target = _aware(value) if not target: return 1 return max(1, math.ceil((target - _now()).total_seconds())) def _lease_until(seconds: int | None = None) -> datetime: return _now() + timedelta(seconds=max(60, int(seconds or settings.VIDEO_UPSCALE_TASK_LEASE_SECONDS or 1800))) def _safe_apply_async( celery_task: Any, *, args: list[Any], queue: str, task: VideoUpscaleOwner | None, upscale_task: VideoUpscaleTask | None, action: str, countdown: int | None = None, kwargs: dict[str, Any] | None = None, ) -> bool: options: dict[str, Any] = {"args": args, "queue": queue} if countdown is not None: options["countdown"] = max(0, int(countdown)) if kwargs: options["kwargs"] = kwargs try: celery_task.apply_async(**options) return True except Exception as exc: log_video_upscale_event( event_type="upscale_queue_enqueue_failed", event_status="failed", task=task, upscale_task=upscale_task, message=str(exc), detail={"action": action, "queue": queue, "countdown": countdown}, error=str(exc), ) return False async def _load_pair( db: AsyncSession, upscale_task_id: str, *, for_update: bool = True, ) -> tuple[VideoUpscaleTask | None, VideoUpscaleOwner | None]: query = select(VideoUpscaleTask).where(VideoUpscaleTask.id == upscale_task_id) if for_update: await apply_short_lock_timeout(db) query = query.with_for_update() result = await db.execute(query.limit(1)) upscale = result.scalar_one_or_none() if not upscale: return None, None owner = await load_upscale_owner(db, upscale, for_update=for_update) return upscale, owner async def _persist_provider_error_payload( db: AsyncSession, *, upscale_task_id: str, payload: dict[str, Any] | None, ) -> None: if not payload: return upscale, _owner = await _load_pair(db, upscale_task_id, for_update=True) if not upscale: return upscale.provider_response_json = json.dumps(payload, ensure_ascii=False, default=str) await db.commit() async def prepare_video_upscale_task( db: AsyncSession, *, task: ChatGenerationTask | None = None, generation_record: GenerationRecord | None = None, source_local_path: str, source_file_size_bytes: int, source_remote_url: str | None = None, source_info: Any | None = None, ) -> VideoUpscaleTask: owner: VideoUpscaleOwner | None = task or generation_record if owner is None: raise RuntimeError("缺少视频超分所有者") if owner.gen_type != GenerationType.VIDEO.value or not bool(owner.video_upscale_enabled_snapshot): raise RuntimeError("当前任务未启用视频超分快照") snapshot = _snapshot(owner) # ffprobe should run before the owner row is locked by the caller. # Keep this fallback for non-generation callers that do not pass probe data. source_info = source_info or await probe_video(source_local_path) remote_url = str(source_remote_url or getattr(owner, "remote_result_url", None) or "").strip() or None signed_at, expires_at = parse_tos_signed_url_expiry(remote_url) owner_filter = ( VideoUpscaleTask.chat_generation_task_id == owner.id if isinstance(owner, ChatGenerationTask) else VideoUpscaleTask.generation_record_id == owner.id ) await apply_short_lock_timeout(db) result = await db.execute( select(VideoUpscaleTask).where(owner_filter).with_for_update().limit(1) ) upscale = result.scalar_one_or_none() if upscale is None: upscale = VideoUpscaleTask( id=generate_id(), chat_generation_task_id=owner.id if isinstance(owner, ChatGenerationTask) else None, generation_record_id=owner.id if isinstance(owner, GenerationRecord) else None, processor_key=str(snapshot.get("processor_key") or ""), target_width=int(snapshot.get("target_width") or 0), target_height=int(snapshot.get("target_height") or 0), ) db.add(upscale) upscale.status = VideoUpscaleTaskStatus.PENDING.value upscale.stage = VideoUpscaleStage.SOURCE_READY.value upscale.source_local_path = source_local_path upscale.source_file_size_bytes = int(source_file_size_bytes or safe_file_size(source_local_path)) upscale.source_width = source_info.width upscale.source_height = source_info.height upscale.source_duration_seconds = float(source_info.duration_seconds) upscale.source_deleted_at = None upscale.source_delete_error = None upscale.source_remote_url = remote_url upscale.source_remote_url_signed_at = signed_at upscale.source_remote_url_expires_at = expires_at upscale.source_remote_url_probe_status = VideoUpscaleProbeStatus.NOT_CHECKED.value upscale.last_error = None upscale.next_retry_at = None upscale.lease_token = None upscale.lease_until = None set_owner_stage(owner, upscale_stage_value(owner, ChatGenerationPipelineStage.UPSCALE_QUEUED)) await db.flush() log_video_upscale_event( event_type="upscale_source_download_success", task=owner, upscale_task=upscale, detail={ "source_local_path": source_local_path, "source_width": source_info.width, "source_height": source_info.height, "source_duration_seconds": source_info.duration_seconds, }, ) return upscale async def enqueue_upscale_task(db: AsyncSession, *, upscale: VideoUpscaleTask, reason: str) -> str: from app.tasks.video_upscale_tasks import execute_local, submit_remote upscale_id = str(upscale.id) processor_key = str(upscale.processor_key) celery_id = f"upscale:{upscale_id}:{uuid.uuid4().hex[:16]}" upscale.celery_task_id = celery_id upscale.status = VideoUpscaleTaskStatus.PENDING.value upscale.stage = VideoUpscaleStage.QUEUED.value upscale.next_retry_at = None log_snapshot = SimpleNamespace( id=upscale_id, chat_generation_task_id=( str(upscale.chat_generation_task_id) if upscale.chat_generation_task_id else None ), generation_record_id=( str(upscale.generation_record_id) if upscale.generation_record_id else None ), processor_key=processor_key, status=VideoUpscaleTaskStatus.PENDING.value, stage=VideoUpscaleStage.QUEUED.value, attempt_count=int(upscale.attempt_count or 0), failure_count=int(upscale.failure_count or 0), provider_task_id=(str(upscale.provider_task_id) if upscale.provider_task_id else None), input_source_type=upscale.input_source_type, target_width=upscale.target_width, target_height=upscale.target_height, ) await db.commit() try: if processor_key in LOCAL_PROCESSOR_KEYS: execute_local.apply_async( args=[upscale_id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_LOCAL.value, task_id=celery_id, ) elif processor_key in REMOTE_PROCESSOR_KEYS: submit_remote.apply_async( args=[upscale_id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_REMOTE.value, task_id=celery_id, ) else: raise RuntimeError(f"未注册的超分处理器: {processor_key}") log_video_upscale_event( event_type="upscale_task_enqueued", upscale_task=log_snapshot, detail={"reason": reason, "celery_task_id": celery_id, "processor_key": processor_key}, ) except Exception as exc: # 数据库状态已经提交，不把队列瞬时异常误判为原视频下载失败或触发退款； # gen_recovery 会扫描 pending/queued 任务并补投。 log_video_upscale_event( event_type="upscale_task_enqueue_failed", event_status="failed", upscale_task=log_snapshot, message=str(exc), detail={"reason": reason, "celery_task_id": celery_id, "processor_key": processor_key}, error=str(exc), ) return celery_id async def _claim( db: AsyncSession, *, upscale_task_id: str, stage: str, chat_stage: str, increment_attempt: bool, execution_token: str, lease_seconds: int | None = None, ) -> tuple[VideoUpscaleTask, VideoUpscaleOwner, dict[str, Any]] | None: upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: return None if upscale.status in {VideoUpscaleTaskStatus.COMPLETED.value, VideoUpscaleTaskStatus.FAILED.value}: return None if not owner_is_generating(task): return None lease_until = _aware(upscale.lease_until) if lease_until and lease_until > _now() and upscale.status == VideoUpscaleTaskStatus.PROCESSING.value: return None snapshot = _snapshot(task) upscale.status = VideoUpscaleTaskStatus.PROCESSING.value upscale.stage = stage upscale.lease_token = execution_token upscale.lease_until = _lease_until(lease_seconds) upscale.started_at = upscale.started_at or _now() upscale.next_retry_at = None if increment_attempt: upscale.attempt_count = int(upscale.attempt_count or 0) + 1 set_owner_stage(task, upscale_stage_value(task, chat_stage)) await db.commit() return upscale, task, snapshot async def _load_owned_pair( db: AsyncSession, *, upscale_task_id: str, execution_token: str, for_update: bool = True, ) -> tuple[VideoUpscaleTask, VideoUpscaleOwner]: upscale, task = await _load_pair(db, upscale_task_id, for_update=for_update) if not upscale or not task: raise RedisExecutionLockLost(f"超分任务或所属任务不存在: {upscale_task_id}") if str(upscale.lease_token or "") != str(execution_token): raise RedisExecutionLockLost(f"超分数据库执行租约已失效: {upscale_task_id}") return upscale, task async def _final_fail( db: AsyncSession, *, upscale_task_id: str, error_message: str, ) -> None: upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: return if upscale.status == VideoUpscaleTaskStatus.COMPLETED.value: return now = _now() upscale.status = VideoUpscaleTaskStatus.FAILED.value upscale.stage = VideoUpscaleStage.FAILED.value upscale.failed_at = now upscale.last_error = error_message upscale.next_retry_at = None upscale.lease_until = None upscale.lease_token = None await mark_owner_upscale_failed(db, task, error_message=error_message) await db.commit() await notify_owner_finished(db, task) log_video_upscale_event( event_type="upscale_retry_exhausted", event_status="failed", task=task, upscale_task=upscale, message=error_message, detail={ "refund_policy": "no_refund", "source_retained": True, "manual_recovery_available": True, "failure_count": upscale.failure_count, }, error=error_message, ) async def _schedule_retry( db: AsyncSession, *, upscale_task_id: str, error_message: str, retry_action: str, retryable: bool = True, ) -> None: from app.tasks.video_upscale_tasks import download_remote_result, execute_local, finalize, poll_remote, submit_remote upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: return snapshot = _snapshot(task) upscale.failure_count = int(upscale.failure_count or 0) + 1 max_failures = _max_failures(snapshot) if not retryable or upscale.failure_count >= max_failures: await db.commit() await _final_fail(db, upscale_task_id=upscale_task_id, error_message=error_message) return next_retry_at = _retry_at(upscale.failure_count) upscale.status = VideoUpscaleTaskStatus.RETRY_WAITING.value upscale.stage = VideoUpscaleStage.RETRY_WAITING.value upscale.last_error = error_message upscale.next_retry_at = next_retry_at upscale.lease_until = None upscale.lease_token = None set_owner_stage(task, upscale_stage_value(task, ChatGenerationPipelineStage.UPSCALE_RETRY_WAITING)) await db.commit() task_map = { "local": (execute_local, settings.VIDEO_UPSCALE_LOCAL_QUEUE), "submit": (submit_remote, settings.VIDEO_UPSCALE_REMOTE_QUEUE), "poll": (poll_remote, settings.VIDEO_UPSCALE_REMOTE_QUEUE), "download": (download_remote_result, settings.VIDEO_UPSCALE_REMOTE_QUEUE), "finalize": (finalize, settings.VIDEO_UPSCALE_LOCAL_QUEUE), } celery_task, queue = task_map[retry_action] enqueue_error: str | None = None try: celery_task.apply_async(args=[upscale_task_id], countdown=_countdown(next_retry_at), queue=queue) except Exception as exc: enqueue_error = str(exc) log_video_upscale_event( event_type="upscale_retry_scheduled" if enqueue_error is None else "upscale_retry_enqueue_failed", event_status="retry_waiting" if enqueue_error is None else "failed", task=task, upscale_task=upscale, message=error_message, detail={ "retry_action": retry_action, "failure_count": upscale.failure_count, "max_failures": max_failures, "next_retry_at": next_retry_at, "enqueue_error": enqueue_error, }, error=enqueue_error or error_message, ) async def _queue_finalize( db: AsyncSession, *, upscale_task_id: str, final_path: str, reason: str, execution_token: str, execution_guard: ExecutionGuard | None = None, ) -> None: await _ensure_guard(execution_guard) upscale, task = await _load_owned_pair( db, upscale_task_id=upscale_task_id, execution_token=execution_token, for_update=True ) if not upscale or not task: return if not is_valid_file(final_path): raise RuntimeError(f"超分最终视频不存在或为空: {final_path}") upscale.final_local_path = final_path upscale.status = VideoUpscaleTaskStatus.PROCESSING.value upscale.stage = VideoUpscaleStage.FINALIZING.value upscale.lease_until = None upscale.lease_token = None upscale.next_retry_at = None set_owner_stage(task, upscale_stage_value(task, ChatGenerationPipelineStage.UPSCALE_FINALIZING)) await db.commit() from app.tasks.video_upscale_tasks import finalize _safe_apply_async( finalize, args=[upscale_task_id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE, task=task, upscale_task=upscale, action="finalize", countdown=max(1, int(settings.VIDEO_UPSCALE_STAGE_HANDOFF_DELAY_SECONDS or 2)), ) log_video_upscale_event( event_type="upscale_finalize_enqueued", event_status="queued", task=task, upscale_task=upscale, detail={"reason": reason, "final_local_path": final_path}, ) async def _finalize_success( db: AsyncSession, *, upscale_task_id: str, final_path: str, execution_token: str, execution_guard: ExecutionGuard | None = None, ) -> None: await _ensure_guard(execution_guard) upscale, task = await _load_owned_pair( db, upscale_task_id=upscale_task_id, execution_token=execution_token, for_update=False ) if not upscale or not task: return snapshot = _snapshot(task) info = await probe_video(final_path) expected_width = int(snapshot.get("target_width") or upscale.target_width or 0) expected_height = int(snapshot.get("target_height") or upscale.target_height or 0) processor_key = str(snapshot.get("processor_key") or upscale.processor_key or "") if processor_key in LOCAL_PROCESSOR_KEYS: if abs(info.width - expected_width) > 2 or abs(info.height - expected_height) > 2: raise RuntimeError( f"本地超分最终视频尺寸不符合快照: 实际 {info.width}x{info.height}，预期 {expected_width}x{expected_height}" ) else: expected_short_edge = int(snapshot.get("target_short_edge_pixels") or min(expected_width, expected_height)) actual_short_edge = min(info.width, info.height) short_edge_tolerance = max(4, int(round(expected_short_edge * 0.03))) if abs(actual_short_edge - expected_short_edge) > short_edge_tolerance: raise RuntimeError( f"火山超分最终视频短边不符合目标档位: 实际 {actual_short_edge}，目标 {expected_short_edge}" ) if expected_width <= 0 or expected_height <= 0 or info.width <= 0 or info.height <= 0: raise RuntimeError("火山超分最终视频宽高无效") expected_ratio = expected_width / expected_height actual_ratio = info.width / info.height ratio_error = abs(actual_ratio - expected_ratio) / expected_ratio if ratio_error > 0.03: raise RuntimeError( f"火山超分最终视频比例异常: 实际 {info.width}:{info.height}，预期约 {expected_width}:{expected_height}" ) try: source_duration = float(upscale.source_duration_seconds or 0) except (TypeError, ValueError): source_duration = 0.0 if source_duration > 0 and info.duration_seconds > 0: duration_tolerance = max(1.0, source_duration * 0.03) if abs(info.duration_seconds - source_duration) > duration_tolerance: raise RuntimeError( f"超分最终视频时长异常: 源视频 {source_duration:.3f}s，最终视频 {info.duration_seconds:.3f}s" ) date_dir = _date_dir(task) cover_url, cover_path = await async_create_video_cover_for_local_video( record_id=task.id, video_path=final_path, date_dir=date_dir, log_prefix=f"超分最终视频封面生成 task_id={task.id}", ) if not cover_url or not cover_path: raise RuntimeError("超分最终视频封面生成失败") await _ensure_guard(execution_guard) upscale, task = await _load_owned_pair( db, upscale_task_id=upscale_task_id, execution_token=execution_token, for_update=True ) if not upscale or not task: return if owner_is_completed(task) and task.video_url: return final_url = f"/generate/videos/{date_dir}/{task.id}.mp4" now = _now() task.video_url = final_url task.video_cover_url = cover_url if isinstance(task, ChatGenerationTask): task.status = ChatGenerationTaskStatus.COMPLETED.value task.pipeline_stage = ChatGenerationPipelineStage.DONE.value else: task.status = "completed" task.pipeline_stage = "done" task.generated_at = now task.error_message = None if isinstance(task, ChatGenerationTask): task.retry_count = int(task.manual_retry_count or 0) upscale.status = VideoUpscaleTaskStatus.COMPLETED.value upscale.stage = VideoUpscaleStage.COMPLETED.value upscale.completed_at = now upscale.final_local_path = final_path upscale.final_resource_url = final_url upscale.final_file_size_bytes = safe_file_size(final_path) upscale.effective_target_width = info.width upscale.effective_target_height = info.height upscale.last_error = None upscale.next_retry_at = None upscale.lease_until = None upscale.lease_token = None if isinstance(task, ChatGenerationTask): await record_chat_task_generated_resource( db, task, resource_url=final_url, storage_path=final_path, file_size_bytes=upscale.final_file_size_bytes, remote_url=upscale.source_remote_url, generated_at=now, ) await sync_chat_generation_task_media_token_snapshot(db, task) else: await record_generation_record_generated_resource( db, task, resource_url=final_url, storage_path=final_path, file_size_bytes=upscale.final_file_size_bytes, remote_url=upscale.source_remote_url, generated_at=now, ) await sync_generation_record_media_token_snapshot(db, task) source_path = str(upscale.source_local_path or "") delete_source_after_success = bool(snapshot.get("delete_source_after_success", True)) if not delete_source_after_success: upscale.source_delete_error = VIDEO_UPSCALE_SOURCE_RETAINED_MARKER await db.commit() if isinstance(task, ChatGenerationTask): await notify_owner_finished(db, task) if delete_source_after_success and source_path and os.path.abspath(source_path) != os.path.abspath(final_path): removed = safe_remove(source_path) cleanup_error = None if removed else f"源视频删除失败: {source_path}" await apply_short_lock_timeout(db) cleanup_result = await db.execute( select(VideoUpscaleTask).where(VideoUpscaleTask.id == upscale_task_id).with_for_update().limit(1) ) cleanup_task = cleanup_result.scalar_one_or_none() if cleanup_task: cleanup_task.source_deleted_at = _now() if removed else None cleanup_task.source_delete_error = cleanup_error await db.commit() if removed: log_video_upscale_event( event_type="upscale_source_cleanup_success", task=task, upscale_task=cleanup_task or upscale, detail={"source_local_path": source_path, "delete_source_after_success": True}, ) else: log_video_upscale_event( event_type="upscale_source_cleanup_failed", event_status="failed", task=task, upscale_task=cleanup_task or upscale, message="超分完成后源视频删除失败", detail={"source_local_path": source_path, "delete_source_after_success": True}, error="source_cleanup_failed", ) elif not delete_source_after_success: log_video_upscale_event( event_type="upscale_source_retained_by_snapshot", event_status="success", task=task, upscale_task=upscale, detail={ "source_local_path": source_path, "delete_source_after_success": False, "source_retained_by_config": True, }, ) log_video_upscale_event( event_type="upscale_success", task=task, upscale_task=upscale, detail={ "final_url": final_url, "cover_url": cover_url, "width": info.width, "height": info.height, "file_size_bytes": upscale.final_file_size_bytes, }, ) async def run_local_upscale( db: AsyncSession, upscale_task_id: str, *, execution_token: str, execution_guard: ExecutionGuard | None = None ) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.LOCAL_PROCESSING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_PROCESSING.value, increment_attempt=True, execution_token=execution_token, lease_seconds=int(settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS or 3600) + 300, ) if not claimed: return upscale, task, snapshot = claimed processor = _processor(snapshot) final_path, _ = _final_video_path(task) log_video_upscale_event(event_type="upscale_local_start", task=task, upscale_task=upscale) try: await execute_local_ffmpeg_crop( source_path=str(upscale.source_local_path or ""), final_path=final_path, target_width=int(snapshot.get("target_width") or upscale.target_width), target_height=int(snapshot.get("target_height") or upscale.target_height), timeout_seconds=int(processor.get("timeout_seconds") or settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS), execution_guard=execution_guard, ) await _ensure_guard(execution_guard) log_video_upscale_event( event_type="upscale_local_success", task=task, upscale_task=upscale, detail={"final_local_path": final_path}, ) except RedisExecutionLockError: await db.rollback() raise except Exception as exc: await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"本地 FFmpeg 超分失败: {exc}", retry_action="local", retryable=True, ) return try: await _queue_finalize( db, upscale_task_id=upscale_task_id, final_path=final_path, reason="local_upscale_completed", execution_token=execution_token, execution_guard=execution_guard, ) except RedisExecutionLockError: await db.rollback() raise except Exception as exc: await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"本地超分结果进入最终化失败: {exc}", retry_action="finalize", retryable=True, ) async def _select_remote_input(db: AsyncSession, upscale: VideoUpscaleTask, processor: dict[str, Any]) -> tuple[str, str]: remote_url = str(upscale.source_remote_url or "").strip() expires_at = _aware(upscale.source_remote_url_expires_at) now = _now() threshold = max(0, int(settings.VIDEO_UPSCALE_REMOTE_URL_PROBE_THRESHOLD_SECONDS or 600)) use_remote = False probe_status = VideoUpscaleProbeStatus.NOT_CHECKED.value # 远程源地址已被火山明确判定不可访问后，后续同一处理器重提必须固定使用本地签名地址。 if int(upscale.input_source_fallback_count or 0) > 0: signed_url = build_local_source_signed_url( str(upscale.source_local_path or ""), int(processor.get("source_url_expire_seconds") or settings.VIDEO_UPSCALE_LOCAL_SIGNED_URL_EXPIRE_SECONDS), ) return signed_url, VideoUpscaleInputSourceType.LOCAL_SIGNED.value if remote_url: if expires_at is not None: remaining = (expires_at - now).total_seconds() if remaining > threshold: use_remote = True elif remaining > 0: use_remote = await probe_remote_url(remote_url) probe_status = VideoUpscaleProbeStatus.SUCCESS.value if use_remote else VideoUpscaleProbeStatus.FAILED.value else: probe_status = VideoUpscaleProbeStatus.EXPIRED.value else: use_remote = await probe_remote_url(remote_url) probe_status = VideoUpscaleProbeStatus.SUCCESS.value if use_remote else VideoUpscaleProbeStatus.UNPARSABLE.value if probe_status != VideoUpscaleProbeStatus.NOT_CHECKED.value: upscale.source_remote_url_last_probe_at = now upscale.source_remote_url_probe_status = probe_status await db.commit() if use_remote: return remote_url, VideoUpscaleInputSourceType.PROVIDER_REMOTE.value signed_url = build_local_source_signed_url( str(upscale.source_local_path or ""), int(processor.get("source_url_expire_seconds") or settings.VIDEO_UPSCALE_LOCAL_SIGNED_URL_EXPIRE_SECONDS), ) return signed_url, VideoUpscaleInputSourceType.LOCAL_SIGNED.value async def run_remote_submit( db: AsyncSession, upscale_task_id: str, *, count_attempt: bool = True, execution_token: str, execution_guard: ExecutionGuard | None = None, ) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.REMOTE_SUBMITTING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_PROCESSING.value, increment_attempt=count_attempt, execution_token=execution_token, lease_seconds=300, ) if not claimed: return upscale, task, snapshot = claimed processor = _processor(snapshot) log_video_upscale_event( event_type="upscale_provider_submit_start", task=task, upscale_task=upscale, detail={ "processor_key": upscale.processor_key, "target_resolution": snapshot.get("target_resolution"), "target_width": snapshot.get("target_width"), "target_height": snapshot.get("target_height"), }, ) try: source_width = int(upscale.source_width or 0) source_height = int(upscale.source_height or 0) source_short_edge = min(source_width, source_height) source_long_edge = max(source_width, source_height) if upscale.processor_key == "volc_large_model_v1": if not (360 <= source_short_edge <= 1080 and 360 <= source_long_edge <= 1920): raise VolcMediaKitError( f"火山画质增强大模型输入尺寸不支持: {source_width}x{source_height}，短边需 360-1080、长边需 360-1920", code="UnsupportedInputResolution", retryable=False, ) source_info = await probe_video(str(upscale.source_local_path or "")) hdr_transfers = {"smpte2084", "arib-std-b67"} if str(source_info.color_transfer or "").strip().lower() in hdr_transfers: raise VolcMediaKitError( f"火山画质增强大模型仅支持 SDR 视频，当前 color_transfer={source_info.color_transfer}", code="UnsupportedHdrInput", retryable=False, ) elif upscale.processor_key in {"volc_standard_v1", "volc_professional_v1"}: if source_short_edge > 1440 or source_long_edge > 2560: raise VolcMediaKitError( f"火山标准版/专业版输入视频最高支持 2K，当前 {source_width}x{source_height}", code="UnsupportedInputResolution", retryable=False, ) source_url, source_type = await _select_remote_input(db, upscale, processor) submit_result = await submit_video_enhance( processor_key=upscale.processor_key, video_url=source_url, target_resolution=str(snapshot.get("target_resolution") or task.resolution or ""), target_width=int(snapshot.get("target_width") or upscale.target_width), target_height=int(snapshot.get("target_height") or upscale.target_height), processor=processor, client_token=f"{upscale.id}-{int(upscale.attempt_count or 0)}-{int(upscale.input_source_fallback_count or 0)}", ) await _ensure_guard(execution_guard) upscale, task = await _load_owned_pair( db, upscale_task_id=upscale_task_id, execution_token=execution_token, for_update=True ) upscale.provider_task_id = submit_result.task_id upscale.provider_submitted_at = _now() upscale.provider_request_json = json.dumps(_sanitize_provider_payload(submit_result.request_payload), ensure_ascii=False, default=str) upscale.provider_response_json = json.dumps(submit_result.response_payload, ensure_ascii=False, default=str) upscale.input_source_type = source_type upscale.status = VideoUpscaleTaskStatus.PROCESSING.value upscale.stage = VideoUpscaleStage.REMOTE_POLLING.value upscale.lease_until = None upscale.lease_token = None set_owner_stage(task, upscale_stage_value(task, ChatGenerationPipelineStage.UPSCALE_POLLING)) await db.commit() from app.tasks.video_upscale_tasks import poll_remote _safe_apply_async( poll_remote, args=[upscale_task_id], countdown=max(5, int(processor.get("poll_interval_seconds") or 30)), queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE, task=task, upscale_task=upscale, action="poll_after_submit", ) log_video_upscale_event( event_type="upscale_provider_submit_success", task=task, upscale_task=upscale, remote_request_id=submit_result.request_id, detail={"provider_task_id": submit_result.task_id, "input_source_type": source_type}, ) except RedisExecutionLockError: await db.rollback() raise except VolcMediaKitError as exc: log_video_upscale_event( event_type="upscale_provider_submit_failed", event_status="failed", task=task, upscale_task=upscale, remote_request_id=exc.request_id, message=str(exc), detail=exc.log_detail(), error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) await _persist_provider_error_payload( db, upscale_task_id=upscale_task_id, payload=exc.response_payload, ) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分提交失败: {exc}", retry_action="submit", retryable=exc.retryable, ) except Exception as exc: log_video_upscale_event( event_type="upscale_provider_submit_failed", event_status="failed", task=task, upscale_task=upscale, message=str(exc), error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分提交失败: {exc}", retry_action="submit", ) async def run_remote_poll( db: AsyncSession, upscale_task_id: str, *, execution_token: str, execution_guard: ExecutionGuard | None = None ) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.REMOTE_POLLING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_POLLING.value, increment_attempt=False, execution_token=execution_token, lease_seconds=300, ) if not claimed: return upscale, task, snapshot = claimed processor = _processor(snapshot) provider_task_id = str(upscale.provider_task_id or "").strip() if not provider_task_id: await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message="火山超分任务缺少 provider_task_id", retry_action="submit", retryable=True, ) return log_video_upscale_event( event_type="upscale_provider_poll_start", task=task, upscale_task=upscale, detail={"provider_task_id": provider_task_id}, ) try: query_result = await query_task( provider_task_id, request_timeout_seconds=int(processor.get("request_timeout_seconds") or 30), ) await _ensure_guard(execution_guard) upscale, task = await _load_owned_pair( db, upscale_task_id=upscale_task_id, execution_token=execution_token, for_update=True ) upscale.provider_response_json = json.dumps(query_result.response_payload, ensure_ascii=False, default=str) upscale.lease_until = None upscale.lease_token = None if query_result.status == "running": provider_submitted_at = _aware(upscale.provider_submitted_at) or _aware(upscale.started_at) or _now() poll_timeout = max(60, int(processor.get("poll_timeout_seconds") or 7200)) if (_now() - provider_submitted_at).total_seconds() > poll_timeout: await db.commit() await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分轮询超过 {poll_timeout} 秒", retry_action="submit", retryable=True, ) return await db.commit() from app.tasks.video_upscale_tasks import poll_remote _safe_apply_async( poll_remote, args=[upscale_task_id], countdown=max(5, int(processor.get("poll_interval_seconds") or 30)), queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE, task=task, upscale_task=upscale, action="poll_running", ) log_video_upscale_event( event_type="upscale_provider_poll_running", event_status="running", task=task, upscale_task=upscale, remote_request_id=query_result.request_id, ) return if query_result.status == "failed": if ( upscale.input_source_type == VideoUpscaleInputSourceType.PROVIDER_REMOTE.value and int(upscale.input_source_fallback_count or 0) < 1 and is_remote_input_access_error(query_result.error) ): upscale.input_source_fallback_count = int(upscale.input_source_fallback_count or 0) + 1 upscale.provider_task_id = None upscale.provider_submitted_at = None upscale.provider_output_url = None upscale.status = VideoUpscaleTaskStatus.PENDING.value upscale.stage = VideoUpscaleStage.QUEUED.value await db.commit() from app.tasks.video_upscale_tasks import submit_remote _safe_apply_async( submit_remote, args=[upscale_task_id], kwargs={"count_attempt": False}, queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE, task=task, upscale_task=upscale, action="submit_local_source_fallback", countdown=max(1, int(settings.VIDEO_UPSCALE_STAGE_HANDOFF_DELAY_SECONDS or 2)), ) log_video_upscale_event( event_type="upscale_source_fallback_local", event_status="retrying", task=task, upscale_task=upscale, detail={"provider_error": query_result.error}, ) return error = query_result.error or {} code = str(error.get("code") or "") error_type = str(error.get("type") or "") retryable = code not in {"InvalidParameter", "Unauthorized", "Forbidden", "NotFound"} and error_type not in { "BadRequest", "AuthError", } log_video_upscale_event( event_type="upscale_provider_task_failed", event_status="failed", task=task, upscale_task=upscale, remote_request_id=query_result.request_id, message=str(error.get("message") or "火山超分任务失败"), detail={ "error_code": code, "error_type": error_type, "error_param": error.get("param"), "provider_error": error, "retryable": retryable, }, error=str(error.get("message") or code), ) await db.commit() await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分任务失败: {code} {error.get('message') or ''}".strip(), retry_action="submit", retryable=retryable, ) return result = query_result.result or {} output_url = str(result.get("video_url") or "").strip() if not output_url: await db.commit() await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message="火山超分任务已完成但未返回 result.video_url", retry_action="poll", retryable=True, ) return upscale.provider_output_url = output_url upscale.provider_output_url_expires_at = ( datetime.fromtimestamp(query_result.expires_at, tz=timezone.utc) if query_result.expires_at else None ) upscale.status = VideoUpscaleTaskStatus.PROCESSING.value upscale.stage = VideoUpscaleStage.RESULT_READY.value set_owner_stage(task, upscale_stage_value(task, ChatGenerationPipelineStage.UPSCALE_DOWNLOADING)) await db.commit() from app.tasks.video_upscale_tasks import download_remote_result _safe_apply_async( download_remote_result, args=[upscale_task_id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE, task=task, upscale_task=upscale, action="download_remote_result", countdown=max(1, int(settings.VIDEO_UPSCALE_STAGE_HANDOFF_DELAY_SECONDS or 2)), ) log_video_upscale_event( event_type="upscale_provider_poll_success", task=task, upscale_task=upscale, remote_request_id=query_result.request_id, detail={"provider_output_url_expires_at": upscale.provider_output_url_expires_at}, ) except RedisExecutionLockError: await db.rollback() raise except VolcMediaKitError as exc: log_video_upscale_event( event_type="upscale_provider_poll_failed", event_status="failed", task=task, upscale_task=upscale, remote_request_id=exc.request_id, message=str(exc), detail=exc.log_detail(), error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) await _persist_provider_error_payload( db, upscale_task_id=upscale_task_id, payload=exc.response_payload, ) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分查询失败: {exc}", retry_action="poll", retryable=exc.retryable, ) except Exception as exc: log_video_upscale_event( event_type="upscale_provider_poll_failed", event_status="failed", task=task, upscale_task=upscale, message=str(exc), error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分查询失败: {exc}", retry_action="poll", ) async def run_finalize_upscale( db: AsyncSession, upscale_task_id: str, *, execution_token: str, execution_guard: ExecutionGuard | None = None ) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.FINALIZING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_FINALIZING.value, increment_attempt=False, execution_token=execution_token, lease_seconds=max(300, int(settings.VIDEO_COVER_TIMEOUT_SECONDS or 15) + 300), ) if not claimed: return upscale, task, _snapshot_data = claimed final_path = str(upscale.final_local_path or "").strip() if not final_path: final_path, _ = _final_video_path(task) log_video_upscale_event( event_type="upscale_finalize_start", task=task, upscale_task=upscale, detail={"final_local_path": final_path}, ) try: await _finalize_success( db, upscale_task_id=upscale_task_id, final_path=final_path, execution_token=execution_token, execution_guard=execution_guard, ) log_video_upscale_event( event_type="upscale_finalize_success", task=task, upscale_task=upscale, detail={"final_local_path": final_path}, ) except RedisExecutionLockError: await db.rollback() raise except Exception as exc: log_video_upscale_event( event_type="upscale_finalize_failed", event_status="failed", task=task, upscale_task=upscale, message=str(exc), detail={"final_local_path": final_path}, error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"超分最终化失败: {exc}", retry_action="finalize", retryable=True, ) async def run_remote_result_download( db: AsyncSession, upscale_task_id: str, *, execution_token: str, execution_guard: ExecutionGuard | None = None ) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.RESULT_DOWNLOADING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_DOWNLOADING.value, increment_attempt=False, execution_token=execution_token, lease_seconds=int(settings.VIDEO_UPSCALE_REMOTE_RESULT_DOWNLOAD_TIMEOUT_SECONDS or 600) + 120, ) if not claimed: return upscale, task, snapshot = claimed expires_at = _aware(upscale.provider_output_url_expires_at) if upscale.provider_task_id and expires_at and (expires_at - _now()).total_seconds() < 2 * 3600: upscale.status = VideoUpscaleTaskStatus.PROCESSING.value upscale.stage = VideoUpscaleStage.REMOTE_POLLING.value upscale.lease_until = None upscale.lease_token = None set_owner_stage(task, upscale_stage_value(task, ChatGenerationPipelineStage.UPSCALE_POLLING)) await db.commit() from app.tasks.video_upscale_tasks import poll_remote _safe_apply_async( poll_remote, args=[upscale_task_id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE, task=task, upscale_task=upscale, action="renew_remote_result", countdown=max(1, int(settings.VIDEO_UPSCALE_STAGE_HANDOFF_DELAY_SECONDS or 2)), ) log_video_upscale_event( event_type="upscale_provider_result_renew_query", event_status="queued", task=task, upscale_task=upscale, detail={"provider_output_url_expires_at": expires_at}, ) return output_url = str(upscale.provider_output_url or "").strip() if not output_url: await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message="火山超分结果下载缺少 provider_output_url", retry_action="poll", ) return final_path, _ = _final_video_path(task) log_video_upscale_event( event_type="upscale_output_download_start", task=task, upscale_task=upscale, detail={"provider_output_url_expires_at": upscale.provider_output_url_expires_at}, ) try: await download_video_to_path( output_url, final_path, int(settings.VIDEO_UPSCALE_REMOTE_RESULT_DOWNLOAD_TIMEOUT_SECONDS or 600), execution_guard=execution_guard, ) await _ensure_guard(execution_guard) log_video_upscale_event( event_type="upscale_output_download_success", task=task, upscale_task=upscale, detail={"final_local_path": final_path, "file_size_bytes": safe_file_size(final_path)}, ) except RedisExecutionLockError: await db.rollback() raise except Exception as exc: log_video_upscale_event( event_type="upscale_output_download_failed", event_status="failed", task=task, upscale_task=upscale, message=str(exc), detail={"provider_output_url_expires_at": upscale.provider_output_url_expires_at}, error=str(exc), ) await db.rollback() await _ensure_guard(execution_guard) expires_at = _aware(upscale.provider_output_url_expires_at) action = "submit" if expires_at and expires_at <= _now() else "download" await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分结果下载失败: {exc}", retry_action=action, retryable=True, ) return try: await _queue_finalize( db, upscale_task_id=upscale_task_id, final_path=final_path, reason="remote_result_download_completed", execution_token=execution_token, execution_guard=execution_guard, ) except RedisExecutionLockError: await db.rollback() raise except Exception as exc: await db.rollback() await _ensure_guard(execution_guard) await _schedule_retry( db, upscale_task_id=upscale_task_id, error_message=f"火山超分结果进入最终化失败: {exc}", retry_action="finalize", retryable=True, ) async def recover_video_upscale_tasks_once(db: AsyncSession) -> dict[str, Any]: now = _now() batch_size = max(1, int(settings.VIDEO_UPSCALE_RECOVERY_BATCH_SIZE or 50)) await apply_short_lock_timeout(db) result = await db.execute( select(VideoUpscaleTask) .where( or_( VideoUpscaleTask.status == VideoUpscaleTaskStatus.PENDING.value, and_( VideoUpscaleTask.status == VideoUpscaleTaskStatus.PROCESSING.value, or_(VideoUpscaleTask.lease_until.is_(None), VideoUpscaleTask.lease_until <= now), ), and_( VideoUpscaleTask.status == VideoUpscaleTaskStatus.RETRY_WAITING.value, or_(VideoUpscaleTask.next_retry_at.is_(None), VideoUpscaleTask.next_retry_at <= now), ), and_( VideoUpscaleTask.status == VideoUpscaleTaskStatus.COMPLETED.value, VideoUpscaleTask.source_local_path.is_not(None), VideoUpscaleTask.source_deleted_at.is_(None), or_( VideoUpscaleTask.source_delete_error.is_(None), VideoUpscaleTask.source_delete_error != VIDEO_UPSCALE_SOURCE_RETAINED_MARKER, ), ), ) ) .order_by(VideoUpscaleTask.updated_at.asc()) .limit(batch_size) .with_for_update(skip_locked=True) ) tasks = list(result.scalars().all()) await db.commit() from app.tasks.video_upscale_tasks import download_remote_result, execute_local, finalize, poll_remote, submit_remote counts: dict[str, int] = {} for item in tasks: action = "" try: if item.status == VideoUpscaleTaskStatus.COMPLETED.value: action = "cleanup" cleanup_upscale, cleanup_chat_task = await _load_pair(db, str(item.id), for_update=True) if not cleanup_upscale or not cleanup_chat_task: continue cleanup_snapshot = _snapshot(cleanup_chat_task) if not bool(cleanup_snapshot.get("delete_source_after_success", True)): cleanup_upscale.source_delete_error = VIDEO_UPSCALE_SOURCE_RETAINED_MARKER await db.commit() counts["cleanup_skipped_by_snapshot"] = counts.get("cleanup_skipped_by_snapshot", 0) + 1 log_video_upscale_event( event_type="upscale_source_retained_by_snapshot", event_status="success", task=cleanup_chat_task, upscale_task=cleanup_upscale, detail={ "source_local_path": cleanup_upscale.source_local_path, "recovery": True, "delete_source_after_success": False, }, ) continue source_path = str(cleanup_upscale.source_local_path or "") removed = safe_remove(source_path) await apply_short_lock_timeout(db) cleanup_result = await db.execute( select(VideoUpscaleTask).where(VideoUpscaleTask.id == item.id).with_for_update().limit(1) ) cleanup_task = cleanup_result.scalar_one_or_none() if cleanup_task: cleanup_task.source_deleted_at = _now() if removed else None cleanup_task.source_delete_error = None if removed else f"源视频删除失败: {source_path}" await db.commit() counts["cleanup_success" if removed else "cleanup_failed"] = counts.get( "cleanup_success" if removed else "cleanup_failed", 0 ) + 1 log_video_upscale_event( event_type="upscale_source_cleanup_success" if removed else "upscale_source_cleanup_failed", event_status="success" if removed else "failed", upscale_task=cleanup_task or item, detail={"source_local_path": source_path, "recovery": True}, error=None if removed else "source_cleanup_failed", ) continue if item.stage == VideoUpscaleStage.FINALIZING.value and is_valid_file(item.final_local_path): action = "finalize" finalize.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE) elif item.processor_key in LOCAL_PROCESSOR_KEYS: action = "local" execute_local.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE) elif item.provider_output_url and item.stage in { VideoUpscaleStage.RESULT_READY.value, VideoUpscaleStage.RESULT_DOWNLOADING.value, }: output_expires_at = _aware(item.provider_output_url_expires_at) if item.provider_task_id and output_expires_at and (output_expires_at - _now()).total_seconds() < 2 * 3600: action = "poll" poll_remote.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE) else: action = "download" download_remote_result.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE) elif item.provider_task_id: action = "poll" poll_remote.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE) else: action = "submit" submit_remote.apply_async(args=[item.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE) counts[action] = counts.get(action, 0) + 1 log_video_upscale_event( event_type="upscale_recovery_enqueued", upscale_task=item, detail={"action": action}, ) except Exception as exc: await db.rollback() counts["enqueue_failed"] = counts.get("enqueue_failed", 0) + 1 log_video_upscale_event( event_type="upscale_recovery_enqueue_failed", event_status="failed", upscale_task=item, message=str(exc), error=str(exc), ) return {"checked": len(tasks), "results": counts} async def reset_failed_upscale_task_for_manual_retry( db: AsyncSession, *, upscale_task_id: str, force_resubmit: bool = False, ) -> VideoUpscaleTask: upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: raise RuntimeError(f"超分任务不存在: {upscale_task_id}") if not is_valid_file(upscale.source_local_path): raise RuntimeError(f"超分源视频不存在，无法人工恢复: {upscale.source_local_path}") restore_owner_for_upscale_retry(task) upscale.status = VideoUpscaleTaskStatus.PENDING.value upscale.stage = VideoUpscaleStage.QUEUED.value upscale.failure_count = 0 upscale.last_error = None upscale.failed_at = None upscale.next_retry_at = None upscale.lease_until = None upscale.lease_token = None upscale.manual_retry_count = int(upscale.manual_retry_count or 0) + 1 if force_resubmit: upscale.provider_task_id = None upscale.provider_submitted_at = None upscale.provider_output_url = None upscale.provider_output_url_expires_at = None await db.commit() return upscale