"""Video deduplication module - compute fingerprints and detect duplicates.""" import hashlib import json import logging import os import subprocess import tempfile from dataclasses import dataclass from typing import Optional import cv2 import numpy as np from celery import Task from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory from app.config import get_settings from app.core.storage import get_storage_service from apps.worker.celery_app import celery_app logger = logging.getLogger(__name__) settings = get_settings() if SessionLocal is None: build_session_factory(settings.database_url) def compute_phash(image: np.ndarray, hash_size: int = 8) -> str: """Compute perceptual hash of an image using DCT.""" # Resize to 32x32 for DCT resized = cv2.resize(image, (hash_size * 4, hash_size * 4)) gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY).astype(np.float32) # Apply 2D DCT dct = cv2.dct(gray) # Take top-left 8x8 low-frequency components dct_low = dct[:hash_size, :hash_size] # Compute median (excluding DC component at [0,0]) dct_low[0, 0] = 0 median = np.median(dct_low) # Generate hash based on comparison with median diff = (dct_low > median).astype(int) hash_str = "".join(str(b) for row in diff for b in row) return hex(int(hash_str, 2))[2:] def hamming_distance(hash1: str, hash2: str) -> int: """Calculate Hamming distance between two hex hashes.""" h1, h2 = int(hash1, 16), int(hash2, 16) return bin(h1 ^ h2).count("1") def compute_color_histogram(image: np.ndarray, bins: int = 32) -> list[float]: """Compute color histogram for an image.""" hist = [] for i in range(3): h = cv2.calcHist([image], [i], None, [bins], [0, 256]) h = cv2.normalize(h, h).flatten() hist.extend(h) return hist @dataclass class VideoFingerprint: """Video fingerprint containing multiple similarity metrics.""" md5: str keyframe_phashes: list[str] color_histograms: list[list[float]] duration: float resolution: tuple[int, int] def to_dict(self) -> dict: return {"md5": self.md5, "keyframe_phashes": self.keyframe_phashes, "color_histograms": self.color_histograms, "duration": self.duration, "resolution": list(self.resolution)} class VideoDeduplicator: """Video deduplication using multiple fingerprint methods.""" PHASH_THRESHOLD = 10 HISTOGRAM_THRESHOLD = 0.85 def compute_fingerprint(self, video_path: str) -> VideoFingerprint: """Compute video fingerprint using MD5, pHash, and color histogram.""" cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise RuntimeError(f"Cannot open video: {video_path}") fps = cap.get(cv2.CAP_PROP_FPS) frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) duration = frame_count / fps if fps > 0 else 0 width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) md5_hash = hashlib.md5() keyframe_phashes = [] color_histograms = [] frame_interval = max(1, frame_count // 10) for i in range(0, frame_count, frame_interval): cap.set(cv2.CAP_PROP_POS_FRAMES, i) ret, frame = cap.read() if not ret: continue _, buffer = cv2.imencode(".jpg", frame) md5_hash.update(buffer) keyframe_phashes.append(compute_phash(frame)) color_histograms.append(compute_color_histogram(frame)) cap.release() return VideoFingerprint(md5=md5_hash.hexdigest(), keyframe_phashes=keyframe_phashes, color_histograms=color_histograms, duration=duration, resolution=(width, height)) def check_duplicate(self, fingerprint: VideoFingerprint, project_id: str, session: Session) -> Optional[dict]: """Check if video is duplicate of existing one. Returns duplicate info if found.""" video_repo = SQLAlchemyGeneratedVideoRepository(session) existing_videos = video_repo.list_by_project(project_id) for existing in existing_videos: if not existing.video_fingerprint: continue ef = existing.video_fingerprint if fingerprint.md5 == ef.get("md5"): return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0} existing_phashes = ef.get("keyframe_phashes", []) if existing_phashes: total_distance = 0 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: return {"duplicate": True, "duplicate_of": existing.id, "reason": "phash_similar", "similarity": 1.0 - (avg_distance / 64)} return None @celery_app.task(bind=True, max_retries=3, name="worker.check_duplicate") def check_duplicate_task(self: Task, generated_video_id: str) -> dict: """Celery task to check if generated video is a duplicate.""" session = SessionLocal() temp_dir = tempfile.mkdtemp() try: video_repo = SQLAlchemyGeneratedVideoRepository(session) storage_service = get_storage_service() deduplicator = VideoDeduplicator() video = video_repo.get(generated_video_id) if video is None: raise ValueError(f"Generated video {generated_video_id} not found") local_path = os.path.join(temp_dir, f"{generated_video_id}.mp4") storage_key = video.file_url.split("/")[-1] storage_service.download_file(f"workspaces/{video.workspace_id}/projects/{video.project_id}/generated/{generated_video_id}/{generated_video_id}.mp4", local_path) fingerprint = deduplicator.compute_fingerprint(local_path) duplicate_result = deduplicator.check_duplicate(fingerprint, video.project_id, session) video.video_fingerprint = fingerprint.to_dict() if duplicate_result: video.is_duplicate = True video.duplicate_of = duplicate_result["duplicate_of"] else: video.is_duplicate = False video.duplicate_of = None video_repo.update(video) session.commit() logger.info(f"Duplicate check completed for video {generated_video_id}: is_duplicate={video.is_duplicate}") return {"ok": True, "video_id": generated_video_id, "is_duplicate": video.is_duplicate, "duplicate_of": video.duplicate_of, "fingerprint": fingerprint.to_dict()} except Exception as e: logger.error(f"Duplicate check failed for {generated_video_id}: {str(e)}") session.rollback() raise self.retry(exc=e, countdown=60) finally: session.close() import shutil shutil.rmtree(temp_dir, ignore_errors=True)