""" Video Comparator Module İki videodan yüz tespiti yaparak ortak kişileri bulur """ import os import cv2 import numpy as np from typing import List, Dict, Tuple, Optional from dataclasses import dataclass from sklearn.metrics.pairwise import cosine_similarity import logging from datetime import datetime import json logger = logging.getLogger(__name__) @dataclass class ComparisonResult: """Karşılaştırma sonuç verisi""" common_faces: List[Dict] # Her iki videoda da bulunan yüzler only_video1: List[Dict] # Sadece video 1'de bulunan yüzler only_video2: List[Dict] # Sadece video 2'de bulunan yüzler total_unique_video1: int total_unique_video2: int common_count: int similarity_threshold: float metadata: Dict class VideoComparator: """ İki videoyu karşılaştırarak ortak yüzleri bulur Mevcut FaceDetector'ı kullanır """ def __init__(self, face_detector, similarity_threshold: float = 0.6): """ Args: face_detector: FaceDetector instance similarity_threshold: Yüz benzerlik eşiği (0-1 arası, varsayılan 0.6) """ self.detector = face_detector self.similarity_threshold = similarity_threshold self.progress_callback = None def set_progress_callback(self, callback): """Progress callback fonksiyonu ayarla""" self.progress_callback = callback def _update_progress(self, value: float, message: str): """Progress güncelleme""" if self.progress_callback: self.progress_callback(value, message) def compare_videos( self, video1_path: str, video2_path: str, is_video1_url: bool = False, is_video2_url: bool = False ) -> Tuple[ComparisonResult, str, List[str]]: """ İki videoyu karşılaştır ve ortak yüzleri bul Args: video1_path: İlk video yolu veya URL video2_path: İkinci video yolu veya URL is_video1_url: İlk video URL mi? is_video2_url: İkinci video URL mi? Returns: Tuple[ComparisonResult, output_dir, saved_images] """ try: logger.info("Video karşılaştırma başlatılıyor...") start_time = datetime.now() # Video 1'i işle self._update_progress(0, "Video 1 işleniyor...") logger.info(f"Video 1 işleniyor: {video1_path}") output_dir1, faces1, metadata1 = self.detector.detect_faces( video1_path, is_url=is_video1_url ) video1_faces = self._load_face_data(output_dir1, metadata1) logger.info(f"Video 1: {len(video1_faces)} benzersiz yüz bulundu") # Video 2'yi işle self._update_progress(50, "Video 2 işleniyor...") logger.info(f"Video 2 işleniyor: {video2_path}") output_dir2, faces2, metadata2 = self.detector.detect_faces( video2_path, is_url=is_video2_url ) video2_faces = self._load_face_data(output_dir2, metadata2) logger.info(f"Video 2: {len(video2_faces)} benzersiz yüz bulundu") # Yüzleri karşılaştır self._update_progress(80, "Yüzler karşılaştırılıyor...") comparison_result = self._compare_face_sets( video1_faces, video2_faces, metadata1, metadata2 ) # Sonuçları kaydet self._update_progress(90, "Sonuçlar kaydediliyor...") output_dir, saved_images = self._save_comparison_results( comparison_result, output_dir1, output_dir2 ) elapsed_time = (datetime.now() - start_time).total_seconds() logger.info(f"Karşılaştırma tamamlandı: {elapsed_time:.1f} saniye") self._update_progress(100, f"✅ Tamamlandı! {comparison_result.common_count} ortak yüz bulundu") return comparison_result, output_dir, saved_images except Exception as e: logger.error(f"Video karşılaştırma hatası: {e}", exc_info=True) raise def _load_face_data(self, output_dir: str, metadata: Dict) -> List[Dict]: """ Bir videodan tespit edilen yüzleri yükle Returns: List of dicts with 'embedding', 'face_path', 'cluster_id', 'quality_score' """ faces = [] for face_info in metadata['faces']: face_path = os.path.join(output_dir, face_info['face_file']) if not os.path.exists(face_path): logger.warning(f"Yüz dosyası bulunamadı: {face_path}") continue # Yüz görselini oku ve embedding çıkar face_img = cv2.imread(face_path) embedding, _ = self.detector.extract_embeddings(face_img) if embedding is not None: faces.append({ 'embedding': embedding, 'face_path': face_path, 'cluster_id': face_info['cluster_id'], 'quality_score': face_info['quality_score'], 'cluster_size': face_info.get('cluster_size', 1), 'bbox': face_info.get('bbox', []) }) return faces def _compare_face_sets( self, video1_faces: List[Dict], video2_faces: List[Dict], metadata1: Dict, metadata2: Dict ) -> ComparisonResult: """ İki video yüz setini karşılaştır """ common_faces = [] only_video1 = [] only_video2 = [] # Video 1'deki her yüz için Video 2'de eşleşme ara matched_video2_indices = set() for v1_face in video1_faces: best_match = None best_similarity = 0 best_v2_idx = -1 # Video 2'deki tüm yüzlerle karşılaştır for v2_idx, v2_face in enumerate(video2_faces): if v2_idx in matched_video2_indices: continue # Zaten eşleşmiş similarity = cosine_similarity( [v1_face['embedding']], [v2_face['embedding']] )[0][0] if similarity > best_similarity: best_similarity = similarity best_match = v2_face best_v2_idx = v2_idx # Eşik üzerinde eşleşme var mı? if best_similarity >= self.similarity_threshold and best_match: matched_video2_indices.add(best_v2_idx) common_faces.append({ 'video1_face': v1_face, 'video2_face': best_match, 'similarity': float(best_similarity), 'match_id': len(common_faces) }) else: only_video1.append({ 'face': v1_face, 'person_id': v1_face['cluster_id'] }) # Video 2'de eşleşmeyen yüzler for v2_idx, v2_face in enumerate(video2_faces): if v2_idx not in matched_video2_indices: only_video2.append({ 'face': v2_face, 'person_id': v2_face['cluster_id'] }) return ComparisonResult( common_faces=common_faces, only_video1=only_video1, only_video2=only_video2, total_unique_video1=len(video1_faces), total_unique_video2=len(video2_faces), common_count=len(common_faces), similarity_threshold=self.similarity_threshold, metadata={ 'video1': { 'path': metadata1['video_path'], 'duration': metadata1['duration'], 'fps': metadata1['fps'] }, 'video2': { 'path': metadata2['video_path'], 'duration': metadata2['duration'], 'fps': metadata2['fps'] }, 'comparison_time': datetime.now().isoformat() } ) def _save_comparison_results( self, result: ComparisonResult, output_dir1: str, output_dir2: str ) -> Tuple[str, List[str]]: """ Karşılaştırma sonuçlarını kaydet Returns: Tuple[output_directory, list_of_saved_image_paths] """ # Ana çıktı dizini oluştur timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_dir = os.path.join( os.path.dirname(output_dir1), f"comparison_{timestamp}" ) os.makedirs(output_dir, exist_ok=True) # Alt dizinler common_dir = os.path.join(output_dir, "common_faces") only_v1_dir = os.path.join(output_dir, "only_video1") only_v2_dir = os.path.join(output_dir, "only_video2") os.makedirs(common_dir, exist_ok=True) os.makedirs(only_v1_dir, exist_ok=True) os.makedirs(only_v2_dir, exist_ok=True) saved_images = [] # Ortak yüzleri kaydet (yan yana) for idx, match in enumerate(result.common_faces): v1_img = cv2.imread(match['video1_face']['face_path']) v2_img = cv2.imread(match['video2_face']['face_path']) # İki resmi yan yana birleştir combined = np.hstack([v1_img, v2_img]) # Benzerlik skorunu ekle similarity = match['similarity'] cv2.putText( combined, f"Similarity: {similarity:.2%}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2 ) output_path = os.path.join(common_dir, f"match_{idx:03d}.jpg") cv2.imwrite(output_path, combined, [cv2.IMWRITE_JPEG_QUALITY, 95]) saved_images.append(output_path) # Sadece Video 1'de olan yüzler for idx, item in enumerate(result.only_video1): face_img = cv2.imread(item['face']['face_path']) output_path = os.path.join(only_v1_dir, f"person_{idx:03d}.jpg") cv2.imwrite(output_path, face_img, [cv2.IMWRITE_JPEG_QUALITY, 95]) # Sadece Video 2'de olan yüzler for idx, item in enumerate(result.only_video2): face_img = cv2.imread(item['face']['face_path']) output_path = os.path.join(only_v2_dir, f"person_{idx:03d}.jpg") cv2.imwrite(output_path, face_img, [cv2.IMWRITE_JPEG_QUALITY, 95]) # Metadata kaydet metadata = { 'comparison_results': { 'common_count': result.common_count, 'only_video1_count': len(result.only_video1), 'only_video2_count': len(result.only_video2), 'total_video1': result.total_unique_video1, 'total_video2': result.total_unique_video2, 'similarity_threshold': result.similarity_threshold }, 'common_faces': [ { 'match_id': m['match_id'], 'similarity': m['similarity'], 'video1_cluster_id': m['video1_face']['cluster_id'], 'video2_cluster_id': m['video2_face']['cluster_id'] } for m in result.common_faces ], 'metadata': result.metadata } metadata_path = os.path.join(output_dir, 'comparison_metadata.json') with open(metadata_path, 'w', encoding='utf-8') as f: json.dump(metadata, f, indent=2, ensure_ascii=False) logger.info(f"Karşılaştırma sonuçları kaydedildi: {output_dir}") return output_dir, saved_images def generate_report(self, result: ComparisonResult) -> str: """ Karşılaştırma raporu oluştur (Markdown formatında) """ report = f""" # 🔍 Video Karşılaştırma Raporu ## 📊 Özet Bilgiler | Metrik | Video 1 | Video 2 | Ortak | |--------|---------|---------|-------| | **Benzersiz Kişi** | {result.total_unique_video1} | {result.total_unique_video2} | {result.common_count} | | **Sadece Bu Videoda** | {len(result.only_video1)} | {len(result.only_video2)} | - | ### 🎯 Karşılaştırma Parametreleri - **Benzerlik Eşiği**: {result.similarity_threshold:.0%} - **Ortak Kişi Oranı**: {(result.common_count / max(result.total_unique_video1, result.total_unique_video2) * 100):.1f}% --- ## 👥 Ortak Yüzler ({result.common_count} kişi) Her iki videoda da bulunan kişiler: """ for idx, match in enumerate(result.common_faces, 1): similarity = match['similarity'] report += f""" **Kişi {idx}** - 🎯 Benzerlik Skoru: {similarity:.1%} - 📹 Video 1 Cluster ID: {match['video1_face']['cluster_id']} - 📹 Video 2 Cluster ID: {match['video2_face']['cluster_id']} """ if result.only_video1: report += f""" --- ## 📹 Sadece Video 1'de Bulunanlar ({len(result.only_video1)} kişi) """ for idx, item in enumerate(result.only_video1, 1): report += f"- Kişi {idx} (Cluster ID: {item['person_id']})\n" if result.only_video2: report += f""" --- ## 📹 Sadece Video 2'de Bulunanlar ({len(result.only_video2)} kişi) """ for idx, item in enumerate(result.only_video2, 1): report += f"- Kişi {idx} (Cluster ID: {item['person_id']})\n" report += """ --- ## ℹ️ Video Bilgileri """ report += f""" ### Video 1 - **Süre**: {result.metadata['video1']['duration']:.1f} saniye - **FPS**: {result.metadata['video1']['fps']} ### Video 2 - **Süre**: {result.metadata['video2']['duration']:.1f} saniye - **FPS**: {result.metadata['video2']['fps']} --- *Rapor Oluşturulma: {result.metadata['comparison_time']}* """ return report