# /home/Autopartes/pos/blueprints/chat_bp.py """Chat blueprint: AI-powered parts lookup via natural language. Endpoints (all under /pos/api/chat): POST / — send a message, get AI response + catalog search results """ from flask import Blueprint, request, jsonify, g from middleware import require_auth from tenant_db import get_master_conn, get_tenant_conn from services import catalog_service, ai_chat chat_bp = Blueprint("chat", __name__, url_prefix="/pos/api/chat") @chat_bp.route("", methods=["POST"]) @require_auth("catalog.view") def chat(): body = request.get_json(force=True) user_message = (body.get("message") or "").strip() if not user_message: return jsonify({"error": "message required"}), 400 history = body.get("history") or [] # Call AI ai_response = ai_chat.chat(user_message, history) search_results = [] vehicle_match = None master = None tenant = None try: # If AI suggests a search query, run it against the catalog search_query = ai_response.get("search_query") vehicle = ai_response.get("vehicle") if search_query or vehicle: master = get_master_conn() tenant = get_tenant_conn(g.tenant_id) branch_id = g.branch_id # Try to resolve vehicle to MYE if vehicle and master: vehicle_match = _resolve_vehicle(master, vehicle) # Run catalog search if we have a search query # Also search if AI identified a vehicle but didn't give a search_query effective_query = search_query if not effective_query and vehicle: # Extract likely part keywords from the user's message import re # Remove brand/model/year from message to get the part description part_words = user_message.lower() for remove in [vehicle.get('brand',''), vehicle.get('model',''), str(vehicle.get('year',''))]: part_words = part_words.replace(remove.lower(), '') part_words = re.sub(r'necesito|quiero|busco|para|un|una|el|la|de|del|los|las|mi|\d{4}', '', part_words).strip() if len(part_words) >= 3: effective_query = part_words if effective_query and tenant: # First: search local inventory try: local_results = _search_local_inventory(tenant, effective_query, search_query or '', branch_id) if local_results: search_results.extend(local_results) except Exception: pass # Then: search TecDoc catalog if master: try: catalog_results = catalog_service.smart_search( master, effective_query, tenant, branch_id, limit=10 ) if catalog_results: # Mark as catalog results and avoid duplicates local_parts = {r.get('part_number', '') for r in search_results} for cr in catalog_results: if cr.get('oem_part_number', '') not in local_parts: cr['source'] = 'catalog' search_results.append(cr) except Exception: pass # search failure is non-fatal except Exception: pass # DB failure is non-fatal for chat finally: if master: try: master.close() except Exception: pass if tenant: try: tenant.close() except Exception: pass return jsonify( { "response": ai_response.get("message", ""), "search_results": search_results, "vehicle": vehicle_match or ai_response.get("vehicle"), } ) def _resolve_vehicle(master_conn, vehicle): """Try to resolve AI-extracted vehicle info to brand_id/model_id in DB.""" brand_name = (vehicle.get("brand") or "").upper().strip() model_name = (vehicle.get("model") or "").strip() year = vehicle.get("year") if not brand_name: return vehicle cur = master_conn.cursor() result = dict(vehicle) try: # Find brand cur.execute( "SELECT id_brand, name_brand FROM brands WHERE UPPER(name_brand) = %s", (brand_name,), ) brand_row = cur.fetchone() if brand_row: result["brand_id"] = brand_row[0] result["brand"] = brand_row[1] # Find model if model_name: cur.execute( """SELECT m.id_model, m.name_model FROM models m WHERE m.brand_id = %s AND UPPER(m.name_model) LIKE %s ORDER BY m.name_model LIMIT 5""", (brand_row[0], f"%{model_name.upper()}%"), ) model_row = cur.fetchone() if model_row: result["model_id"] = model_row[0] result["model"] = model_row[1] # Find year -> MYE if year: cur.execute( """SELECT mye.id_mye, y.year_car, e.name_engine, mye.trim_level FROM model_year_engine mye JOIN years y ON y.id_year = mye.year_id JOIN engines e ON e.id_engine = mye.engine_id WHERE mye.model_id = %s AND y.year_car = %s ORDER BY e.name_engine LIMIT 10""", (model_row[0], int(year)), ) mye_rows = cur.fetchall() if mye_rows: result["mye_options"] = [ { "mye_id": r[0], "year": r[1], "engine": r[2], "trim": r[3], } for r in mye_rows ] except Exception: pass finally: cur.close() return result def _search_local_inventory(tenant_conn, query_en, query_es, branch_id): """Search tenant's local inventory by part name/number in both English and Spanish.""" cur = tenant_conn.cursor() results = [] try: # Search by part_number, name, or brand — try both English and Spanish terms terms = set() terms.add(query_en) if query_es: terms.add(query_es) where_parts = [] params = [] for term in terms: if not term: continue where_parts.append("(i.part_number ILIKE %s OR i.name ILIKE %s OR i.brand ILIKE %s)") params.extend([f'%{term}%', f'%{term}%', f'%{term}%']) if not where_parts: return [] where = " OR ".join(where_parts) if branch_id: where = f"({where}) AND i.branch_id = %s" params.append(branch_id) cur.execute(f""" SELECT i.id, i.part_number, i.name, i.brand, i.price_1, i.price_2, i.price_3, i.cost, COALESCE((SELECT SUM(quantity) FROM inventory_operations WHERE inventory_id = i.id), 0) as stock FROM inventory i WHERE i.is_active = true AND ({where}) ORDER BY i.name LIMIT 10 """, params) for r in cur.fetchall(): results.append({ 'source': 'local', 'inventory_id': r[0], 'part_number': r[1], 'oem_part_number': r[1], 'name_part': r[2], 'brand': r[3], 'price_1': float(r[4]) if r[4] else 0, 'price_2': float(r[5]) if r[5] else 0, 'price_3': float(r[6]) if r[6] else 0, 'cost': float(r[7]) if r[7] else 0, 'local_stock': r[8], }) except Exception: pass finally: cur.close() return results