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

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")

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)

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"]

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)

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

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(),