Autoparts-DB/pos/services/semantic_search.py

# /home/Autopartes/pos/services/semantic_search.py
"""Búsqueda semántica sobre el catálogo: rerank + embeddings vía API QWEN.

El proveedor QWEN (``QWEN_API_URL``) expone, además de chat, dos endpoints
compatibles con OpenAI/Cohere y ya verificados en producción:

    POST {QWEN_API_URL}/rerank
        {"model": "rerank", "query": "...", "documents": ["a", "b", ...]}
        → {"results": [{"index": 1, "relevance_score": 0.70, "document": {...}}, ...]}
        (ordenado por relevancia descendente)

    POST {QWEN_API_URL}/embeddings
        {"model": "qwen3-embedding", "input": "texto"}
        → {"data": [{"embedding": [4096 floats]}, ...]}  (OpenAI-compatible)

Regla de oro: **ninguna función lanza excepción**. Si falta configuración, hay
timeout o cualquier error de red/API, ``rerank()`` devuelve el orden original y
``embed()``/``find_similar()`` devuelven ``None``. Así el buscador del POS
nunca se rompe ni se cae por culpa del rerank; en el peor caso la búsqueda es
la de siempre.

Variables de entorno (ver ``pos/config.py``):
    QWEN_API_URL               URL base del proveedor (igual que ``ai_chat``)
    QWEN_API_KEY               API key (igual que ``ai_chat``)
    QWEN_RERANK_MODEL          default ``rerank``
    QWEN_EMBED_MODEL           default ``qwen3-embedding``
    SEARCH_RERANK_ENABLED      default ``true``  — scope ``catalog`` (público)
    SEARCH_RERANK_POS_ENABLED  default ``false`` — scope ``pos`` (caja/POS)
    SEARCH_RERANK_TIMEOUT      default ``2`` segundos (ambos scopes)
    SEARCH_RERANK_MAX_CANDIDATES default ``50`` documentos por llamada

Nota de rendimiento (medido 2026-10-11 contra api.nan.builders, RTT ~40 ms):
el endpoint /rerank tarda ~0.6–1.8 s independientemente del número de
documentos (5 a 51). Por eso el POS va apagado por defecto: su búsqueda en
Meilisearch ya responde en ~18 ms y el rerank la encarecería 30–80×.
"""

import math
import os

import requests

try:  # config del POS; con sys.path del dashboard puede resolver el config raíz
    from config import QWEN_API_URL, QWEN_API_KEY
    from config import QWEN_RERANK_MODEL, QWEN_EMBED_MODEL
    from config import SEARCH_RERANK_ENABLED, SEARCH_RERANK_POS_ENABLED
    from config import SEARCH_RERANK_TIMEOUT, SEARCH_RERANK_MAX_CANDIDATES
except ImportError:
    QWEN_API_URL = os.environ.get('QWEN_API_URL', '')
    QWEN_API_KEY = os.environ.get('QWEN_API_KEY', '')
    QWEN_RERANK_MODEL = os.environ.get('QWEN_RERANK_MODEL', 'rerank')
    QWEN_EMBED_MODEL = os.environ.get('QWEN_EMBED_MODEL', 'qwen3-embedding')
    SEARCH_RERANK_ENABLED = os.environ.get('SEARCH_RERANK_ENABLED', 'true').lower() == 'true'
    SEARCH_RERANK_POS_ENABLED = os.environ.get('SEARCH_RERANK_POS_ENABLED', 'false').lower() == 'true'
    SEARCH_RERANK_TIMEOUT = float(os.environ.get('SEARCH_RERANK_TIMEOUT', '2'))
    SEARCH_RERANK_MAX_CANDIDATES = int(os.environ.get('SEARCH_RERANK_MAX_CANDIDATES', '50'))

DEFAULT_RERANK_MODEL = 'rerank'
DEFAULT_EMBED_MODEL = 'qwen3-embedding'
DEFAULT_TIMEOUT = 2.0
DEFAULT_MAX_CANDIDATES = 50

# Scopes del flag: 'catalog' = catálogo público / API / dashboard;
# 'pos' = caja y chat del POS, donde la latencia es crítica.
SCOPE_CATALOG = 'catalog'
SCOPE_POS = 'pos'


def is_configured():
    """True si hay URL y API key (condición previa a cualquier llamada)."""
    return bool((QWEN_API_URL or '').strip() and (QWEN_API_KEY or '').strip())


def is_rerank_enabled(scope=SCOPE_CATALOG):
    """Flag de rerank para un scope.

    ``SEARCH_RERANK_ENABLED`` es el interruptor maestro (default ``true``) y
    aplica al scope ``catalog``. El POS requiere además
    ``SEARCH_RERANK_POS_ENABLED=true`` (default ``false``) porque el rerank
    agrega cientos de ms a una búsqueda que hoy es de ~18 ms.
    """
    if not SEARCH_RERANK_ENABLED:
        return False
    if (scope or SCOPE_CATALOG).lower() == SCOPE_POS:
        return bool(SEARCH_RERANK_POS_ENABLED)
    return True


def rerank_model():
    return QWEN_RERANK_MODEL or DEFAULT_RERANK_MODEL


def embed_model():
    return QWEN_EMBED_MODEL or DEFAULT_EMBED_MODEL


def max_candidates():
    """Tope de documentos por llamada (acota payload y latencia)."""
    try:
        value = int(SEARCH_RERANK_MAX_CANDIDATES)
    except (TypeError, ValueError):
        return DEFAULT_MAX_CANDIDATES
    return value if value > 0 else DEFAULT_MAX_CANDIDATES


def _timeout(timeout=None):
    if timeout is not None:
        return timeout
    try:
        return float(SEARCH_RERANK_TIMEOUT)
    except (TypeError, ValueError):
        return DEFAULT_TIMEOUT


def _endpoint(path):
    url = (QWEN_API_URL or '').strip()
    if not url:
        return None
    return url.rstrip('/') + path


def _headers():
    return {
        'Authorization': f"Bearer {QWEN_API_KEY}",
        'Content-Type': 'application/json',
    }


def _identity_order(count):
    return [{'index': i, 'relevance_score': None} for i in range(count)]


def rerank(query, documents, top_n=None, timeout=None):
    """Reordena ``documents`` por relevancia semántica respecto a ``query``.

    Args:
        query: texto de búsqueda del usuario.
        documents: lista de textos candidatos.
        top_n: opcional, trunca el resultado a los N más relevantes. Por
            defecto devuelve **todos** los documentos: el caller pagina/muestra
            y no debe perder items.
        timeout: segundos (default ``SEARCH_RERANK_TIMEOUT``).

    Returns:
        list[dict]: ``[{'index': <int>, 'relevance_score': <float|None>}, ...]``
        en orden de relevancia descendente. En degradación (sin URL/key,
        timeout, HTTP error, respuesta vacía o índice inválido) devuelve el
        orden original con ``relevance_score=None``. Nunca lanza.
    """
    docs = list(documents or [])
    identity = _identity_order(len(docs))
    if not docs or not query:
        return identity[:top_n] if top_n else identity

    url = _endpoint('/rerank')
    if not url or not (QWEN_API_KEY or '').strip():
        return identity[:top_n] if top_n else identity

    try:
        resp = requests.post(
            url,
            headers=_headers(),
            json={
                'model': rerank_model(),
                'query': query,
                'documents': docs,
            },
            timeout=_timeout(timeout),
        )
        if resp.status_code >= 400:
            print(f"[semantic_search] rerank HTTP {resp.status_code}: {resp.text[:200]}")
            return identity[:top_n] if top_n else identity
        results = (resp.json() or {}).get('results') or []
    except Exception as e:
        print(f"[semantic_search] rerank error: {e}")
        return identity[:top_n] if top_n else identity

    ordered = []
    seen = set()
    for item in results:
        try:
            idx = int(item.get('index'))
        except (TypeError, ValueError, AttributeError):
            continue
        if idx < 0 or idx >= len(docs) or idx in seen:
            continue
        seen.add(idx)
        score = item.get('relevance_score')
        ordered.append({
            'index': idx,
            'relevance_score': float(score) if isinstance(score, (int, float)) else None,
        })

    if not ordered:
        return identity[:top_n] if top_n else identity

    # Documentos que la API no devolvió (p.ej. si aplicó su propio tope)
    # conservan su orden original al final — nunca se pierden.
    for i in range(len(docs)):
        if i not in seen:
            ordered.append({'index': i, 'relevance_score': None})

    return ordered[:top_n] if top_n else ordered


def rerank_order(query, documents, top_n=None, timeout=None):
    """Igual que ``rerank()`` pero devuelve solo los índices reordenados."""
    return [r['index'] for r in rerank(query, documents, top_n=top_n, timeout=timeout)]


def embed(texts, timeout=None):
    """Devuelve los embeddings de ``texts`` vía el endpoint /embeddings.

    Args:
        texts: str o iterable de str.
        timeout: segundos (default ``SEARCH_RERANK_TIMEOUT``).

    Returns:
        list[list[float]] en el mismo orden que la entrada, o ``None`` si no
        hay configuración, hay timeout, error HTTP o respuesta malformada.
        Nunca lanza.
    """
    if isinstance(texts, str):
        items = [texts]
    else:
        items = list(texts or [])
    if not items:
        return []

    url = _endpoint('/embeddings')
    if not url or not (QWEN_API_KEY or '').strip():
        return None

    try:
        resp = requests.post(
            url,
            headers=_headers(),
            json={'model': embed_model(), 'input': items},
            timeout=_timeout(timeout),
        )
        if resp.status_code >= 400:
            print(f"[semantic_search] embeddings HTTP {resp.status_code}: {resp.text[:200]}")
            return None
        data = (resp.json() or {}).get('data') or []
        vectors = [row.get('embedding') for row in data]
        if len(vectors) != len(items) or any(not isinstance(v, list) or not v for v in vectors):
            print(f"[semantic_search] embeddings respuesta inesperada ({len(vectors)} de {len(items)})")
            return None
        return vectors
    except Exception as e:
        print(f"[semantic_search] embeddings error: {e}")
        return None


def cosine_similarity(vec_a, vec_b):
    """Similitud coseno entre dos vectores (None si no es calculable)."""
    if not vec_a or not vec_b or len(vec_a) != len(vec_b):
        return None
    try:
        dot = 0.0
        norm_a = 0.0
        norm_b = 0.0
        for a, b in zip(vec_a, vec_b):
            dot += a * b
            norm_a += a * a
            norm_b += b * b
        if norm_a <= 0 or norm_b <= 0:
            return None
        return dot / (math.sqrt(norm_a) * math.sqrt(norm_b))
    except (TypeError, ValueError):
        return None


def similar_texts(text_a, text_b, timeout=None):
    """Similitud coseno entre dos textos (None si no hay embeddings)."""
    vectors = embed([text_a, text_b], timeout=timeout)
    if not vectors:
        return None
    return cosine_similarity(vectors[0], vectors[1])


def find_similar(query, candidates, top_n=None, timeout=None):
    """Ordena ``candidates`` por similitud coseno con ``query``.

    Returns:
        list[dict] ``[{'index', 'similarity'}, ...]`` ordenado desc, o ``None``
        si no hay embeddings disponibles. Nunca lanza.
    """
    items = list(candidates or [])
    if not items or not query:
        return None
    vectors = embed([query] + items, timeout=timeout)
    if not vectors:
        return None
    query_vec = vectors[0]
    scored = []
    for i, vec in enumerate(vectors[1:]):
        sim = cosine_similarity(query_vec, vec)
        if sim is not None:
            scored.append({'index': i, 'similarity': sim})
    if not scored:
        return None
    scored.sort(key=lambda r: r['similarity'], reverse=True)
    return scored[:top_n] if top_n else scored