#!/usr/bin/env python3 """ Smart Search - Akıllı Arama Servisi =================================== Query Expander + Hybrid Search entegrasyonu. Akış: 1. Kullanıcı Türkçe sorgu yazar 2. Query Expander düzeltir ve İngilizce terimleri ekler 3. Query Analyzer marka/model çıkarır 4. Qdrant'ta hybrid search (filter + semantic) 5. Sonuçlar döner Kullanım: from smart_search import SmartSearchService service = SmartSearchService() results = service.search("jcb 330 bom silindir kaçırıyor") """ import os import time from typing import Dict, List, Optional, Any from dataclasses import dataclass, field try: from qdrant_client import QdrantClient from qdrant_client.models import Filter, FieldCondition, MatchText, MatchValue except ImportError: print("❌ qdrant-client yüklü değil: pip install qdrant-client") raise try: from openai import OpenAI except ImportError: print("❌ openai yüklü değil: pip install openai") raise from query_analyzer import QueryAnalyzer, QueryAnalysis from smart_translator import SmartTranslator # ==================== CONFIGURATION ==================== QDRANT_HOST = os.getenv("QDRANT_HOST", "10.10.10.25") QDRANT_PORT = int(os.getenv("QDRANT_PORT", "6333")) COLLECTION_NAME = os.getenv("QDRANT_COLLECTION", "machine_docs") EMBEDDING_MODEL = "text-embedding-3-large" EMBEDDING_DIM = 3072 DEFAULT_LIMIT = 50 # Daha fazla sonuç MAX_LIMIT = 100 @dataclass class SearchResult: """Tek bir arama sonucu""" id: str score: float pdf_filename: str page_number: Optional[int] text: str brand: Optional[str] = None model: Optional[str] = None pdf_path: str = "" def to_dict(self) -> Dict: # 10 satır yaklaşık 1000-1200 karakter text_limit = 1200 return { "id": self.id, "score": self.score, "pdf_filename": self.pdf_filename, "page_number": self.page_number, "text": self.text[:text_limit] + "..." if len(self.text) > text_limit else self.text, "brand": self.brand, "model": self.model, "pdf_path": self.pdf_path } @dataclass class SmartSearchResponse: """Akıllı arama yanıtı""" # Giriş original_query: str # İşleme relevant_terms: Dict[str, str] analysis: QueryAnalysis search_query: str # Qdrant'a gönderilen son sorgu # Sonuçlar results: List[SearchResult] total_found: int # Meta search_time_ms: int filter_applied: bool fallback_used: bool def to_dict(self) -> Dict: return { "original_query": self.original_query, "english_terms": list(self.relevant_terms.values()), "detected": { "brand": self.analysis.brand, "model": self.analysis.model, "doc_type": self.analysis.doc_type }, "search_query": self.search_query, "results": [r.to_dict() for r in self.results], "total_found": self.total_found, "search_time_ms": self.search_time_ms, "filter_applied": self.filter_applied, "fallback_used": self.fallback_used } class SmartSearchService: """ Akıllı Arama Servisi Türkçe giriş -> Sözlük çevirisi -> Marka/Model tespiti -> Hybrid Search """ def __init__( self, qdrant_host: str = QDRANT_HOST, qdrant_port: int = QDRANT_PORT, collection_name: str = COLLECTION_NAME, openai_api_key: str = None, dictionary_path: str = None ): # OpenAI client api_key = openai_api_key or os.getenv("OPENAI_API_KEY") if not api_key: raise ValueError("OPENAI_API_KEY gerekli!") self.openai = OpenAI(api_key=api_key) # Qdrant client self.qdrant = QdrantClient(host=qdrant_host, port=qdrant_port, timeout=30) self.collection_name = collection_name # Smart Translator (Context ve Terimler için) self.translator = SmartTranslator() # Query Analyzer (marka/model tespiti) self.analyzer = QueryAnalyzer() # Bağlantı testi self._verify_connection() def _verify_connection(self): """Qdrant bağlantısını doğrula""" try: info = self.qdrant.get_collection(self.collection_name) print(f"✓ Qdrant bağlantısı OK ({info.points_count:,} vektör)") except Exception as e: raise ConnectionError(f"Qdrant bağlantı hatası: {e}") def _get_embedding(self, text: str) -> List[float]: """Metni vektöre çevir""" response = self.openai.embeddings.create( input=text, model=EMBEDDING_MODEL ) return response.data[0].embedding def _build_filter(self, analysis: QueryAnalysis, pdf_filter: str = None) -> Optional[Filter]: """Qdrant filtresi oluştur""" conditions = [] if analysis.brand: conditions.append( FieldCondition( key="pdf_path", match=MatchText(text=analysis.brand) ) ) if analysis.model: conditions.append( FieldCondition( key="pdf_path", match=MatchText(text=analysis.model) ) ) if pdf_filter: # Native Filename Filtering # MatchText kullanıyoruz (Daha esnek - Token bazlı) # MatchValue çok katı olduğu için karakter farklarında (TR/EN) sorun çıkarabiliyor import os filename = os.path.basename(pdf_filter) conditions.append( FieldCondition( key="pdf_filename", match=MatchValue(value=filename) # Text yerine Value kullanıyoruz (Tam Eşleşme) ) ) if not conditions: return None return Filter(must=conditions) def _search_qdrant( self, query_vector: List[float], filter_obj: Optional[Filter], limit: int ) -> List[Dict]: """Qdrant'ta arama yap""" try: # Qdrant client v1.7+ uses query_points results = self.qdrant.query_points( collection_name=self.collection_name, query=query_vector, query_filter=filter_obj, limit=limit, with_payload=True ) return [ { "id": str(hit.id), "score": hit.score, "payload": hit.payload } for hit in results.points ] except Exception as e: print(f"⚠️ Qdrant arama hatası: {e}") return [] def search( self, query: str, limit: int = DEFAULT_LIMIT, use_filter: bool = True, use_expansion: bool = True, fallback_on_empty: bool = True, pdf_filter: str = None ) -> SmartSearchResponse: """ Akıllı arama yap. Args: query: Kullanıcı sorgusu (Türkçe, hatalar olabilir) limit: Maksimum sonuç sayısı use_filter: Marka/Model filtreleme kullan use_expansion: Sözlük genişletme kullan fallback_on_empty: Filtre sonuç vermezse genel aramaya düş Returns: SmartSearchResponse nesnesi """ start_time = time.time() # 1. Sorgudaki terimleri bul relevant_terms = self.translator.get_relevant_terms(query) # İngilizce terimler english_terms = list(relevant_terms.values()) # 2. Sorguyu analiz et analysis = self.analyzer.analyze(query) # 3. ÇİFT DİLLİ ARAMA - Hem Türkçe hem İngilizce ile ara # Bazı dokümanlar Türkçe (HİDROMEK), bazıları İngilizce (JCB, CAT) # Orijinal Türkçe sorgu (Artık düzeltme yapmıyoruz, orijinali kullanıyoruz) turkish_query = query # İngilizce çeviri (Terimler varsa onları kullan, yoksa genel çeviri yapabiliriz ama şimdilik terim eklemesi yapalım) if False: # KULLANICI İSTEĞİ: Çeviriyi devrenden çıkar, saf sorgu kullan # 1. Yazım hatalarını düzelt ve İngilizce karşılığını bul (LLM) # "hyriloik pumpa" -> "hydraulic pump" english_query = self.translator.correct_terminology(query) print(f"DEBUG SEARCH: Raw='{query}' -> Corrected EN='{english_query}'") # Eğer düzeltme başarısız olduysa veya orijinalle aynıysa # Yine de vektör aramasında şansımız artar english_terms = [english_query] # Vektör üretimi için liste formatı else: print(f"DEBUG SEARCH: Raw Query Used='{query}' (No Translation)") english_query = query english_terms = [query] # 4. Her iki dilde embedding oluştur ve ara filter_obj = None filter_applied = False if use_filter and (analysis.has_filters() or pdf_filter): filter_obj = self._build_filter(analysis, pdf_filter) filter_applied = True # İngilizce Vektör (Her zaman var) search_vector = self._get_embedding(english_query) # 5. Qdrant'ta ara # (Eskiden iki ayrı arama yapıyorduk, şimdi tek ve güçlü bir vektörle arayalım) raw_results = self._search_qdrant(search_vector, filter_obj, limit) fallback_used = False search_query = f"Query: {query} | Vector: {english_query}" # 7. Fallback: Filtre sonuç vermezse genel arama # 7. Fallback: Filtre sonuç vermezse genel arama # AMA: Kullanıcı spesifik bir PDF seçtiyse, genel arama YAPMAMALİYİZ. # Çünkü kullanıcı "Bu PDF'te var mı?" diye soruyor, "Dünyada var mı?" diye sormuyor. if not raw_results and filter_applied and fallback_on_empty and not pdf_filter: print(f"⚠️ Filtre (Marka/Model) sonuç vermedi, genel aramaya düşülüyor...") # Fallback durumunda yine aynı güçlü vektörü kullanalım (search_vector) raw_results = self._search_qdrant(search_vector, None, limit) fallback_used = True filter_applied = False elif not raw_results and pdf_filter: print(f"ℹ️ PDF Filtresi ile sonuç bulunamadı. Genel aramaya DÜŞÜLMÜYOR (Kullanıcı tercihi).") # --------------------------------------------------------- # 📌 PHRASE BOOSTING (Tamlama Güçlendirme) 🚀 # "Hydraulic pump" geçiyorsa, sadece "pump" geçenlerin üzerine çıkar. # --------------------------------------------------------- if raw_results and english_terms: print(f"🔍 Phrase Boosting V2 uygulanıyor. Terimler: {english_terms}") for r in raw_results: text_content = r.get("payload", {}).get("text", "").lower() current_score = r["score"] # Boost mantığı total_boost = 0.0 for term in english_terms: # term örneğin: "travel pump" term = term.lower().strip() if not term: continue # 1. TAM ÖBEK EŞLEŞMESİ (En Değerli) # "travel pump" olarak yan yana geçiyorsa if term in text_content: total_boost += 0.45 # Çok yüksek puan # 2. KELİME BAZLI EŞLEŞME (Parçala ve Yönet) # Eğer terim birden fazla kelimeyse ("travel pump"), kelimelerine ayır if ' ' in term: sub_words = term.split() for i, word in enumerate(sub_words): if word in text_content: # İlk kelime genelde niteleyicidir (Travel, Hydraulic, Fuel) -> Daha değerli if i == 0: total_boost += 0.15 else: # İkinci kelime genelde nesnedir (Pump, Valve, Tank) -> Daha az değerli total_boost += 0.05 # Maksimum boost limiti (abartmamak için) if total_boost > 0.6: total_boost = 0.6 r["score"] += total_boost # Debug log (Sadece ilk 3 sonuç için) # if raw_results.index(r) < 3: # print(f" -> Doc: {r.get('pdf_filename')} | Base: {current_score:.4f} + Boost: {total_boost:.4f} = {r['score']:.4f}") # Puanlar değiştiği için listeyi TEKRAR SIRALA raw_results.sort(key=lambda x: x["score"], reverse=True) # --------------------------------------------------------- # 8. Sonuçları dönüştür results = [] for r in raw_results: payload = r.get("payload", {}) # Qdrant payload alanları: pdf_filename öncelikli olmalı # Bazı kayıtlarda source='viewer_v2' olabiliyor, bu yüzden pdf_filename varsa onu kullan pdf_filename = payload.get("pdf_filename") or payload.get("source", "") if pdf_filename == "viewer_v2": # viewer_v2 ise ve pdf_filename yoksa, path'den çıkar path = payload.get("file_path") or payload.get("pdf_path", "") if path: pdf_filename = os.path.basename(path) result = SearchResult( id=r["id"], score=r["score"], pdf_filename=pdf_filename, page_number=payload.get("page", payload.get("page_number")), text=payload.get("text", payload.get("content", "")), brand=payload.get("brand"), model=payload.get("model"), pdf_path=payload.get("file_path", payload.get("pdf_path", "")) ) results.append(result) elapsed_ms = int((time.time() - start_time) * 1000) return SmartSearchResponse( original_query=query, relevant_terms=relevant_terms, analysis=analysis, search_query=search_query, results=results, total_found=len(results), search_time_ms=elapsed_ms, filter_applied=filter_applied, fallback_used=fallback_used ) def get_page_context( self, pdf_path: str, page_number: int, context_pages: int = 1 ) -> List[SearchResult]: """ Belirli bir sayfanın çevresindeki sayfaları getir. Args: pdf_path: PDF dosya yolu page_number: Sayfa numarası context_pages: Kaç sayfa öncesi ve sonrasını getir Returns: Liste halinde sayfa sonuçları """ # Sayfa aralığını hesapla start_page = max(1, page_number - context_pages) end_page = page_number + context_pages conditions = [ FieldCondition(key="pdf_path", match=MatchText(text=pdf_path)), ] filter_obj = Filter(must=conditions) try: results = self.qdrant.scroll( collection_name=self.collection_name, scroll_filter=filter_obj, limit=100, # Yeterli sayfa almak için with_payload=True, with_vectors=False ) # Sayfa aralığına göre filtrele page_results = [] for point in results[0]: page = point.payload.get("page_number", 0) if start_page <= page <= end_page: page_results.append(SearchResult( id=str(point.id), score=1.0, pdf_filename=point.payload.get("pdf_filename", ""), page_number=page, text=point.payload.get("text", ""), pdf_path=point.payload.get("pdf_path", "") )) # Sayfa numarasına göre sırala page_results.sort(key=lambda x: x.page_number or 0) return page_results except Exception as e: print(f"⚠️ Sayfa context hatası: {e}") return [] def close(self): """Kaynakları temizle""" pass # ==================== CLI INTERFACE ==================== def main(): """CLI arayüzü""" import argparse parser = argparse.ArgumentParser(description="Smart Search Service") parser.add_argument("query", nargs="?", help="Arama sorgusu") parser.add_argument("-l", "--limit", type=int, default=5, help="Sonuç limiti") parser.add_argument("--no-filter", action="store_true", help="Filtrelemeyi devre dışı bırak") parser.add_argument("--no-expand", action="store_true", help="Sözlük genişletmeyi devre dışı bırak") args = parser.parse_args() # API Key kontrolü if not os.getenv("OPENAI_API_KEY"): print("❌ OPENAI_API_KEY ortam değişkeni ayarlanmalı!") print(" export OPENAI_API_KEY='sk-...'") return if not args.query: # İnteraktif mod print("\n" + "=" * 70) print(" 🔍 SMART SEARCH SERVICE") print(" Türkçe sorgu yazın, sistem sözlükten çevirir ve arar") print("=" * 70) service = SmartSearchService() while True: try: query = input("\n📝 Sorgu (q=çıkış): ").strip() if query.lower() == 'q': break if not query: continue response = service.search( query, limit=args.limit, use_filter=not args.no_filter, use_expansion=not args.no_expand ) print(f"\n{'=' * 60}") print(f"🔤 Orijinal: {response.original_query}") print(f"🇬🇧 İngilizce Terimler: {list(response.relevant_terms.values())}") print(f"🏭 Marka: {response.analysis.brand or '-'}") print(f"🔢 Model: {response.analysis.model or '-'}") print(f"🔍 Arama: {response.search_query[:60]}...") print(f"⏱️ Süre: {response.search_time_ms}ms") print(f"🎯 Filter: {'Evet' if response.filter_applied else 'Hayır'}") print(f"\n📋 Sonuçlar ({response.total_found}):") for i, r in enumerate(response.results, 1): print(f" {i}. [{r.score:.3f}] {r.pdf_filename}") print(f" 📄 Sayfa: {r.page_number}") print(f" 📝 {r.text[:100]}...") except KeyboardInterrupt: break except Exception as e: print(f"❌ Hata: {e}") import traceback traceback.print_exc() service.close() print("\n👋 Güle güle!") else: # Tek sorgu modu service = SmartSearchService() response = service.search( args.query, limit=args.limit, use_filter=not args.no_filter, use_expansion=not args.no_expand ) # JSON çıktı import json print(json.dumps(response.to_dict(), ensure_ascii=False, indent=2)) service.close() if __name__ == "__main__": main()