From 82a71b349881d47beee23bbd2956d9aa4aaa9b8a Mon Sep 17 00:00:00 2001 From: CI Test Date: Fri, 26 Jun 2026 21:23:39 +0800 Subject: [PATCH] chore: squash merge feature/voice-dedup into develop (resolve conflicts) - Add voice extraction and deduplication functionality - Add dedup.py for video deduplication - Add voice_extraction.py task for worker - Resolve CI/CD and deploy workflow conflicts with develop version - Keep new voice dedup features from feature branch --- .../versions/007_add_video_dedup_fields.py | 38 ++++ apps/worker/video_processing/dedup.py | 180 ++++++++++++++++++ apps/worker/worker_app/tasks/__init__.py | 3 +- .../worker_app/tasks/voice_extraction.py | 122 ++++++++++++ .../generated_video_repository.py | 9 + packages/domain/generated_video.py | 13 +- 6 files changed, 359 insertions(+), 6 deletions(-) create mode 100644 alembic/versions/007_add_video_dedup_fields.py create mode 100644 apps/worker/video_processing/dedup.py create mode 100644 apps/worker/worker_app/tasks/voice_extraction.py diff --git a/alembic/versions/007_add_video_dedup_fields.py b/alembic/versions/007_add_video_dedup_fields.py new file mode 100644 index 000000000..6c28e116e --- /dev/null +++ b/alembic/versions/007_add_video_dedup_fields.py @@ -0,0 +1,38 @@ +"""Add video fingerprint and duplicate detection fields to generated_videos table. + +Revision ID: 007 +Revises: 006 +Create Date: 2024-06-26 +""" +import sqlalchemy as sa + +from alembic import op + +revision = "007" +down_revision = "006" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Add video_fingerprint column as JSON text + op.add_column( + "generated_videos", + sa.Column("video_fingerprint", sa.Text(), nullable=True) + ) + # Add is_duplicate column + op.add_column( + "generated_videos", + sa.Column("is_duplicate", sa.Boolean(), nullable=False, server_default="false") + ) + # Add duplicate_of column for tracking original video + op.add_column( + "generated_videos", + sa.Column("duplicate_of", sa.String(32), nullable=True) + ) + + +def downgrade() -> None: + op.drop_column("generated_videos", "duplicate_of") + op.drop_column("generated_videos", "is_duplicate") + op.drop_column("generated_videos", "video_fingerprint") diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py new file mode 100644 index 000000000..492ef0995 --- /dev/null +++ b/apps/worker/video_processing/dedup.py @@ -0,0 +1,180 @@ +"""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.""" + image = cv2.resize(image, (hash_size * 4, hash_size * 4)) + gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) + avg = gray.mean() + diff = (gray > avg).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) diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py index 882e56981..83cab8b39 100644 --- a/apps/worker/worker_app/tasks/__init__.py +++ b/apps/worker/worker_app/tasks/__init__.py @@ -4,5 +4,6 @@ from .classification import classify_asset from .generation import generate_video from .health import healthcheck from .ingest import ingest_asset +from .voice_extraction import extract_voice_task, extract_background_task -__all__ = ["classify_asset", "generate_video", "healthcheck", "ingest_asset"] +__all__ = ["classify_asset", "generate_video", "healthcheck", "ingest_asset", "extract_voice_task", "extract_background_task"] diff --git a/apps/worker/worker_app/tasks/voice_extraction.py b/apps/worker/worker_app/tasks/voice_extraction.py new file mode 100644 index 000000000..552c49173 --- /dev/null +++ b/apps/worker/worker_app/tasks/voice_extraction.py @@ -0,0 +1,122 @@ +"""Voice extraction tasks - extract voice tracks and background music from videos.""" +import json +import logging +import os +import subprocess +import tempfile +from typing import Optional + +from celery import Task +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository +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 .celery_app import celery_app + +logger = logging.getLogger(__name__) +settings = get_settings() +if SessionLocal is None: + build_session_factory(settings.database_url) + + +class VoiceExtractor: + """Extract voice tracks and background music from videos using FFmpeg.""" + + @staticmethod + def _run_ffmpeg(cmd: list[str]) -> subprocess.CompletedProcess: + """Run FFmpeg command and return result.""" + logger.info(f"Running FFmpeg: {chr(39).join(cmd)}") + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + logger.error(f"FFmpeg error: {result.stderr}") + raise RuntimeError(f"FFmpeg failed: {result.stderr}") + return result + + def extract_voice(self, input_path: str, output_path: str, highpass: int = 200, bandpass_freq: int = 300, bandpass_width: int = 3000, noise_reduction: int = 20) -> str: + """Extract voice track from video using FFmpeg.""" + os.makedirs(os.path.dirname(output_path), exist_ok=True) + cmd = ["ffmpeg", "-y", "-i", input_path, "-af", f"highpass=f={highpass},afftdn=bn={noise_reduction},bandpass=f={bandpass_freq}:width_type=h:width={bandpass_width},loudnorm", "-vn", "-acodec", "libmp3lame", "-q:a", "2", output_path] + self._run_ffmpeg(cmd) + logger.info(f"Voice extracted to: {output_path}") + return output_path + + def extract_background(self, input_path: str, output_path: str, lowpass: int = 200) -> str: + """Extract background music from video.""" + os.makedirs(os.path.dirname(output_path), exist_ok=True) + cmd = ["ffmpeg", "-y", "-i", input_path, "-af", f"lowpass=f={lowpass},loudnorm", "-vn", "-acodec", "libmp3lame", "-q:a", "2", output_path] + self._run_ffmpeg(cmd) + logger.info(f"Background extracted to: {output_path}") + return output_path + + +@celery_app.task(bind=True, max_retries=3, name="worker.extract_voice") +def extract_voice_task(self: Task, asset_id: str) -> dict: + session = SessionLocal() + temp_dir = tempfile.mkdtemp() + try: + asset_repo = SQLAlchemyAssetRepository(session) + storage_service = get_storage_service() + extractor = VoiceExtractor() + asset = asset_repo.get(asset_id) + if asset is None: + raise ValueError(f"Asset {asset_id} not found") + local_video_path = os.path.join(temp_dir, f"{asset_id}.mp4") + storage_service.download_file(asset.storage_key, local_video_path) + voice_output_path = os.path.join(temp_dir, f"{asset_id}_voice.mp3") + extractor.extract_voice(local_video_path, voice_output_path) + voice_storage_key = f"assets/{asset_id}/voice.mp3" + storage_service.upload_file(voice_output_path, voice_storage_key) + voice_url = storage_service.get_url(voice_storage_key) + if asset.metadata is None: + asset.metadata = {} + asset.metadata["voice_url"] = voice_url + asset_repo.update(asset) + session.commit() + logger.info(f"Voice extraction completed for asset {asset_id}: {voice_url}") + return {"ok": True, "asset_id": asset_id, "voice_url": voice_url} + except Exception as e: + logger.error(f"Voice extraction failed for {asset_id}: {str(e)}") + session.rollback() + raise self.retry(exc=e, countdown=60) + finally: + session.close() + import shutil + shutil.rmtree(temp_dir, ignore_errors=True) + + +@celery_app.task(bind=True, max_retries=3, name="worker.extract_background") +def extract_background_task(self: Task, asset_id: str) -> dict: + session = SessionLocal() + temp_dir = tempfile.mkdtemp() + try: + asset_repo = SQLAlchemyAssetRepository(session) + storage_service = get_storage_service() + extractor = VoiceExtractor() + asset = asset_repo.get(asset_id) + if asset is None: + raise ValueError(f"Asset {asset_id} not found") + local_video_path = os.path.join(temp_dir, f"{asset_id}.mp4") + storage_service.download_file(asset.storage_key, local_video_path) + bg_output_path = os.path.join(temp_dir, f"{asset_id}_background.mp3") + extractor.extract_background(local_video_path, bg_output_path) + bg_storage_key = f"assets/{asset_id}/background.mp3" + storage_service.upload_file(bg_output_path, bg_storage_key) + bg_url = storage_service.get_url(bg_storage_key) + if asset.metadata is None: + asset.metadata = {} + asset.metadata["background_url"] = bg_url + asset_repo.update(asset) + session.commit() + logger.info(f"Background extraction completed for asset {asset_id}: {bg_url}") + return {"ok": True, "asset_id": asset_id, "background_url": bg_url} + except Exception as e: + logger.error(f"Background extraction failed for {asset_id}: {str(e)}") + session.rollback() + raise self.retry(exc=e, countdown=60) + finally: + session.close() + import shutil + shutil.rmtree(temp_dir, ignore_errors=True) diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py index bfe1056ab..dfe6d4042 100644 --- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py +++ b/packages/adapters/sqlalchemy_impl/generated_video_repository.py @@ -27,6 +27,9 @@ class SQLAlchemyGeneratedVideoRepository: status=video.status, review_status=video.review_status, generation_params=json.dumps(video.generation_params, ensure_ascii=False), + video_fingerprint=json.dumps(video.video_fingerprint) if video.video_fingerprint else None, + is_duplicate=video.is_duplicate, + duplicate_of=video.duplicate_of, generated_at=video.generated_at, created_at=video.created_at, ) @@ -54,6 +57,9 @@ class SQLAlchemyGeneratedVideoRepository: 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, ) @@ -65,6 +71,9 @@ class SQLAlchemyGeneratedVideoRepository: model.status = video.status model.review_status = video.review_status model.generation_params = json.dumps(video.generation_params, ensure_ascii=False) + 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 self.session.add(model) self.session.commit() return video diff --git a/packages/domain/generated_video.py b/packages/domain/generated_video.py index aa807158d..427beac0b 100644 --- a/packages/domain/generated_video.py +++ b/packages/domain/generated_video.py @@ -23,6 +23,9 @@ class GeneratedVideo: status: str = "completed" review_status: str = "pending_review" generation_params: dict[str, Any] = field(default_factory=dict) + video_fingerprint: dict[str, Any] | None = None + is_duplicate: bool = False + duplicate_of: str | None = None generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @@ -44,15 +47,15 @@ class GeneratedVideo: generation_params: dict[str, Any] | None = None, ) -> "GeneratedVideo": if not workspace_id.strip(): - raise ValueError("workspace_id 不能为空") + raise ValueError("workspace_id cannot be empty") if not project_id.strip(): - raise ValueError("project_id 不能为空") + raise ValueError("project_id cannot be empty") if not generation_task_id.strip(): - raise ValueError("generation_task_id 不能为空") + raise ValueError("generation_task_id cannot be empty") if not name.strip(): - raise ValueError("name 不能为空") + raise ValueError("name cannot be empty") if not file_url.strip(): - raise ValueError("file_url 不能为空") + raise ValueError("file_url cannot be empty") return cls( id=uuid4().hex, workspace_id=workspace_id.strip(),