""" VideoDetection Pipeline Orchestrator. Điều phối toàn bộ luồng Ingestion -> Feature Extraction -> Indexing -> Alignment. """ from __future__ import annotations import datetime import logging from pathlib import Path from typing import Dict, List, Optional, Tuple import numpy as np from .alignment.matcher import MatchSegment, TemporalMatcher from .audio.neuralfp_extractor import NeuralFPExtractor from .core.config import CONFIG from .core.paths import DATA_DIR, TARGET_DIR, TEMP_DIR, clean_temp_dir from .indexing.database import OwnerDatabase, VideoRecord from .indexing.feature_store import CachedVideoFeatures, FeatureStore, SUPPORTED_EXTENSIONS from .indexing.sqlite_db import VideoDatabase from .ingestion.audio_loader import AudioChunkExtractor from .ingestion.video_loader import VideoClipExtractor, get_video_metadata from .ingestion.youtube_downloader import ( DEFAULT_SECTION_1H, YouTubeDownloadResult, download_youtube_video as dl_yt_video, ) from .visual.vjepa_extractor import VJEPAExtractor logger = logging.getLogger(__name__) class VideoDetectionPipeline: def __init__(self) -> None: self.db = OwnerDatabase(data_dir=DATA_DIR) # Trích xuất dữ liệu thô self.clip_extractor = VideoClipExtractor( clip_frames=CONFIG.visual.get("frames_per_clip", 16), target_size=CONFIG.visual.get("frame_size", 224), stride_sec=CONFIG.visual.get("clip_duration_sec", 1.0), ) self.chunk_extractor = AudioChunkExtractor( sample_rate=CONFIG.audio.get("sample_rate", 16000), hop_duration_sec=CONFIG.audio.get("hop_duration_sec", 0.5), ) # AI Models device = CONFIG.hardware.get("device", "auto") precision = CONFIG.hardware.get("precision", "fp16") self.visual_extractor = VJEPAExtractor( model_name=CONFIG.visual.get("model_name", "vjepa2_vitl"), device=device, precision=precision, embedding_dim=1024, ) self.audio_extractor = NeuralFPExtractor( sample_rate=CONFIG.audio.get("sample_rate", 16000), fingerprint_dim=CONFIG.audio.get("fingerprint_dim", 128), device=device, ) # Feature Store tự động cache và đồng bộ self.feature_store = FeatureStore( clip_extractor=self.clip_extractor, chunk_extractor=self.chunk_extractor, visual_extractor=self.visual_extractor, audio_extractor=self.audio_extractor, ) # Temporal Matcher self.matcher = TemporalMatcher( db=self.db, visual_threshold=float(CONFIG.visual.get("similarity_threshold", 0.78)), audio_threshold=float(CONFIG.audio.get("similarity_threshold", 0.82)), min_match_duration_sec=float(CONFIG.alignment.get("min_match_duration_sec", 3.0)), fusion_weight_visual=float(CONFIG.alignment.get("fusion_weight_visual", 0.65)), fusion_weight_audio=float(CONFIG.alignment.get("fusion_weight_audio", 0.35)), ) # CSDL SQLite quản lý metadata và vectors nhị phân theo tên video self.sql_db = VideoDatabase(DATA_DIR / "videodetection.db") def extract_features( self, video_path: Path, progress_callback=None, ) -> Tuple[List[Tuple[float, np.ndarray]], List[Tuple[float, np.ndarray]]]: """ Trích xuất đặc trưng Visual (V-JEPA 2) và Audio (NeuralFP). """ visual_features: List[Tuple[float, np.ndarray]] = [] audio_features: List[Tuple[float, np.ndarray]] = [] if progress_callback: progress_callback(10, f"Đang trích xuất video clips: {video_path.name}") # 1. Trích xuất clips hình ảnh for t_sec, clip in self.clip_extractor.extract_clips(video_path): emb = self.visual_extractor.extract(clip) visual_features.append((t_sec, emb)) if progress_callback: progress_callback(50, f"Đang trích xuất audio fingerprints: {video_path.name}") # 2. Trích xuất audio chunks for t_sec, chunk in self.chunk_extractor.extract_chunks(video_path): fp = self.audio_extractor.extract(chunk) audio_features.append((t_sec, fp)) if progress_callback: progress_callback(80, "Hoàn tất trích xuất đặc trưng") return visual_features, audio_features def index_reference_video( self, video_path: Path, progress_callback=None, ) -> VideoRecord: """ Nạp một video của chủ sở hữu vào cơ sở dữ liệu Data/. """ meta = get_video_metadata(video_path) video_id = video_path.stem vis_feats, aud_feats = self.extract_features(video_path, progress_callback) now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") record = VideoRecord( video_id=video_id, filename=video_path.name, rel_path=str(video_path), duration_sec=meta.duration_sec, visual_clips_count=len(vis_feats), audio_chunks_count=len(aud_feats), indexed_at=now_str, ) self.db.add_reference(record, vis_feats, aud_feats) self.db.save() if CONFIG.ui.get("auto_clean_temp", True): clean_temp_dir() if progress_callback: progress_callback(100, f"Đã lưu video '{video_path.name}' vào cơ sở dữ liệu.") return record def scan_target_video( self, target_video_path: Path, progress_callback=None, ) -> List[MatchSegment]: """ Quét một video mục tiêu để tìm kiếm đoạn trùng lặp với kho dữ liệu trong Data/. """ if self.db.visual_index.total == 0 and self.db.audio_index.total == 0: logger.warning("Cơ sở dữ liệu Data/ đang rỗng. Chưa có video chủ sở hữu nào được nạp.") return [] vis_feats, aud_feats = self.extract_features(target_video_path, progress_callback) if progress_callback: progress_callback(85, "Đang đối soát & căn chỉnh thời gian (Temporal Alignment)...") matches = self.matcher.match(vis_feats, aud_feats) if CONFIG.ui.get("auto_clean_temp", True): clean_temp_dir() if progress_callback: progress_callback(100, f"Hoàn tất quét {target_video_path.name}: tìm thấy {len(matches)} đoạn trùng lặp.") return matches def scan_all_targets(self, progress_callback=None) -> Dict[str, List[MatchSegment]]: """ Quét tất cả các file video nằm trong thư mục Target/. """ valid_extensions = {".mp4", ".mkv", ".avi", ".mov", ".flv", ".webm", ".ts"} target_files = [ f for f in TARGET_DIR.iterdir() if f.is_file() and f.suffix.lower() in valid_extensions ] results: Dict[str, List[MatchSegment]] = {} total = len(target_files) if total == 0: logger.info("Thư mục Target/ hiện không có video nào cần quét.") return results for idx, file_path in enumerate(target_files): def file_progress(pct, msg): if progress_callback: overall_pct = int(((idx * 100) + pct) / total) progress_callback(overall_pct, f"[{idx+1}/{total}] {msg}") matches = self.scan_target_video(file_path, file_progress) results[file_path.name] = matches return results # ==================== CƠ CHẾ TỰ ĐỘNG ĐỒNG BỘ & CACHE MỚI ==================== def sync_data_repository(self, progress_callback=None) -> Tuple[int, int, List[CachedVideoFeatures]]: """ Tự động quét và đồng bộ các video gốc người dùng copy vào thư mục Data/. Trích xuất đặc trưng V-JEPA 2 + NeuralFP và lưu vào SQLite Database theo tên video. """ new_cnt, cache_cnt, all_feats = self.feature_store.sync_directory(DATA_DIR, progress_callback) # Đồng bộ vào CSDL SQLite Data/videodetection.db for feat in all_feats: try: self.sql_db.upsert_video(Path(feat.video_path), category="reference", features=feat) except Exception as e: logger.error(f"Lỗi lưu video {feat.filename} vào SQLite: {e}") return new_cnt, cache_cnt, all_feats def sync_target_repository(self, progress_callback=None) -> Tuple[int, int, List[CachedVideoFeatures]]: """ Tự động quét và đồng bộ các video mục tiêu người dùng copy vào thư mục Target/. """ new_cnt, cache_cnt, all_feats = self.feature_store.sync_directory(TARGET_DIR, progress_callback) for feat in all_feats: try: self.sql_db.upsert_video(Path(feat.video_path), category="target", features=feat) except Exception as e: logger.error(f"Lỗi lưu video {feat.filename} vào SQLite: {e}") return new_cnt, cache_cnt, all_feats def get_cached_references(self) -> List[CachedVideoFeatures]: """Lấy tất cả các video đã cache trong Data/ (ưu tiên đọc từ SQLite/Cache).""" refs: List[CachedVideoFeatures] = [] for f in DATA_DIR.iterdir(): if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS: feats = self.sql_db.get_features_by_name(f.name) or self.feature_store.load_features(f) if feats: refs.append(feats) return refs def get_cached_targets(self) -> List[CachedVideoFeatures]: """Lấy tất cả các video đã cache trong Target/.""" targets: List[CachedVideoFeatures] = [] for f in TARGET_DIR.iterdir(): if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS: feats = self.sql_db.get_features_by_name(f.name) or self.feature_store.load_features(f) if feats: targets.append(feats) return targets def compare_target_with_references( self, target_video_path: Path, selected_reference_ids: Optional[List[str]] = None, progress_callback=None, ) -> List[MatchSegment]: """ Thực hiện so sánh linh hoạt: - 1 video target vs Cả kho Data/ (khi selected_reference_ids is None hoặc rỗng) - 1 video target vs 1 vài video tham chiếu cụ thể do người dùng tích chọn Lưu kết quả đối soát vào bảng comparison_history trong CSDL SQLite. """ if progress_callback: progress_callback(10, f"Đang nạp đặc trưng target: {target_video_path.name}") target_feats = self.sql_db.get_features_by_name(target_video_path.name) or self.feature_store.load_features(target_video_path) if not target_feats or not self.feature_store.is_cached_and_valid(target_video_path): target_feats = self.feature_store.process_and_cache_video(target_video_path, progress_callback) self.sql_db.upsert_video(target_video_path, category="target", features=target_feats) all_refs = self.get_cached_references() if not all_refs: _, _, all_refs = self.sync_data_repository(progress_callback) if not all_refs: logger.warning("Kho Data/ chưa có video tham chiếu nào.") return [] if selected_reference_ids: refs_to_compare = [r for r in all_refs if r.video_id in selected_reference_ids or r.filename in selected_reference_ids] else: refs_to_compare = all_refs if progress_callback: progress_callback(70, f"Đang đối soát {target_video_path.name} với {len(refs_to_compare)} video tham chiếu...") matches = self.matcher.match_cached(target_feats, refs_to_compare) # Ghi nhận kết quả vào SQLite database history if matches: try: self.sql_db.log_match_history(target_video_path.name, matches) except Exception as e: logger.error(f"Lỗi ghi lịch sử đối soát vào SQLite: {e}") if progress_callback: progress_callback(100, f"Đối soát hoàn tất: phát hiện {len(matches)} đoạn trùng lặp.") return matches def download_youtube_video( self, url_or_id: str, section: Optional[str] = DEFAULT_SECTION_1H, dest: str | Path = "target", auto_sync: bool = False, progress_callback=None, ) -> YouTubeDownloadResult: """ Tải video hoặc 1 đoạn chỉ định (mặc định 1h đầu) từ YouTube vào thư mục Target/ hoặc Data/. Hỗ trợ tự động đồng bộ (auto_sync) để trích xuất V-JEPA 2 & NeuralFP ngay sau khi tải. """ res = dl_yt_video( url_or_id=url_or_id, section=section, dest=dest, progress_callback=progress_callback, ) if res.success and auto_sync: dest_lower = str(dest).lower() if "data" in dest_lower: if progress_callback: progress_callback(90, "Đang đồng bộ và trích xuất đặc trưng vào Data/...") self.sync_data_repository(progress_callback) else: if progress_callback: progress_callback(90, "Đang đồng bộ và trích xuất đặc trưng vào Target/...") self.sync_target_repository(progress_callback) return res