Autoparts-DB/pos/services/cross_sell.py

"""Cross-sell recommendations for the POS cart ("esto + esto").

Given the inventory_ids already in the cart, suggest complementary parts
from two independent sources, merged and ranked:

  1. Curated part kits — ``services.part_kits.KIT_SUGGESTIONS`` (Spanish
     aftermarket knowledge; works from day one, no sales data required).
  2. Real co-occurrence — inventory items that historically appear in the
     same ``sale_id`` as one of the cart items (learns from actual sales).

Items that already have stock are preferred, and items already in the cart
are never suggested.

Public API:
    get_cross_sell(conn, inventory_ids, limit=5) -> list[dict]
"""

from services.part_kits import get_kit_suggestions

# Same accent-stripping used by catalog_service so that "Batería" in the
# inventory matches the unaccented keywords in KIT_SUGGESTIONS.
_ACCENTS = [
    ('Á', 'A'), ('É', 'E'), ('Í', 'I'), ('Ó', 'O'), ('Ú', 'U'),
    ('À', 'A'), ('È', 'E'), ('Ì', 'I'), ('Ò', 'O'), ('Ù', 'U'),
    ('Ä', 'A'), ('Ë', 'E'), ('Ï', 'I'), ('Ö', 'O'), ('Ü', 'U'), ('Ñ', 'N'),
]

# Ranking weights: stock first (a suggestion you cannot sell is useless),
# then source (real co-occurrence beats the curated kit), then evidence.
_W_STOCK = 1000
_W_COOC = 100
_W_KIT = 50
_W_FREQ = 10


def _norm_text(value):
    """Upper-case and strip accents for keyword comparison."""
    text = (value or '').upper()
    for a, b in _ACCENTS:
        text = text.replace(a, b)
    return text


def _norm_expr(column):
    """SQL expression that upper-cases and strips accents from a column."""
    expr = f"UPPER({column})"
    for a, b in _ACCENTS:
        expr = f"REPLACE({expr}, '{a}', '{b}')"
    return expr


def _cart_items(conn, inventory_ids):
    """Load the cart items (id, name) so we can derive kit keywords."""
    cur = conn.cursor()
    try:
        cur.execute(
            "SELECT id, part_number, name FROM inventory WHERE id = ANY(%s)",
            (list(inventory_ids),),
        )
        return cur.fetchall()
    finally:
        cur.close()


def _kit_candidates(conn, cart_items, exclude_ids, limit):
    """Inventory rows whose name matches a curated kit keyword.

    Returns list of (inventory_id, part_number, name, price, stock, reason).
    """
    keyword_sources = {}
    patterns = []
    for _inv_id, _pn, name in cart_items:
        for keyword in get_kit_suggestions(name):
            norm = _norm_text(keyword)
            if not norm or norm in keyword_sources:
                continue
            keyword_sources[norm] = (name, keyword)
            patterns.append(f'%{norm}%')
    if not patterns:
        return []

    name_expr = _norm_expr('i.name')
    sql = f"""
        SELECT DISTINCT ON (i.id)
               i.id, i.part_number, i.name, i.price_1,
               COALESCE(s.stock, 0) AS stock, {name_expr} AS name_norm
        FROM inventory i
        LEFT JOIN inventory_stock_summary s ON s.inventory_id = i.id
        WHERE i.is_active = true
          AND i.id <> ALL(%s)
          AND {name_expr} LIKE ANY(%s)
        ORDER BY i.id, i.name
        LIMIT %s
    """
    cur = conn.cursor()
    try:
        cur.execute(sql, (list(exclude_ids), patterns, limit * 4))
        rows = cur.fetchall()
    finally:
        cur.close()

    out = []
    for inv_id, part_number, name, price, stock, name_norm in rows:
        source_name, keyword = None, None
        for norm, (src, kw) in keyword_sources.items():
            if norm in (name_norm or ''):
                source_name, keyword = src, kw
                break
        if source_name is None:
            continue
        out.append((inv_id, part_number, name, price, stock,
                    f"Complemento habitual de {source_name} ({keyword})"))
    return out


def _cooccurrence_candidates(conn, inventory_ids, exclude_ids, limit):
    """Inventory items historically sold in the same sale as a cart item.

    Returns list of (inventory_id, part_number, name, price, stock, freq, paired_with).
    """
    cur = conn.cursor()
    try:
        cur.execute("""
            SELECT si2.inventory_id,
                   COUNT(DISTINCT si2.sale_id) AS freq,
                   MIN(si1.name) AS paired_with
            FROM sale_items si1
            JOIN sale_items si2
              ON si2.sale_id = si1.sale_id
             AND si2.inventory_id IS NOT NULL
             AND si2.inventory_id <> ALL(%s)
            WHERE si1.inventory_id = ANY(%s)
            GROUP BY si2.inventory_id
            ORDER BY freq DESC, si2.inventory_id
            LIMIT %s
        """, (list(exclude_ids), list(inventory_ids), limit * 4))
        pairs = cur.fetchall()
        if not pairs:
            return []

        ids = [r[0] for r in pairs]
        cur.execute("""
            SELECT i.id, i.part_number, i.name, i.price_1,
                   COALESCE(s.stock, 0) AS stock
            FROM inventory i
            LEFT JOIN inventory_stock_summary s ON s.inventory_id = i.id
            WHERE i.is_active = true AND i.id = ANY(%s)
        """, (ids,))
        by_id = {r[0]: r for r in cur.fetchall()}
    finally:
        cur.close()

    out = []
    for inv_id, freq, paired_with in pairs:
        row = by_id.get(inv_id)
        if not row:
            continue
        _rid, part_number, name, price, stock = row
        out.append((inv_id, part_number, name, price, stock, freq, paired_with))
    return out


def get_cross_sell(conn, inventory_ids, limit=5):
    """Suggest complementary parts for the given cart contents.

    Args:
        conn: psycopg2 connection to the tenant DB.
        inventory_ids: iterable of inventory ids currently in the cart.
        limit: max number of suggestions.

    Returns:
        list of dicts: {inventory_id, name, part_number, price, reason, stock, source}
    """
    try:
        cart_ids = [int(i) for i in (inventory_ids or [])]
    except (TypeError, ValueError):
        cart_ids = []
    cart_ids = list(dict.fromkeys(cart_ids))[:50]
    if not cart_ids:
        return []

    scored = {}  # inventory_id -> dict with score + payload

    try:
        cart_items = _cart_items(conn, cart_ids)
    except Exception:
        conn.rollback()
        return []

    # 1. Co-occurrence (real sales) — highest evidence weight.
    try:
        for inv_id, part_number, name, price, stock, freq, paired_with in \
                _cooccurrence_candidates(conn, cart_ids, cart_ids, limit):
            score = (stock > 0) * _W_STOCK + _W_COOC + min(freq, 5) * _W_FREQ
            scored[inv_id] = {
                'inventory_id': inv_id,
                'name': name,
                'part_number': part_number,
                'price': float(price) if price is not None else None,
                'stock': int(stock or 0),
                'reason': f"Suele venderse con {paired_with or 'este artículo'}",
                'source': 'cooccurrence',
                '_score': score,
            }
    except Exception:
        conn.rollback()

    # 2. Curated kits — fill in whatever co-occurrence did not cover.
    try:
        for inv_id, part_number, name, price, stock, reason in \
                _kit_candidates(conn, cart_items, cart_ids, limit):
            score = (stock > 0) * _W_STOCK + _W_KIT
            if inv_id in scored:
                # Already suggested by real sales: bump evidence, keep reason.
                scored[inv_id]['_score'] += _W_KIT
                continue
            scored[inv_id] = {
                'inventory_id': inv_id,
                'name': name,
                'part_number': part_number,
                'price': float(price) if price is not None else None,
                'stock': int(stock or 0),
                'reason': reason,
                'source': 'kit',
                '_score': score,
            }
    except Exception:
        conn.rollback()

    ordered = sorted(scored.values(), key=lambda d: d['_score'], reverse=True)
    for item in ordered:
        item.pop('_score', None)
    return ordered[:max(1, limit)]