from __future__ import annotations import json import math import os import uuid from datetime import datetime, timedelta, timezone 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.ai.task_group_service import aggregate_parent_for_child from app.services.generation.module_hook_service import notify_chat_generation_task_finished 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 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: 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, ) -> 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) source_info = 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 ) 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 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=upscale, 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=upscale, 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, 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 = uuid.uuid4().hex 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 _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() 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, ) -> None: upscale, task = await _load_pair(db, upscale_task_id, 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", ) 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) -> None: upscale, task = await _load_pair(db, upscale_task_id, 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("超分最终视频封面生成失败") upscale, task = await _load_pair(db, upscale_task_id, 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 = 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) await notify_chat_generation_task_finished(db, task) await aggregate_parent_for_child(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 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}" 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) -> 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, 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), ) log_video_upscale_event( event_type="upscale_local_success", task=task, upscale_task=upscale, detail={"final_local_path": final_path}, ) except Exception as exc: await db.rollback() 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", ) except Exception as exc: await db.rollback() 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) -> 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, 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)}", ) upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: return 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 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 _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 _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) -> 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, 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), ) upscale, task = await _load_pair(db, upscale_task_id, for_update=True) if not upscale or not task: return 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", ) 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", ) 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 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 _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 _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) -> None: claimed = await _claim( db, upscale_task_id=upscale_task_id, stage=VideoUpscaleStage.FINALIZING.value, chat_stage=ChatGenerationPipelineStage.UPSCALE_FINALIZING.value, increment_attempt=False, 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) log_video_upscale_event( event_type="upscale_finalize_success", task=task, upscale_task=upscale, detail={"final_local_path": final_path}, ) 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 _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) -> 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, 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", ) 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), ) 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 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() 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", ) except Exception as exc: await db.rollback() 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)) 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) 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