Autoparts-DB/pos/services/part_vision.py

# /home/Autopartes/pos/services/part_vision.py
"""A4 — Vision grading for used/yonke parts.

Given a photo of a used part (plus optional metadata: name, category,
declared condition) a VLM suggests a condition grade:

    A = excelente   (sin daños, funcional con seguridad)
    B = bueno       (desgaste normal de uso, funcional)
    C = regular     (desgaste/defecto menor, requiere revisión o reparación)
    D = con daño    (daño estructural visible / funcionalidad dudosa)

Backends (reused from ``services.ai_chat``): QWEN -> Hermes -> OpenRouter.
When no backend answers (e.g. the current QWEN key returns 401) the function
degrades gracefully: it never raises and returns a dict with
``condition_grade_suggested=None`` and a human-readable ``error``.

Public API:
    suggest_grade(image_bytes=None, image_base64=None, name=None,
                  category=None, condition='usado', notes=None) -> dict
"""

import base64
import json
import re

GRADES = ('A', 'B', 'C', 'D')

GRADE_RUBRIC = """Escala de grados para piezas usadas de yonke:
- A (excelente): sin daños visibles, no requiere reparación; funcionamiento prácticamente seguro.
- B (bueno): desgaste normal por uso, sin daños estructurales; funciona; puede tener detalles cosméticos (rayones leves, óxido superficial).
- C (regular): desgaste notable o defecto menor que puede requerir reparación o cambio de componentes secundarios; funcionalidad probable pero a verificar.
- D (con daño): daño visible estructural, pieza rota/incompleta/muy corroída, o funcionalidad dudosa; se vende como refacción o no funcional."""

GRADE_INSTRUCTIONS = """Evalúa la pieza de la foto y asigna un grado.

Reglas:
1. Basa el grado SOLO en lo observable en la foto. Si la foto no muestra la pieza, está borrosa o no es de una autoparte, baja la confianza y explícalo en "notes".
2. Sé conservador: ante duda razonable entre dos grados, asigna el peor (menor).
3. "defects" enumera defectos concretos y visibles (en español). Vacío si no hay.
4. "confidence" es tu certeza global del grado (0.0 a 1.0).

Responde EXCLUSIVAMENTE con JSON válido, sin texto adicional:
{
  "condition_grade_suggested": "A" | "B" | "C" | "D",
  "condition_suggested": "usado",
  "defects": ["..."],
  "confidence": 0.0,
  "notes": "una o dos frases en español"
}"""


def _build_prompt(name=None, category=None, condition=None):
    """Build the user prompt embedding the piece metadata."""
    lines = [GRADE_INSTRUCTIONS, "", GRADE_RUBRIC, ""]
    meta = []
    if name:
        meta.append(f"- Nombre declarado: {name}")
    if category:
        meta.append(f"- Categoría: {category}")
    if condition:
        meta.append(f"- Condición declarada: {condition}")
    if meta:
        lines.append("Metadatos de la pieza:")
        lines.extend(meta)
    else:
        lines.append("No hay metadatos de la pieza; evalúa solo lo visible.")
    return "\n".join(lines)


def _empty_result():
    return {
        'condition_grade_suggested': None,
        'condition_suggested': 'usado',
        'defects': [],
        'confidence': 0.0,
        'notes': '',
        'source': None,
        'error': None,
    }


def _extract_json_object(text):
    """Best-effort JSON extraction from an LLM reply.

    Handles markdown fences, leading/trailing prose and trailing commas.
    Returns a dict or None.
    """
    if not text:
        return None
    s = str(text).strip()
    if s.startswith('```'):
        s = re.sub(r'^```[A-Za-z0-9_-]*\s*', '', s)
        s = re.sub(r'```\s*$', '', s).strip()
    try:
        parsed = json.loads(s)
        return parsed if isinstance(parsed, dict) else None
    except (json.JSONDecodeError, ValueError):
        pass
    m = re.search(r'\{.*\}', s, re.DOTALL)
    if not m:
        return None
    frag = m.group(0)
    for candidate in (frag, re.sub(r',\s*([}\]])', r'\1', frag)):
        try:
            parsed = json.loads(candidate)
            if isinstance(parsed, dict):
                return parsed
        except (json.JSONDecodeError, ValueError):
            continue
    return None


def _normalize_grade(data):
    """Coerce a raw model dict into the public grade schema."""
    out = _empty_result()
    if not isinstance(data, dict):
        return out

    raw_grade = (data.get('condition_grade_suggested')
                 or data.get('condition_grade')
                 or data.get('grade'))
    grade = str(raw_grade).strip().upper()[:1] if raw_grade else None
    out['condition_grade_suggested'] = grade if grade in GRADES else None

    cond = str(data.get('condition_suggested') or data.get('condition') or 'usado').strip().lower()
    out['condition_suggested'] = 'reconstruido' if 'reconstr' in cond else 'usado'

    defects = data.get('defects')
    if isinstance(defects, str):
        defects = re.split(r'[;\n]|,', defects)
    if isinstance(defects, list):
        out['defects'] = [str(d).strip() for d in defects if str(d).strip()][:20]

    try:
        conf = float(data.get('confidence'))
        if conf > 1:
            conf = conf / 100.0
        out['confidence'] = max(0.0, min(1.0, conf))
    except (TypeError, ValueError):
        out['confidence'] = 0.0

    notes = data.get('notes') or data.get('message') or data.get('reason') or ''
    out['notes'] = str(notes).strip()[:1000]
    return out


def _is_usable(parsed):
    """True when a normalized dict carries a real grading signal.

    Deliberately ignores ``notes``: the shared vision helper falls back to a
    plain ``{"message": ...}`` dict on backend errors, and treating that as a
    valid grade would hide outages behind a misleading answer.
    """
    return bool(parsed.get('condition_grade_suggested') or parsed.get('defects'))


def _backends():
    """Ordered vision backends (url, key, model, tag) from ai_chat config."""
    from services import ai_chat

    out = []
    if getattr(ai_chat, 'QWEN_ENABLED', False):
        out.append((ai_chat.QWEN_CHAT_URL, ai_chat.QWEN_API_KEY, ai_chat.QWEN_MODEL, 'qwen'))
    if getattr(ai_chat, 'HERMES_ENABLED', False):
        out.append((ai_chat.HERMES_CHAT_URL, ai_chat.HERMES_API_KEY, ai_chat.HERMES_MODEL, 'hermes'))
    if getattr(ai_chat, 'OPENROUTER_API_KEY', None):
        out.append((ai_chat.OPENROUTER_URL, ai_chat.OPENROUTER_API_KEY, ai_chat.VISION_MODEL, 'openrouter'))
    return out


def suggest_grade(image_bytes=None, image_base64=None, name=None, category=None,
                  condition='usado', notes=None):
    """Suggest a condition grade for a used part photo.

    Returns:
        {
          "condition_grade_suggested": "A"|"B"|"C"|"D"|None,
          "condition_suggested": "usado"|"reconstruido",
          "defects": [str, ...],
          "confidence": float,          # 0.0 - 1.0
          "notes": str,
          "source": "qwen"|"hermes"|"openrouter"|None,
          "error": str|None             # set when no grade could be produced
        }
    """
    result = _empty_result()

    if not image_bytes and not image_base64:
        result['error'] = 'No se proporcionó imagen de la pieza.'
        return result

    b64 = image_base64
    if not b64:
        try:
            b64 = base64.b64encode(image_bytes).decode('ascii')
        except Exception as e:  # pragma: no cover - defensive
            result['error'] = f'Imagen inválida: {e}'
            return result
    b64_clean = b64.split(',', 1)[1] if b64.startswith('data:') else b64

    prompt = _build_prompt(name=name, category=category, condition=condition)

    # 1) Reuse the shared vision entry point (handles its own JSON parsing).
    try:
        from services import ai_chat
        res = ai_chat.chat_with_image(prompt, b64_clean)
        if isinstance(res, dict):
            parsed = _normalize_grade(res)
            if not _is_usable(parsed) and res.get('message'):
                extracted = _extract_json_object(res.get('message'))
                if extracted:
                    parsed = _normalize_grade(extracted)
            if _is_usable(parsed):
                parsed['source'] = 'vision'
                return parsed
    except Exception as e:
        print(f"[PART_VISION] chat_with_image failed: {e}")

    # 2) Direct OpenAI-compatible call with each backend's own tag.
    backends = _backends()
    if not backends:
        result['error'] = ('Servicio de visión no configurado: sin backend de IA disponible '
                           '(QWEN/Hermes/OpenRouter).')
        return result

    messages = [
        {"role": "system", "content": "Eres un perito de autopartes usadas. Responde SOLO en JSON válido."},
        {"role": "user", "content": [
            {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64," + b64_clean}},
            {"type": "text", "text": prompt},
        ]},
    ]

    last_error = None
    soft = None  # valid JSON with notes but no grade (model refused/unsure)
    for url, key, model, tag in backends:
        try:
            from services import ai_chat
            res = ai_chat._post_chat_completion(
                url, key, model, messages, max_tokens=500, temperature=0.1, timeout=45
            )
        except Exception as e:
            last_error = f'{tag}: {e}'
            continue
        if not res:
            last_error = f'{tag}: sin respuesta (ver logs [AI])'
            continue
        parsed = _extract_json_object(res.get('content'))
        if not parsed:
            last_error = f'{tag}: respuesta no-JSON'
            continue
        normalized = _normalize_grade(parsed)
        if _is_usable(normalized):
            normalized['source'] = tag
            return normalized
        if soft is None and normalized.get('notes'):
            normalized['source'] = tag
            soft = normalized
        last_error = f'{tag}: JSON sin grado'

    if soft is not None:
        soft['error'] = 'El modelo no asignó un grado; revisa notes.'
        return soft

    result['error'] = ('El servicio de visión no pudo graduar la pieza '
                       f'({last_error or "sin backend disponible"}).')
    return result