From 312dad7497b1006e7313261d2e4e46e6e68f46a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?API=E6=96=87=E6=A1=A3=E7=BB=B4=E6=8A=A4Agent?= Date: Sun, 28 Jun 2026 15:41:34 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20Phase=202=20=E6=9F=A5=E9=87=8D?= =?UTF-8?q?=E5=90=8E=E7=AB=AF=20API=20=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 六边形架构分层实现： Domain 层: - DuplicationRecord / DuplicateSegment 实体 - 状态机生命周期 (pending→processing→completed/failed) Ports 层: - DuplicationRecordRepository Protocol Adapters 层: - SQLAlchemyDuplicationRecordRepository 完整实现 - DuplicationRecordModel / DuplicationSegmentModel ORM 模型 Application 层: - UploadForDuplicationUseCase (上传查重) - ListDuplicationRecordsUseCase (记录列表) - GetDuplicationDetailUseCase (详情查询) - DeleteDuplicationRecordUseCase (删除) - RetryDuplicationUseCase (重新查重) API 层: - POST /api/v1/duplication/upload (上传视频查重) - GET /api/v1/duplication/records (记录列表) - GET /api/v1/duplication/records/{id} (详情含片段) - DELETE /api/v1/duplication/records/{id} (删除) - POST /api/v1/duplication/records/{id}/retry (重新查重) Alembic 迁移: - 011_add_duplication_tables.py (duplication_records + duplication_segments) API 契约与前端 feat/phase2-duplication-ui (PR#75) 完全对齐。 Co-Authored-By: Claude Fable 5 --- .../versions/011_add_duplication_tables.py | 73 ++++++ apps/api/app/api/router.py | 6 + apps/api/app/api/routes/duplication.py | 220 ++++++++++++++++++ apps/api/app/dependencies.py | 11 + apps/api/app/schemas/duplication.py | 45 ++++ .../sqlalchemy_impl/duplication_repository.py | 134 +++++++++++ packages/adapters/sqlalchemy_impl/models.py | 32 +++ packages/application/__init__.py | 14 ++ packages/application/duplication.py | 90 +++++++ packages/domain/__init__.py | 3 + packages/domain/duplication.py | 112 +++++++++ packages/ports/duplication_repository.py | 18 ++ 12 files changed, 758 insertions(+) create mode 100644 alembic/versions/011_add_duplication_tables.py create mode 100644 apps/api/app/api/routes/duplication.py create mode 100644 apps/api/app/schemas/duplication.py create mode 100644 packages/adapters/sqlalchemy_impl/duplication_repository.py create mode 100644 packages/application/duplication.py create mode 100644 packages/domain/duplication.py create mode 100644 packages/ports/duplication_repository.py diff --git a/alembic/versions/011_add_duplication_tables.py b/alembic/versions/011_add_duplication_tables.py new file mode 100644 index 000000000..cea50acbf --- /dev/null +++ b/alembic/versions/011_add_duplication_tables.py @@ -0,0 +1,73 @@ +"""Phase 2 - 查重功能：duplication_records + duplication_segments + +Revision ID: 011 +Revises: 010 +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 = "011" +down_revision = "010" +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")) diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 049378ae5..b750819cc 100644 --- a/apps/api/app/api/router.py +++ b/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"], +) diff --git a/apps/api/app/api/routes/duplication.py b/apps/api/app/api/routes/duplication.py new file mode 100644 index 000000000..2c5c4fcec --- /dev/null +++ b/apps/api/app/api/routes/duplication.py @@ -0,0 +1,220 @@ +"""查重 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() + + +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="文件名不能为空", + ) + + # 读取文件内容并上传到 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) + 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=file.content_type or "video/mp4", + ) + 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="已重新提交查重", + ) diff --git a/apps/api/app/dependencies.py b/apps/api/app/dependencies.py index fe6a993e2..79bcdffdc 100644 --- a/apps/api/app/dependencies.py +++ b/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: diff --git a/apps/api/app/schemas/duplication.py b/apps/api/app/schemas/duplication.py new file mode 100644 index 000000000..48416cfbe --- /dev/null +++ b/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 diff --git a/packages/adapters/sqlalchemy_impl/duplication_repository.py b/packages/adapters/sqlalchemy_impl/duplication_repository.py new file mode 100644 index 000000000..9211d8525 --- /dev/null +++ b/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, + ) diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 390db0dd4..8cb94e29f 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/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) + diff --git a/packages/application/__init__.py b/packages/application/__init__.py index 960a30c11..161b22e17 100644 --- a/packages/application/__init__.py +++ b/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", ] diff --git a/packages/application/duplication.py b/packages/application/duplication.py new file mode 100644 index 000000000..4879a509a --- /dev/null +++ b/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 diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py index 9ac6df2e0..7ab16d99a 100644 --- a/packages/domain/__init__.py +++ b/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", diff --git a/packages/domain/duplication.py b/packages/domain/duplication.py new file mode 100644 index 000000000..d83785e13 --- /dev/null +++ b/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) diff --git a/packages/ports/duplication_repository.py b/packages/ports/duplication_repository.py new file mode 100644 index 000000000..40e9bfcb9 --- /dev/null +++ b/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: ... -- 2.54.0 From ea0bde374237c37d497cbd930f9205dcb51f04fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?API=E6=96=87=E6=A1=A3=E7=BB=B4=E6=8A=A4Agent?= Date: Sun, 28 Jun 2026 15:58:07 +0800 Subject: [PATCH 2/2] fix: add security validations to duplication upload API P0-1: Add MIME type whitelist validation (video files only) - Accept: mp4, mpeg, mov, avi, webm, mkv, 3gp - Reject non-video files with 415 Unsupported Media Type P0-2: Add file size limit validation - Use OSS_DIRECT_UPLOAD_MAX_MB config (default 800MB) - Check Content-Length header before reading file - Verify actual file size after reading - Return 413 Request Entity Too Large if exceeded Reference: apps/api/app/api/routes/upload.py implementation --- ...ables.py => 012_add_duplication_tables.py} | 8 +-- apps/api/app/api/routes/duplication.py | 54 ++++++++++++++++++- 2 files changed, 56 insertions(+), 6 deletions(-) rename alembic/versions/{011_add_duplication_tables.py => 012_add_duplication_tables.py} (97%) diff --git a/alembic/versions/011_add_duplication_tables.py b/alembic/versions/012_add_duplication_tables.py similarity index 97% rename from alembic/versions/011_add_duplication_tables.py rename to alembic/versions/012_add_duplication_tables.py index cea50acbf..8940b0e7c 100644 --- a/alembic/versions/011_add_duplication_tables.py +++ b/alembic/versions/012_add_duplication_tables.py @@ -1,7 +1,7 @@ """Phase 2 - 查重功能：duplication_records + duplication_segments -Revision ID: 011 -Revises: 010 +Revision ID: 012 +Revises: 011 Create Date: 2026-06-28 This migration creates two new tables: @@ -12,8 +12,8 @@ from alembic import op import sqlalchemy as sa # revision identifiers -revision = "011" -down_revision = "010" +revision = "012" +down_revision = "011" branch_labels = None depends_on = None diff --git a/apps/api/app/api/routes/duplication.py b/apps/api/app/api/routes/duplication.py index 2c5c4fcec..c9d8b832c 100644 --- a/apps/api/app/api/routes/duplication.py +++ b/apps/api/app/api/routes/duplication.py @@ -30,6 +30,32 @@ 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( @@ -86,6 +112,21 @@ async def upload_for_duplication( 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("\\", "_") @@ -94,6 +135,15 @@ async def upload_for_duplication( 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, @@ -104,7 +154,7 @@ async def upload_for_duplication( storage_service.upload_file( content, storage_key, - content_type=file.content_type or "video/mp4", + content_type=validated_content_type, ) except Exception as exc: raise HTTPException( @@ -176,7 +226,7 @@ def delete_duplication_record( 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: -- 2.54.0