diff --git a/alembic/versions/059_add_duplicate_rate_to_generated_videos.py b/alembic/versions/059_add_duplicate_rate_to_generated_videos.py new file mode 100644 index 000000000..06bf3b897 --- /dev/null +++ b/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") diff --git a/apps/api/app/api/routes/assets.py b/apps/api/app/api/routes/assets.py index 80a199aff..e096f2285 100755 --- a/apps/api/app/api/routes/assets.py +++ b/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 = [ diff --git a/apps/api/app/api/routes/videos.py b/apps/api/app/api/routes/videos.py index e01c94e7c..cba5a2298 100644 --- a/apps/api/app/api/routes/videos.py +++ b/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), ) diff --git a/apps/api/app/schemas/video_center.py b/apps/api/app/schemas/video_center.py index 059f4f249..d349cfd01 100755 --- a/apps/api/app/schemas/video_center.py +++ b/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): diff --git a/apps/api/app/services/asset_segment_tracker.py b/apps/api/app/services/asset_segment_tracker.py index 88ffd07eb..06b7273c6 100644 --- a/apps/api/app/services/asset_segment_tracker.py +++ b/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 diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index 2fa6642d3..16992ab30 100755 --- a/apps/worker/video_processing/dedup.py +++ b/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: diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py index 401c09f87..8f17d7ac3 100755 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/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( diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py index 44569958f..6867e5b33 100755 --- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py +++ b/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, ) diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index df5cf79f3..240bed750 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/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): diff --git a/packages/domain/generated_video.py b/packages/domain/generated_video.py index 9d15e8c69..9f345b59c 100755 --- a/packages/domain/generated_video.py +++ b/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)) diff --git a/tests/unit/test_duplicate_rate.py b/tests/unit/test_duplicate_rate.py new file mode 100644 index 000000000..b9d6fd2ed --- /dev/null +++ b/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 diff --git a/tests/unit/test_high_use_exclusion.py b/tests/unit/test_high_use_exclusion.py new file mode 100644 index 000000000..70d72d94b --- /dev/null +++ b/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}