xiaoxia-saas/packages/domain/duplication.py

"""查重记录领域实体。"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4


@dataclass(slots=True)
class DuplicateSegment:
    """重复片段 — 描述上传视频中的一段与已有视频的匹配关系。"""

    id: str
    source_start: float
    source_end: float
    matched_video_id: str
    matched_video_name: str
    matched_start: float
    matched_end: float
    similarity: float  # 0-100

    @classmethod
    def create(
        cls,
        source_start: float,
        source_end: float,
        matched_video_id: str,
        matched_video_name: str,
        matched_start: float,
        matched_end: float,
        similarity: float,
    ) -> "DuplicateSegment":
        if source_start < 0 or source_end <= source_start:
            raise ValueError("invalid source segment range")
        if matched_start < 0 or matched_end <= matched_start:
            raise ValueError("invalid matched segment range")
        if not 0 <= similarity <= 100:
            raise ValueError("similarity must be between 0 and 100")
        return cls(
            id=uuid4().hex,
            source_start=source_start,
            source_end=source_end,
            matched_video_id=matched_video_id,
            matched_video_name=matched_video_name,
            matched_start=matched_start,
            matched_end=matched_end,
            similarity=similarity,
        )


@dataclass(slots=True)
class DuplicationRecord:
    """查重记录 — 一次视频查重请求的完整生命周期。"""

    id: str
    user_id: str
    filename: str
    file_size: int
    storage_key: str  # OSS 对象键
    duration_seconds: float = 0.0
    status: str = "pending"  # pending / processing / completed / failed
    duplicate_rate: float | None = None  # 0-100
    duplicate_count: int = 0
    # #1661 手动查重：视觉相似度（归一化 0~1）/ 匹配视频数
    visual_similarity: float | None = None
    match_count: int | None = None
    video_fingerprint: dict[str, Any] | None = None
    error_message: str = ""
    segments: list[DuplicateSegment] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @classmethod
    def create(
        cls,
        user_id: str,
        filename: str,
        file_size: int,
        storage_key: str,
        *,
        duration_seconds: float = 0.0,
    ) -> "DuplicationRecord":
        if not user_id.strip():
            raise ValueError("user_id cannot be empty")
        if not filename.strip():
            raise ValueError("filename cannot be empty")
        if file_size <= 0:
            raise ValueError("file_size must be positive")
        return cls(
            id=uuid4().hex,
            user_id=user_id.strip(),
            filename=filename.strip(),
            file_size=file_size,
            storage_key=storage_key,
            duration_seconds=duration_seconds,
        )

    def mark_processing(self) -> None:
        self.status = "processing"
        self.updated_at = datetime.now(timezone.utc)

    def mark_completed(
        self,
        duplicate_rate: float,
        duplicate_count: int,
        segments: list[DuplicateSegment],
        *,
        visual_similarity: float | None = None,
        match_count: int | None = None,
    ) -> None:
        if not 0 <= duplicate_rate <= 100:
            raise ValueError("duplicate_rate must be between 0 and 100")
        self.status = "completed"
        self.duplicate_rate = duplicate_rate
        self.duplicate_count = duplicate_count
        self.segments = segments
        self.visual_similarity = visual_similarity
        self.match_count = match_count
        self.updated_at = datetime.now(timezone.utc)

    def mark_failed(self, error_message: str) -> None:
        self.status = "failed"
        self.error_message = error_message
        self.updated_at = datetime.now(timezone.utc)

    def can_retry(self) -> bool:
        """
        判断是否可重试。

        仅 failed 状态的记录允许重试，防止误操作已完成的记录。

        Returns:
            True 表示可以重试
        """
        return self.status == "failed"

    def reset_for_retry(self) -> None:
        """
        重置记录以重新查重。

        清空查重结果和错误信息，状态回到 pending。
        调用前应先通过 can_retry() 确认状态合法。
        """
        self.status = "pending"
        self.duplicate_rate = None
        self.duplicate_count = 0
        self.visual_similarity = None
        self.match_count = None
        self.error_message = ""
        self.segments = []
        self.video_fingerprint = None
        self.updated_at = datetime.now(timezone.utc)