#!/usr/bin/env python3 """ Usta Assistant - Web Arayüzü v3 =============================== Marka/Model seçimli, PDF görüntülemeli, DeepL destekli web uygulaması. Çalıştırma: python3 usta_web.py Erişim: http://localhost:5001 veya http://yedekparcaservis.com.tr/usta/ """ import os import json from flask import Flask, render_template_string, request, jsonify # Modülleri import et # Modülleri import et from smart_search import SmartSearchService from vision_analyzer import VisionAnalyzer from smart_translator import SmartTranslator from analytics_db import AnalyticsDB app = Flask(__name__) # Global servisler search_service = None vision_analyzer = None translator = None analytics_db = None # API Keys OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "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") DEEPL_API_KEY = os.getenv("DEEPL_API_KEY", "b121dc7e-8e98-427f-8984-54c4d4f0851e") # Marka/Model verileri BRANDS_FILE = os.path.join(os.path.dirname(__file__), "brands_models.json") # HTML Template HTML_TEMPLATE = """ 👷 Usta Assistant

👷 Usta Assistant

İş Makinası Servis Manueli Asistanı

🔧 Makine ve Doküman Seçimi

1️⃣ Marka

2️⃣ Model

3️⃣ Doküman

📋 Sonuçlar

Arama sonuçları burada görüntülenecek

📄 PDF Görüntüleyici

Sayfa -

👈 Soldaki sonuçlardan birine tıklayın

📚 Teknik Sözlük

➕ Yeni Terim Öner

❓ Bu sayfa hakkında soru sorun

🤖 AI Cevabı

🔤 Çeviri Karşılaştırması

📕 Sözlük: -

🌐 DeepL: -

🔀 Hibrit: -

""" def init_services(): """Servisleri başlat""" global search_service, vision_analyzer, translator, analytics_db if search_service is None: search_service = SmartSearchService(openai_api_key=OPENAI_API_KEY) if vision_analyzer is None: # Dictionary path'i SmartTranslator kullanacak vision_analyzer = VisionAnalyzer(openai_api_key=OPENAI_API_KEY) if translator is None: translator = SmartTranslator(api_key=OPENAI_API_KEY) if analytics_db is None: analytics_db = AnalyticsDB() def load_brands(): """Marka/Model verilerini yükle""" try: with open(BRANDS_FILE, 'r', encoding='utf-8') as f: return json.load(f) except: return {"brands": {}, "document_types": {}} @app.route('/usta/') def index(): """Ana sayfa""" return render_template_string(HTML_TEMPLATE) @app.route('/usta/api/brands') def api_brands(): """Marka listesi API - JSON katalogdan""" try: catalog_path = os.path.join(os.path.dirname(__file__), 'pdf_catalog.json') if os.path.exists(catalog_path): with open(catalog_path, 'r', encoding='utf-8') as f: catalog = json.load(f) # Katalogdaki markaları ve modellerini döndür brands_data = {} for brand_name, brand_data in catalog.get('brands', {}).items(): models = list(brand_data.get('models', {}).keys()) brands_data[brand_name] = sorted(models) return jsonify(brands_data) else: # Fallback - eski dosyadan oku data = load_brands() return jsonify(data) except Exception as e: return jsonify({'error': str(e)}) @app.route('/usta/api/pdfs', methods=['POST']) def api_pdfs(): """Marka/Model için PDF listesi API - JSON katalogdan""" try: data = request.get_json() brand = data.get('brand', '').upper() model = data.get('model', '').upper() if not brand: return jsonify({'success': False, 'error': 'Marka seçilmeli'}) # JSON katalog yükle catalog_path = os.path.join(os.path.dirname(__file__), 'pdf_catalog.json') if not os.path.exists(catalog_path): return jsonify({'success': False, 'error': 'PDF katalogu bulunamadı. build_pdf_catalog.py çalıştırın.'}) with open(catalog_path, 'r', encoding='utf-8') as f: catalog = json.load(f) # Marka ara (case-insensitive) brand_data = None for b_name, b_data in catalog.get('brands', {}).items(): if brand in b_name.upper() or b_name.upper() in brand: brand_data = b_data break if not brand_data: return jsonify({'success': True, 'brand': brand, 'model': model, 'total': 0, 'pdfs': []}) # Model ara pdf_list = [] if model: # Belirli model for m_name, m_data in brand_data.get('models', {}).items(): if model in m_name.upper() or m_name.upper() in model: pdf_list.extend(m_data.get('pdfs', [])) else: # Tüm modeller for m_name, m_data in brand_data.get('models', {}).items(): pdf_list.extend(m_data.get('pdfs', [])) # Sırala pdf_list.sort(key=lambda x: (x.get('type', ''), x.get('filename', ''))) return jsonify({ 'success': True, 'brand': brand, 'model': model, 'total': len(pdf_list), 'pdfs': pdf_list }) except Exception as e: import traceback return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/pdf-pages', methods=['POST']) def api_pdf_pages(): """PDF sayfa sayısını ve bilgilerini döndür""" try: init_services() data = request.get_json() pdf_path = data.get('pdf_path', '') if not pdf_path: return jsonify({'success': False, 'error': 'PDF yolu gerekli'}) # PDF yolunu düzelt if pdf_path.startswith('/pdfs/'): pdf_path = pdf_path.replace('/pdfs/', '/mnt/pdfs/') import fitz if not os.path.exists(pdf_path): # Alternatif yol dene alt_path = pdf_path.replace('/home/poyraz/pdfs/', '/mnt/pdfs/') if os.path.exists(alt_path): pdf_path = alt_path else: return jsonify({'success': False, 'error': f'PDF bulunamadı: {pdf_path}'}) doc = fitz.open(pdf_path) page_count = len(doc) doc.close() return jsonify({ 'success': True, 'pdf_path': pdf_path, 'page_count': page_count }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/speed-test') def speed_test(): """Hızlı test sayfası""" with open('/var/www/html/PEPCVSON/templates/speed_test.html', 'r', encoding='utf-8') as f: return f.read() @app.route('/usta/api/search', methods=['POST']) def api_search(): """Arama API - DeepL destekli metin çevirisi + Analytics""" try: init_services() data = request.get_json() query = data.get('query', '') brand = data.get('brand', '').upper() model = data.get('model', '').upper() model = data.get('model', '').upper() pdf_filter = data.get('pdf_filter', '') limit = data.get('limit', 50) session_id = data.get('session_id', '') if not query: return jsonify({'success': False, 'error': 'Sorgu boş olamaz'}) # Daha fazla sonuç al (filtreleme sonrası azalabilir) # PDF Filtresi için sadece dosya adını al pdf_filter_filename = os.path.basename(pdf_filter) if pdf_filter else None print(f"DEBUG SEARCH REQ: Filter Raw='{pdf_filter}' -> Filename='{pdf_filter_filename}'") # Filtre varsa limit 100 yeterli search_limit = 100 if pdf_filter_filename else (limit * 5) # Native Filtering Kullan result = search_service.search(query, limit=search_limit, pdf_filter=pdf_filter_filename) # Marka/Model filtresi uygula (PDF yolunda ara) # Türkçe karakterleri normalize et def normalize_turkish(text): """Türkçe karakterleri normalize et (hem büyük hem küçük harf uyumu)""" if not text: return "" text = text.upper() # Türkçe karakter dönüşümleri tr_map = { 'İ': 'I', 'Ş': 'S', 'Ğ': 'G', 'Ü': 'U', 'Ö': 'O', 'Ç': 'C', 'ı': 'I', 'ş': 'S', 'ğ': 'G', 'ü': 'U', 'ö': 'O', 'ç': 'C' } for tr_char, en_char in tr_map.items(): text = text.replace(tr_char, en_char) return text brand_normalized = normalize_turkish(brand) model_normalized = normalize_turkish(model) filtered_results = [] for r in result.results: # Path normalizasyonu - Veritabanındaki farklı formatları eşle # Hata düzeltme: pdf_path değişkenini burada tanımlıyoruz pdf_path = normalize_turkish(r.pdf_path) if r.pdf_path else "" pdf_name = normalize_turkish(r.pdf_filename) if r.pdf_filename else "" db_path = r.pdf_path # Veritabanında "pdfs/" ile başlıyorsa temizle if db_path.startswith("pdfs/"): db_path = db_path[5:] elif db_path.startswith("/home/poyraz/pdfs/"): db_path = db_path[18:] # Filter path normalizasyonu filter_path_clean = pdf_filter if filter_path_clean.startswith("/mnt/pdfs/"): filter_path_clean = filter_path_clean[10:] elif filter_path_clean.startswith("pdfs/"): filter_path_clean = filter_path_clean[5:] # PDF Filtresi (Dosya adı ve Path Eşleşmesi) if pdf_filter: # 1. Dosya Adı Kontrolü (En güvenilir) filter_filename = os.path.basename(pdf_filter) db_filename = r.pdf_filename or os.path.basename(db_path) # Debug Check print(f"DEBUG FILTER: Filter={filter_filename} | DB_File={db_filename} | DB_Path={db_path}") # Native Filtering kullandığımız için Python filtrelemeyi kaldırıyoruz # Ancak yine de emin olmak için basit bir kontrol bırakabiliriz if pdf_filter: filter_filename = os.path.basename(pdf_filter) db_filename = r.pdf_filename or os.path.basename(r.pdf_path or "") # Çok gevşek kontrol (büyük/küçük harf vs. SmartSearch halletmiş olmalı ama) if filter_filename.lower() not in db_filename.lower(): # Yine de sonuç gelmişse yazdıralım, belki varyasyondur print(f"INFO: Native filter result '{db_filename}' vs target '{filter_filename}'") # Filtre kontrolü if brand_normalized and brand_normalized not in pdf_path and brand_normalized not in pdf_name: continue if model_normalized and model_normalized not in pdf_path and model_normalized not in pdf_name: continue filtered_results.append(r) if len(filtered_results) >= limit: break # Sonuçları işle ve metinleri çevir results_data = [] for r in filtered_results: r_dict = r.to_dict() # Metin varsa DeepL ile çevir (10 satır = ~1000 karakter) original_text = r_dict.get('text', '') if original_text and translator: try: # SmartTranslator ile çeviri trans_result = translator.translate(original_text, "EN", "TR") r_dict['text_tr'] = trans_result.translated_text r_dict['text_original'] = original_text # Orijinali sakla except Exception as te: print(f"Çeviri hatası: {te}") r_dict['text_tr'] = original_text # Hata olursa orijinali kullan r_dict['text_original'] = original_text else: r_dict['text_tr'] = original_text r_dict['text_original'] = original_text results_data.append(r_dict) # Analytics kaydı if analytics_db and results_data: top_result = results_data[0] if results_data else {} analytics_db.log_search( brand=brand, model=model, query_text=query, query_english=result.search_query if hasattr(result, 'search_query') else '', results_count=len(results_data), top_result_pdf=top_result.get('pdf_path', ''), top_result_page=top_result.get('page_number', 0), top_result_score=top_result.get('score', 0), search_time_ms=result.search_time_ms, session_id=session_id ) return jsonify({ 'success': True, 'query': query, 'brand': brand, 'model': model, 'search_time_ms': result.search_time_ms, 'results': results_data }) except Exception as e: import traceback traceback.print_exc() return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/ask', methods=['POST']) def api_ask(): """Sayfa analiz API - DeepL destekli + Analytics""" import time start_time = time.time() try: init_services() data = request.get_json() pdf_path = data.get('pdf_path', '') page_number = data.get('page_number', 1) question = data.get('question', '') brand = data.get('brand', '') model = data.get('model', '') session_id = data.get('session_id', '') if not pdf_path or not question: return jsonify({'success': False, 'error': 'PDF yolu ve soru gerekli'}) # Vision analizi yap result = vision_analyzer.analyze_page(pdf_path, page_number, question) if result.error: return jsonify({'success': False, 'error': result.error}) # AI cevabı İngilizce geldi, şimdi çevirelim english_answer = result.answer # SmartTranslator ile çevir (Sözlük destekli) trans_result = translator.translate(english_answer, "EN", "TR") best_answer = trans_result.translated_text used_terms = [{"en": k, "tr": v} for k, v in trans_result.used_terms.items()] # Eski yapıya uyumluluk (Frontend için) translations = { 'dictionary': 'Smart Translate', 'deepl': '-', 'hybrid': best_answer } response_time_ms = int((time.time() - start_time) * 1000) # Analytics kaydı if analytics_db: analytics_db.log_question( brand=brand, model=model, pdf_path=pdf_path, page_number=page_number, question=question, question_english=result.translated_question, answer_english=english_answer, answer_turkish=best_answer, confidence=result.confidence, safety_warnings=','.join(result.safety_warnings) if result.safety_warnings else '', response_time_ms=response_time_ms, session_id=session_id ) return jsonify({ 'success': True, 'answer': best_answer, 'answer_english': english_answer, 'confidence': result.confidence, 'safety_warnings': result.safety_warnings, 'referenced_items': result.referenced_items, 'translations': translations, 'used_terms': used_terms[:10] # Max 10 terim göster }) except Exception as e: import traceback traceback.print_exc() return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/dictionary/search', methods=['POST']) def api_dictionary_search(): """Sözlük arama API""" try: init_services() data = request.get_json() query = data.get('query', '').strip().lower() lang = data.get('lang', 'both') # 'tr', 'en', 'both' limit = data.get('limit', 50) if not query or len(query) < 2: return jsonify({'success': False, 'error': 'En az 2 karakter girin'}) results = [] # Sözlükten ara if translator and hasattr(translator, '_tr_to_en_cache') and hasattr(translator, '_en_to_tr_cache'): # Türkçe -> İngilizce if lang in ['tr', 'both']: for tr_term, en_list in translator._tr_to_en_cache.items(): if query in tr_term: for en in en_list[:3]: results.append({ 'tr': tr_term, 'en': en, 'direction': 'tr_to_en', 'match_type': 'exact' if tr_term == query else 'partial' }) # İngilizce -> Türkçe if lang in ['en', 'both']: for en_term, tr_list in translator._en_to_tr_cache.items(): if query in en_term: for tr in tr_list[:3]: results.append({ 'tr': tr, 'en': en_term, 'direction': 'en_to_tr', 'match_type': 'exact' if en_term == query else 'partial' }) # Sonuçları sırala (exact match önce) results.sort(key=lambda x: (0 if x['match_type'] == 'exact' else 1, len(x['en']))) # Tekrarları kaldır seen = set() unique_results = [] for r in results: key = f"{r['en']}|{r['tr']}" if key not in seen: seen.add(key) unique_results.append(r) return jsonify({ 'success': True, 'query': query, 'total': len(unique_results), 'results': unique_results[:limit] }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/dictionary/add', methods=['POST']) def api_dictionary_add(): """Sözlüğe terim ekleme önerisi""" try: init_services() data = request.get_json() english_term = data.get('english', '').strip() turkish_term = data.get('turkish', '').strip() context = data.get('context', '') if not english_term or not turkish_term: return jsonify({'success': False, 'error': 'Her iki terim de gerekli'}) if analytics_db: analytics_db.add_dictionary_suggestion( english_term=english_term, turkish_term=turkish_term, suggested_by='user', context=context ) return jsonify({ 'success': True, 'message': 'Terim öneriniz kaydedildi. Teşekkürler!' }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/dictionary/stats') def api_dictionary_stats(): """Sözlük istatistikleri""" try: init_services() stats = { 'total_terms': 0, 'tr_to_en': 0, 'en_to_tr': 0, 'pending_suggestions': 0 } if translator: if hasattr(translator, '_tr_to_en_cache'): stats['tr_to_en'] = len(translator._tr_to_en_cache) if hasattr(translator, '_en_to_tr_cache'): stats['en_to_tr'] = len(translator._en_to_tr_cache) stats['total_terms'] = stats['tr_to_en'] + stats['en_to_tr'] if analytics_db: pending = analytics_db.get_pending_suggestions(limit=1000) stats['pending_suggestions'] = len(pending) return jsonify({'success': True, 'stats': stats}) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/rate-term', methods=['POST']) def api_rate_term(): """Terim rating API""" try: init_services() data = request.get_json() english_term = data.get('english_term', '') turkish_term = data.get('turkish_term', '') rating = data.get('rating', 3) suggestion = data.get('suggestion', '') brand = data.get('brand', '') model = data.get('model', '') if not english_term: return jsonify({'success': False, 'error': 'Terim gerekli'}) # Analytics kaydı if analytics_db: analytics_db.log_translation_feedback( feedback_type='term', original_text=english_term, translated_text=turkish_term, rating=rating, user_suggestion=suggestion, brand=brand, model=model ) # Öneri varsa sözlük önerisine ekle if suggestion: analytics_db.add_dictionary_suggestion( english_term=english_term, turkish_term=suggestion, suggested_by='user', context=f"{brand} {model}" ) return jsonify({ 'success': True, 'message': 'Teşekkürler! Geri bildiriminiz kaydedildi.' }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/usta/api/page-image', methods=['POST']) def api_page_image(): """PDF sayfasını PNG olarak döndür""" try: import fitz import base64 data = request.get_json() pdf_path = data.get('pdf_path', '') page_number = data.get('page_number', 1) if not pdf_path: return jsonify({'success': False, 'error': 'PDF yolu gerekli'}) # PDF yolunu çöz if pdf_path.startswith("/home/poyraz/pdfs/"): pdf_path = pdf_path.replace("/home/poyraz/pdfs/", "/mnt/pdfs/") elif not os.path.isabs(pdf_path): if pdf_path.startswith("pdfs/"): pdf_path = pdf_path[5:] pdf_path = os.path.join("/mnt/pdfs", pdf_path) if not os.path.exists(pdf_path): return jsonify({'success': False, 'error': f'PDF bulunamadı'}) # PDF'yi aç ve sayfayı render et doc = fitz.open(pdf_path) page_idx = page_number - 1 if page_idx < 0 or page_idx >= len(doc): doc.close() return jsonify({'success': False, 'error': f'Geçersiz sayfa: {page_number}'}) page = doc[page_idx] # 1.5x zoom ile render mat = fitz.Matrix(1.5, 1.5) pix = page.get_pixmap(matrix=mat) png_bytes = pix.tobytes("png") total_pages = len(doc) doc.close() # Base64 encode base64_image = base64.b64encode(png_bytes).decode('utf-8') return jsonify({ 'success': True, 'image': f'data:image/png;base64,{base64_image}', 'page_number': page_number, 'total_pages': total_pages }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) if __name__ == '__main__': print("\n" + "=" * 60) print(" 👷 USTA ASSISTANT WEB v2.0") print("=" * 60) print() print(f"🌐 Erişim: http://0.0.0.0:5001/usta/") print() # 0.0.0.0 ile dış dünyaya açıyoruz app.run(host='0.0.0.0', port=5001, debug=False)