diff --git a/alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py b/alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py
new file mode 100644
index 000000000..3af61f178
--- /dev/null
+++ b/alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py
@@ -0,0 +1,33 @@
+"""Add file_hash to assets and ingest_jobs
+
+Revision ID: 031
+Revises: 030
+Create Date: 2026-07-07
+
+为素材去重检测功能添加 file_hash 字段。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "031"
+down_revision = "030"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("assets", sa.Column("file_hash", sa.String(64), nullable=True))
+    op.create_index(op.f("ix_assets_file_hash"), "assets", ["file_hash"])
+
+    op.add_column("ingest_jobs", sa.Column("file_hash", sa.String(64), nullable=True))
+    op.create_index(op.f("ix_ingest_jobs_file_hash"), "ingest_jobs", ["file_hash"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_ingest_jobs_file_hash"), table_name="ingest_jobs")
+    op.drop_column("ingest_jobs", "file_hash")
+
+    op.drop_index(op.f("ix_assets_file_hash"), table_name="assets")
+    op.drop_column("assets", "file_hash")
diff --git a/alembic/versions/032_add_asset_select_mode_to_generation_tasks.py b/alembic/versions/032_add_asset_select_mode_to_generation_tasks.py
new file mode 100644
index 000000000..842863817
--- /dev/null
+++ b/alembic/versions/032_add_asset_select_mode_to_generation_tasks.py
@@ -0,0 +1,28 @@
+"""Add asset_select_mode to generation_tasks
+
+Revision ID: 032
+Revises: 031
+Create Date: 2026-07-07
+
+素材库自动匹配功能：为 generation_tasks 表添加 asset_select_mode 字段。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "032"
+down_revision = "031"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("asset_select_mode", sa.String(20), nullable=False, server_default=""),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generation_tasks", "asset_select_mode")
diff --git a/alembic/versions/033_add_batch_id_to_generation_tasks.py b/alembic/versions/033_add_batch_id_to_generation_tasks.py
new file mode 100644
index 000000000..843dfab63
--- /dev/null
+++ b/alembic/versions/033_add_batch_id_to_generation_tasks.py
@@ -0,0 +1,31 @@
+"""Add batch_id to generation_tasks
+
+Revision ID: 033
+Revises: 032
+Create Date: 2026-07-07
+
+视频查重功能：为 generation_tasks 表添加 batch_id 字段，
+用于关联同一次批量生成请求中的多个任务。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "033"
+down_revision = "032"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("batch_id", sa.String(32), nullable=False, server_default=""),
+    )
+    op.create_index(op.f("ix_generation_tasks_batch_id"), "generation_tasks", ["batch_id"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_generation_tasks_batch_id"), table_name="generation_tasks")
+    op.drop_column("generation_tasks", "batch_id")
diff --git a/apps/api/app/api/routes/chunked_upload.py b/apps/api/app/api/routes/chunked_upload.py
index 058b1a884..cf8aaae4c 100644
--- a/apps/api/app/api/routes/chunked_upload.py
+++ b/apps/api/app/api/routes/chunked_upload.py
@@ -19,6 +19,7 @@ from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import (
     get_asset_library_repository,
+    get_asset_repository,
     get_ingest_job_repository,
     get_project_repository,
 )
@@ -360,6 +361,7 @@ async def complete_chunked_upload(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> ChunkedUploadCompleteResponse:
@@ -423,6 +425,29 @@ async def complete_chunked_upload(
             content_type=meta["content_type"],
         )
 
+        # ── 素材去重检测：同素材库 + 同 file_hash 视为重复 ──
+        if request.file_hash:
+            existing = asset_repository.find_by_library_and_file_hash(
+                library_id=request.library_id,
+                file_hash=request.file_hash,
+            )
+            if existing is not None:
+                logger.info(
+                    "素材去重命中(chunked): library=%s hash=%s existing_asset=%s",
+                    request.library_id,
+                    request.file_hash,
+                    existing.id,
+                )
+                meta["status"] = "completed"
+                _save_upload_meta(upload_id, meta)
+                return ChunkedUploadCompleteResponse(
+                    storage_key=storage_key,
+                    ingest_job_id="",
+                    url=file_url,
+                    duplicated=True,
+                    asset_id=existing.id,
+                )
+
         # Create ingest job
         use_case = SubmitIngestJobUseCase(ingest_job_repository)
         job = use_case.execute(
@@ -430,6 +455,7 @@ async def complete_chunked_upload(
                 project_id=meta["project_id"],
                 library_id=meta["library_id"],
                 storage_key=storage_key,
+                file_hash=request.file_hash,
             )
         )
         celery_app.send_task("worker.ingest_asset", args=[job.id])
diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py
index cfd47edbd..b4ece7f3e 100644
--- a/apps/api/app/api/routes/generation_tasks.py
+++ b/apps/api/app/api/routes/generation_tasks.py
@@ -1,3 +1,5 @@
+import random
+import uuid
 from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -14,6 +16,7 @@ from app.schemas.generated_video import (
     ListGeneratedVideosResponse,
 )
 from app.schemas.generation_task import (
+    BatchGenerationTaskResponse,
     CreateGenerationTaskRequest,
     GenerationTaskResponse,
     ListGenerationTasksResponse,
@@ -51,6 +54,8 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         title_ids=task.title_ids,
         voice_ids=task.voice_ids,
         source_edit_plan_id=task.source_edit_plan_id or "",
+        asset_select_mode=getattr(task, "asset_select_mode", ""),
+        batch_id=getattr(task, "batch_id", ""),
         status=task.status,
         progress=task.progress,
         result_count=task.result_count,
@@ -85,6 +90,49 @@ def _ensure_library_has_ready_video_assets(assets) -> None:
         )
 
 
+def _select_assets_from_library(
+    assets: list,
+    mode: str,
+    count: int,
+) -> list[str]:
+    """根据选取模式从素材库中选取 ready 状态的视频素材 ID。
+
+    Args:
+        assets: 素材库中所有素材（Asset 实体列表）
+        mode: 选取模式 — all=全部, random=随机, smart=按质量评分
+        count: 选取数量，0 表示全部（仅 random/smart 模式有效）
+
+    Returns:
+        选中的素材 ID 列表
+    """
+    ready_video_assets = [a for a in assets if a.status.value == "ready" and a.mime_type.startswith("video")]
+
+    if not ready_video_assets:
+        return []
+
+    if mode == "random":
+        selected = (
+            ready_video_assets if count <= 0 else random.sample(ready_video_assets, min(count, len(ready_video_assets)))
+        )
+        return [a.id for a in selected]
+
+    if mode == "smart":
+        # 按质量分降序排列（质量分高的优先），质量分相同时按时长降序
+        sorted_assets = sorted(
+            ready_video_assets,
+            key=lambda a: (
+                a.quality_score if a.quality_score is not None else 0.0,
+                a.duration if a.duration is not None else 0.0,
+            ),
+            reverse=True,
+        )
+        selected = sorted_assets if count <= 0 else sorted_assets[:count]
+        return [a.id for a in selected]
+
+    # 默认 all 模式：返回全部 ready 视频素材
+    return [a.id for a in ready_video_assets]
+
+
 def _resolve_project_and_library(
     request: CreateGenerationTaskRequest,
     project_repository: Any,
@@ -122,7 +170,7 @@ def _resolve_project_and_library(
     return project_id, asset_library_id
 
 
-@router.post("/tasks", response_model=GenerationTaskResponse)
+@router.post("/tasks", response_model=BatchGenerationTaskResponse)
 def create_generation_task(
     request: CreateGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -130,12 +178,13 @@ def create_generation_task(
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
     asset_repository: Any = Depends(get_asset_repository),
-) -> GenerationTaskResponse:
+) -> BatchGenerationTaskResponse:
     project_id, asset_library_id = _resolve_project_and_library(
         request, project_repository, asset_library_repository, asset_repository, authenticated_user
     )
 
     # asset_library 存在性校验（仅在提供了 asset_library_id 时）
+    resolved_asset_ids: list[str] = list(request.asset_ids)
     if asset_library_id:
         library = asset_library_repository.get(asset_library_id)
         if library is None or (project_id and library.project_id != project_id):
@@ -144,23 +193,42 @@ def create_generation_task(
         assets = asset_repository.find_by_library(asset_library_id)
         _ensure_library_has_ready_video_assets(assets)
 
+        # 素材库自动匹配：当未显式指定 asset_ids 时，按模式自动选取
+        if not resolved_asset_ids:
+            resolved_asset_ids = _select_assets_from_library(
+                assets,
+                mode=request.asset_select_mode,
+                count=request.asset_select_count,
+            )
+
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
-    task = use_case.execute(
-        CreateGenerationTaskCommand(
-            project_id=project_id,
-            asset_library_id=asset_library_id,
-            strategy_id=request.strategy_id,
-            voice_library_id=request.voice_library_id,
-            template_id=request.template_id,
-            asset_ids=request.asset_ids,
-            title_ids=request.title_ids,
-            voice_ids=request.voice_ids,
-            created_by_user_id=authenticated_user.user.id,
-            source_edit_plan_id=request.source_edit_plan_id,
+    count = request.count
+    created_tasks = []
+    # 同批次任务共享 batch_id，用于视频查重时批次内比对
+    batch_id = uuid.uuid4().hex if count > 1 else ""
+
+    for _ in range(count):
+        task = use_case.execute(
+            CreateGenerationTaskCommand(
+                project_id=project_id,
+                asset_library_id=asset_library_id,
+                strategy_id=request.strategy_id,
+                voice_library_id=request.voice_library_id,
+                template_id=request.template_id,
+                asset_ids=resolved_asset_ids,
+                title_ids=request.title_ids,
+                voice_ids=request.voice_ids,
+                created_by_user_id=authenticated_user.user.id,
+                source_edit_plan_id=request.source_edit_plan_id,
+                asset_select_mode=request.asset_select_mode,
+                batch_id=batch_id,
+            )
         )
-    )
-    celery_app.send_task("worker.generate_video", args=[task.id])
-    return _to_generation_task_response(task)
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        created_tasks.append(task)
+
+    items = [_to_generation_task_response(t) for t in created_tasks]
+    return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
 @router.get("/tasks", response_model=ListGenerationTasksResponse)
@@ -237,6 +305,7 @@ def retry_generation_task(
             voice_ids=task.voice_ids,
             created_by_user_id=authenticated_user.user.id,
             source_edit_plan_id=task.source_edit_plan_id or "",
+            asset_select_mode=getattr(task, "asset_select_mode", ""),
         )
     )
     celery_app.send_task("worker.generate_video", args=[retried.id])
diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py
index 304afefe9..7679630c3 100644
--- a/apps/api/app/api/routes/upload.py
+++ b/apps/api/app/api/routes/upload.py
@@ -8,6 +8,7 @@ from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import (
     get_asset_library_repository,
+    get_asset_repository,
     get_ingest_job_repository,
     get_project_repository,
 )
@@ -99,6 +100,7 @@ def _submit_ingest_job(
     library_id: str,
     storage_key: str,
     ingest_job_repository: Any,
+    file_hash: str = "",
 ) -> Any:
     use_case = SubmitIngestJobUseCase(ingest_job_repository)
     job = use_case.execute(
@@ -106,6 +108,7 @@ def _submit_ingest_job(
             project_id=project_id,
             library_id=library_id,
             storage_key=storage_key,
+            file_hash=file_hash,
         )
     )
     celery_app.send_task("worker.ingest_asset", args=[job.id])
@@ -176,6 +179,7 @@ async def complete_direct_upload(
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> DirectUploadCompleteResponse:
     """确认浏览器直传完成并创建导入任务。"""
@@ -199,11 +203,32 @@ async def complete_direct_upload(
     if not file_exists:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Uploaded file not found")
 
+    # ── 素材去重检测：同素材库 + 同 file_hash 视为重复 ──
+    if request.file_hash:
+        existing = asset_repository.find_by_library_and_file_hash(
+            library_id=request.library_id,
+            file_hash=request.file_hash,
+        )
+        if existing is not None:
+            logger.info(
+                "素材去重命中: library=%s hash=%s existing_asset=%s",
+                request.library_id,
+                request.file_hash,
+                existing.id,
+            )
+            return DirectUploadCompleteResponse(
+                storage_key=normalized_key,
+                ingest_job_id="",
+                duplicated=True,
+                asset_id=existing.id,
+            )
+
     job = _submit_ingest_job(
         project_id=request.project_id,
         library_id=request.library_id,
         storage_key=normalized_key,
         ingest_job_repository=ingest_job_repository,
+        file_hash=request.file_hash,
     )
     return DirectUploadCompleteResponse(storage_key=normalized_key, ingest_job_id=job.id)
 
@@ -218,15 +243,38 @@ async def upload_asset(
     project_id: str = Form(..., min_length=1, description="项目 ID"),
     library_id: str = Form(..., min_length=1, description="素材库 ID"),
     file: UploadFile = File(..., description="要上传的文件（视频、音频、图片等）"),
+    file_hash: str = Form(default="", description="文件 MD5 哈希，用于去重检测"),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> UploadAssetResponse:
     """上传素材文件并触发导入流水线。"""
     _require_project_and_library(project_id, library_id, project_repository, asset_library_repository)
 
+    # ── 素材去重检测：上传前检查同素材库 + 同 file_hash ──
+    if file_hash:
+        existing = asset_repository.find_by_library_and_file_hash(
+            library_id=library_id,
+            file_hash=file_hash,
+        )
+        if existing is not None:
+            logger.info(
+                "素材去重命中(multipart): library=%s hash=%s existing_asset=%s",
+                library_id,
+                file_hash,
+                existing.id,
+            )
+            return UploadAssetResponse(
+                storage_key=existing.storage_key,
+                ingest_job_id="",
+                url="",
+                duplicated=True,
+                asset_id=existing.id,
+            )
+
     # P2-5: 服务端验证 MIME 类型
     validated_content_type = _validate_mime_type(file.content_type)
 
@@ -255,6 +303,7 @@ async def upload_asset(
         library_id=library_id,
         storage_key=storage_key,
         ingest_job_repository=ingest_job_repository,
+        file_hash=file_hash,
     )
 
     return UploadAssetResponse(
diff --git a/apps/api/app/schemas/chunked_upload.py b/apps/api/app/schemas/chunked_upload.py
index 8c5cbef78..884b1f481 100644
--- a/apps/api/app/schemas/chunked_upload.py
+++ b/apps/api/app/schemas/chunked_upload.py
@@ -36,9 +36,12 @@ class ChunkedUploadStatusResponse(BaseModel):
 class ChunkedUploadCompleteRequest(BaseModel):
     project_id: str = Field(..., min_length=1, description="Project ID")
     library_id: str = Field(..., min_length=1, description="Asset library ID")
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class ChunkedUploadCompleteResponse(BaseModel):
     storage_key: str = Field(..., description="Storage key")
     ingest_job_id: str = Field(..., description="Ingest job ID")
     url: str = Field(..., description="File URL")
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py
index 8e1fc7a74..e75479954 100644
--- a/apps/api/app/schemas/generation_task.py
+++ b/apps/api/app/schemas/generation_task.py
@@ -21,6 +21,16 @@ class CreateGenerationTaskRequest(BaseModel):
     voice_ids: list[str] = Field(default_factory=list)
     # ── 来源剪辑计划 ──
     source_edit_plan_id: str = ""
+    # ── 批量生成 ──
+    count: int = Field(default=1, ge=1, le=50, description="批量生成数量，默认1，最大50")
+    # ── 素材库自动匹配 ──
+    asset_select_mode: str = Field(
+        default="all",
+        description="素材选取模式：all=全部ready视频, random=随机选取, smart=智能匹配（按质量/时长评分）",
+    )
+    asset_select_count: int = Field(
+        default=0, ge=0, le=100, description="选取数量，0表示全部（仅 random/smart 模式有效）"
+    )
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -46,12 +56,21 @@ class GenerationTaskResponse(BaseModel):
     title_ids: list[str] = Field(default_factory=list)
     voice_ids: list[str] = Field(default_factory=list)
     source_edit_plan_id: str = ""
+    asset_select_mode: str = ""
+    batch_id: str = ""
     status: str
     progress: float
     result_count: int
     error_message: str
 
 
+class BatchGenerationTaskResponse(BaseModel):
+    """批量生成任务响应。"""
+
+    items: list[GenerationTaskResponse]
+    total: int
+
+
 class ListGenerationTasksResponse(BaseModel):
     """用户级生成任务列表响应（跨 project）。"""
 
diff --git a/apps/api/app/schemas/upload.py b/apps/api/app/schemas/upload.py
index 90a7a026d..b9763606c 100644
--- a/apps/api/app/schemas/upload.py
+++ b/apps/api/app/schemas/upload.py
@@ -6,12 +6,7 @@ class UploadAssetRequest(BaseModel):
 
     project_id: str = Field(..., min_length=1, description="项目 ID")
     library_id: str = Field(..., min_length=1, description="素材库 ID")
-
-
-class UploadAssetResponse(BaseModel):
-    storage_key: str
-    ingest_job_id: str
-    url: str = Field(..., description="Public URL of uploaded file")
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadPrepareRequest(BaseModel):
@@ -20,6 +15,7 @@ class DirectUploadPrepareRequest(BaseModel):
     filename: str = Field(..., min_length=1, max_length=255)
     content_type: str = Field(default="application/octet-stream", min_length=1, max_length=100)
     file_size: int = Field(..., gt=0)
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadPrepareResponse(BaseModel):
@@ -35,8 +31,19 @@ class DirectUploadCompleteRequest(BaseModel):
     project_id: str = Field(..., min_length=1)
     library_id: str = Field(..., min_length=1)
     storage_key: str = Field(..., min_length=1, max_length=255)
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadCompleteResponse(BaseModel):
     storage_key: str
     ingest_job_id: str
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
+
+
+class UploadAssetResponse(BaseModel):
+    storage_key: str
+    ingest_job_id: str
+    url: str = Field(..., description="Public URL of uploaded file")
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
diff --git a/apps/web/src/pages/assets/AssetLibrary.tsx b/apps/web/src/pages/assets/AssetLibrary.tsx
index e7129fff7..80aef0597 100644
--- a/apps/web/src/pages/assets/AssetLibrary.tsx
+++ b/apps/web/src/pages/assets/AssetLibrary.tsx
@@ -110,14 +110,16 @@ const mapAsset = (item: ApiAssetItem): AssetItem => {
     item.classification_status ?? undefined,
   );
   const metadata = item.metadata || {};
+  const kind = inferKind(item.mime_type || "");
   return {
     id: item.id,
     name: item.name,
-    kind: inferKind(item.mime_type || ""),
+    kind,
+    // 视频类型不能用 file_url 做缩略图（是视频文件，<img> 无法渲染）
     thumbUrl:
       (item.thumbnail_url as string | undefined) ||
-      (item.file_url as string | undefined) ||
-      (metadata.thumbnail_url as string | undefined),
+      (metadata.thumbnail_url as string | undefined) ||
+      (kind !== "video" ? (item.file_url as string | undefined) : undefined),
     fileUrl:
       (item.file_url as string | undefined) ||
       (metadata.file_url as string | undefined),
@@ -361,6 +363,7 @@ const AssetLibrary: React.FC = () => {
 
   /* 上传 */
   const [uploading, setUploading] = useState(false);
+  const [uploadProgress, setUploadProgress] = useState(0);
 
   /* 新建素材库 */
   const [createModalOpen, setCreateModalOpen] = useState(false);
@@ -434,11 +437,16 @@ const AssetLibrary: React.FC = () => {
     }
 
     setUploading(true);
+    setUploadProgress(0);
     try {
       if (file.size > LARGE_FILE_THRESHOLD) {
         message.info(`大文件 "${file.name}" 将使用直传上传`);
       }
-      await uploadAssetDirect({ file, library_id: effectiveLibId });
+      await uploadAssetDirect({
+        file,
+        library_id: effectiveLibId,
+        onProgress: (pct) => setUploadProgress(pct),
+      });
       message.success(`"${file.name}" 上传成功`);
       queryClient.invalidateQueries({ queryKey: ["assets"] });
       queryClient.invalidateQueries({ queryKey: ["asset-libraries"] });
@@ -446,6 +454,7 @@ const AssetLibrary: React.FC = () => {
       message.error(`"${file.name}" 上传失败`);
     } finally {
       setUploading(false);
+      setUploadProgress(0);
     }
     return false;
   };
@@ -532,6 +541,54 @@ const AssetLibrary: React.FC = () => {
 
   return (
     <div className="xx-assets-page">
+      {/* ─── 上传进度弹窗（圆形动画 + 百分比） ─── */}
+      <AntModal
+        open={uploading}
+        footer={null}
+        closable={false}
+        centered
+        width={260}
+        maskClosable={false}
+        className="xx-upload-progress-modal"
+      >
+        <div className="xx-upload-progress-body">
+          <svg
+            className="xx-upload-progress-ring"
+            viewBox="0 0 120 120"
+            width={120}
+            height={120}
+          >
+            {/* 背景圆环 */}
+            <circle
+              cx="60"
+              cy="60"
+              r="52"
+              fill="none"
+              stroke="var(--border-primary, #e5e7eb)"
+              strokeWidth="8"
+            />
+            {/* 进度圆弧 */}
+            <circle
+              cx="60"
+              cy="60"
+              r="52"
+              fill="none"
+              stroke="var(--primary-color, #6366f1)"
+              strokeWidth="8"
+              strokeLinecap="round"
+              strokeDasharray={`${2 * Math.PI * 52}`}
+              strokeDashoffset={`${2 * Math.PI * 52 * (1 - uploadProgress / 100)}`}
+              transform="rotate(-90 60 60)"
+              style={{ transition: "stroke-dashoffset 0.3s ease" }}
+            />
+          </svg>
+          <div className="xx-upload-progress-text">
+            <span className="xx-upload-progress-pct">{uploadProgress}%</span>
+            <span className="xx-upload-progress-label">上传中…</span>
+          </div>
+        </div>
+      </AntModal>
+
       {/* 两栏布局 */}
       <div className="xx-assets-layout">
         {/* ─── 左侧：素材库列表 ─── */}
diff --git a/apps/web/src/pages/assets/assets.css b/apps/web/src/pages/assets/assets.css
index 3333f1306..63e46e653 100644
--- a/apps/web/src/pages/assets/assets.css
+++ b/apps/web/src/pages/assets/assets.css
@@ -587,3 +587,40 @@
     grid-template-columns: 1fr;
   }
 }
+
+/* ─── 上传进度弹窗 ─── */
+.xx-upload-progress-modal .ant-modal-content {
+  padding: 24px 16px 20px;
+  border-radius: 16px;
+}
+
+.xx-upload-progress-body {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  gap: 16px;
+  padding: 8px 0;
+}
+
+.xx-upload-progress-ring {
+  display: block;
+}
+
+.xx-upload-progress-text {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  gap: 4px;
+}
+
+.xx-upload-progress-pct {
+  font-size: 22px;
+  font-weight: 700;
+  color: var(--primary-color, #6366f1);
+  line-height: 1;
+}
+
+.xx-upload-progress-label {
+  font-size: 13px;
+  color: var(--text-secondary, #6b7280);
+}
diff --git a/apps/web/src/pages/generate/GeneratePage.tsx b/apps/web/src/pages/generate/GeneratePage.tsx
index 6e9dc0342..1c6ad79ea 100644
--- a/apps/web/src/pages/generate/GeneratePage.tsx
+++ b/apps/web/src/pages/generate/GeneratePage.tsx
@@ -17,13 +17,13 @@ import {
   ShareAltOutlined,
   SaveOutlined,
   PlusOutlined,
+  MinusOutlined,
   CloseOutlined,
 } from "@ant-design/icons";
 import type { AssetItem } from "@/api/assets";
 import { getAssets, getAssetLibraries } from "@/api/assets";
 import { createEditPlan, generateEditPlan } from "@/api/editPlans";
 import { getEditingTemplates } from "@/api/editingPlanner";
-import { MODE_LABELS, type TemplateMode } from "@/api/editingPlanner";
 import { getTitles } from "@/api/titles";
 import apiClient from "@/api/client";
 import { fetchPresetVoices } from "@/api/voices";
@@ -54,13 +54,6 @@ const MODE_GRADIENTS: Record<string, string> = {
   voice_over: "linear-gradient(135deg, #6366f1, #4f46e5)",
   voice_pip: "linear-gradient(135deg, #10b981, #059669)",
 };
-const MODE_ABBRS: Record<string, string> = {
-  pip: "PIP",
-  one_take: "ONE",
-  voice_over: "VOI",
-  voice_pip: "VP",
-};
-
 /* ── 配音预设卡片：从 API 动态生成，不再硬编码 ── */
 const VOICE_GENDER_ICON: Record<string, string> = {
   female: "🎀",
@@ -122,6 +115,8 @@ const GeneratePage: React.FC = () => {
 
   /* ── 素材 ── */
   const [selectedMaterials, setSelectedMaterials] = useState<string[]>([]);
+  /* 素材选择模式：手动选择 / 自动匹配 */
+  const [materialMode, setMaterialMode] = useState<"manual" | "auto">("manual");
 
   /* ── 标题 ── */
   const [title, setTitle] = useState("");
@@ -146,6 +141,9 @@ const GeneratePage: React.FC = () => {
   );
   const [customVoiceText, setCustomVoiceText] = useState("");
 
+  /* ── 生成数量 ── */
+  const [generateCount, setGenerateCount] = useState(1);
+
   /* ── 克隆声音 ── */
   const [selectedClonedVoice, setSelectedClonedVoice] = useState<string>("");
   const [cloneModalOpen, setCloneModalOpen] = useState(false);
@@ -456,11 +454,17 @@ const GeneratePage: React.FC = () => {
   }, [navigate]);
 
   const handleGenerate = useCallback(async () => {
+    console.log("[handleGenerate] 开始生成, 参数:", {
+      title,
+      selectedTemplate,
+      selectedMaterials,
+      voiceMode,
+    });
     if (!title.trim()) {
       message.warning("请先选择或输入标题");
       return;
     }
-    if (selectedMaterials.length === 0) {
+    if (materialMode === "manual" && selectedMaterials.length === 0) {
       message.warning("请至少选择一个素材");
       return;
     }
@@ -498,6 +502,8 @@ const GeneratePage: React.FC = () => {
           duration,
           auto_subtitles: autoSubtitles,
           bgm,
+          generate_count: generateCount,
+          material_mode: materialMode,
         },
         total_duration: duration,
         source_edit_plan_id: editPlanId || undefined,
@@ -560,20 +566,34 @@ const GeneratePage: React.FC = () => {
         typeof setInterval
       >;
     } catch (err: unknown) {
-      console.error("生成失败:", err);
+      console.error("[handleGenerate] 生成失败:", err);
       setGenerating(false);
       // 提取 axios 响应中的后端错误信息
       const axiosErr = err as {
         response?: {
-          data?: { message?: string; error?: string; detail?: string };
+          data?: {
+            message?: string;
+            error?: string;
+            detail?: string;
+            msg?: string;
+          };
         };
+        message?: string;
       };
       const backendMsg =
         axiosErr.response?.data?.message ||
         axiosErr.response?.data?.error ||
         axiosErr.response?.data?.detail ||
+        axiosErr.response?.data?.msg ||
+        axiosErr.message ||
         "";
-      message.error(backendMsg || "生成失败，请重试");
+      console.error(
+        "[handleGenerate] 错误信息:",
+        backendMsg,
+        "完整错误:",
+        axiosErr,
+      );
+      message.error(backendMsg || "生成失败，请检查网络后重试或联系管理员");
     }
   }, [
     title,
@@ -590,6 +610,8 @@ const GeneratePage: React.FC = () => {
     bgm,
     editPlanId,
     selectedTemplate,
+    generateCount,
+    materialMode,
   ]);
 
   /* ── 步骤导航 ── */
@@ -598,7 +620,11 @@ const GeneratePage: React.FC = () => {
       message.warning("请先选择一个模板");
       return;
     }
-    if (currentStep === 2 && selectedMaterials.length === 0) {
+    if (
+      currentStep === 2 &&
+      materialMode === "manual" &&
+      selectedMaterials.length === 0
+    ) {
       message.warning("请至少选择一个素材");
       return;
     }
@@ -609,7 +635,13 @@ const GeneratePage: React.FC = () => {
     if (currentStep < 5) {
       setCurrentStep((s) => s + 1);
     }
-  }, [currentStep, selectedTemplate, selectedMaterials.length, title]);
+  }, [
+    currentStep,
+    selectedTemplate,
+    selectedMaterials.length,
+    title,
+    materialMode,
+  ]);
 
   const goPrev = useCallback(() => {
     if (currentStep > 1) {
@@ -669,11 +701,10 @@ const GeneratePage: React.FC = () => {
                   background: MODE_GRADIENTS[tpl.mode] || MODE_GRADIENTS.pip,
                 }}
               >
-                {MODE_ABBRS[tpl.mode] || "TPL"}
+                🎬
               </div>
               <h4>{tpl.name}</h4>
               <p>
-                {MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode} ·{" "}
                 {tpl.estimated_duration}s · {tpl.segments.length}片段
               </p>
               {tpl.tags.length > 0 && (
@@ -708,11 +739,31 @@ const GeneratePage: React.FC = () => {
     </div>
   );
 
-  /** 步骤 2：选择素材 */
+  /** 步骤 2：选择素材（双模式：手动选择 / 自动匹配） */
   const renderStep2 = () => (
     <div className="xx-form-section">
       <h3>📦 选择素材</h3>
-      <div className="xx-form-field">
+
+      {/* ── 模式切换 Tab ── */}
+      <div className="xx-material-mode-tabs">
+        <button
+          className={`xx-material-mode-tab ${materialMode === "manual" ? "active" : ""}`}
+          onClick={() => setMaterialMode("manual")}
+          type="button"
+        >
+          手动选择素材
+        </button>
+        <button
+          className={`xx-material-mode-tab ${materialMode === "auto" ? "active" : ""}`}
+          onClick={() => setMaterialMode("auto")}
+          type="button"
+        >
+          选择素材库自动匹配
+        </button>
+      </div>
+
+      {/* ── 素材库选择（两种模式共用） ── */}
+      <div className="xx-form-field" style={{ marginTop: 12 }}>
         <label>选择素材库</label>
         <select
           value={selectedLibraryId}
@@ -725,90 +776,138 @@ const GeneratePage: React.FC = () => {
           ))}
         </select>
       </div>
-      <div
-        style={{
-          marginTop: 14,
-          display: "flex",
-          alignItems: "center",
-          gap: 10,
-        }}
-      >
-        <span className="xx-pill xx-pill-ok">
-          已选 {selectedMaterials.length} 个素材
-        </span>
-        <Text style={{ color: "var(--text-tertiary, #94a3b8)", fontSize: 13 }}>
-          系统将自动选择最合适的素材
-        </Text>
-      </div>
 
-      {/* 素材列表 */}
-      <div style={{ marginTop: 14 }}>
-        {materialsLoading ? (
-          <Text style={{ color: "var(--text-secondary)", padding: "16px 0" }}>
-            加载素材中…
-          </Text>
-        ) : materials.length === 0 ? (
-          <Text style={{ color: "var(--text-secondary)", padding: "16px 0" }}>
-            暂无素材，请先在素材库中上传
-          </Text>
-        ) : (
-          <div style={{ display: "flex", flexDirection: "column", gap: 6 }}>
-            {materials.map((m) => {
-              const checked = selectedMaterials.includes(m.id);
-              return (
-                <label
-                  key={m.id}
-                  style={{
-                    display: "flex",
-                    alignItems: "center",
-                    gap: 10,
-                    padding: "8px 12px",
-                    background: checked
-                      ? "var(--primary-soft, #eef2ff)"
-                      : "#f8fafc",
-                    borderRadius: 10,
-                    cursor: "pointer",
-                    border: checked
-                      ? "1px solid var(--primary-color, #4f46e5)"
-                      : "1px solid transparent",
-                    transition: "all 0.15s ease",
-                  }}
-                >
-                  <input
-                    type="checkbox"
-                    checked={checked}
-                    onChange={() => {
-                      setSelectedMaterials((prev) =>
-                        prev.includes(m.id)
-                          ? prev.filter((id) => id !== m.id)
-                          : [...prev, m.id],
-                      );
-                    }}
-                    style={{ accentColor: "var(--primary-color, #4f46e5)" }}
-                  />
-                  <span
-                    style={{
-                      fontSize: 13,
-                      color: "var(--text-primary)",
-                      flex: 1,
-                    }}
-                  >
-                    {m.name}
-                  </span>
-                  <span
-                    style={{
-                      fontSize: 11,
-                      color: "var(--text-tertiary, #94a3b8)",
-                    }}
-                  >
-                    {m.mime_type.split("/")[1].toUpperCase()}
-                  </span>
-                </label>
-              );
-            })}
+      {/* ── 手动选择模式 ── */}
+      {materialMode === "manual" && (
+        <>
+          <div
+            style={{
+              marginTop: 14,
+              display: "flex",
+              alignItems: "center",
+              gap: 10,
+            }}
+          >
+            <span className="xx-pill xx-pill-ok">
+              已选 {selectedMaterials.length} 个素材
+            </span>
           </div>
-        )}
-      </div>
+
+          {/* 素材列表 */}
+          <div style={{ marginTop: 14 }}>
+            {materialsLoading ? (
+              <Text
+                style={{ color: "var(--text-secondary)", padding: "16px 0" }}
+              >
+                加载素材中…
+              </Text>
+            ) : materials.length === 0 ? (
+              <Text
+                style={{ color: "var(--text-secondary)", padding: "16px 0" }}
+              >
+                暂无素材，请先在素材库中上传
+              </Text>
+            ) : (
+              <div style={{ display: "flex", flexDirection: "column", gap: 6 }}>
+                {materials.map((m) => {
+                  const checked = selectedMaterials.includes(m.id);
+                  return (
+                    <label
+                      key={m.id}
+                      style={{
+                        display: "flex",
+                        alignItems: "center",
+                        gap: 10,
+                        padding: "8px 12px",
+                        background: checked
+                          ? "var(--primary-soft, #eef2ff)"
+                          : "#f8fafc",
+                        borderRadius: 10,
+                        cursor: "pointer",
+                        border: checked
+                          ? "1px solid var(--primary-color, #4f46e5)"
+                          : "1px solid transparent",
+                        transition: "all 0.15s ease",
+                      }}
+                    >
+                      <input
+                        type="checkbox"
+                        checked={checked}
+                        onChange={() => {
+                          setSelectedMaterials((prev) =>
+                            prev.includes(m.id)
+                              ? prev.filter((id) => id !== m.id)
+                              : [...prev, m.id],
+                          );
+                        }}
+                        style={{ accentColor: "var(--primary-color, #4f46e5)" }}
+                      />
+                      <span
+                        style={{
+                          fontSize: 13,
+                          color: "var(--text-primary)",
+                          flex: 1,
+                        }}
+                      >
+                        {m.name}
+                      </span>
+                      <span
+                        style={{
+                          fontSize: 11,
+                          color: "var(--text-tertiary, #94a3b8)",
+                        }}
+                      >
+                        {m.mime_type.split("/")[1].toUpperCase()}
+                      </span>
+                    </label>
+                  );
+                })}
+              </div>
+            )}
+          </div>
+        </>
+      )}
+
+      {/* ── 自动匹配模式 ── */}
+      {materialMode === "auto" && (
+        <div className="xx-auto-match-card">
+          <div className="xx-auto-match-icon">🤖</div>
+          <div className="xx-auto-match-body">
+            <h4 className="xx-auto-match-title">智能素材匹配</h4>
+            <p className="xx-auto-match-desc">
+              系统将根据所选模板和标题，从素材库中自动分析并匹配最合适的素材进行视频生成。
+              无需手动挑选，AI
+              会综合素材质量、时长、内容相关性等维度进行智能筛选。
+            </p>
+            <div className="xx-auto-match-features">
+              <span className="xx-auto-match-feature">📊 质量评分筛选</span>
+              <span className="xx-auto-match-feature">🎯 内容相关性匹配</span>
+              <span className="xx-auto-match-feature">⏱️ 时长智能分配</span>
+            </div>
+          </div>
+          {materialsLoading ? (
+            <Text
+              style={{
+                color: "var(--text-secondary)",
+                fontSize: 12,
+                marginTop: 8,
+              }}
+            >
+              扫描素材库中…
+            </Text>
+          ) : (
+            <Text
+              style={{
+                color: "var(--text-tertiary, #94a3b8)",
+                fontSize: 12,
+                marginTop: 8,
+              }}
+            >
+              当前素材库共 {materials.length} 个素材可供匹配
+            </Text>
+          )}
+        </div>
+      )}
     </div>
   );
 
@@ -1270,7 +1369,9 @@ const GeneratePage: React.FC = () => {
         <div className="xx-summary-row">
           <span className="xx-summary-label">素材</span>
           <span className="xx-summary-value">
-            {selectedMaterials.length} 个素材
+            {materialMode === "auto"
+              ? "自动匹配"
+              : `${selectedMaterials.length} 个素材`}
           </span>
         </div>
         <div className="xx-summary-row">
@@ -1281,6 +1382,29 @@ const GeneratePage: React.FC = () => {
           <span className="xx-summary-label">配音</span>
           <span className="xx-summary-value">{getVoiceName()}</span>
         </div>
+        <div className="xx-summary-row">
+          <span className="xx-summary-label">生成数量</span>
+          <span className="xx-summary-value">
+            <div className="xx-count-stepper">
+              <button
+                className="xx-count-stepper-btn"
+                disabled={generateCount <= 1 || generating}
+                onClick={() => setGenerateCount((c) => Math.max(1, c - 1))}
+              >
+                <MinusOutlined />
+              </button>
+              <span className="xx-count-stepper-value">{generateCount}</span>
+              <button
+                className="xx-count-stepper-btn"
+                disabled={generateCount >= 10 || generating}
+                onClick={() => setGenerateCount((c) => Math.min(10, c + 1))}
+              >
+                <PlusOutlined />
+              </button>
+              <span className="xx-count-stepper-hint">条视频</span>
+            </div>
+          </span>
+        </div>
       </div>
 
       {/* 生成进度 */}
diff --git a/apps/web/src/pages/generate/generate.css b/apps/web/src/pages/generate/generate.css
index 82052f16b..8f915d49e 100644
--- a/apps/web/src/pages/generate/generate.css
+++ b/apps/web/src/pages/generate/generate.css
@@ -1074,3 +1074,141 @@
     opacity: 1;
   }
 }
+
+/* ── 生成数量步进器 ── */
+.xx-count-stepper {
+  display: inline-flex;
+  align-items: center;
+  gap: 6px;
+}
+
+.xx-count-stepper-btn {
+  display: inline-flex;
+  align-items: center;
+  justify-content: center;
+  width: 28px;
+  height: 28px;
+  border: 1px solid var(--border-primary, #e2e8f0);
+  border-radius: 8px;
+  background: var(--bg-surface, #fff);
+  color: var(--text-secondary, #64748b);
+  font-size: 13px;
+  cursor: pointer;
+  transition: all 0.15s;
+}
+
+.xx-count-stepper-btn:hover:not(:disabled) {
+  border-color: var(--primary-400, #818cf8);
+  color: var(--primary-600, #4f46e5);
+  background: var(--primary-50, #eef2ff);
+}
+
+.xx-count-stepper-btn:disabled {
+  opacity: 0.35;
+  cursor: not-allowed;
+}
+
+.xx-count-stepper-value {
+  min-width: 24px;
+  text-align: center;
+  font-size: 16px;
+  font-weight: 600;
+  color: var(--text-primary, #1e293b);
+}
+
+.xx-count-stepper-hint {
+  font-size: 12px;
+  color: var(--text-tertiary, #94a3b8);
+  margin-left: 2px;
+}
+
+/* ── 素材选择模式切换 Tab ── */
+.xx-material-mode-tabs {
+  display: flex;
+  gap: 0;
+  border: 1px solid var(--border-primary, #e2e8f0);
+  border-radius: 10px;
+  overflow: hidden;
+  margin-bottom: 4px;
+}
+
+.xx-material-mode-tab {
+  flex: 1;
+  padding: 10px 16px;
+  font-size: 13px;
+  font-weight: 500;
+  border: none;
+  background: var(--bg-surface, #fff);
+  color: var(--text-secondary, #64748b);
+  cursor: pointer;
+  transition: all 0.2s ease;
+  text-align: center;
+}
+
+.xx-material-mode-tab:first-child {
+  border-right: 1px solid var(--border-primary, #e2e8f0);
+}
+
+.xx-material-mode-tab:hover {
+  background: var(--primary-50, #eef2ff);
+  color: var(--primary-600, #4f46e5);
+}
+
+.xx-material-mode-tab.active {
+  background: var(--primary-500, #6366f1);
+  color: #fff;
+  font-weight: 600;
+}
+
+/* ── 自动匹配卡片 ── */
+.xx-auto-match-card {
+  margin-top: 14px;
+  padding: 20px;
+  background: linear-gradient(135deg, #f0f4ff 0%, #faf5ff 100%);
+  border: 1px solid var(--border-primary, #e2e8f0);
+  border-radius: 14px;
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  text-align: center;
+}
+
+.xx-auto-match-icon {
+  font-size: 36px;
+  margin-bottom: 10px;
+}
+
+.xx-auto-match-body {
+  width: 100%;
+}
+
+.xx-auto-match-title {
+  font-size: 15px;
+  font-weight: 600;
+  color: var(--text-primary, #1e293b);
+  margin: 0 0 8px;
+}
+
+.xx-auto-match-desc {
+  font-size: 13px;
+  color: var(--text-secondary, #64748b);
+  line-height: 1.6;
+  margin: 0 0 14px;
+}
+
+.xx-auto-match-features {
+  display: flex;
+  flex-wrap: wrap;
+  justify-content: center;
+  gap: 8px;
+}
+
+.xx-auto-match-feature {
+  display: inline-block;
+  padding: 4px 12px;
+  font-size: 12px;
+  color: var(--primary-600, #4f46e5);
+  background: rgba(255, 255, 255, 0.8);
+  border: 1px solid var(--border-light, #f1f5f9);
+  border-radius: 20px;
+}
diff --git a/apps/web/src/pages/products/ProductLibrary.tsx b/apps/web/src/pages/products/ProductLibrary.tsx
index d0aa34e58..b992c853f 100644
--- a/apps/web/src/pages/products/ProductLibrary.tsx
+++ b/apps/web/src/pages/products/ProductLibrary.tsx
@@ -737,11 +737,19 @@ const ProductLibrary: React.FC = () => {
 
   // ── Error 状态 ──
   if (isError) {
+    console.error("[ProductLibrary] 加载失败:", error);
+    const errorMsg = error?.message || "加载失败";
+    // 区分 404 和其他错误
+    const is404 = errorMsg.includes("404") || errorMsg.includes("Not Found");
     return (
       <div className="xx-products-page">
         <div className="xx-products-empty">
-          <div className="xx-products-empty-icon">❌</div>
-          <p>{error?.message || "加载失败"}</p>
+          <div className="xx-products-empty-icon">{is404 ? "🔍" : "❌"}</div>
+          <p>
+            {is404
+              ? "成片库功能正在建设中，敬请期待"
+              : errorMsg || "加载失败，请稍后重试"}
+          </p>
           <Button
             buttonType="primary"
             buttonSize="sm"
diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py
index 137585563..9e7a992d3 100644
--- a/apps/worker/video_processing/dedup.py
+++ b/apps/worker/video_processing/dedup.py
@@ -206,6 +206,68 @@ class VideoDeduplicator:
 
         return None
 
+    def check_batch_duplicate(
+        self,
+        fingerprint: VideoFingerprint,
+        batch_id: str,
+        current_video_id: str,
+        session: Session,
+    ) -> Optional[dict]:
+        """检查视频是否与同批次内其他视频重复。
+
+        逻辑与 check_duplicate 一致（MD5 + pHash），但搜索范围限定为同 batch_id 的视频。
+
+        Args:
+            fingerprint: 待检测视频的指纹
+            batch_id: 批次 ID
+            current_video_id: 当前视频 ID（排除自身）
+            session: 数据库会话
+
+        Returns:
+            重复信息字典，或 None 表示未找到重复
+        """
+        video_repo = SQLAlchemyGeneratedVideoRepository(session)
+        batch_videos = video_repo.list_by_batch(batch_id)
+
+        for existing in batch_videos:
+            if existing.id == current_video_id:
+                continue
+            if not existing.video_fingerprint:
+                continue
+
+            ef = existing.video_fingerprint
+
+            if fingerprint.md5 == ef.get("md5"):
+                return {
+                    "duplicate": True,
+                    "duplicate_of": existing.id,
+                    "reason": "batch_exact_md5_match",
+                    "similarity": 1.0,
+                }
+
+            existing_phashes = ef.get("keyframe_phashes", [])
+            if not existing_phashes:
+                continue
+
+            min_distances = []
+            for phash in fingerprint.keyframe_phashes:
+                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
+                min_distances.append(min(distances))
+            avg_distance = sum(min_distances) / len(min_distances) if min_distances else 100
+
+            if avg_distance >= self.PHASH_THRESHOLD:
+                continue
+
+            phash_similarity = 1.0 - (avg_distance / 64)
+            return {
+                "duplicate": True,
+                "duplicate_of": existing.id,
+                "reason": "batch_phash_similar",
+                "similarity": phash_similarity,
+            }
+
+        return None
+
     @staticmethod
     def _average_histogram_similarity(histograms_a: list[list[float]], histograms_b: list[list[float]]) -> float:
         """
diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py
index 373a29913..08ff3119b 100755
--- a/apps/worker/worker_app/tasks/generation.py
+++ b/apps/worker/worker_app/tasks/generation.py
@@ -139,14 +139,16 @@ def _download_library_assets(
     asset_library_id: str,
     temp_path: Path,
     video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
+    asset_ids: list[str] | None = None,
 ) -> list[str]:
     """
-    从素材库下载所有视频素材
+    从素材库下载视频素材
 
     Args:
         asset_library_id: 素材库 ID
         temp_path: 临时目录路径
         video_extensions: 支持的视频扩展名
+        asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材
 
     Returns:
         下载成功的视频文件路径列表
@@ -161,16 +163,15 @@ def _download_library_assets(
 
         try:
             # 查询素材库中的视频素材
-            assets = (
-                session.query(AssetModel)
-                .filter(
-                    AssetModel.asset_library_id == asset_library_id,
-                    AssetModel.status == "ready",
-                    AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
-                )
-                .order_by(AssetModel.created_at)
-                .all()
+            query = session.query(AssetModel).filter(
+                AssetModel.asset_library_id == asset_library_id,
+                AssetModel.status == "ready",
+                AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
             )
+            # 如果指定了 asset_ids，则只下载这些素材
+            if asset_ids:
+                query = query.filter(AssetModel.id.in_(asset_ids))
+            assets = query.order_by(AssetModel.created_at).all()
 
             if not assets:
                 logger.info(f"No video assets found in library {asset_library_id}")
@@ -259,6 +260,8 @@ def generate_video(self, task_id: str) -> dict:
         asset_library_id = gen_task.asset_library_id
         voice_library_id = gen_task.voice_library_id or ""
         mode = gen_task.strategy_id or "one_take"
+        task_asset_ids = list(gen_task.asset_ids or [])
+        batch_id = getattr(gen_task, "batch_id", "") or ""
     finally:
         session.close()
 
@@ -275,8 +278,8 @@ def generate_video(self, task_id: str) -> dict:
             temp_path = Path(temp_dir)
             output_path = temp_path / output_name
 
-            # 从素材库下载视频素材
-            downloaded_videos = _download_library_assets(asset_library_id, temp_path)
+            # 从素材库下载视频素材（如果任务指定了 asset_ids 则只下载这些）
+            downloaded_videos = _download_library_assets(asset_library_id, temp_path, asset_ids=task_asset_ids or None)
 
             audio_path = None
             if voice_library_id:
@@ -297,6 +300,32 @@ def generate_video(self, task_id: str) -> dict:
             file_size = output_path.stat().st_size
             duration = _probe_duration(output_path)
 
+            # 上传到 OSS
+            bucket = _oss_bucket()
+            if bucket:
+                try:
+                    bucket.put_object_from_file(storage_key, str(output_path))
+                except Exception as oss_err:
+                    logger.warning(f"OSS upload failed: {oss_err}")
+
+            # 构建视频 URL
+            if bucket:
+                file_url = f"{PUBLIC_API_BASE_URL}/{storage_key}"
+            else:
+                file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}"
+
+            # 创建 GeneratedVideo 记录 + 查重
+            _create_video_record_and_dedup(
+                task_id=task_id,
+                project_id=project_id,
+                batch_id=batch_id,
+                file_url=file_url,
+                file_size=file_size,
+                duration=duration,
+                video_path=str(output_path),
+                mode=editing_mode.value,
+            )
+
             return {
                 "status": "completed",
                 "task_id": task_id,
@@ -314,3 +343,85 @@ def generate_video(self, task_id: str) -> dict:
             "task_id": task_id,
             "error": str(error),
         }
+
+
+def _create_video_record_and_dedup(
+    *,
+    task_id: str,
+    project_id: str,
+    batch_id: str,
+    file_url: str,
+    file_size: int,
+    duration: float,
+    video_path: str,
+    mode: str,
+) -> None:
+    """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。"""
+    from uuid import uuid4
+
+    from video_processing.dedup import VideoDeduplicator
+    from worker_app.db import SessionLocal
+
+    from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+        SQLAlchemyGeneratedVideoRepository,
+    )
+    from packages.domain import GeneratedVideo
+
+    session = SessionLocal()
+    try:
+        video_id = uuid4().hex
+        generated_video = GeneratedVideo(
+            id=video_id,
+            project_id=project_id,
+            generation_task_id=task_id,
+            name=f"generated-{task_id[:8]}.mp4",
+            file_url=file_url,
+            file_size=file_size,
+            duration=duration,
+            width=OUTPUT_WIDTH,
+            height=OUTPUT_HEIGHT,
+            fps=OUTPUT_FPS,
+            status="completed",
+            generation_params={"mode": mode},
+        )
+
+        video_repo = SQLAlchemyGeneratedVideoRepository(session)
+        video_repo.create(generated_video)
+
+        # 计算视频指纹
+        deduplicator = VideoDeduplicator()
+        try:
+            fingerprint = deduplicator.compute_fingerprint(video_path)
+        except Exception as fp_err:
+            logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}")
+            session.commit()
+            return
+
+        generated_video.video_fingerprint = fingerprint.to_dict()
+
+        # (a) 历史成片查重
+        duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session)
+
+        # (b) 批次内查重（仅当有 batch_id 时）
+        if not duplicate_result and batch_id:
+            duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)
+
+        if duplicate_result:
+            generated_video.is_duplicate = True
+            generated_video.duplicate_of = duplicate_result["duplicate_of"]
+            logger.info(
+                f"Duplicate detected: {video_id} -> {duplicate_result['duplicate_of']} "
+                f"(reason={duplicate_result['reason']}, similarity={duplicate_result['similarity']:.3f})"
+            )
+        else:
+            generated_video.is_duplicate = False
+            generated_video.duplicate_of = None
+
+        video_repo.update(generated_video)
+        session.commit()
+        logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})")
+    except Exception as e:
+        logger.error(f"Failed to create video record / dedup for task {task_id}: {e}")
+        session.rollback()
+    finally:
+        session.close()
diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py
index 8c084e828..e131cec5c 100755
--- a/apps/worker/worker_app/tasks/ingest.py
+++ b/apps/worker/worker_app/tasks/ingest.py
@@ -179,6 +179,7 @@ def ingest_asset(job_id: str) -> dict:
             width=int(metadata.get("width", 0)),
             height=int(metadata.get("height", 0)),
             status=AssetStatus.READY,
+            file_hash=job.file_hash,
         )
         asset_repo.create(asset)
 
diff --git a/docs/schema-metadata-snapshot.json b/docs/schema-metadata-snapshot.json
index 769beb8ff..1b947366f 100644
--- a/docs/schema-metadata-snapshot.json
+++ b/docs/schema-metadata-snapshot.json
@@ -274,6 +274,14 @@
           "type": "VARCHAR(36)",
           "unique": false
         },
+        {
+          "index": true,
+          "name": "file_hash",
+          "nullable": true,
+          "primary_key": false,
+          "type": "VARCHAR(64)",
+          "unique": false
+        },
         {
           "index": false,
           "name": "metadata",
@@ -321,6 +329,13 @@
           "name": "ix_assets_created_at",
           "unique": false
         },
+        {
+          "columns": [
+            "file_hash"
+          ],
+          "name": "ix_assets_file_hash",
+          "unique": false
+        },
         {
           "columns": [
             "file_type"
@@ -1502,6 +1517,22 @@
           "type": "VARCHAR(32)",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "asset_select_mode",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(20)",
+          "unique": false
+        },
+        {
+          "index": true,
+          "name": "batch_id",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(32)",
+          "unique": false
+        },
         {
           "index": false,
           "name": "metadata",
@@ -1527,6 +1558,13 @@
           "name": "ix_generation_tasks_asset_library_id",
           "unique": false
         },
+        {
+          "columns": [
+            "batch_id"
+          ],
+          "name": "ix_generation_tasks_batch_id",
+          "unique": false
+        },
         {
           "columns": [
             "created_by_user_id"
@@ -1632,6 +1670,14 @@
           "type": "VARCHAR(32)",
           "unique": false
         },
+        {
+          "index": true,
+          "name": "file_hash",
+          "nullable": true,
+          "primary_key": false,
+          "type": "VARCHAR(64)",
+          "unique": false
+        },
         {
           "index": false,
           "name": "created_at",
@@ -1650,6 +1696,13 @@
         }
       ],
       "indexes": [
+        {
+          "columns": [
+            "file_hash"
+          ],
+          "name": "ix_ingest_jobs_file_hash",
+          "unique": false
+        },
         {
           "columns": [
             "library_id"
diff --git a/packages/adapters/in_memory/asset_repository.py b/packages/adapters/in_memory/asset_repository.py
index 6a3a64043..02f323736 100755
--- a/packages/adapters/in_memory/asset_repository.py
+++ b/packages/adapters/in_memory/asset_repository.py
@@ -76,3 +76,16 @@ class InMemoryAssetRepository:
         tag_set = set(tag_ids)
         items = [a for a in self._assets.values() if tag_set.issubset(set(a.tag_ids))]
         return items[skip : skip + limit]
+
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        if not file_hash:
+            return None
+        for asset in self._assets.values():
+            if asset.library_id == library_id and asset.file_hash == file_hash:
+                return asset
+        return None
diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py
index ab6083043..b340b4d75 100644
--- a/packages/adapters/sqlalchemy_impl/asset_repository.py
+++ b/packages/adapters/sqlalchemy_impl/asset_repository.py
@@ -83,6 +83,7 @@ class SQLAlchemyAssetRepository:
             classification_result=(json.dumps(asset.metadata) if asset.metadata else None),
             quality_score=asset.quality_score,
             uploaded_by_user_id=asset.uploaded_by_user_id or "system",
+            file_hash=asset.file_hash or None,
             created_at=asset.created_at,
             updated_at=now,
         )
@@ -110,6 +111,7 @@ class SQLAlchemyAssetRepository:
         model.classification_result = json.dumps(asset.metadata) if asset.metadata else None
         model.quality_score = asset.quality_score
         model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id
+        model.file_hash = asset.file_hash or model.file_hash
         model.updated_at = datetime.now(timezone.utc)
         self.session.flush()
         self._sync_asset_tags(asset.id, asset.tag_ids)
@@ -230,6 +232,7 @@ class SQLAlchemyAssetRepository:
             classification_status=ClassificationStatus(model.classification_status),
             quality_score=model.quality_score,
             uploaded_by_user_id=model.uploaded_by_user_id,
+            file_hash=model.file_hash or "",
             metadata=metadata,
             tag_ids=tag_ids,
             created_at=model.created_at,
@@ -267,3 +270,23 @@ class SQLAlchemyAssetRepository:
             return []
         models = self.session.query(AssetModel).filter(AssetModel.id.in_(ids)).offset(skip).limit(limit).all()
         return [self._to_domain(m) for m in models]
+
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        if not file_hash:
+            return None
+        model = (
+            self.session.query(AssetModel)
+            .filter(
+                AssetModel.asset_library_id == library_id,
+                AssetModel.file_hash == file_hash,
+            )
+            .first()
+        )
+        if model is None:
+            return None
+        return self._to_domain(model)
diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py
index 830faaed5..3730d04bf 100644
--- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py
+++ b/packages/adapters/sqlalchemy_impl/generated_video_repository.py
@@ -87,3 +87,39 @@ class SQLAlchemyGeneratedVideoRepository:
             .all()
         )
         return [self.get(model.id) for model in models if self.get(model.id) is not None]
+
+    def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]:
+        """通过 batch_id 查找同批次生成的所有视频（跨 generation_task 关联查询）。"""
+        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+        task_ids = (
+            self.session.query(GenerationTaskModel.id).filter(GenerationTaskModel.batch_id == batch_id).subquery()
+        )
+        models = (
+            self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.generation_task_id.in_(task_ids)).all()
+        )
+        return [self._to_domain(model) for model in models]
+
+    @staticmethod
+    def _to_domain(model: GeneratedVideoModel) -> GeneratedVideo:
+        return GeneratedVideo(
+            id=model.id,
+            project_id=model.project_id,
+            generation_task_id=model.generation_task_id,
+            name=model.name,
+            file_url=model.file_url,
+            file_size=int(model.file_size or 0),
+            duration=model.duration,
+            thumbnail_url=model.thumbnail_url,
+            width=int(model.width or 0),
+            height=int(model.height or 0),
+            fps=model.fps,
+            status=getattr(model, "status", "completed"),
+            review_status=getattr(model, "review_status", "pending_review"),
+            generation_params=json.loads(getattr(model, "generation_params", "{}") or "{}"),
+            video_fingerprint=json.loads(getattr(model, "video_fingerprint", "null") or "null"),
+            is_duplicate=getattr(model, "is_duplicate", False),
+            duplicate_of=getattr(model, "duplicate_of", None),
+            generated_at=model.generated_at,
+            created_at=model.created_at,
+        )
diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py
index 12c439292..b6cadb880 100644
--- a/packages/adapters/sqlalchemy_impl/generation_task_repository.py
+++ b/packages/adapters/sqlalchemy_impl/generation_task_repository.py
@@ -25,6 +25,8 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         completed_at=model.completed_at,
         created_by_user_id=model.created_by_user_id,
         source_edit_plan_id=model.source_edit_plan_id or "",
+        asset_select_mode=model.asset_select_mode or "",
+        batch_id=model.batch_id or "",
         created_at=model.created_at,
     )
 
@@ -52,6 +54,8 @@ class SQLAlchemyGenerationTaskRepository:
             completed_at=task.completed_at,
             created_by_user_id=task.created_by_user_id,
             source_edit_plan_id=task.source_edit_plan_id or None,
+            asset_select_mode=task.asset_select_mode or "",
+            batch_id=task.batch_id or "",
             created_at=task.created_at,
         )
         self.session.add(model)
@@ -123,5 +127,7 @@ class SQLAlchemyGenerationTaskRepository:
         model.started_at = task.started_at
         model.completed_at = task.completed_at
         model.source_edit_plan_id = task.source_edit_plan_id or None
+        model.asset_select_mode = task.asset_select_mode or ""
+        model.batch_id = task.batch_id or ""
         self.session.commit()
         return task
diff --git a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py
index ed3a2ff49..f16dc735d 100644
--- a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py
+++ b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py
@@ -17,6 +17,7 @@ class SQLAlchemyIngestJobRepository:
             status=job.status.value,
             error_message=job.error_message,
             result_asset_id=job.result_asset_id,
+            file_hash=job.file_hash,
             created_at=job.created_at,
             updated_at=job.updated_at,
         )
@@ -36,6 +37,7 @@ class SQLAlchemyIngestJobRepository:
             status=IngestJobStatus(model.status),
             error_message=model.error_message,
             result_asset_id=model.result_asset_id,
+            file_hash=model.file_hash or "",
             created_at=model.created_at,
             updated_at=model.updated_at,
         )
@@ -51,6 +53,7 @@ class SQLAlchemyIngestJobRepository:
         model.status = job.status.value
         model.error_message = job.error_message
         model.result_asset_id = job.result_asset_id
+        model.file_hash = job.file_hash
         model.updated_at = job.updated_at
         self.session.commit()
         return job
diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py
index 27a07513e..90a714bdf 100755
--- a/packages/adapters/sqlalchemy_impl/models.py
+++ b/packages/adapters/sqlalchemy_impl/models.py
@@ -85,6 +85,7 @@ class AssetModel(Base):
     classification_result = Column(Text, nullable=True)
     quality_score = Column(Float, nullable=True)
     uploaded_by_user_id = Column(String(36), nullable=False)
+    file_hash = Column(String(64), nullable=True, index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), index=True)
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
@@ -210,6 +211,7 @@ class IngestJobModel(Base):
     status = Column(String(20), nullable=False, default="pending")
     error_message = Column(Text, nullable=False, default="")
     result_asset_id = Column(String(32), nullable=False, default="")
+    file_hash = Column(String(64), nullable=True, index=True)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 
@@ -251,6 +253,8 @@ class GenerationTaskModel(Base):
     completed_at = Column(DateTime, nullable=True)
     created_by_user_id = Column(String(32), nullable=False, default="", index=True)
     source_edit_plan_id = Column(String(32), nullable=True, index=True)
+    asset_select_mode = Column(String(20), nullable=False, default="")
+    batch_id = Column(String(32), nullable=False, default="", index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 
diff --git a/packages/application/generation_tasks.py b/packages/application/generation_tasks.py
index b3a20a30e..894b44032 100644
--- a/packages/application/generation_tasks.py
+++ b/packages/application/generation_tasks.py
@@ -19,6 +19,8 @@ class CreateGenerationTaskCommand:
     voice_ids: list[str] = field(default_factory=list)
     created_by_user_id: str = ""
     source_edit_plan_id: str = ""
+    asset_select_mode: str = ""
+    batch_id: str = ""
 
 
 class CreateGenerationTaskUseCase:
@@ -44,6 +46,8 @@ class CreateGenerationTaskUseCase:
             completed_at=None,
             created_by_user_id=command.created_by_user_id,
             source_edit_plan_id=command.source_edit_plan_id,
+            asset_select_mode=command.asset_select_mode,
+            batch_id=command.batch_id,
         )
         return self.generation_task_repository.create(task)
 
diff --git a/packages/application/ingest_jobs.py b/packages/application/ingest_jobs.py
index bbb05553e..75a708de7 100644
--- a/packages/application/ingest_jobs.py
+++ b/packages/application/ingest_jobs.py
@@ -11,6 +11,7 @@ class SubmitIngestJobCommand:
     project_id: str
     library_id: str
     storage_key: str
+    file_hash: str = ""
 
 
 class SubmitIngestJobUseCase:
@@ -22,5 +23,6 @@ class SubmitIngestJobUseCase:
             project_id=command.project_id,
             library_id=command.library_id,
             storage_key=command.storage_key,
+            file_hash=command.file_hash,
         )
         return self.ingest_job_repository.create(job)
diff --git a/packages/domain/entities.py b/packages/domain/entities.py
index b31897d15..72e2f97fd 100644
--- a/packages/domain/entities.py
+++ b/packages/domain/entities.py
@@ -161,6 +161,7 @@ class Asset:
     classification_status: ClassificationStatus = ClassificationStatus.PENDING
     quality_score: float | None = None
     uploaded_by_user_id: str = ""
+    file_hash: str = ""
     metadata: dict[str, Any] = field(default_factory=dict)
     tag_ids: list[str] = field(default_factory=list)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -187,6 +188,7 @@ class Asset:
         classification_status: ClassificationStatus = ClassificationStatus.PENDING,
         quality_score: float | None = None,
         uploaded_by_user_id: str = "",
+        file_hash: str = "",
     ) -> "Asset":
         clean_name = name.strip()
         if not clean_name:
@@ -213,6 +215,7 @@ class Asset:
             classification_status=classification_status,
             quality_score=quality_score,
             uploaded_by_user_id=uploaded_by_user_id.strip(),
+            file_hash=file_hash.strip(),
             metadata=metadata or {},
             tag_ids=[],
         )
@@ -243,6 +246,7 @@ class IngestJob:
     status: IngestJobStatus = IngestJobStatus.PENDING
     error_message: str = ""
     result_asset_id: str = ""
+    file_hash: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
@@ -252,6 +256,7 @@ class IngestJob:
         project_id: str,
         library_id: str,
         storage_key: str,
+        file_hash: str = "",
     ) -> "IngestJob":
         if not project_id.strip():
             raise ValueError("project_id 不能为空")
@@ -264,4 +269,5 @@ class IngestJob:
             project_id=project_id.strip(),
             library_id=library_id.strip(),
             storage_key=storage_key.strip(),
+            file_hash=file_hash.strip(),
         )
diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py
index 7df475d27..f9f0b2541 100644
--- a/packages/domain/generation_task.py
+++ b/packages/domain/generation_task.py
@@ -43,6 +43,8 @@ class GenerationTask:
     completed_at: datetime | None = None
     source_edit_plan_id: str = ""
     created_by_user_id: str = ""
+    asset_select_mode: str = ""
+    batch_id: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @classmethod
@@ -59,6 +61,8 @@ class GenerationTask:
         voice_ids: list[str] | None = None,
         created_by_user_id: str = "",
         source_edit_plan_id: str = "",
+        asset_select_mode: str = "",
+        batch_id: str = "",
     ) -> "GenerationTask":
         if not project_id.strip() and not template_id.strip():
             raise ValueError("project_id 或 template_id 至少需要提供一个")
@@ -76,4 +80,6 @@ class GenerationTask:
             voice_ids=list(voice_ids) if voice_ids else [],
             created_by_user_id=created_by_user_id.strip(),
             source_edit_plan_id=source_edit_plan_id.strip(),
+            asset_select_mode=asset_select_mode,
+            batch_id=batch_id,
         )
diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py
index 9f96d1f1c..b2dda54e9 100644
--- a/packages/ports/asset_repository.py
+++ b/packages/ports/asset_repository.py
@@ -93,3 +93,12 @@ class AssetRepository(ABC):
     ) -> list[Asset]:
         """查找包含所有指定标签的素材。"""
         pass
+
+    @abstractmethod
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        pass
diff --git a/packages/ports/generated_video_repository.py b/packages/ports/generated_video_repository.py
index 400107c2b..f9aac9c21 100644
--- a/packages/ports/generated_video_repository.py
+++ b/packages/ports/generated_video_repository.py
@@ -13,3 +13,5 @@ class GeneratedVideoRepository(Protocol):
     def list_by_project(self, project_id: str) -> list[GeneratedVideo]: ...
 
     def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]: ...
+
+    def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]: ...
diff --git a/tests/unit/test_asset_dedup.py b/tests/unit/test_asset_dedup.py
new file mode 100644
index 000000000..a1cbc9b72
--- /dev/null
+++ b/tests/unit/test_asset_dedup.py
@@ -0,0 +1,341 @@
+"""
+素材重复上传检测 单元测试
+
+覆盖：
+- 表单上传（multipart）命中去重 → 直接返回已有 asset_id，不上传 OSS
+- 直传 OSS complete 命中去重 → 直接返回已有 asset_id，不创建 ingest job
+- 未命中去重 → 正常创建 ingest job
+- file_hash 为空 → 跳过去重检测
+- IngestJob 透传 file_hash 到 Asset
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, IngestJob, Project
+
+# ---------------------------------------------------------------------------
+# Stub repositories
+# ---------------------------------------------------------------------------
+
+
+class StubProjectRepository:
+    def __init__(self, projects: dict | None = None):
+        self._projects = projects or {}
+
+    def get(self, project_id: str):
+        return self._projects.get(project_id)
+
+    def find_by_id(self, project_id: str):
+        return self._projects.get(project_id)
+
+
+class StubAssetLibraryRepository:
+    def __init__(self, libraries: dict | None = None):
+        self._libraries = libraries or {}
+
+    def find_by_project(self, project_id: str, kind=None) -> list:
+        items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
+        if kind is not None:
+            items = [lib for lib in items if lib.kind == kind]
+        return items
+
+
+class StubAssetRepository:
+    """支持 find_by_library_and_file_hash 去重检测。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self._assets = assets or []
+
+    def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
+        for a in self._assets:
+            if a.library_id == library_id and a.file_hash == file_hash:
+                return a
+        return None
+
+    def create(self, asset: Asset) -> Asset:
+        self._assets.append(asset)
+        return asset
+
+
+class StubIngestJobRepository:
+    def __init__(self):
+        self._jobs: dict[str, IngestJob] = {}
+
+    def create(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        return job
+
+    def get(self, job_id: str) -> IngestJob | None:
+        return self._jobs.get(job_id)
+
+    def update(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        return job
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+DUPE_HASH = "a" * 32
+
+
+def _make_project(id: str = "proj-1", owner_user_id: str = "user-1") -> Project:
+    return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
+
+
+def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
+    return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=AssetLibraryKind.VIDEO)
+
+
+def _make_existing_asset(
+    id: str = "existing-asset-1",
+    library_id: str = "lib-1",
+    file_hash: str = DUPE_HASH,
+) -> Asset:
+    return Asset(
+        id=id,
+        project_id="proj-1",
+        library_id=library_id,
+        name="existing.mp4",
+        storage_key="uploads/existing/existing.mp4",
+        mime_type="video/mp4",
+        file_hash=file_hash,
+        status=AssetStatus.READY,
+    )
+
+
+def _build_app(
+    project_repo=None,
+    library_repo=None,
+    asset_repo=None,
+    ingest_repo=None,
+    storage=None,
+):
+    from app.api.routes.upload import router
+    from app.auth import AuthenticatedUser, get_current_user
+    from app.core.storage import get_storage_service
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_ingest_job_repository,
+        get_project_repository,
+    )
+
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+
+    project_repo = project_repo or StubProjectRepository()
+    library_repo = library_repo or StubAssetLibraryRepository()
+    asset_repo = asset_repo or StubAssetRepository()
+    ingest_repo = ingest_repo or StubIngestJobRepository()
+    storage = storage or MagicMock()
+    storage.is_configured = True
+    storage._normalize_storage_key = lambda key: key
+    storage.file_exists = lambda key: True
+    storage.upload_file = MagicMock(return_value="https://oss.example.com/file.mp4")
+
+    mock_user = MagicMock(spec=AuthenticatedUser)
+    mock_user.id = "user-1"
+    mock_user.email = "test@example.com"
+
+    app.dependency_overrides[get_current_user] = lambda: mock_user
+    app.dependency_overrides[get_project_repository] = lambda: project_repo
+    app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
+    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
+    app.dependency_overrides[get_storage_service] = lambda: storage
+
+    return app
+
+
+def _client(**kwargs) -> TestClient:
+    return TestClient(_build_app(**kwargs))
+
+
+# ---------------------------------------------------------------------------
+# 测试用例
+# ---------------------------------------------------------------------------
+
+
+class TestMultipartUploadDedup:
+    """表单上传（POST /api/v1/assets）去重检测。"""
+
+    def test_dedup_hit_returns_existing_asset(self):
+        """file_hash 命中已有素材 → 返回 duplicated=true + asset_id，不上传 OSS。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": DUPE_HASH,
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is True
+        assert body["asset_id"] == existing.id
+        assert body["ingest_job_id"] == ""
+
+    def test_dedup_miss_creates_ingest_job(self):
+        """file_hash 未命中 → 正常上传并创建 ingest job。"""
+        project = _make_project()
+        library = _make_library()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),  # 无已有素材
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": "b" * 32,  # 新的 hash
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+        assert body["ingest_job_id"] != ""
+
+    def test_empty_hash_skips_dedup(self):
+        """file_hash 为空 → 跳过去重检测，直接上传。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                # 不传 file_hash
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+
+
+class TestDirectUploadDedup:
+    """直传 OSS complete（POST /api/v1/direct/complete）去重检测。"""
+
+    def test_dedup_hit_returns_existing_asset(self):
+        """complete 阶段 file_hash 命中 → 返回 duplicated=true。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1/direct/complete",
+            json={
+                "project_id": project.id,
+                "library_id": library.id,
+                "storage_key": "uploads/abc/test.mp4",
+                "file_hash": DUPE_HASH,
+            },
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is True
+        assert body["asset_id"] == existing.id
+        assert body["ingest_job_id"] == ""
+
+    def test_dedup_miss_creates_ingest_job(self):
+        """complete 阶段 file_hash 未命中 → 创建 ingest job。"""
+        project = _make_project()
+        library = _make_library()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),
+        )
+
+        resp = client.post(
+            "/api/v1/direct/complete",
+            json={
+                "project_id": project.id,
+                "library_id": library.id,
+                "storage_key": "uploads/abc/test.mp4",
+                "file_hash": "c" * 32,
+            },
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+        assert body["ingest_job_id"] != ""
+
+
+class TestIngestJobFileHashPassthrough:
+    """file_hash 从上传接口透传到 IngestJob。"""
+
+    def test_ingest_job_stores_file_hash(self):
+        """上传时传入的 file_hash 应保存到 IngestJob 实体。"""
+        project = _make_project()
+        library = _make_library()
+        ingest_repo = StubIngestJobRepository()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),
+            ingest_repo=ingest_repo,
+        )
+
+        new_hash = "d" * 32
+        client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": new_hash,
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+
+        # 验证 IngestJob 存储了 file_hash
+        assert len(ingest_repo._jobs) == 1
+        job = list(ingest_repo._jobs.values())[0]
+        assert job.file_hash == new_hash
diff --git a/tests/unit/test_asset_select_mode.py b/tests/unit/test_asset_select_mode.py
new file mode 100644
index 000000000..98469e7fa
--- /dev/null
+++ b/tests/unit/test_asset_select_mode.py
@@ -0,0 +1,176 @@
+"""
+素材库自动匹配 单元测试
+
+覆盖：
+- all 模式：返回全部 ready 视频素材 ID
+- random 模式：随机选取 N 个
+- smart 模式：按质量分/时长评分降序选取
+- 无 ready 视频素材时返回空列表
+- count=0 时返回全部（random/smart 模式）
+- 非视频素材和非 ready 状态素材被过滤
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from app.api.routes.generation_tasks import _select_assets_from_library
+
+from packages.domain import Asset, AssetStatus
+
+
+def _asset(
+    id: str,
+    name: str,
+    mime_type: str = "video/mp4",
+    status: AssetStatus = AssetStatus.READY,
+    quality_score: float | None = None,
+    duration: float | None = None,
+) -> Asset:
+    a = Asset.create(
+        project_id="proj-1",
+        library_id="lib-1",
+        name=name,
+        storage_key=f"uploads/{name}",
+        mime_type=mime_type,
+        file_size=1024,
+        status=status,
+        quality_score=quality_score,
+        duration=duration,
+    )
+    # create() 会覆盖 id，手动设置
+    a.id = id
+    return a
+
+
+class TestSelectAssetsAllMode:
+    """all 模式：返回全部 ready 视频素材。"""
+
+    def test_returns_all_ready_video_assets(self):
+        assets = [
+            _asset("a1", "v1.mp4"),
+            _asset("a2", "v2.mp4"),
+            _asset("a3", "v3.mp4"),
+        ]
+        result = _select_assets_from_library(assets, mode="all", count=0)
+        assert sorted(result) == ["a1", "a2", "a3"]
+
+    def test_ignores_count_in_all_mode(self):
+        assets = [
+            _asset("a1", "v1.mp4"),
+            _asset("a2", "v2.mp4"),
+        ]
+        result = _select_assets_from_library(assets, mode="all", count=1)
+        assert len(result) == 2
+
+    def test_filters_non_video_assets(self):
+        assets = [
+            _asset("a1", "v1.mp4", mime_type="video/mp4"),
+            _asset("a2", "img.jpg", mime_type="image/jpeg"),
+            _asset("a3", "v2.mov", mime_type="video/quicktime"),
+        ]
+        result = _select_assets_from_library(assets, mode="all", count=0)
+        assert sorted(result) == ["a1", "a3"]
+
+    def test_filters_non_ready_assets(self):
+        assets = [
+            _asset("a1", "v1.mp4", status=AssetStatus.READY),
+            _asset("a2", "v2.mp4", status=AssetStatus.UPLOADING),
+            _asset("a3", "v3.mp4", status=AssetStatus.PROCESSING),
+        ]
+        result = _select_assets_from_library(assets, mode="all", count=0)
+        assert result == ["a1"]
+
+    def test_empty_library_returns_empty(self):
+        result = _select_assets_from_library([], mode="all", count=0)
+        assert result == []
+
+    def test_no_ready_video_returns_empty(self):
+        assets = [
+            _asset("a1", "v1.mp4", status=AssetStatus.UPLOADING),
+            _asset("a2", "img.jpg", mime_type="image/jpeg", status=AssetStatus.READY),
+        ]
+        result = _select_assets_from_library(assets, mode="all", count=0)
+        assert result == []
+
+
+class TestSelectAssetsRandomMode:
+    """random 模式：随机选取 N 个。"""
+
+    def test_random_selects_exact_count(self):
+        assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(10)]
+        result = _select_assets_from_library(assets, mode="random", count=3)
+        assert len(result) == 3
+        assert all(rid in [a.id for a in assets] for rid in result)
+
+    def test_random_count_zero_returns_all(self):
+        assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(5)]
+        result = _select_assets_from_library(assets, mode="random", count=0)
+        assert len(result) == 5
+
+    def test_random_count_exceeds_total_returns_all(self):
+        assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(3)]
+        result = _select_assets_from_library(assets, mode="random", count=100)
+        assert len(result) == 3
+
+
+class TestSelectAssetsSmartMode:
+    """smart 模式：按质量分/时长评分降序选取。"""
+
+    def test_smart_sorts_by_quality_score_desc(self):
+        assets = [
+            _asset("low", "low.mp4", quality_score=0.3),
+            _asset("high", "high.mp4", quality_score=0.9),
+            _asset("mid", "mid.mp4", quality_score=0.6),
+        ]
+        result = _select_assets_from_library(assets, mode="smart", count=0)
+        assert result == ["high", "mid", "low"]
+
+    def test_smart_tiebreak_by_duration_desc(self):
+        assets = [
+            _asset("short", "short.mp4", quality_score=0.8, duration=10.0),
+            _asset("long", "long.mp4", quality_score=0.8, duration=60.0),
+        ]
+        result = _select_assets_from_library(assets, mode="smart", count=0)
+        assert result == ["long", "short"]
+
+    def test_smart_with_count_limits_results(self):
+        assets = [
+            _asset("a1", "v1.mp4", quality_score=0.9),
+            _asset("a2", "v2.mp4", quality_score=0.7),
+            _asset("a3", "v3.mp4", quality_score=0.5),
+        ]
+        result = _select_assets_from_library(assets, mode="smart", count=2)
+        assert result == ["a1", "a2"]
+
+    def test_smart_null_quality_treated_as_zero(self):
+        assets = [
+            _asset("scored", "scored.mp4", quality_score=0.5),
+            _asset("unscored", "unscored.mp4", quality_score=None),
+        ]
+        result = _select_assets_from_library(assets, mode="smart", count=0)
+        assert result == ["scored", "unscored"]
+
+    def test_smart_count_zero_returns_all_sorted(self):
+        assets = [
+            _asset("a1", "v1.mp4", quality_score=0.1),
+            _asset("a2", "v2.mp4", quality_score=0.9),
+            _asset("a3", "v3.mp4", quality_score=0.5),
+        ]
+        result = _select_assets_from_library(assets, mode="smart", count=0)
+        assert result == ["a2", "a3", "a1"]
+
+
+class TestSelectAssetsDefaultMode:
+    """默认模式（未知 mode 字符串）应回退到 all。"""
+
+    def test_unknown_mode_falls_back_to_all(self):
+        assets = [
+            _asset("a1", "v1.mp4"),
+            _asset("a2", "v2.mp4"),
+        ]
+        result = _select_assets_from_library(assets, mode="unknown", count=0)
+        assert len(result) == 2
diff --git a/tests/unit/test_dedup_engine.py b/tests/unit/test_dedup_engine.py
index 2a92f7540..5b7e17633 100644
--- a/tests/unit/test_dedup_engine.py
+++ b/tests/unit/test_dedup_engine.py
@@ -461,3 +461,171 @@ class TestVideoDeduplicatorCheckDuplicate:
             assert result["similarity"] == 1.0  # avg_distance = 0
         finally:
             self._restore_repo(mod, orig)
+
+
+class TestVideoDeduplicatorCheckBatchDuplicate:
+    """VideoDeduplicator.check_batch_duplicate() 测试。
+
+    批次内查重逻辑与历史查重一致（MD5 + pHash），但搜索范围限定为同 batch_id 的视频。
+    """
+
+    @pytest.fixture
+    def deduplicator(self):
+        return VideoDeduplicator()
+
+    @pytest.fixture
+    def mock_session(self):
+        return MagicMock()
+
+    def _make_batch_video(self, video_id, md5, phashes=None):
+        video = MagicMock()
+        video.id = video_id
+        video.video_fingerprint = {
+            "md5": md5,
+            "keyframe_phashes": phashes or [],
+            "color_histograms": [],
+        }
+        return video
+
+    def _patch_repo(self, mock_repo):
+        import apps.worker.video_processing.dedup as dedup_module
+
+        original = dedup_module.SQLAlchemyGeneratedVideoRepository
+        dedup_module.SQLAlchemyGeneratedVideoRepository = MagicMock(return_value=mock_repo)
+        return original, dedup_module
+
+    def _restore_repo(self, dedup_module, original):
+        dedup_module.SQLAlchemyGeneratedVideoRepository = original
+
+    def test_batch_exact_md5_match(self, deduplicator, mock_session):
+        """批次内 MD5 完全匹配应返回 duplicate。"""
+        other = self._make_batch_video("vid-other", "abc123")
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [other]
+
+        fingerprint = VideoFingerprint(
+            md5="abc123",
+            keyframe_phashes=["ff"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is not None
+            assert result["duplicate"] is True
+            assert result["reason"] == "batch_exact_md5_match"
+            assert result["similarity"] == 1.0
+            assert result["duplicate_of"] == "vid-other"
+        finally:
+            self._restore_repo(mod, orig)
+
+    def test_batch_phash_similar(self, deduplicator, mock_session):
+        """批次内 pHash 距离 < 阈值应判定为重复。"""
+        other = self._make_batch_video("vid-other", "md5_diff", phashes=["abcdef01"])
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [other]
+
+        fingerprint = VideoFingerprint(
+            md5="md5_new",
+            keyframe_phashes=["abcdef01"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is not None
+            assert result["duplicate"] is True
+            assert result["reason"] == "batch_phash_similar"
+        finally:
+            self._restore_repo(mod, orig)
+
+    def test_batch_excludes_self(self, deduplicator, mock_session):
+        """批次查重应排除自身视频。"""
+        self_video = self._make_batch_video("vid-self", "abc123")
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [self_video]
+
+        fingerprint = VideoFingerprint(
+            md5="abc123",
+            keyframe_phashes=["ff"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is None
+        finally:
+            self._restore_repo(mod, orig)
+
+    def test_batch_no_match(self, deduplicator, mock_session):
+        """批次内无重复时应返回 None。"""
+        other = self._make_batch_video("vid-other", "md5_a", phashes=["0000000000000000"])
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [other]
+
+        fingerprint = VideoFingerprint(
+            md5="md5_b",
+            keyframe_phashes=["ffffffffffffffff"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is None
+        finally:
+            self._restore_repo(mod, orig)
+
+    def test_batch_empty_returns_none(self, deduplicator, mock_session):
+        """空批次应返回 None。"""
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = []
+
+        fingerprint = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=["ff"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is None
+        finally:
+            self._restore_repo(mod, orig)
+
+    def test_batch_skips_no_fingerprint(self, deduplicator, mock_session):
+        """批次内无指纹的视频应被跳过。"""
+        other = MagicMock()
+        other.id = "vid-other"
+        other.video_fingerprint = None
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [other]
+
+        fingerprint = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=["ff"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1280, 720),
+        )
+
+        orig, mod = self._patch_repo(mock_repo)
+        try:
+            result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
+            assert result is None
+        finally:
+            self._restore_repo(mod, orig)