xiaoxia-saas/packages/application/generated_video_finalize.py

"""#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,
    }