"""Cached warehouse search to reduce API calls and token usage""" import time import re import json import requests from typing import Dict, List, Optional, Tuple # Cache configuration CACHE_DURATION = 43200 # 12 hours (12 * 60 * 60) cache = { 'warehouse_xml': {'data': None, 'time': 0}, 'trek_xml': {'data': None, 'time': 0}, 'products_summary': {'data': None, 'time': 0}, 'simple_searches': {} # Cache for specific searches } def get_cached_warehouse_xml() -> str: """Get warehouse XML with caching""" current_time = time.time() if cache['warehouse_xml']['data'] and (current_time - cache['warehouse_xml']['time'] < CACHE_DURATION): print("📦 Using cached warehouse XML") return cache['warehouse_xml']['data'] print("📡 Fetching fresh warehouse XML...") url = 'https://video.trek-turkey.com/bizimhesap-warehouse-xml-b2b-api-v2.php' response = requests.get(url, verify=False, timeout=15) cache['warehouse_xml']['data'] = response.text cache['warehouse_xml']['time'] = current_time return response.text def get_cached_trek_xml() -> str: """Get Trek XML with caching""" current_time = time.time() if cache['trek_xml']['data'] and (current_time - cache['trek_xml']['time'] < CACHE_DURATION): print("🚴 Using cached Trek XML") return cache['trek_xml']['data'] print("📡 Fetching fresh Trek XML...") url = 'https://www.trekbisiklet.com.tr/output/8582384479' response = requests.get(url, verify=False, timeout=15) cache['trek_xml']['data'] = response.content cache['trek_xml']['time'] = current_time return response.content def simple_product_search(query: str) -> Optional[List[Dict]]: """ Simple local search without GPT-5 Returns product info if exact/close match found """ query_upper = query.upper() query_parts = query_upper.split() # Get cached products summary if not cache['products_summary']['data'] or \ (time.time() - cache['products_summary']['time'] > CACHE_DURATION): # Build products summary from cached XML build_products_summary() products_summary = cache['products_summary']['data'] # Exact product name patterns exact_patterns = { 'MADONE SL 6': lambda p: 'MADONE SL 6' in p['name'], 'MADONE SL 7': lambda p: 'MADONE SL 7' in p['name'], 'MARLIN 5': lambda p: 'MARLIN 5' in p['name'], 'MARLIN 6': lambda p: 'MARLIN 6' in p['name'], 'MARLIN 7': lambda p: 'MARLIN 7' in p['name'], 'DOMANE SL 5': lambda p: 'DOMANE SL 5' in p['name'], 'CHECKPOINT': lambda p: 'CHECKPOINT' in p['name'], 'FX': lambda p: p['name'].startswith('FX'), 'DUAL SPORT': lambda p: 'DUAL SPORT' in p['name'], 'RAIL': lambda p: 'RAIL' in p['name'], 'POWERFLY': lambda p: 'POWERFLY' in p['name'], } # Check for exact patterns for pattern, matcher in exact_patterns.items(): if pattern in query_upper: matching = [p for p in products_summary if matcher(p)] if matching: print(f"✅ Found {len(matching)} products via simple search (no GPT-5 needed)") return matching # Check for simple one-word queries if len(query_parts) == 1: matching = [p for p in products_summary if query_parts[0] in p['name']] if matching and len(matching) < 20: # If reasonable number of matches print(f"✅ Found {len(matching)} products via simple search (no GPT-5 needed)") return matching return None # Need GPT-5 for complex queries def build_products_summary(): """Build products summary from cached XMLs""" xml_text = get_cached_warehouse_xml() # Extract products product_pattern = r'(.*?)' all_products = re.findall(product_pattern, xml_text, re.DOTALL) products_summary = [] for i, product_block in enumerate(all_products): name_match = re.search(r'', product_block) variant_match = re.search(r'', product_block) if name_match: warehouses_with_stock = [] warehouse_regex = r'.*?.*?(.*?).*?' warehouses = re.findall(warehouse_regex, product_block, re.DOTALL) for wh_name, wh_stock in warehouses: try: if int(wh_stock.strip()) > 0: warehouses_with_stock.append(wh_name) except: pass product_info = { "index": i, "name": name_match.group(1), "variant": variant_match.group(1) if variant_match else "", "warehouses": warehouses_with_stock } products_summary.append(product_info) cache['products_summary']['data'] = products_summary cache['products_summary']['time'] = time.time() print(f"📊 Built products summary: {len(products_summary)} products") def should_use_gpt5(query: str) -> bool: """Determine if query needs GPT-5""" query_lower = query.lower() # Complex queries that need GPT-5 gpt5_triggers = [ 'öneri', 'tavsiye', 'bütçe', 'karşılaştır', 'hangisi', 'ne önerirsin', 'yardım', 'en iyi', 'en ucuz', 'en pahalı', 'kaç tane', 'toplam', 'fark' ] for trigger in gpt5_triggers: if trigger in query_lower: return True # If simple search found results, don't use GPT-5 if simple_product_search(query): return False return True # Default to GPT-5 for uncertain cases # Usage example def smart_warehouse_search(query: str) -> List[str]: """ Smart search with caching and minimal GPT-5 usage """ # Check simple search cache first cache_key = query.lower() if cache_key in cache['simple_searches']: cached_result = cache['simple_searches'][cache_key] if time.time() - cached_result['time'] < CACHE_DURATION: print(f"✅ Using cached result for '{query}'") return cached_result['data'] # Try simple search simple_results = simple_product_search(query) if simple_results: # Format and cache the results formatted_results = format_simple_results(simple_results) cache['simple_searches'][cache_key] = { 'data': formatted_results, 'time': time.time() } return formatted_results # Fall back to GPT-5 if needed print(f"🤖 Using GPT-5 for complex query: '{query}'") # Call existing GPT-5 function here return None # Would call get_warehouse_stock_smart_with_price def format_simple_results(products: List[Dict]) -> List[str]: """Format simple search results""" if not products: return ["Ürün bulunamadı"] result = ["Bulunan ürünler:"] # Group by product name product_groups = {} for p in products: if p['name'] not in product_groups: product_groups[p['name']] = [] product_groups[p['name']].append(p) for product_name, variants in product_groups.items(): result.append(f"\n{product_name}:") for v in variants: if v['variant']: warehouses_str = ", ".join([w.replace('MAGAZA DEPO', '').strip() for w in v['warehouses']]) result.append(f"• {v['variant']}: {warehouses_str if warehouses_str else 'Stokta yok'}") return result