Autoparts-DB/pos/services/yonke_ai.py

# /home/Autopartes/pos/services/yonke_ai.py
"""Non-conversational AI helpers for the Nexus Yonke module.

Four capabilities, each one isolated and best-effort:

  * ``identify_piece``   — vision: name + category of a photo
  * ``suggest_price``    — rule engine over cost/grade/category + local comparables
  * ``find_duplicate_groups`` — semantic grouping of used-piece names
  * ``validate_photo``   — vision: is it an auto part? is the photo usable?

Golden rule (same as ``semantic_search``): **no function raises**. When the AI
provider is not configured or does not answer, each function returns a
well-formed dict with ``available=False`` so the caller can keep going.
"""

import base64
import json
import re
import statistics

# Multipliers applied to a reference price depending on the piece grade.
GRADE_FACTOR = {'A': 1.15, 'B': 1.0, 'C': 0.8, 'D': 0.6}
# Target markup over acquisition cost for a grade-B used piece.
TARGET_MARKUP = 1.6
# Weight of the local market reference vs. the cost-based reference.
MARKET_WEIGHT = 0.6


def _coerce_float(val, default=0.0):
    try:
        return float(val)
    except (TypeError, ValueError):
        return default


def _b64(image_bytes):
    return base64.b64encode(image_bytes).decode('ascii')


def _extract_json(text):
    """Return the first JSON object found in ``text``, or None."""
    if not text:
        return None
    stripped = str(text).strip()
    if stripped.startswith('```'):
        lines = stripped.split('\n')
        stripped = '\n'.join(lines[1:-1] if len(lines) > 2 else lines)
    try:
        parsed = json.loads(stripped)
        return parsed if isinstance(parsed, dict) else None
    except Exception:
        pass
    match = re.search(r'\{.*\}', stripped, re.DOTALL)
    if match:
        try:
            parsed = json.loads(match.group(0))
            return parsed if isinstance(parsed, dict) else None
        except Exception:
            return None
    return None


def _normalize_confidence(value):
    conf = _coerce_float(value, 0.0)
    if conf > 1:
        conf = conf / 100.0
    return round(max(0.0, min(1.0, conf)), 2)


def _vision_json(user_message, image_bytes=None, image_base64=None):
    """Ask the vision model for a JSON object. Returns (dict|None, error|None)."""
    image = image_base64 or (_b64(image_bytes) if image_bytes else None)
    if not image:
        return None, 'no image'
    try:
        from services import ai_chat
    except Exception as e:  # pragma: no cover
        return None, f'ai_chat unavailable: {e}'
    try:
        result = ai_chat.chat_with_image(user_message, image)
    except Exception as e:
        return None, str(e)
    if not isinstance(result, dict):
        return None, 'unexpected AI response'
    if 'name' in result or 'category' in result or 'is_part' in result:
        return result, None
    if result.get('message'):
        parsed = _extract_json(result['message'])
        if parsed:
            return parsed, None
    return None, result.get('message') or 'AI did not return JSON'


def identify_piece(image_bytes=None, image_base64=None, hint_name=None):
    """Identify an auto part from a photo.

    Returns ``{name, category, confidence, available, source}``; ``available``
    is False when the provider is unreachable (the endpoint then answers 503).
    """
    empty = {'name': None, 'category': None, 'confidence': 0.0,
             'available': False, 'source': 'vision'}
    if not image_bytes and not image_base64:
        return dict(empty, error='no image')

    prompt = (
        'Eres un experto en autopartes (partes usadas de deshuesadero). '
        'Identifica la pieza de la imagen. Responde SOLO con JSON válido, sin texto extra: '
        '{"name": "nombre en español", "category": "categoría (Frenos, Motor, Eléctrico, '
        'Suspensión, Carrocería, Enfriamiento, Luces, Transmisión, etc.)", '
        '"confidence": 0.0}'
    )
    if hint_name:
        prompt += f' Pista del usuario: "{hint_name}".'

    parsed, error = _vision_json(prompt, image_bytes, image_base64)
    if not parsed:
        return dict(empty, error=error or 'AI unavailable')

    name = (parsed.get('name') or '').strip() or None
    category = (parsed.get('category') or '').strip() or None
    if isinstance(name, str) and name.lower() in ('null', 'none', 'desconocido'):
        name = None
    return {
        'name': name,
        'category': category,
        'confidence': _normalize_confidence(parsed.get('confidence')),
        'available': bool(name or category),
        'source': 'vision',
    }


def validate_photo(image_bytes=None, image_base64=None):
    """Validate that a photo shows an auto part with usable quality.

    Returns ``{ok, issues, quality, available}``. When the provider is down the
    photo is NOT validated: ``available=False`` and ``ok=True`` (do not block).
    """
    if not image_bytes and not image_base64:
        return {'ok': False, 'issues': ['No se recibió imagen'], 'available': False}

    prompt = (
        'Evalúa esta foto destinada a un catálogo de autopartes usadas. Responde SOLO '
        'con JSON válido: {"is_part": true, "quality": "good", '
        '"issues": [], "confidence": 0.0}. '
        '"quality" debe ser "good", "fair" o "poor". En "issues" enumera problemas '
        '(borrosa, oscura, no se ve la pieza, fondo con ruido, etc.).'
    )
    parsed, error = _vision_json(prompt, image_bytes, image_base64)
    if not parsed:
        return {'ok': True, 'issues': [], 'available': False,
                'error': error or 'AI unavailable'}

    is_part = parsed.get('is_part')
    if isinstance(is_part, str):
        is_part = is_part.strip().lower() in ('true', '1', 'yes', 'sí', 'si')
    quality = (parsed.get('quality') or '').strip().lower() or None
    issues = parsed.get('issues') or []
    if isinstance(issues, str):
        issues = [issues]
    issues = [str(i).strip() for i in issues if str(i).strip()]

    if is_part is False:
        issues.append('No parece una autoparte')
    if quality == 'poor':
        issues.append('Calidad de imagen deficiente')

    return {
        'ok': bool(is_part) and quality != 'poor',
        'issues': issues,
        'quality': quality,
        'confidence': _normalize_confidence(parsed.get('confidence')),
        'available': True,
    }


def _local_comparables(conn, piece, limit=100):
    """Median price_1 of active local items sharing category or name tokens."""
    name = (piece.get('name') or '').strip()
    tokens = [t for t in re.split(r'\W+', name.lower()) if len(t) >= 4][:3]
    patterns = [f'%{t}%' for t in tokens]
    category_id = piece.get('category_id')
    piece_id = piece.get('id') or 0

    cur = conn.cursor()
    try:
        if category_id is None:
            cur.execute("SELECT category_id FROM inventory WHERE id = %s", (piece_id,))
            row = cur.fetchone()
            category_id = row[0] if row else None
        cur.execute("""
            SELECT i.price_1
            FROM inventory i
            WHERE i.is_active = true AND i.price_1 > 0
              AND i.id <> %s
              AND (i.category_id = %s OR i.name ILIKE ANY(%s))
            LIMIT %s
        """, (piece_id, category_id, patterns, limit))
        prices = [float(r[0]) for r in cur.fetchall() if r[0] is not None]
    except Exception:
        try:
            conn.rollback()
        except Exception:
            pass
        prices = []
    finally:
        cur.close()
    return prices


def suggest_price(conn, piece):
    """Suggest a sale price for a used piece.

    Rule engine: local comparables (median price of active items in the same
    category / with a shared name token) scaled by grade, blended with a
    cost-based reference (acquisition cost × grade factor × target markup).

    Returns ``{suggested_price, low, high, basis, available}``.
    """
    piece = piece or {}
    grade = (piece.get('condition_grade') or 'B').upper()
    grade = grade if grade in GRADE_FACTOR else 'B'
    factor = GRADE_FACTOR[grade]

    cost = _coerce_float(piece.get('acquisition_cost') or piece.get('cost') or 0)
    current_price = _coerce_float(piece.get('price_1') or 0)

    comparables = _local_comparables(conn, piece)
    market_ref = statistics.median(comparables) if comparables else None
    market_price = market_ref * factor if market_ref else None
    cost_price = cost * factor * TARGET_MARKUP if cost > 0 else None

    parts = []
    if market_price is not None:
        parts.append((MARKET_WEIGHT, market_price))
        parts.append((1 - MARKET_WEIGHT, cost_price if cost_price else market_price))
    elif cost_price is not None:
        parts = [(1.0, cost_price)]

    if parts:
        total_w = sum(w for w, _ in parts)
        suggested = sum(w * v for w, v in parts) / total_w
        available = True
    else:
        suggested = current_price
        available = bool(current_price)

    suggested = round(suggested, 2)
    low = round(suggested * 0.85, 2)
    high = round(suggested * 1.2, 2)

    basis_bits = [f'grado {grade}', f'costo {cost:.2f}']
    if comparables:
        basis_bits.append(
            f'{len(comparables)} comparables locales (mediana {market_ref:.2f})'
        )
    else:
        basis_bits.append('sin comparables locales')
    if not available:
        basis_bits.append('sin costo ni comparables: se usó el precio actual')

    return {
        'suggested_price': suggested,
        'low': low,
        'high': high,
        'basis': ', '.join(basis_bits),
        'available': available,
        'grade': grade,
        'comparables': len(comparables),
    }


def find_duplicate_groups(conn, threshold=0.85, limit=300):
    """Group used pieces whose names are semantically almost identical.

    Uses ``semantic_search.embed`` (one call for every name) and cosine
    similarity, then union-find. Returns ``{'groups': [[{id,part_number,name,
    donor_vin}]]}``. Degrades to ``groups=[]`` + ``available=False`` when
    embeddings are not available.
    """
    try:
        threshold = float(threshold)
    except (TypeError, ValueError):
        threshold = 0.85
    threshold = max(0.0, min(1.0, threshold))

    cur = conn.cursor()
    cur.execute("""
        SELECT id, part_number, name, donor_vin
        FROM inventory
        WHERE is_active = true AND condition IN ('usado', 'reconstruido')
          AND name IS NOT NULL AND name <> ''
        ORDER BY id
        LIMIT %s
    """, (limit,))
    rows = cur.fetchall()
    cur.close()

    items = [{'id': r[0], 'part_number': r[1], 'name': r[2], 'donor_vin': r[3]}
             for r in rows]
    if len(items) < 2:
        return {'groups': [], 'available': True, 'threshold': threshold}

    try:
        from services import semantic_search
        vectors = semantic_search.embed([it['name'] for it in items])
    except Exception:
        vectors = None
    if not vectors or len(vectors) != len(items):
        return {'groups': [], 'available': False, 'threshold': threshold,
                'error': 'embeddings no disponibles'}

    parent = list(range(len(items)))

    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(a, b):
        ra, rb = find(a), find(b)
        if ra != rb:
            parent[rb] = ra

    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            sim = semantic_search.cosine_similarity(vectors[i], vectors[j])
            if sim is not None and sim >= threshold:
                union(i, j)

    grouped = {}
    for idx in range(len(items)):
        grouped.setdefault(find(idx), []).append(items[idx])
    groups = [g for g in grouped.values() if len(g) >= 2]
    groups.sort(key=len, reverse=True)
    return {'groups': groups, 'available': True, 'threshold': threshold}