alembic/versions/012_add_duplication_tables.py

@@ -0,0 +1,73 @@
+"""Phase 2 - 查重功能：duplication_records + duplication_segments
+
+Revision ID: 012
+Revises: 011
+Create Date: 2026-06-28
+
+This migration creates two new tables:
+1. duplication_records — 查重记录主表
+2. duplication_segments — 重复片段详情表
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers
+revision = "012"
+down_revision = "011"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # ── 1. Create duplication_records table ──
+
+    conn.execute(sa.text("""
+        CREATE TABLE IF NOT EXISTS duplication_records (
+            id VARCHAR(36) PRIMARY KEY,
+            user_id VARCHAR(36) NOT NULL,
+            filename VARCHAR(500) NOT NULL,
+            file_size INTEGER NOT NULL,
+            storage_key VARCHAR(500) NOT NULL,
+            duration_seconds FLOAT NOT NULL DEFAULT 0,
+            status VARCHAR(20) NOT NULL DEFAULT 'pending',
+            duplicate_rate FLOAT,
+            duplicate_count INTEGER NOT NULL DEFAULT 0,
+            video_fingerprint TEXT,
+            error_message TEXT NOT NULL DEFAULT '',
+            created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMP NOT NULL DEFAULT NOW()
+        )
+    """))
+    conn.execute(sa.text(
+        "CREATE INDEX IF NOT EXISTS ix_duplication_records_user_id ON duplication_records(user_id)"
+    ))
+    conn.execute(sa.text(
+        "CREATE INDEX IF NOT EXISTS ix_duplication_records_status ON duplication_records(status)"
+    ))
+
+    # ── 2. Create duplication_segments table ──
+
+    conn.execute(sa.text("""
+        CREATE TABLE IF NOT EXISTS duplication_segments (
+            id VARCHAR(36) PRIMARY KEY,
+            record_id VARCHAR(36) NOT NULL,
+            source_start FLOAT NOT NULL,
+            source_end FLOAT NOT NULL,
+            matched_video_id VARCHAR(36) NOT NULL,
+            matched_video_name VARCHAR(500) NOT NULL DEFAULT '',
+            matched_start FLOAT NOT NULL,
+            matched_end FLOAT NOT NULL,
+            similarity FLOAT NOT NULL
+        )
+    """))
+    conn.execute(sa.text(
+        "CREATE INDEX IF NOT EXISTS ix_duplication_segments_record_id ON duplication_segments(record_id)"
+    ))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("DROP TABLE IF EXISTS duplication_segments"))
+    conn.execute(sa.text("DROP TABLE IF EXISTS duplication_records"))

apps/api/app/api/router.py

@@ -4,6 +4,7 @@ from app.api.routes.assets import router as assets_router
 from app.api.routes.auth import router as auth_router
 from app.api.routes.chunked_upload import router as chunked_upload_router
 from app.api.routes.classification_jobs import router as classification_jobs_router
+from app.api.routes.duplication import router as duplication_router
 from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.titles import router as titles_router
 from app.api.routes.voices import router as voices_router
@@ -86,3 +87,8 @@ api_router.include_router(
     prefix="/voices",
     tags=["VoiceLibrary"],
 )
+api_router.include_router(
+    duplication_router,
+    prefix="/duplication",
+    tags=["Duplication"],
+)

apps/api/app/api/routes/duplication.py

@@ -0,0 +1,270 @@
+"""查重 API 路由。"""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import uuid4
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_duplication_repository
+from app.schemas.duplication import (
+    DuplicationDetailResponse,
+    DuplicationRecordResponse,
+    DuplicationUploadResponse,
+    DuplicateSegmentResponse,
+)
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
+
+from packages.application import (
+    DeleteDuplicationRecordUseCase,
+    GetDuplicationDetailUseCase,
+    ListDuplicationRecordsUseCase,
+    RetryDuplicationUseCase,
+    UploadForDuplicationCommand,
+    UploadForDuplicationUseCase,
+)
+from packages.domain.duplication import DuplicationRecord
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# 查重功能只接受视频文件
+ALLOWED_VIDEO_MIME_TYPES = frozenset({
+    "video/mp4", "video/mpeg", "video/quicktime", "video/x-msvideo",
+    "video/webm", "video/x-matroska", "video/3gpp",
+})
+
+
+def _validate_video_mime_type(content_type: str | None) -> str:
+    """验证视频文件的 MIME 类型，如果无效则抛出异常。"""
+    if not content_type:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Content-Type header is required",
+        )
+    
+    # 处理带参数的类型，如 "video/mp4; charset=utf-8"
+    base_type = content_type.split(";")[0].strip().lower()
+    
+    if base_type not in ALLOWED_VIDEO_MIME_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+            detail=f"只支持视频文件。支持的类型: mp4, mpeg, mov, avi, webm, mkv, 3gp",
+        )
+    
+    return base_type
+
+
+def _to_record_response(record: DuplicationRecord) -> DuplicationRecordResponse:
+    return DuplicationRecordResponse(
+        id=record.id,
+        filename=record.filename,
+        file_size=record.file_size,
+        duration_seconds=record.duration_seconds,
+        status=record.status,
+        duplicate_rate=record.duplicate_rate,
+        duplicate_count=record.duplicate_count,
+        created_at=record.created_at.isoformat(),
+        updated_at=record.updated_at.isoformat(),
+    )
+
+
+def _to_detail_response(record: DuplicationRecord) -> DuplicationDetailResponse:
+    return DuplicationDetailResponse(
+        id=record.id,
+        filename=record.filename,
+        file_size=record.file_size,
+        duration_seconds=record.duration_seconds,
+        status=record.status,
+        duplicate_rate=record.duplicate_rate,
+        duplicate_count=record.duplicate_count,
+        created_at=record.created_at.isoformat(),
+        updated_at=record.updated_at.isoformat(),
+        segments=[
+            DuplicateSegmentResponse(
+                id=seg.id,
+                source_start=seg.source_start,
+                source_end=seg.source_end,
+                matched_video_id=seg.matched_video_id,
+                matched_video_name=seg.matched_video_name,
+                matched_start=seg.matched_start,
+                matched_end=seg.matched_end,
+                similarity=seg.similarity,
+            )
+            for seg in record.segments
+        ],
+    )
+
+
+@router.post("/upload", response_model=DuplicationUploadResponse)
+async def upload_for_duplication(
+    file: UploadFile = File(..., description="要查重的视频文件"),
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> DuplicationUploadResponse:
+    """上传视频进行查重。"""
+    if file.filename is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="文件名不能为空",
+        )
+
+    # P0-1: 验证 MIME 类型（只接受视频文件）
+    validated_content_type = _validate_video_mime_type(file.content_type)
+
+    # P0-2: 验证文件大小（参考 OSS_DIRECT_UPLOAD_MAX_MB）
+    from app.config import get_settings
+    settings = get_settings()
+    max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024
+    
+    # 先检查 Content-Length header（如果可用）
+    if file.size is not None and file.size > max_size_bytes:
+        raise HTTPException(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
+        )
+
+    # 读取文件内容并上传到 OSS
+    file_id = uuid4().hex[:8]
+    safe_filename = file.filename.replace("/", "_").replace("\\", "_")
+    storage_key = f"duplication/{file_id}/{safe_filename}"
+
+    try:
+        content = await file.read()
+        file_size = len(content)
+        
+        # 再次检查实际文件大小
+        if file_size > max_size_bytes:
+            raise HTTPException(
+                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+                detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
+            )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"读取文件失败: {exc}",
+        ) from exc
+
+    try:
+        storage_service.upload_file(
+            content,
+            storage_key,
+            content_type=validated_content_type,
+        )
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"文件上传失败: {exc}",
+        ) from exc
+
+    use_case = UploadForDuplicationUseCase(duplication_repository)
+    record = use_case.execute(
+        UploadForDuplicationCommand(
+            user_id=authenticated_user.user.id,
+            filename=file.filename,
+            file_size=file_size,
+            storage_key=storage_key,
+        )
+    )
+
+    logger.info(
+        "Duplication upload: record=%s file=%s user=%s",
+        record.id,
+        file.filename,
+        authenticated_user.user.id,
+    )
+
+    return DuplicationUploadResponse(
+        id=record.id,
+        status=record.status,
+        message=f'文件 "{file.filename}" 已上传，正在查重中...',
+    )
+
+
+@router.get("/records", response_model=list[DuplicationRecordResponse])
+def list_duplication_records(
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> list[DuplicationRecordResponse]:
+    """获取当前用户的查重记录列表。"""
+    use_case = ListDuplicationRecordsUseCase(duplication_repository)
+    records = use_case.execute(authenticated_user.user.id)
+    return [_to_record_response(r) for r in records]
+
+
+@router.get("/records/{record_id}", response_model=DuplicationDetailResponse)
+def get_duplication_detail(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> DuplicationDetailResponse:
+    """获取查重记录详情（含重复片段）。"""
+    use_case = GetDuplicationDetailUseCase(duplication_repository)
+    record = use_case.execute(record_id)
+    if record is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+    if record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+    return _to_detail_response(record)
+
+
+@router.delete("/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_duplication_record(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> None:
+    """删除查重记录。"""
+    # 检查记录是否存在且属于当前用户
+    detail_uc = GetDuplicationDetailUseCase(duplication_repository)
+    record = detail_uc.execute(record_id)
+    if record is None or record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    use_case = DeleteDuplicationRecordUseCase(duplication_repository)
+    use_case.execute(record_id)
+
+
+@router.post("/records/{record_id}/retry", response_model=DuplicationUploadResponse)
+def retry_duplication(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> DuplicationUploadResponse:
+    """重新提交查重。"""
+    # 检查记录存在且属于当前用户
+    detail_uc = GetDuplicationDetailUseCase(duplication_repository)
+    record = detail_uc.execute(record_id)
+    if record is None or record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    use_case = RetryDuplicationUseCase(duplication_repository)
+    updated = use_case.execute(record_id)
+    if updated is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    return DuplicationUploadResponse(
+        id=updated.id,
+        status=updated.status,
+        message="已重新提交查重",
+    )

apps/api/app/dependencies.py

@@ -21,6 +21,9 @@ from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRe
 from packages.adapters.sqlalchemy_impl.classification_job_repository import (
     SQLAlchemyClassificationJobRepository,
 )
+from packages.adapters.sqlalchemy_impl.duplication_repository import (
+    SQLAlchemyDuplicationRecordRepository,
+)
 from packages.adapters.sqlalchemy_impl.generated_video_repository import (
     SQLAlchemyGeneratedVideoRepository,
 )
@@ -45,6 +48,7 @@ from packages.ports.asset_repository import AssetRepository
 from packages.ports.asset_library_repository import AssetLibraryRepository
 from packages.ports.user_repository import UserRepository
 from packages.ports.classification_job_repository import ClassificationJobRepository
+from packages.ports.duplication_repository import DuplicationRecordRepository
 from packages.ports.generation_task_repository import GenerationTaskRepository
 from packages.ports.title_library_repository import TitleLibraryRepository
 from packages.ports.voice_library_repository import VoiceLibraryRepository
@@ -106,6 +110,13 @@ def get_generated_video_repository(
     return SQLAlchemyGeneratedVideoRepository(session)
 
 
+def get_duplication_repository(
+    session: Session = Depends(get_db_session),
+) -> SQLAlchemyDuplicationRecordRepository:
+    """Provide the SQLAlchemy duplication record repository implementation."""
+    return SQLAlchemyDuplicationRecordRepository(session)
+
+
 def get_project_repository(
     session: Session = Depends(get_db_session),
 ) -> SQLAlchemyProjectRepository:

apps/api/app/schemas/duplication.py

@@ -0,0 +1,45 @@
+"""查重 API Pydantic schemas。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class DuplicateSegmentResponse(BaseModel):
+    """重复片段响应。"""
+
+    id: str
+    source_start: float
+    source_end: float
+    matched_video_id: str
+    matched_video_name: str
+    matched_start: float
+    matched_end: float
+    similarity: float
+
+
+class DuplicationRecordResponse(BaseModel):
+    """查重记录响应（列表项）。"""
+
+    id: str
+    filename: str
+    file_size: int
+    duration_seconds: float = 0.0
+    status: str = "pending"
+    duplicate_rate: float | None = None
+    duplicate_count: int = 0
+    created_at: str
+    updated_at: str
+
+
+class DuplicationDetailResponse(DuplicationRecordResponse):
+    """查重详情响应（含重复片段）。"""
+
+    segments: list[DuplicateSegmentResponse] = Field(default_factory=list)
+
+
+class DuplicationUploadResponse(BaseModel):
+    """上传查重响应。"""
+
+    id: str
+    status: str
+    message: str

packages/adapters/sqlalchemy_impl/duplication_repository.py

@@ -0,0 +1,134 @@
+"""查重记录 SQLAlchemy 仓库实现。"""
+from __future__ import annotations
+
+import json
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import DuplicationRecordModel, DuplicationSegmentModel
+from packages.domain.duplication import DuplicateSegment, DuplicationRecord
+
+
+class SQLAlchemyDuplicationRecordRepository:
+    def __init__(self, session: Session):
+        self.session = session
+
+    def create(self, record: DuplicationRecord) -> DuplicationRecord:
+        model = DuplicationRecordModel(
+            id=record.id,
+            user_id=record.user_id,
+            filename=record.filename,
+            file_size=record.file_size,
+            storage_key=record.storage_key,
+            duration_seconds=record.duration_seconds,
+            status=record.status,
+            duplicate_rate=record.duplicate_rate,
+            duplicate_count=record.duplicate_count,
+            video_fingerprint=json.dumps(record.video_fingerprint) if record.video_fingerprint else None,
+            error_message=record.error_message,
+            created_at=record.created_at,
+            updated_at=record.updated_at,
+        )
+        self.session.add(model)
+        self.session.commit()
+        return record
+
+    def get(self, record_id: str) -> DuplicationRecord | None:
+        model = self.session.query(DuplicationRecordModel).filter(
+            DuplicationRecordModel.id == record_id
+        ).first()
+        if model is None:
+            return None
+        return self._to_domain(model)
+
+    def list_by_user(self, user_id: str, *, offset: int = 0, limit: int = 50) -> list[DuplicationRecord]:
+        models = (
+            self.session.query(DuplicationRecordModel)
+            .filter(DuplicationRecordModel.user_id == user_id)
+            .order_by(DuplicationRecordModel.created_at.desc())
+            .offset(offset)
+            .limit(limit)
+            .all()
+        )
+        return [self._to_domain(m) for m in models]
+
+    def update(self, record: DuplicationRecord) -> DuplicationRecord:
+        model = self.session.query(DuplicationRecordModel).filter(
+            DuplicationRecordModel.id == record.id
+        ).first()
+        if model is None:
+            return record
+        model.status = record.status
+        model.duplicate_rate = record.duplicate_rate
+        model.duplicate_count = record.duplicate_count
+        model.video_fingerprint = json.dumps(record.video_fingerprint) if record.video_fingerprint else None
+        model.error_message = record.error_message
+        model.updated_at = record.updated_at
+
+        # 更新 segments：先删后建
+        self.session.query(DuplicationSegmentModel).filter(
+            DuplicationSegmentModel.record_id == record.id
+        ).delete()
+        for seg in record.segments:
+            seg_model = DuplicationSegmentModel(
+                id=seg.id,
+                record_id=record.id,
+                source_start=seg.source_start,
+                source_end=seg.source_end,
+                matched_video_id=seg.matched_video_id,
+                matched_video_name=seg.matched_video_name,
+                matched_start=seg.matched_start,
+                matched_end=seg.matched_end,
+                similarity=seg.similarity,
+            )
+            self.session.add(seg_model)
+
+        self.session.commit()
+        return record
+
+    def delete(self, record_id: str) -> bool:
+        count = self.session.query(DuplicationRecordModel).filter(
+            DuplicationRecordModel.id == record_id
+        ).delete()
+        self.session.query(DuplicationSegmentModel).filter(
+            DuplicationSegmentModel.record_id == record_id
+        ).delete()
+        self.session.commit()
+        return count > 0
+
+    def _to_domain(self, model: DuplicationRecordModel) -> DuplicationRecord:
+        segment_models = (
+            self.session.query(DuplicationSegmentModel)
+            .filter(DuplicationSegmentModel.record_id == model.id)
+            .all()
+        )
+        segments = [
+            DuplicateSegment(
+                id=s.id,
+                source_start=s.source_start,
+                source_end=s.source_end,
+                matched_video_id=s.matched_video_id,
+                matched_video_name=s.matched_video_name,
+                matched_start=s.matched_start,
+                matched_end=s.matched_end,
+                similarity=s.similarity,
+            )
+            for s in segment_models
+        ]
+        fp_raw = getattr(model, "video_fingerprint", None)
+        return DuplicationRecord(
+            id=model.id,
+            user_id=model.user_id,
+            filename=model.filename,
+            file_size=int(model.file_size or 0),
+            storage_key=model.storage_key,
+            duration_seconds=model.duration_seconds,
+            status=model.status,
+            duplicate_rate=model.duplicate_rate,
+            duplicate_count=int(model.duplicate_count or 0),
+            video_fingerprint=json.loads(fp_raw) if fp_raw else None,
+            error_message=getattr(model, "error_message", ""),
+            segments=segments,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )

packages/adapters/sqlalchemy_impl/models.py

@@ -222,3 +222,35 @@ class VoiceLibraryModel(Base):
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 
+
+class DuplicationRecordModel(Base):
+    __tablename__ = "duplication_records"
+
+    id = Column(String(36), primary_key=True)
+    user_id = Column(String(36), nullable=False, index=True)
+    filename = Column(String(500), nullable=False)
+    file_size = Column(Integer, nullable=False)
+    storage_key = Column(String(500), nullable=False)
+    duration_seconds = Column(Float, nullable=False, default=0)
+    status = Column(String(20), nullable=False, default="pending", index=True)
+    duplicate_rate = Column(Float, nullable=True)
+    duplicate_count = Column(Integer, nullable=False, default=0)
+    video_fingerprint = Column(Text, nullable=True)
+    error_message = Column(Text, nullable=False, default="")
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+    updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+
+
+class DuplicationSegmentModel(Base):
+    __tablename__ = "duplication_segments"
+
+    id = Column(String(36), primary_key=True)
+    record_id = Column(String(36), nullable=False, index=True)
+    source_start = Column(Float, nullable=False)
+    source_end = Column(Float, nullable=False)
+    matched_video_id = Column(String(36), nullable=False)
+    matched_video_name = Column(String(500), nullable=False, default="")
+    matched_start = Column(Float, nullable=False)
+    matched_end = Column(Float, nullable=False)
+    similarity = Column(Float, nullable=False)
+

packages/application/__init__.py

@@ -6,6 +6,14 @@ from .asset_libraries import (
     ListAssetLibrariesUseCase,
 )
 from .assets import CreateAssetCommand, CreateAssetUseCase, ListAssetsUseCase
+from .duplication import (
+    DeleteDuplicationRecordUseCase,
+    GetDuplicationDetailUseCase,
+    ListDuplicationRecordsUseCase,
+    RetryDuplicationUseCase,
+    UploadForDuplicationCommand,
+    UploadForDuplicationUseCase,
+)
 from .classification_jobs import (
     SubmitClassificationJobCommand,
     SubmitClassificationJobUseCase,
@@ -33,16 +41,22 @@ __all__ = [
     "CreateGenerationTaskUseCase",
     "CreateProjectCommand",
     "CreateProjectUseCase",
+    "DeleteDuplicationRecordUseCase",
+    "GetDuplicationDetailUseCase",
     "GetGeneratedVideoDownloadUrlUseCase",
     "GetGeneratedVideoUseCase",
     "GetProjectUseCase",
     "ListAssetLibrariesUseCase",
     "ListAssetsUseCase",
+    "ListDuplicationRecordsUseCase",
     "ListGeneratedVideosByTaskUseCase",
     "ListGeneratedVideosUseCase",
     "ListProjectsUseCase",
+    "RetryDuplicationUseCase",
     "SubmitClassificationJobCommand",
     "SubmitClassificationJobUseCase",
     "SubmitIngestJobCommand",
     "SubmitIngestJobUseCase",
+    "UploadForDuplicationCommand",
+    "UploadForDuplicationUseCase",
 ]

packages/application/duplication.py

@@ -0,0 +1,90 @@
+"""查重应用层用例。"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from packages.domain.duplication import DuplicationRecord
+from packages.ports.duplication_repository import DuplicationRecordRepository
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class UploadForDuplicationCommand:
+    user_id: str
+    filename: str
+    file_size: int
+    storage_key: str
+    duration_seconds: float = 0.0
+
+
+class UploadForDuplicationUseCase:
+    """上传视频进行查重 — 创建查重记录并返回。"""
+
+    def __init__(self, duplication_repository: DuplicationRecordRepository):
+        self.duplication_repository = duplication_repository
+
+    def execute(self, command: UploadForDuplicationCommand) -> DuplicationRecord:
+        record = DuplicationRecord.create(
+            user_id=command.user_id,
+            filename=command.filename,
+            file_size=command.file_size,
+            storage_key=command.storage_key,
+            duration_seconds=command.duration_seconds,
+        )
+        record = self.duplication_repository.create(record)
+        logger.info("Duplication record created: %s for user %s", record.id, record.user_id)
+        return record
+
+
+class ListDuplicationRecordsUseCase:
+    """获取用户查重记录列表。"""
+
+    def __init__(self, duplication_repository: DuplicationRecordRepository):
+        self.duplication_repository = duplication_repository
+
+    def execute(self, user_id: str, *, offset: int = 0, limit: int = 50) -> list[DuplicationRecord]:
+        if not user_id.strip():
+            raise ValueError("user_id 不能为空")
+        return self.duplication_repository.list_by_user(user_id.strip(), offset=offset, limit=limit)
+
+
+class GetDuplicationDetailUseCase:
+    """获取查重记录详情（含重复片段）。"""
+
+    def __init__(self, duplication_repository: DuplicationRecordRepository):
+        self.duplication_repository = duplication_repository
+
+    def execute(self, record_id: str) -> DuplicationRecord | None:
+        return self.duplication_repository.get(record_id)
+
+
+class DeleteDuplicationRecordUseCase:
+    """删除查重记录。"""
+
+    def __init__(self, duplication_repository: DuplicationRecordRepository):
+        self.duplication_repository = duplication_repository
+
+    def execute(self, record_id: str) -> bool:
+        return self.duplication_repository.delete(record_id)
+
+
+class RetryDuplicationUseCase:
+    """重新提交查重 — 将记录状态重置为 pending。"""
+
+    def __init__(self, duplication_repository: DuplicationRecordRepository):
+        self.duplication_repository = duplication_repository
+
+    def execute(self, record_id: str) -> DuplicationRecord | None:
+        record = self.duplication_repository.get(record_id)
+        if record is None:
+            return None
+        record.status = "pending"
+        record.error_message = ""
+        record.duplicate_rate = None
+        record.duplicate_count = 0
+        record.segments = []
+        record = self.duplication_repository.update(record)
+        logger.info("Duplication record %s reset to pending for retry", record_id)
+        return record

packages/domain/__init__.py

@@ -17,6 +17,7 @@ from .entities import (
     Project,
     User,
 )
+from .duplication import DuplicateSegment, DuplicationRecord
 from .generated_video import GeneratedVideo
 from .generation_task import GenerationTask, GenerationTaskStatus
 from .title_library import TitleLibraryItem
@@ -31,6 +32,8 @@ __all__ = [
     "ClassificationJob",
     "ClassificationJobStatus",
     "ClassificationStatus",
+    "DuplicateSegment",
+    "DuplicationRecord",
     "EditingMode",
     "GeneratedVideo",
     "GenerationTask",

packages/domain/duplication.py

@@ -0,0 +1,112 @@
+"""查重记录领域实体。"""
+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
+    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]) -> 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.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)

packages/ports/duplication_repository.py

@@ -0,0 +1,18 @@
+"""查重记录仓库端口（Protocol）。"""
+from __future__ import annotations
+
+from typing import Protocol
+
+from packages.domain.duplication import DuplicationRecord
+
+
+class DuplicationRecordRepository(Protocol):
+    def create(self, record: DuplicationRecord) -> DuplicationRecord: ...
+
+    def get(self, record_id: str) -> DuplicationRecord | None: ...
+
+    def list_by_user(self, user_id: str, *, offset: int = 0, limit: int = 50) -> list[DuplicationRecord]: ...
+
+    def update(self, record: DuplicationRecord) -> DuplicationRecord: ...
+
+    def delete(self, record_id: str) -> bool: ...