Create services/search.py

services/search.py

@@ -0,0 +1,172 @@
+import numpy as np
+import json
+import sqlite3
+from pathlib import Path
+from typing import List, Dict, Tuple
+import faiss
+from loguru import logger
+
+class VectorSearchService:
+    def __init__(self, db_path="database/identities.db", vector_dim=512):
+        self.db_path = db_path
+        self.vector_dim = vector_dim
+        self.index = None
+        self.id_to_index = {}
+        self.index_to_id = {}
+        
+        # تهيئة FAISS index
+        self.init_index()
+        
+    def init_index(self):
+        """تهيئة فهرس FAISS للبحث السريع"""
+        try:
+            # استخدام IndexFlatIP (Inner Product) للتشابه
+            self.index = faiss.IndexFlatIP(self.vector_dim)
+            logger.info("✅ FAISS index initialized successfully")
+            self.load_existing_embeddings()
+        except Exception as e:
+            logger.error(f"❌ Failed to initialize FAISS: {e}")
+            self.index = None
+    
+    def load_existing_embeddings(self):
+        """تحميل المتجهات الموجودة من قاعدة البيانات"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
+            cursor.execute("SELECT id, embedding FROM identities")
+            results = cursor.fetchall()
+            conn.close()
+            
+            vectors = []
+            for idx, (identity_id, emb_json) in enumerate(results):
+                embedding = np.array(json.loads(emb_json)).astype(np.float32)
+                vectors.append(embedding)
+                self.id_to_index[identity_id] = idx
+                self.index_to_id[idx] = identity_id
+            
+            if vectors:
+                vectors_array = np.vstack(vectors)
+                self.index.add(vectors_array)
+                logger.info(f"✅ Loaded {len(vectors)} existing embeddings")
+                
+        except Exception as e:
+            logger.warning(f"⚠️ No existing embeddings loaded: {e}")
+    
+    def add_embedding(self, identity_id: str, embedding: np.ndarray):
+        """إضافة متجه جديد إلى الفهرس"""
+        if self.index is None:
+            logger.warning("⚠️ FAISS index not available, using fallback")
+            return
+        
+        embedding = embedding.astype(np.float32).reshape(1, -1)
+        
+        # إضافة إلى FAISS
+        idx = self.index.ntotal
+        self.index.add(embedding)
+        
+        # تحديث التعيينات
+        self.id_to_index[identity_id] = idx
+        self.index_to_id[idx] = identity_id
+        
+        logger.info(f"✅ Added embedding for {identity_id} at index {idx}")
+    
+    def search(self, query_embedding: np.ndarray, k: int = 5, threshold: float = 0.6) -> List[Dict]:
+        """البحث عن أكثر الوجوه تشابهاً"""
+        if self.index is None or self.index.ntotal == 0:
+            logger.warning("⚠️ No embeddings in index, returning empty results")
+            return []
+        
+        query_embedding = query_embedding.astype(np.float32).reshape(1, -1)
+        
+        # البحث في FAISS
+        similarities, indices = self.index.search(query_embedding, min(k, self.index.ntotal))
+        
+        results = []
+        for similarity, idx in zip(similarities[0], indices[0]):
+            if idx == -1:
+                continue
+                
+            identity_id = self.index_to_id.get(int(idx))
+            if identity_id and similarity >= threshold:
+                # جلب معلومات إضافية من قاعدة البيانات
+                conn = sqlite3.connect(self.db_path)
+                cursor = conn.cursor()
+                cursor.execute("SELECT name, metadata FROM identities WHERE id = ?", (identity_id,))
+                row = cursor.fetchone()
+                conn.close()
+                
+                if row:
+                    results.append({
+                        'identity_id': identity_id,
+                        'name': row[0],
+                        'similarity': float(similarity),
+                        'metadata': json.loads(row[1]) if row[1] else {}
+                    })
+        
+        return results
+    
+    def remove_embedding(self, identity_id: str):
+        """حذف متجه من الفهرس"""
+        if identity_id not in self.id_to_index:
+            logger.warning(f"⚠️ Identity {identity_id} not found in index")
+            return
+        
+        # FAISS لا يدعم الحذف المباشر، نحتاج لإعادة بناء الفهرس
+        logger.info("🔄 Rebuilding index after removal...")
+        self.rebuild_index()
+    
+    def rebuild_index(self):
+        """إعادة بناء الفهرس بالكامل"""
+        self.init_index()
+        self.load_existing_embeddings()
+        logger.info("✅ Index rebuilt successfully")
+    
+    def get_stats(self) -> Dict:
+        """إحصائيات عن الفهرس"""
+        return {
+            'total_vectors': self.index.ntotal if self.index else 0,
+            'vector_dimension': self.vector_dim,
+            'index_type': type(self.index).__name__ if self.index else 'None',
+            'is_ready': self.index is not None and self.index.ntotal > 0
+        }
+
+class SimpleVectorSearch:
+    """نسخة بسيطة للبحث عندما لا يتوفر FAISS"""
+    
+    def __init__(self, db_path="database/identities.db"):
+        self.db_path = db_path
+        
+    def search(self, query_embedding: np.ndarray, k: int = 5, threshold: float = 0.6) -> List[Dict]:
+        """بحث خطي بسيط"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
+            cursor.execute("SELECT id, name, embedding, metadata FROM identities")
+            results = cursor.fetchall()
+            conn.close()
+            
+            similarities = []
+            for identity_id, name, emb_json, metadata in results:
+                db_embedding = np.array(json.loads(emb_json))
+                similarity = np.dot(query_embedding, db_embedding)
+                similarities.append((identity_id, name, similarity, metadata))
+            
+            # ترتيب النتائج
+            similarities.sort(key=lambda x: x[2], reverse=True)
+            
+            # تصفية حسب العتبة
+            results_list = []
+            for identity_id, name, similarity, metadata in similarities[:k]:
+                if similarity >= threshold:
+                    results_list.append({
+                        'identity_id': identity_id,
+                        'name': name,
+                        'similarity': float(similarity),
+                        'metadata': json.loads(metadata) if metadata else {}
+                    })
+            
+            return results_list
+            
+        except Exception as e:
+            logger.error(f"❌ Simple search failed: {e}")
+            return []