#!/usr/bin/env python3 """ Query Analyzer - Sorgu Analiz Modülü ==================================== Kullanıcının girdiği metinden Marka, Model ve Doküman Tipi bilgisini çıkarır. Kullanım: from query_analyzer import QueryAnalyzer analyzer = QueryAnalyzer() result = analyzer.analyze("JCB 330 hidrolik pompa arızası") # {'brand': 'JCB', 'model': '330', 'doc_type': 'arıza', 'cleaned_query': 'hidrolik pompa arızası'} """ import os import re import json from typing import Dict, List, Optional, Tuple from dataclasses import dataclass @dataclass class QueryAnalysis: """Sorgu analiz sonucu""" original_query: str brand: Optional[str] = None brand_aliases: List[str] = None model: Optional[str] = None doc_type: Optional[str] = None cleaned_query: str = "" confidence: float = 0.0 def to_dict(self) -> Dict: return { "original_query": self.original_query, "brand": self.brand, "model": self.model, "doc_type": self.doc_type, "cleaned_query": self.cleaned_query, "confidence": self.confidence } def has_filters(self) -> bool: """Filtrelenebilir bilgi var mı?""" return self.brand is not None or self.model is not None class QueryAnalyzer: """Sorgu Analizörü - Marka/Model/Tip çıkarma""" def __init__(self, config_path: str = None): """ Args: config_path: brands_models.json dosya yolu """ if config_path is None: script_dir = os.path.dirname(os.path.abspath(__file__)) config_path = os.path.join(script_dir, "brands_models.json") self.config = self._load_config(config_path) self.brands = self.config.get("brands", {}) self.doc_types = self.config.get("document_types", {}) # Alias → Brand lookup tablosu oluştur self.alias_to_brand = {} for brand, data in self.brands.items(): self.alias_to_brand[brand.lower()] = brand for alias in data.get("aliases", []): self.alias_to_brand[alias.lower()] = brand # Tüm modelleri düz listeye çıkar (marka ile birlikte) self.model_to_brand = {} for brand, data in self.brands.items(): for model in data.get("models", []): model_lower = model.lower() if model_lower not in self.model_to_brand: self.model_to_brand[model_lower] = [] self.model_to_brand[model_lower].append(brand) def _load_config(self, path: str) -> Dict: """Config dosyasını yükle""" try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except FileNotFoundError: print(f"⚠️ Config dosyası bulunamadı: {path}") return {"brands": {}, "document_types": {}} except json.JSONDecodeError as e: print(f"⚠️ Config parse hatası: {e}") return {"brands": {}, "document_types": {}} def analyze(self, query: str) -> QueryAnalysis: """ Sorguyu analiz et ve marka/model/tip çıkar. Args: query: Kullanıcı sorgusu Returns: QueryAnalysis nesnesi """ if not query: return QueryAnalysis(original_query="") query_lower = query.lower().strip() tokens = self._tokenize(query_lower) # Sonuç nesnesi result = QueryAnalysis(original_query=query) # 1. Marka tespit et brand, brand_match = self._detect_brand(query_lower, tokens) if brand: result.brand = brand result.brand_aliases = self.brands.get(brand, {}).get("aliases", []) # 2. Model tespit et model, model_match = self._detect_model(query_lower, tokens, brand) if model: result.model = model # Model bulunduysa ama marka bulunamadıysa, modelden markayı çıkar if not result.brand and model.lower() in self.model_to_brand: possible_brands = self.model_to_brand[model.lower()] if len(possible_brands) == 1: result.brand = possible_brands[0] # 3. Doküman tipi tespit et doc_type = self._detect_doc_type(query_lower) if doc_type: result.doc_type = doc_type # 4. Temizlenmiş sorgu oluştur (marka/model çıkarılmış) result.cleaned_query = self._clean_query(query, brand_match, model_match) # 5. Güven skoru hesapla result.confidence = self._calculate_confidence(result) return result def _tokenize(self, text: str) -> List[str]: """Metni token'lara ayır""" # Alfanumerik olmayan karakterleri boşlukla değiştir text = re.sub(r'[^a-z0-9\s\-]', ' ', text) return text.split() def _detect_brand(self, query_lower: str, tokens: List[str]) -> Tuple[Optional[str], Optional[str]]: """ Marka tespit et. Returns: (brand_name, matched_text) tuple """ # Önce tam eşleşme dene (uzun alias'lar önce) sorted_aliases = sorted(self.alias_to_brand.keys(), key=len, reverse=True) for alias in sorted_aliases: # Kelime sınırı ile eşleşme pattern = r'\b' + re.escape(alias) + r'\b' match = re.search(pattern, query_lower) if match: return self.alias_to_brand[alias], match.group() return None, None def _detect_model(self, query_lower: str, tokens: List[str], brand: Optional[str]) -> Tuple[Optional[str], Optional[str]]: """ Model tespit et. Returns: (model_name, matched_text) tuple """ # Eğer marka biliniyorsa, sadece o markanın modellerini ara if brand: brand_models = self.brands.get(brand, {}).get("models", []) for model in sorted(brand_models, key=len, reverse=True): pattern = r'\b' + re.escape(model.lower()) + r'\b' match = re.search(pattern, query_lower) if match: return model, match.group() # Genel model araması # Tipik model pattern'leri: PC200, ZX330, 320D, D6, vb. model_patterns = [ r'\b([a-z]{1,3}\s*\d{2,4}[a-z]?(?:-\d+)?)\b', # PC200, ZX330, 320D r'\b(\d{2,4}[a-z]?(?:-\d+)?)\b', # 330, 200-8 r'\b([a-z]\d{1,2})\b', # D6, D7 ] for pattern in model_patterns: matches = re.findall(pattern, query_lower) for match in matches: match_clean = match.replace(' ', '').upper() # Model veritabanında var mı? if match_clean.lower() in self.model_to_brand: return match_clean, match return None, None def _detect_doc_type(self, query_lower: str) -> Optional[str]: """Doküman tipini tespit et""" for doc_type, keywords in self.doc_types.items(): for keyword in keywords: if keyword.lower() in query_lower: return doc_type return None def _clean_query(self, query: str, brand_match: Optional[str], model_match: Optional[str]) -> str: """Sorgudan marka ve model bilgisini çıkar""" cleaned = query if brand_match: # Case-insensitive replace pattern = re.compile(re.escape(brand_match), re.IGNORECASE) cleaned = pattern.sub('', cleaned) if model_match: pattern = re.compile(re.escape(model_match), re.IGNORECASE) cleaned = pattern.sub('', cleaned) # Fazla boşlukları temizle cleaned = re.sub(r'\s+', ' ', cleaned).strip() return cleaned def _calculate_confidence(self, result: QueryAnalysis) -> float: """Analiz güven skoru hesapla (0.0 - 1.0)""" score = 0.0 if result.brand: score += 0.4 if result.model: score += 0.4 if result.doc_type: score += 0.2 return min(score, 1.0) def extract_from_path(self, file_path: str) -> Dict[str, Optional[str]]: """ Dosya yolundan marka ve model bilgisi çıkar. Örnek path: pdfs/İŞ MAKİNASI GRUBU/JCB/TELESKOPİK YÜKLEYİCİLER/330/330 SERVİS MANUELİ.pdf Returns: {'brand': 'JCB', 'model': '330', 'category': 'TELESKOPİK YÜKLEYİCİLER'} """ result = {"brand": None, "model": None, "category": None} if not file_path: return result # Path'i parçalara ayır parts = file_path.replace('\\', '/').split('/') for i, part in enumerate(parts): part_upper = part.upper() # Marka kontrolü if part_upper in self.brands: result["brand"] = part_upper # Marka'dan sonraki klasör kategori olabilir if i + 1 < len(parts) - 1: next_part = parts[i + 1] if not any(c.isdigit() for c in next_part[:3]): # Model değilse result["category"] = next_part # Kategori'den sonraki klasör model olabilir if i + 2 < len(parts) - 1: model_candidate = parts[i + 2] if any(c.isdigit() for c in model_candidate): result["model"] = model_candidate continue # Alias kontrolü part_lower = part.lower() if part_lower in self.alias_to_brand: result["brand"] = self.alias_to_brand[part_lower] continue # Dosya adından model çıkarmayı dene if not result["model"] and parts: filename = parts[-1] # Dosya adındaki ilk sayısal kısmı model olarak al model_match = re.search(r'\b([A-Z]{0,3}\d{2,4}[A-Z]?(?:-\d+)?)\b', filename, re.IGNORECASE) if model_match: result["model"] = model_match.group(1).upper() return result # ==================== TEST ==================== def test_analyzer(): """Test fonksiyonu""" analyzer = QueryAnalyzer() test_queries = [ "JCB 330 hidrolik pompa arızası", "komatsu pc200 motor yağ kapasitesi", "Caterpillar 320D parça kataloğu", "hidrolik pompa arıza teşhis", "Liebherr LHM 150 servis bakım", "ZF 4WG94 şanzıman tamir", "Volvo EC210 elektrik şeması", "PC400 motor problemi", # Marka yok ama model var ] print("=" * 60) print(" QUERY ANALYZER TEST") print("=" * 60) for query in test_queries: result = analyzer.analyze(query) print(f"\n📝 Sorgu: {query}") print(f" Marka: {result.brand or '-'}") print(f" Model: {result.model or '-'}") print(f" Tip: {result.doc_type or '-'}") print(f" Temiz: {result.cleaned_query}") print(f" Güven: {result.confidence:.0%}") # Path extraction testi print("\n" + "=" * 60) print(" PATH EXTRACTION TEST") print("=" * 60) test_paths = [ "pdfs/İŞ MAKİNASI GRUBU/JCB/TELESKOPİK YÜKLEYİCİLER/330/330 SERVİS MANUELİ.pdf", "pdfs/İŞ MAKİNASI GRUBU/KOMATSU/EKSKAVATÖR/PC200-8/PC200-8 PARÇA KATALOĞU.pdf", "pdfs/FORKLİFT GRUBU/HYUNDAI/HDF50/HDF50 SERVİS MANUELİ.pdf", ] for path in test_paths: result = analyzer.extract_from_path(path) print(f"\n📁 Path: ...{path[-50:]}") print(f" Marka: {result['brand'] or '-'}") print(f" Model: {result['model'] or '-'}") print(f" Kategori: {result['category'] or '-'}") if __name__ == "__main__": test_analyzer()