206 lines
7.8 KiB
Python
206 lines
7.8 KiB
Python
"""#2024: 视频生成 finalize 入库用例。
|
||
|
||
Worker 渲染+上传完成后不自动入库，只把渲染产物与查重结果保存到
|
||
GenerationTask.extra_meta["rendered_output"]，并标记为 awaiting_cover。
|
||
用户点「完成」时由 API 调用本用例：创建 GeneratedVideo 记录（复用预计算查重结果）、
|
||
推进任务到 completed，返回新记录 id。
|
||
|
||
设计原则：finalize 必须快速（仅 DB 写入，不下载视频、不重算指纹）——
|
||
所有耗时操作（指纹计算、历史/批次查重）都在 worker 渲染阶段预完成。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class RenderedOutput:
|
||
"""Worker 预计算并写入 extra_meta 的渲染产物+查重结果。"""
|
||
|
||
file_url: str
|
||
file_size: int = 0
|
||
duration: float = 0.0
|
||
width: int = 1280
|
||
height: int = 720
|
||
fps: float = 25.0
|
||
name: str = ""
|
||
thumbnail_url: str = ""
|
||
mode: str = "narrative"
|
||
batch_id: str = ""
|
||
project_id: str = ""
|
||
user_id: str = ""
|
||
# 查重结果（worker 预计算）
|
||
fingerprint_dict: dict[str, Any] | None = None
|
||
fingerprint_chunks: list[dict[str, Any]] | None = None
|
||
is_duplicate: bool = False
|
||
duplicate_of: str | None = None
|
||
duplicate_rate: float | None = None
|
||
match_count: int | None = None
|
||
visual_similarity: float | None = None
|
||
video_fingerprint_md5: str = ""
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: dict[str, Any]) -> "RenderedOutput":
|
||
if not isinstance(data, dict):
|
||
raise ValueError("rendered_output must be a dict")
|
||
# fingerprint_chunks 历史上有两种位置：
|
||
# 1) 顶层 ``fingerprint_chunks``（由 compute_render_fingerprint_and_dedup 直接返回）
|
||
# 2) 嵌套在 ``fingerprint_dict["chunks"]``（VideoFingerprint.to_dict() 序列化的结构）
|
||
# 顶层优先；顶层为空时回退到嵌套位置，兼容旧数据。
|
||
fp_dict = data.get("fingerprint_dict") or {}
|
||
chunks_raw = data.get("fingerprint_chunks")
|
||
if not chunks_raw and isinstance(fp_dict, dict):
|
||
chunks_raw = fp_dict.get("chunks")
|
||
# md5 同样可能在顶层或嵌套在 fingerprint_dict 内（历史数据兼容）
|
||
md5_value = data.get("video_fingerprint_md5")
|
||
if not md5_value and isinstance(fp_dict, dict):
|
||
md5_value = fp_dict.get("md5")
|
||
return cls(
|
||
file_url=str(data.get("file_url") or ""),
|
||
file_size=int(data.get("file_size") or 0),
|
||
duration=float(data.get("duration") or 0.0),
|
||
width=int(data.get("width") or 1280),
|
||
height=int(data.get("height") or 720),
|
||
fps=float(data.get("fps") or 25.0),
|
||
name=str(data.get("name") or ""),
|
||
thumbnail_url=str(data.get("thumbnail_url") or ""),
|
||
mode=str(data.get("mode") or "narrative"),
|
||
batch_id=str(data.get("batch_id") or ""),
|
||
project_id=str(data.get("project_id") or ""),
|
||
user_id=str(data.get("user_id") or ""),
|
||
fingerprint_dict=fp_dict or None,
|
||
fingerprint_chunks=chunks_raw if isinstance(chunks_raw, list) else None,
|
||
is_duplicate=bool(data.get("is_duplicate", False)),
|
||
duplicate_of=data.get("duplicate_of"),
|
||
duplicate_rate=_safe_float(data.get("duplicate_rate")),
|
||
match_count=_safe_int(data.get("match_count")),
|
||
visual_similarity=_safe_float(data.get("visual_similarity")),
|
||
video_fingerprint_md5=str(md5_value or ""),
|
||
)
|
||
|
||
|
||
def _safe_float(v) -> float | None:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _safe_int(v) -> int | None:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return int(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def finalize_generated_video(
|
||
*,
|
||
task,
|
||
session: Session,
|
||
effective_cover_url: str = "",
|
||
custom_name: str | None = None,
|
||
) -> dict:
|
||
"""将 awaiting_cover 的任务正式入库。
|
||
|
||
从 ``task.extra_meta["rendered_output"]`` 读取 worker 预存的渲染结果与查重数据，
|
||
创建 GeneratedVideo 记录并 commit；调用方负责将 task 推进到 completed 并 update。
|
||
|
||
Returns:
|
||
{"video_id": str, "is_duplicate": bool, "duplicate_of": str|None}
|
||
"""
|
||
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
||
SQLAlchemyGeneratedVideoRepository,
|
||
)
|
||
from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel
|
||
from packages.domain.generated_video import GeneratedVideo
|
||
|
||
meta = dict(task.extra_meta or {})
|
||
rendered_dict = meta.get("rendered_output") or {}
|
||
rendered = RenderedOutput.from_dict(rendered_dict)
|
||
|
||
if not rendered.file_url.strip():
|
||
raise ValueError(f"task {task.id} rendered_output.file_url 为空，无法 finalize")
|
||
|
||
video_id = uuid4().hex
|
||
_custom = (custom_name or "").strip() if custom_name else ""
|
||
video_name = _custom or (rendered.name.strip() or f"generated-{task.id[:8]}.mp4")
|
||
|
||
generated_video = GeneratedVideo(
|
||
id=video_id,
|
||
project_id=(rendered.project_id or task.project_id or "").strip(),
|
||
user_id=(rendered.user_id or task.created_by_user_id or "").strip(),
|
||
generation_task_id=task.id,
|
||
name=video_name,
|
||
file_url=rendered.file_url.strip(),
|
||
file_size=rendered.file_size,
|
||
duration=rendered.duration,
|
||
width=rendered.width,
|
||
height=rendered.height,
|
||
fps=rendered.fps,
|
||
status="completed",
|
||
generation_params={"mode": rendered.mode},
|
||
thumbnail_url=effective_cover_url or rendered.thumbnail_url or None,
|
||
video_fingerprint=rendered.fingerprint_dict,
|
||
is_duplicate=rendered.is_duplicate,
|
||
duplicate_of=rendered.duplicate_of,
|
||
duplicate_rate=rendered.duplicate_rate,
|
||
match_count=rendered.match_count,
|
||
visual_similarity=rendered.visual_similarity,
|
||
created_at=datetime.now(UTC),
|
||
generated_at=datetime.now(UTC),
|
||
)
|
||
|
||
# 写入分片指纹（worker 预序列化的 chunk 列表）
|
||
if rendered.fingerprint_chunks:
|
||
try:
|
||
chunk_models = []
|
||
for c in rendered.fingerprint_chunks:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
chunk_models.append(
|
||
VideoFingerprintChunkModel(
|
||
id=uuid4().hex,
|
||
video_id=video_id,
|
||
project_id=generated_video.project_id,
|
||
user_id=generated_video.user_id,
|
||
start_time_ms=int(c.get("start_time_ms", 0)),
|
||
end_time_ms=int(c.get("end_time_ms", 0)),
|
||
phash_binary=str(c.get("phash_binary", "")),
|
||
color_histogram=[float(v) for v in (c.get("color_histogram") or [])],
|
||
frame_count=int(c.get("frame_count", 0)),
|
||
)
|
||
)
|
||
if chunk_models:
|
||
session.bulk_save_objects(chunk_models)
|
||
except Exception as chunk_err:
|
||
logger.warning("Failed to persist fingerprint chunks for video %s: %s", video_id, chunk_err)
|
||
|
||
video_repo = SQLAlchemyGeneratedVideoRepository(session)
|
||
video_repo.create(generated_video)
|
||
session.commit()
|
||
logger.info(
|
||
"[finalize] GeneratedVideo created: %s (task=%s, dup=%s, cover=%s)",
|
||
video_id,
|
||
task.id,
|
||
rendered.is_duplicate,
|
||
bool(effective_cover_url),
|
||
)
|
||
return {
|
||
"video_id": video_id,
|
||
"is_duplicate": rendered.is_duplicate,
|
||
"duplicate_of": rendered.duplicate_of,
|
||
}
|