Autoparts-DB/pos/blueprints/chat_bp.py

# /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 master and tenant:
            try:
                results = catalog_service.smart_search(
                    master, effective_query, tenant, branch_id, limit=10
                )
                search_results = results if results else []
            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