"""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