alembic/versions/059_add_duplicate_rate_to_generated_videos.py

@@ -0,0 +1,23 @@
+"""add duplicate_rate to generated_videos
+
+Revision ID: 059_duplicate_rate
+Revises: 058_uq_asset_lib_project_kind
+Create Date: 2026-08-31
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "059_duplicate_rate"
+down_revision = "058_uq_asset_lib_project_kind"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("generated_videos", sa.Column("duplicate_rate", sa.Float(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("generated_videos", "duplicate_rate")

apps/api/app/api/routes/assets.py

@@ -26,7 +26,7 @@ from app.schemas.asset import (
     UpdateAssetReviewRequest,
 )
 from app.schemas.tag import TagAssetsRequest
-from app.services.asset_segment_tracker import compute_asset_availability
+from app.services.asset_segment_tracker import compute_asset_availability, get_asset_recent_use_counts
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 
 from packages.domain.smart_match import smart_select_assets
@@ -608,6 +608,38 @@ def smart_match_assets(
             continue
         filtered_results.append(r)
 
+    # 高频使用排除：同一素材在最近 5 个视频中出现超过 3 次则排除
+    MAX_RECENT_USE_COUNT = 3
+    if filtered_results:
+        asset_ids = [getattr(r.asset, "id", "") for r in filtered_results if getattr(r.asset, "id", "")]
+        if asset_ids:
+            try:
+                use_counts = get_asset_recent_use_counts(
+                    db=asset_repository.session,
+                    asset_ids=asset_ids,
+                    recent_video_count=5,
+                )
+                high_use_excluded = set()
+                for r in filtered_results:
+                    aid = getattr(r.asset, "id", "")
+                    count = use_counts.get(aid, 0)
+                    if count > MAX_RECENT_USE_COUNT:
+                        logger.info(
+                            "smart-match 排除高频使用素材: asset_id=%s use_count=%d limit=%d",
+                            aid, count, MAX_RECENT_USE_COUNT,
+                        )
+                        high_use_excluded.add(id(r))
+                    else:
+                        pass
+                # 如果排除后不够 limit，放宽到不限制
+                remaining = [r for r in filtered_results if id(r) not in high_use_excluded]
+                if len(remaining) >= request.limit:
+                    filtered_results = remaining
+                else:
+                    logger.info("smart-match 高频排除后素材不足(%d<%d)，保留全部", len(remaining), request.limit)
+            except Exception:
+                logger.warning("smart-match 高频使用查询失败，跳过排除", exc_info=True)
+
     # 扁平结构：SmartMatchItem 继承 AssetResponse，素材字段直接在条目顶层，
     # 前端无需解析 item.asset 包装层，item.id / item.usable / 余量字段直接可读
     items = [

apps/api/app/api/routes/videos.py

@@ -52,6 +52,7 @@ def _to_video_response(item, storage: OSSStorageService | None = None) -> VideoI
         generation_params=item.generation_params,
         download_url=download_url,
         generated_at=format_utc_datetime(item.generated_at) if hasattr(item, "generated_at") else "",
+        duplicate_rate=getattr(item, "duplicate_rate", None),
     )
 
 

apps/api/app/schemas/video_center.py

@@ -22,6 +22,7 @@ class VideoItemResponse(BaseModel):
     generation_params: dict = Field(default_factory=dict)
     download_url: str | None = None
     generated_at: str = ""
+    duplicate_rate: float | None = None
 
 
 class ListVideosResponse(BaseModel):

apps/api/app/services/asset_segment_tracker.py

@@ -438,3 +438,57 @@ def make_reuse_callback(
         return result
 
     return _reuse
+
+
+def get_asset_recent_use_counts(
+    db: Session,
+    asset_ids: list[str],
+    recent_video_count: int = 5,
+) -> dict[str, int]:
+    """统计每个素材在最近 N 个不同 plan_id 中的使用次数。
+
+    遍历素材 metadata 中的 used_time_ranges，统计有多少个不同的 plan_id（去重），
+    返回 {asset_id: count}。只统计最近 recent_video_count 个不同 plan_id 的使用次数。
+
+    Args:
+        db: 数据库会话
+        asset_ids: 素材 ID 列表
+        recent_video_count: 统计最近多少个不同 plan_id
+
+    Returns:
+        {asset_id: 在最近 recent_video_count 个 plan 中的使用次数}
+    """
+    if not asset_ids:
+        return {}
+
+    result: dict[str, int] = {}
+    models = db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
+    for model in models:
+        meta = _read_meta(model)
+        ranges = meta.get(USED_RANGES_KEY) or []
+        if not ranges:
+            result[model.id] = 0
+            continue
+
+        # 按 created_at 倒序收集不同 plan_id
+        sorted_ranges = sorted(
+            ranges,
+            key=lambda r: r.get("created_at") or "",
+            reverse=True,
+        )
+        recent_plan_ids: set[str] = set()
+        for r in sorted_ranges:
+            plan_id = r.get("plan_id")
+            if plan_id:
+                recent_plan_ids.add(plan_id)
+            if len(recent_plan_ids) >= recent_video_count:
+                break
+
+        result[model.id] = len(recent_plan_ids)
+
+    # 未找到的素材计为 0
+    for aid in asset_ids:
+        if aid not in result:
+            result[aid] = 0
+
+    return result

apps/worker/video_processing/dedup.py

@@ -308,6 +308,71 @@ class VideoDeduplicator:
 
         return sum(similarities) / len(similarities) if similarities else 0.0
 
+    def compute_duplicate_rate(
+        self,
+        fingerprint: VideoFingerprint,
+        project_id: str,
+        current_video_id: str | None,
+        session: Session,
+    ) -> float:
+        """计算当前视频与项目内已有视频的最高相似度百分比。
+
+        遍历项目内所有其他有指纹的视频，对每个计算相似度：
+        - MD5 精确匹配 → 100%
+        - pHash 相似度 → (1.0 - avg_distance / 64) * 100
+        取最高值作为 duplicate_rate（0~100）。
+        如果没有其他视频可比较，返回 0.0。
+
+        Args:
+            fingerprint: 当前视频的指纹
+            project_id: 项目 ID
+            current_video_id: 当前视频 ID（排除自身，可为 None）
+            session: 数据库会话
+
+        Returns:
+            duplicate_rate: 0~100 的浮点数
+        """
+        # 限制查询最近 100 个视频，避免大项目内存溢出
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        recent_models = (
+            session.query(GeneratedVideoModel)
+            .filter(GeneratedVideoModel.project_id == project_id)
+            .order_by(GeneratedVideoModel.generated_at.desc())
+            .limit(100)
+            .all()
+        )
+        video_repo = SQLAlchemyGeneratedVideoRepository(session)
+        existing_videos = [video_repo._to_domain(m) for m in recent_models]
+
+        max_similarity = 0.0
+        for existing in existing_videos:
+            if current_video_id and existing.id == current_video_id:
+                continue
+            if not existing.video_fingerprint:
+                continue
+
+            ef = existing.video_fingerprint
+
+            # MD5 精确匹配 → 100%
+            if fingerprint.md5 == ef.get("md5"):
+                return 100.0
+
+            # pHash 相似度
+            existing_phashes = ef.get("keyframe_phashes", [])
+            if not existing_phashes or not fingerprint.keyframe_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 64
+            similarity = (1.0 - avg_distance / 64) * 100
+            max_similarity = max(max_similarity, similarity)
+
+        return round(max(max_similarity, 0.0), 2)
+
 
 @celery_app.task(bind=True, max_retries=3, name="worker.check_duplicate")
 def check_duplicate_task(self: Task, generated_video_id: str) -> dict:

apps/worker/video_processing/dedup_helpers.py

@@ -121,6 +121,15 @@ def create_video_record_and_dedup(
             generated_video.is_duplicate = False
             generated_video.duplicate_of = None
 
+        # 计算重复率百分比（与项目内所有已有视频对比取最高相似度）
+        try:
+            dup_rate = deduplicator.compute_duplicate_rate(fingerprint, project_id, video_id, session)
+            generated_video.duplicate_rate = dup_rate
+            logger.info("Duplicate rate for %s: %.2f%%", video_id, dup_rate)
+        except Exception as rate_err:
+            logger.warning("Failed to compute duplicate_rate for %s: %s", video_id, rate_err)
+            generated_video.duplicate_rate = None
+
         video_repo.update(generated_video)
         session.commit()
         logger.info(

packages/adapters/sqlalchemy_impl/generated_video_repository.py

@@ -30,6 +30,7 @@ class SQLAlchemyGeneratedVideoRepository:
             video_fingerprint=json.dumps(video.video_fingerprint) if video.video_fingerprint else None,
             is_duplicate=video.is_duplicate,
             duplicate_of=video.duplicate_of,
+            duplicate_rate=video.duplicate_rate,
             generated_at=video.generated_at,
             created_at=video.created_at,
         )
@@ -60,6 +61,7 @@ class SQLAlchemyGeneratedVideoRepository:
             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),
+            duplicate_rate=getattr(model, "duplicate_rate", None),
             generated_at=model.generated_at,
             created_at=model.created_at,
         )
@@ -74,6 +76,7 @@ class SQLAlchemyGeneratedVideoRepository:
         model.video_fingerprint = json.dumps(video.video_fingerprint) if video.video_fingerprint else None
         model.is_duplicate = video.is_duplicate
         model.duplicate_of = video.duplicate_of
+        model.duplicate_rate = video.duplicate_rate
         self.session.add(model)
         self.session.commit()
         return video
@@ -204,6 +207,7 @@ class SQLAlchemyGeneratedVideoRepository:
             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),
+            duplicate_rate=getattr(model, "duplicate_rate", None),
             generated_at=model.generated_at,
             created_at=model.created_at,
         )

packages/adapters/sqlalchemy_impl/models.py

@@ -338,6 +338,7 @@ class GeneratedVideoModel(Base):
     video_fingerprint = Column(Text, nullable=True)
     is_duplicate = Column(Boolean, nullable=False, default=False)
     duplicate_of = Column(String(36), nullable=True)
+    duplicate_rate = Column(Float, nullable=True)
 
 
 class TitleLibraryModel(Base):

packages/domain/generated_video.py

@@ -26,6 +26,7 @@ class GeneratedVideo:
     video_fingerprint: dict[str, Any] | None = None
     is_duplicate: bool = False
     duplicate_of: str | None = None
+    duplicate_rate: float | None = None
     generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 

tests/unit/test_duplicate_rate.py

@@ -0,0 +1,221 @@
+"""Tests for duplicate_rate computation and API response."""
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Mock cv2 and numpy before any imports that need them
+sys.modules.setdefault("cv2", MagicMock())
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "api"))
+sys.path.insert(0, str(ROOT / "packages"))
+sys.path.insert(0, str(ROOT / "apps" / "worker"))
+
+
+class TestComputeDuplicateRate:
+    """Test VideoDeduplicator.compute_duplicate_rate."""
+
+    def _make_fingerprint(self, md5="abc123", phashes=None):
+        from video_processing.dedup import VideoFingerprint
+
+        return VideoFingerprint(
+            md5=md5,
+            keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
+            color_histograms=[],
+            duration=10.0,
+            resolution=(1920, 1080),
+        )
+
+    def _make_existing_video(self, vid, fingerprint_dict):
+        from packages.domain import GeneratedVideo
+
+        return GeneratedVideo(
+            id=vid,
+            project_id="proj1",
+            generation_task_id="task1",
+            name=f"video-{vid}",
+            file_url=f"https://example.com/{vid}.mp4",
+            file_size=1000,
+            duration=10.0,
+            width=1920,
+            height=1080,
+            fps=25.0,
+            video_fingerprint=fingerprint_dict,
+        )
+
+    def test_no_existing_videos_returns_zero(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint()
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = (
+                []
+            )
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        assert rate == 0.0
+
+    def test_md5_match_returns_100(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint(md5="exact_match_md5")
+        session = MagicMock()
+
+        existing = self._make_existing_video("existing1", {"md5": "exact_match_md5", "keyframe_phashes": ["aa"]})
+        # Create a mock model with the domain attributes
+        mock_model = MagicMock(spec=GeneratedVideoModel)
+        mock_model.id = existing.id
+        mock_model.project_id = existing.project_id
+        mock_model.video_fingerprint = existing.video_fingerprint
+        mock_model.generated_at = "2026-01-01"
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo._to_domain.return_value = existing
+            # Mock the session.query chain
+            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
+                mock_model
+            ]
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        assert rate == 100.0
+
+    def test_phash_similarity_computed(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint(md5="different_md5", phashes=["ff00ff00ff00ff00"])
+        session = MagicMock()
+
+        existing = self._make_existing_video(
+            "existing1",
+            {"md5": "other_md5", "keyframe_phashes": ["ff00ff00ff00ff03"]},
+        )
+        mock_model = MagicMock(spec=GeneratedVideoModel)
+        mock_model.id = existing.id
+        mock_model.project_id = existing.project_id
+        mock_model.video_fingerprint = existing.video_fingerprint
+        mock_model.generated_at = "2026-01-01"
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo._to_domain.return_value = existing
+            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
+                mock_model
+            ]
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        # hamming distance = 2, similarity = (1 - 2/64) * 100 = 96.875
+        assert rate == pytest.approx(96.88, abs=0.1)
+
+    def test_excludes_self_video(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint(md5="same_md5")
+        session = MagicMock()
+
+        self_video = self._make_existing_video("vid1", {"md5": "same_md5", "keyframe_phashes": ["aa"]})
+        mock_model = MagicMock(spec=GeneratedVideoModel)
+        mock_model.id = self_video.id
+        mock_model.project_id = self_video.project_id
+        mock_model.video_fingerprint = self_video.video_fingerprint
+        mock_model.generated_at = "2026-01-01"
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo._to_domain.return_value = self_video
+            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
+                mock_model
+            ]
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        assert rate == 0.0
+
+    def test_takes_max_similarity(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint(md5="new_md5", phashes=["ff00ff00ff00ff00"])
+        session = MagicMock()
+
+        existing1 = self._make_existing_video("e1", {"md5": "md5_1", "keyframe_phashes": ["ff00ff00ff00ff0f"]})
+        existing2 = self._make_existing_video("e2", {"md5": "md5_2", "keyframe_phashes": ["ff00ff00ff00ff01"]})
+        mock_model1 = MagicMock(spec=GeneratedVideoModel)
+        mock_model1.id = existing1.id
+        mock_model1.project_id = existing1.project_id
+        mock_model1.video_fingerprint = existing1.video_fingerprint
+        mock_model1.generated_at = "2026-01-02"
+        mock_model2 = MagicMock(spec=GeneratedVideoModel)
+        mock_model2.id = existing2.id
+        mock_model2.project_id = existing2.project_id
+        mock_model2.video_fingerprint = existing2.video_fingerprint
+        mock_model2.generated_at = "2026-01-01"
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo._to_domain.side_effect = [existing1, existing2]
+            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
+                mock_model1,
+                mock_model2,
+            ]
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        # max similarity: e2 distance=1, (1-1/64)*100 = 98.4375
+        assert rate == pytest.approx(98.44, abs=0.1)
+
+
+class TestDuplicateRateAPI:
+    """Test that duplicate_rate is returned in API responses."""
+
+    def test_video_item_response_has_duplicate_rate(self):
+        from app.schemas.video_center import VideoItemResponse
+
+        resp = VideoItemResponse(
+            id="v1",
+            project_id="p1",
+            generation_task_id="t1",
+            name="test.mp4",
+            file_url="https://example.com/test.mp4",
+            file_size=1000,
+            duration=10.0,
+            width=1920,
+            height=1080,
+            fps=25.0,
+            duplicate_rate=75.5,
+        )
+        assert resp.duplicate_rate == 75.5
+
+    def test_video_item_response_duplicate_rate_default_none(self):
+        from app.schemas.video_center import VideoItemResponse
+
+        resp = VideoItemResponse(
+            id="v1",
+            project_id="p1",
+            generation_task_id="t1",
+            name="test.mp4",
+            file_url="https://example.com/test.mp4",
+            file_size=1000,
+            duration=10.0,
+            width=1920,
+            height=1080,
+            fps=25.0,
+        )
+        assert resp.duplicate_rate is None

tests/unit/test_high_use_exclusion.py

@@ -0,0 +1,93 @@
+"""Tests for get_asset_recent_use_counts and smart-match high-use exclusion."""
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "api"))
+sys.path.insert(0, str(ROOT / "packages"))
+
+
+class TestGetAssetRecentUseCounts:
+    """Test asset_segment_tracker.get_asset_recent_use_counts."""
+
+    def _make_asset_model(self, asset_id, used_time_ranges=None):
+        from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+        model = AssetModel(
+            id=asset_id,
+            name=f"asset-{asset_id}",
+            file_type="video",
+            status="ready",
+            asset_library_id="lib1",
+            project_id="proj1",
+            file_size=1000,
+            file_url=f"https://example.com/{asset_id}.mp4",
+            uploaded_by_user_id="user1",
+        )
+        model.classification_result = json.dumps(
+            {
+                "used_time_ranges": used_time_ranges or [],
+            }
+        )
+        return model
+
+    def test_empty_asset_ids_returns_empty(self):
+        from app.services.asset_segment_tracker import get_asset_recent_use_counts
+
+        db = MagicMock()
+        result = get_asset_recent_use_counts(db, [])
+        assert result == {}
+
+    def test_no_usage_returns_zero(self):
+        from app.services.asset_segment_tracker import get_asset_recent_use_counts
+
+        db = MagicMock()
+        model = self._make_asset_model("a1", [])
+        db.query.return_value.filter.return_value.all.return_value = [model]
+
+        result = get_asset_recent_use_counts(db, ["a1"])
+        assert result == {"a1": 0}
+
+    def test_counts_distinct_plan_ids(self):
+        from app.services.asset_segment_tracker import get_asset_recent_use_counts
+
+        db = MagicMock()
+        ranges = [
+            {"start": 0, "end": 5, "plan_id": "plan1", "created_at": "2026-08-01T00:00:00"},
+            {"start": 5, "end": 10, "plan_id": "plan1", "created_at": "2026-08-01T00:01:00"},
+            {"start": 0, "end": 5, "plan_id": "plan2", "created_at": "2026-08-02T00:00:00"},
+            {"start": 0, "end": 5, "plan_id": "plan3", "created_at": "2026-08-03T00:00:00"},
+        ]
+        model = self._make_asset_model("a1", ranges)
+        db.query.return_value.filter.return_value.all.return_value = [model]
+
+        result = get_asset_recent_use_counts(db, ["a1"])
+        assert result == {"a1": 3}
+
+    def test_limits_to_recent_n(self):
+        from app.services.asset_segment_tracker import get_asset_recent_use_counts
+
+        db = MagicMock()
+        ranges = [
+            {"start": 0, "end": 5, "plan_id": f"plan{i}", "created_at": f"2026-08-{i+1:02d}T00:00:00"}
+            for i in range(10)
+        ]
+        model = self._make_asset_model("a1", ranges)
+        db.query.return_value.filter.return_value.all.return_value = [model]
+
+        result = get_asset_recent_use_counts(db, ["a1"], recent_video_count=5)
+        assert result["a1"] == 5
+
+    def test_missing_asset_defaults_to_zero(self):
+        from app.services.asset_segment_tracker import get_asset_recent_use_counts
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.all.return_value = []
+
+        result = get_asset_recent_use_counts(db, ["missing_asset"])
+        assert result == {"missing_asset": 0}