From 078128f7b144911ff856d2041547be11e31c2da1 Mon Sep 17 00:00:00 2001 From: Alvaro Romero Date: Fri, 3 Jul 2026 13:12:11 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20VIN-specific=20data=20=E2=80=94=20recal?= =?UTF-8?q?l=20check,=20safety=20ratings,=20full=20VPIC=20specs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 22 ++++- src/ai/ollama_analyzer.py | 46 ++++++++- src/api/nhtsa.py | 102 +++++++++++++++----- src/report/pdf.py | 191 ++++++++++++++++++++++++++++++++++---- 4 files changed, 315 insertions(+), 46 deletions(-) diff --git a/main.py b/main.py index 8ab492b..dcd153b 100644 --- a/main.py +++ b/main.py @@ -19,7 +19,10 @@ from fastapi.templating import Jinja2Templates from src.ai.ollama_analyzer import analyze_vehicle from src.api.epa import fetch_epa_data -from src.api.nhtsa import decode_vin, fetch_complaints, fetch_investigations, fetch_recalls +from src.api.nhtsa import ( + decode_vin, fetch_complaints, fetch_investigations, fetch_recalls, + fetch_vin_recalls, fetch_safety_ratings, +) from src.report.pdf import generate_pdf from src.utils.risk import calculate_risk, compute_extended_fields from src.utils.validator import validate_vin @@ -114,15 +117,20 @@ async def fetch( yield evt(f"✅ Vehículo: {year} {make} {model}", "success") await asyncio.sleep(0.05) - # 3 — Recalls + Complaints + Investigations (paralelo) - yield evt("⏳ Consultando recalls, quejas e investigaciones...", "progress") - recalls, complaints, invests = await asyncio.gather( + # 3 — Recalls + Complaints + Investigations + VIN check + Safety (paralelo) + yield evt("⏳ Consultando NHTSA: recalls, quejas, VIN check, seguridad...", "progress") + recalls, complaints, invests, vin_recalls, safety_ratings = await asyncio.gather( fetch_recalls(make, model, year), fetch_complaints(make, model, year), fetch_investigations(make, model, year), + fetch_vin_recalls(vin_clean), + fetch_safety_ratings(year, make, model), ) + open_count = len(vin_recalls) + stars = safety_ratings.get("overall", "") + stars_txt = f" · {stars}★ seguridad" if stars and stars != "NR" else "" yield evt( - f"✅ {len(recalls)} recalls · {len(complaints)} quejas · {len(invests)} investigaciones", + f"✅ {len(recalls)} recalls · {len(complaints)} quejas · {open_count} VIN-recalls{stars_txt}", "success", ) await asyncio.sleep(0.05) @@ -185,6 +193,8 @@ async def fetch( "risk": risk, "recalls": recalls, "complaints": complaints, + "vin_recalls": vin_recalls, + "safety_ratings": safety_ratings, "critical_recall_count": extended["critical_recall_count"], "engine_complaints": extended["engine_complaints"], "transmission_complaints": extended["transmission_complaints"], @@ -212,6 +222,8 @@ async def fetch( "recalls": recalls, "complaints": complaints, "investigations": invests, + "vin_recalls": vin_recalls, + "safety_ratings": safety_ratings, "epa": epa, "risk": risk, "ai_analysis": ai_analysis, diff --git a/src/ai/ollama_analyzer.py b/src/ai/ollama_analyzer.py index f0949b1..ea624a7 100644 --- a/src/ai/ollama_analyzer.py +++ b/src/ai/ollama_analyzer.py @@ -48,6 +48,39 @@ def _build_prompt(data: dict) -> str: trans_q = data.get("transmission_complaints", 0) alerts_text = "; ".join(alerts) if alerts else "Ninguna" + # VIN-specific recalls + vin_recalls = data.get("vin_recalls", []) + open_vr = [r for r in vin_recalls + if not (r.get("RemedyAvailability") or "").lower().startswith("remedy")] + vin_recall_text = ( + f" ⚠ {len(open_vr)} SIN REMEDIAR de {len(vin_recalls)} total" + if vin_recalls else " Ninguno registrado para este VIN" + ) + + # Safety ratings + sr = data.get("safety_ratings", {}) + if sr: + safety_text = ( + f" Global: {sr.get('overall','NR')}★ | " + f"Frontal conductor: {sr.get('frontal_driver','NR')}★ | " + f"Lateral: {sr.get('side_driver','NR')}★ | " + f"Vuelco: {sr.get('rollover','NR')}★" + ) + else: + safety_text = " Sin datos de seguridad NHTSA" + + # VPIC specs + cyl = vpic.get("Cylinders", "") + disp = vpic.get("DisplacementL", "") + hp = vpic.get("Horsepower", "") + specs = " / ".join(filter(None, [ + f"{cyl} cil." if cyl else "", + f"{disp}L" if disp else "", + f"{hp}HP" if hp else "", + vpic.get("FuelType", ""), + vpic.get("DriveType", ""), + ])) or "N/A" + vehicle = f"{s(vpic.get('ModelYear'))} {s(vpic.get('Make'))} {s(vpic.get('Model'))} {s(vpic.get('Trim'))}".strip() return f"""Eres un experto en evaluación de vehículos usados en subastas americanas (Copart, IAAI). @@ -60,8 +93,11 @@ Sé directo, técnico y específico. Explica el SIGNIFICADO real de cada dato, n ═══════════════════════════════════════════ DATOS DEL VEHÍCULO ═══════════════════════════════════════════ -Vehículo : {vehicle} +Vehículo : {vehicle} VIN : {s(data.get('vin'))} +Specs : {specs} +Carrocería: {s(vpic.get('BodyClass'))} · {s(vpic.get('Doors'))} puertas +Transmisión: {s(vpic.get('Transmission'))} Odómetro : {s(data.get('odometer'))} millas Daño prim.: {s(data.get('primary_damage'))} Daño sec. : {s(data.get('secondary_damage'))} @@ -69,9 +105,15 @@ Título : {s(data.get('title'))} Bid actual: ${s(data.get('bid'))} Score riesgo: {risk.get('score', 'N/A')}/100 -RECALLS NHTSA ({risk.get('recall_count', 0)} total, {critical_count} críticos de motor/transmisión/frenos): +RECALLS NHTSA DEL MODELO ({risk.get('recall_count', 0)} total, {critical_count} críticos): {recalls_text} +RECALLS ESPECÍFICOS DE ESTE VIN: +{vin_recall_text} + +CALIFICACIONES DE SEGURIDAD (5★ NHTSA): +{safety_text} + QUEJAS DE PROPIETARIOS ({risk.get('complaint_count', 0)} total): {complaint_breakdown} → Quejas de motor: {engine_q} diff --git a/src/api/nhtsa.py b/src/api/nhtsa.py index 3d26e6d..d330fff 100644 --- a/src/api/nhtsa.py +++ b/src/api/nhtsa.py @@ -1,30 +1,31 @@ import httpx -_VPIC_BASE = "https://vpic.nhtsa.dot.gov/api/vehicles" +_VPIC_BASE = "https://vpic.nhtsa.dot.gov/api/vehicles" _NHTSA_BASE = "https://api.nhtsa.gov" -_VPIC_FIELDS = { - "Make", "Model", "Model Year", "Trim", "Body Class", - "Displacement (L)", "Engine Number of Cylinders", "Fuel Type - Primary", - "Drive Type", "Transmission Style", "Plant Country", - "Electrification Level", "Engine Model", +_FIELD_MAP = { + "Make": "Make", + "Model": "Model", + "Model Year": "ModelYear", + "Trim": "Trim", + "Body Class": "BodyClass", + "Doors": "Doors", + "Displacement (L)": "DisplacementL", + "Engine Number of Cylinders": "Cylinders", + "Engine Horsepower": "Horsepower", + "Engine Model": "EngineModel", + "Fuel Type - Primary": "FuelType", + "Drive Type": "DriveType", + "Transmission Style": "Transmission", + "Brake System Type": "BrakeSystem", + "Gross Vehicle Weight Rating": "GVWR", + "Plant Country": "PlantCountry", + "Plant City": "PlantCity", + "Electrification Level": "EVLevel", + "Series": "Series", } -_FIELD_MAP = { - "Make": "Make", - "Model": "Model", - "Model Year": "ModelYear", - "Trim": "Trim", - "Body Class": "BodyClass", - "Displacement (L)": "DisplacementL", - "Engine Number of Cylinders": "Cylinders", - "Fuel Type - Primary": "FuelType", - "Drive Type": "DriveType", - "Transmission Style": "Transmission", - "Plant Country": "PlantCountry", - "Electrification Level": "EVLevel", - "Engine Model": "EngineModel", -} +_SKIP_VALS = {"Not Applicable", "null", "None", "0", ""} async def decode_vin(vin: str) -> dict: @@ -41,12 +42,69 @@ async def decode_vin(vin: str) -> dict: for item in data.get("Results", []): var = item.get("Variable", "") val = (item.get("Value") or "").strip() - if var in _FIELD_MAP and val and val not in ("Not Applicable", "null", "None"): + if var in _FIELD_MAP and val and val not in _SKIP_VALS: result[_FIELD_MAP[var]] = val return result +async def fetch_vin_recalls(vin: str) -> list: + """Returns recalls that apply to this specific VIN (not just make/model/year).""" + url = f"{_NHTSA_BASE}/recalls/recallsByVehicleId" + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.get(url, params={"vin": vin}) + resp.raise_for_status() + return resp.json().get("results", []) + except Exception: + return [] + + +async def fetch_safety_ratings(year: str, make: str, model: str) -> dict: + """Returns NHTSA 5-star crash test ratings for the vehicle.""" + if not (year and make and model): + return {} + try: + async with httpx.AsyncClient(timeout=15) as client: + # Step 1: get list of vehicle variants + r1 = await client.get( + f"{_NHTSA_BASE}/SafetyRatings/modelyear/{year}/make/{make}/model/{model}" + ) + r1.raise_for_status() + variants = r1.json().get("Results", []) + if not variants: + return {} + + vehicle_id = variants[0].get("VehicleId") + if not vehicle_id: + return {} + + # Step 2: get ratings for that variant + r2 = await client.get(f"{_NHTSA_BASE}/SafetyRatings/VehicleId/{vehicle_id}") + r2.raise_for_status() + results = r2.json().get("Results", []) + if not results: + return {} + + r = results[0] + return { + "vehicle_desc": r.get("VehicleDescription", ""), + "overall": r.get("OverallRating", "NR"), + "frontal_driver": r.get("FrontCrashDriversideRating", "NR"), + "frontal_passenger": r.get("FrontCrashPassengersideRating", "NR"), + "side_driver": r.get("SideCrashDriversideRating", "NR"), + "side_passenger": r.get("SideCrashPassengersideRating", "NR"), + "rollover": r.get("RolloverRating", "NR"), + "rollover_risk_pct": r.get("RolloverPossibility", ""), + "pole_crash": r.get("SidePoleCrashRating", "NR"), + "esc": r.get("NHTSAElectronicStabilityControl", ""), + "fcw": r.get("NHTSAForwardCollisionWarning", ""), + "ldw": r.get("NHTSALaneDepartureWarning", ""), + } + except Exception: + return {} + + async def fetch_recalls(make: str, model: str, year: str) -> list: url = f"{_NHTSA_BASE}/recalls/recallsByVehicle" params = {"make": make, "model": model, "modelYear": year} diff --git a/src/report/pdf.py b/src/report/pdf.py index ebd916e..4afc0bb 100644 --- a/src/report/pdf.py +++ b/src/report/pdf.py @@ -145,14 +145,37 @@ def _complaints_chart(complaints: list) -> io.BytesIO | None: # ── Main generator ──────────────────────────────────────────────────────────── +def _stars(rating: str, max_stars: int = 5) -> str: + """Return filled/empty star string, e.g. '★★★★☆ (4)'.""" + try: + n = int(rating) + return "★" * n + "☆" * (max_stars - n) + f" ({n}/{max_stars})" + except (ValueError, TypeError): + return "NR (sin datos)" + + +def _star_color(rating: str) -> HexColor: + try: + n = int(rating) + if n >= 4: + return C_GREEN + if n == 3: + return C_YELL + return C_RED + except (ValueError, TypeError): + return DGRAY + + def generate_pdf(data: dict) -> str: - vin = data["vin"] - vpic = data.get("vpic", {}) - recalls = data.get("recalls", []) - complaints = data.get("complaints", []) - invests = data.get("investigations", []) - epa = data.get("epa", {}) - risk = data.get("risk", {}) + vin = data["vin"] + vpic = data.get("vpic", {}) + recalls = data.get("recalls", []) + complaints = data.get("complaints", []) + invests = data.get("investigations", []) + vin_recalls = data.get("vin_recalls", []) + safety_ratings = data.get("safety_ratings", {}) + epa = data.get("epa", {}) + risk = data.get("risk", {}) make = vpic.get("Make", "Unknown") model = vpic.get("Model", "Unknown") @@ -214,17 +237,35 @@ def generate_pdf(data: dict) -> str: except Exception: photo_elem = None + # Build engine description + cyl = vpic.get("Cylinders", "") + disp = vpic.get("DisplacementL", "") + hp = vpic.get("Horsepower", "") + eng_model = vpic.get("EngineModel", "") + engine_str = " / ".join(filter(None, [ + f"{cyl} cil." if cyl else "", + f"{disp} L" if disp else "", + f"{hp} HP" if hp else "", + eng_model, + ])) or "—" + + plant_str = " / ".join(filter(None, [ + vpic.get("PlantCity", ""), vpic.get("PlantCountry", "") + ])) or "—" + id_rows = [ ["Campo", "Valor"], ["Año / Make / Model", f"{year} {make} {model}"], - ["Trim", vpic.get("Trim", "—")], - ["Carrocería", vpic.get("BodyClass", "—")], - ["Motor", f"{vpic.get('Cylinders','—')} cil. — {vpic.get('DisplacementL','—')} L"], + ["Trim / Serie", " / ".join(filter(None, [vpic.get("Trim",""), vpic.get("Series","")])) or "—"], + ["Carrocería / Puertas", f"{vpic.get('BodyClass','—')} · {vpic.get('Doors','—')} puertas"], + ["Motor", engine_str], ["Combustible", vpic.get("FuelType", "—")], ["Tracción", vpic.get("DriveType", "—")], ["Transmisión", vpic.get("Transmission", "—")], - ["Planta", vpic.get("PlantCountry", "—")], - ["EV Level", vpic.get("EVLevel", "—")], + ["Frenos", vpic.get("BrakeSystem", "—")], + ["GVWR", vpic.get("GVWR", "—")], + ["Planta ensamble", plant_str], + ["EV / Híbrido", vpic.get("EVLevel", "—")], ] if photo_elem: @@ -275,8 +316,43 @@ def generate_pdf(data: dict) -> str: r_rows.append([P(i), P(campaign), P(comp), P(summary)]) story.append(_table(r_rows, [0.25*inch, 1.0*inch, 1.9*inch, W-3.15*inch])) + # ── VIN-specific recall status ──────────────────────────────────────────── + story += _section("4 — ESTADO DE RECALL ESPECÍFICO PARA ESTE VIN") + + if vin_recalls: + open_vr = [r for r in vin_recalls if not (r.get("RemedyAvailability") or "").lower().startswith("remedy")] + fixed_vr = [r for r in vin_recalls if r not in open_vr] + + if open_vr: + story.append(_badge( + f"⚠️ {len(open_vr)} RECALL(S) ABIERTO(S) SIN REMEDIAR EN ESTE VIN", + "red", + )) + else: + story.append(_badge( + f"✅ {len(vin_recalls)} recall(s) registrado(s) — con remedio disponible", + "green", + )) + story.append(Spacer(1, 6)) + + PV = lambda txt, bold=False: Paragraph( + str(txt), + ParagraphStyle("vr", fontSize=7, fontName="Helvetica-Bold" if bold else "Helvetica", + textColor=NAVY2, leading=9, wordWrap="LTR"), + ) + vr_rows = [[PV("Campaña", True), PV("Componente", True), PV("Remedio disponible", True), PV("Consecuencia", True)]] + for r in vin_recalls[:10]: + campaign = r.get("NHTSACampaignNumber") or "—" + comp = (r.get("Component") or "—")[:80] + remedy = (r.get("RemedyAvailability") or "Pendiente")[:60] + conseq = (r.get("Consequence") or r.get("Summary") or "—")[:120] + vr_rows.append([PV(campaign), PV(comp), PV(remedy), PV(conseq)]) + story.append(_table(vr_rows, [0.9*inch, 1.7*inch, 1.4*inch, W-4.0*inch])) + else: + story.append(_badge("✅ Sin recalls registrados para este número de VIN", "green")) + # ── Complaints ──────────────────────────────────────────────────────────── - story += _section("4 — QUEJAS DE PROPIETARIOS") + story += _section("5 — QUEJAS DE PROPIETARIOS") story.append(Paragraph( f"Total de quejas registradas: {len(complaints)}", @@ -293,7 +369,7 @@ def generate_pdf(data: dict) -> str: story.append(Paragraph("Sin quejas registradas.", STYLE_BODY)) # ── Investigations ──────────────────────────────────────────────────────── - story += _section("5 — INVESTIGACIONES DE SEGURIDAD") + story += _section("6 — INVESTIGACIONES DE SEGURIDAD") if invests: PI = lambda txt, bold=False: Paragraph( @@ -312,7 +388,7 @@ def generate_pdf(data: dict) -> str: story.append(Paragraph("Sin investigaciones activas.", STYLE_BODY)) # ── EPA ─────────────────────────────────────────────────────────────────── - story += _section("6 — EFICIENCIA EPA") + story += _section("7 — EFICIENCIA EPA") epa_rows = [ ["Métrica", "Valor"], @@ -326,8 +402,89 @@ def generate_pdf(data: dict) -> str: epa_rows.append(["Autonomía eléctrica", f"{epa['range_miles']} mi"]) story.append(_table(epa_rows, [2 * inch, W - 2 * inch])) + # ── Safety Ratings ──────────────────────────────────────────────────────── + story += _section("8 — CALIFICACIONES DE SEGURIDAD NHTSA (5 ESTRELLAS)") + + STYLE_STAR = ParagraphStyle("star", fontSize=13, fontName="Helvetica-Bold", + textColor=NAVY2, alignment=TA_CENTER, leading=16) + STYLE_STAR_LBL = ParagraphStyle("starlbl", fontSize=7, fontName="Helvetica", + textColor=DGRAY, alignment=TA_CENTER) + + if safety_ratings: + vdesc = safety_ratings.get("vehicle_desc", f"{year} {make} {model}") + story.append(Paragraph(f"Datos de: {vdesc}", STYLE_BODY)) + story.append(Spacer(1, 6)) + + def _star_cell(label: str, rating: str) -> list: + color = _star_color(rating) + star_p = ParagraphStyle("sc", fontSize=12, fontName="Helvetica-Bold", + textColor=color, alignment=TA_CENTER, leading=15) + lbl_p = ParagraphStyle("sl", fontSize=6.5, fontName="Helvetica", + textColor=DGRAY, alignment=TA_CENTER) + return [Paragraph(_stars(rating), star_p), Paragraph(label, lbl_p)] + + # Overall rating — full-width badge + overall = safety_ratings.get("overall", "NR") + ov_color = _star_color(overall) + ov_p = ParagraphStyle("ov", fontSize=14, fontName="Helvetica-Bold", + textColor=C_WHITE, alignment=TA_CENTER) + ov_tbl = Table([[Paragraph(f"CALIFICACIÓN GLOBAL: {_stars(overall)}", ov_p)]], + colWidths=[W]) + ov_tbl.setStyle(TableStyle([ + ("BACKGROUND", (0, 0), (-1, -1), ov_color), + ("TOPPADDING", (0, 0), (-1, -1), 8), + ("BOTTOMPADDING", (0, 0), (-1, -1), 8), + ])) + story.append(ov_tbl) + story.append(Spacer(1, 8)) + + # Detail grid: 3 columns × 2 rows + col_w = W / 3 + detail_data = [ + [ + _star_cell("Choque Frontal\n(Conductor)", safety_ratings.get("frontal_driver", "NR")), + _star_cell("Choque Frontal\n(Pasajero)", safety_ratings.get("frontal_passenger", "NR")), + _star_cell("Choque Lateral\n(Conductor)", safety_ratings.get("side_driver", "NR")), + ], + [ + _star_cell("Choque Lateral\n(Pasajero)", safety_ratings.get("side_passenger", "NR")), + _star_cell("Poste Lateral", safety_ratings.get("pole_crash", "NR")), + _star_cell("Riesgo de Vuelco", safety_ratings.get("rollover", "NR")), + ], + ] + for row in detail_data: + row_tbl = Table(row, colWidths=[col_w, col_w, col_w]) + row_tbl.setStyle(TableStyle([ + ("GRID", (0, 0), (-1, -1), 0.4, MGRAY), + ("BACKGROUND", (0, 0), (-1, -1), LGRAY), + ("TOPPADDING", (0, 0), (-1, -1), 6), + ("BOTTOMPADDING", (0, 0), (-1, -1), 6), + ("ALIGN", (0, 0), (-1, -1), "CENTER"), + ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), + ])) + story.append(row_tbl) + story.append(Spacer(1, 4)) + + # Safety tech row + tech_items = [] + if safety_ratings.get("esc"): + tech_items.append(f"ESC: {safety_ratings['esc']}") + if safety_ratings.get("fcw"): + tech_items.append(f"FCW: {safety_ratings['fcw']}") + if safety_ratings.get("ldw"): + tech_items.append(f"LDW: {safety_ratings['ldw']}") + if safety_ratings.get("rollover_risk_pct"): + tech_items.append(f"Riesgo vuelco estadístico: {safety_ratings['rollover_risk_pct']}%") + if tech_items: + story.append(Paragraph("Tecnologías de seguridad: " + " · ".join(tech_items), STYLE_BODY)) + else: + story.append(Paragraph( + "Sin calificaciones de seguridad disponibles para este vehículo en NHTSA.", + STYLE_BODY, + )) + # ── Risk Score ──────────────────────────────────────────────────────────── - story += _section("7 — ANÁLISIS DE RIESGO") + story += _section("9 — ANÁLISIS DE RIESGO") score = risk.get("score", 0) level = risk.get("level", "N/A") @@ -344,7 +501,7 @@ def generate_pdf(data: dict) -> str: story.append(_table(f_rows, [W - 1.2 * inch, 1.2 * inch])) # ── AI Analysis ─────────────────────────────────────────────────────────── - story += _section("8 — ANÁLISIS INTELIGENTE (IA Local)") + story += _section("10 — ANÁLISIS INTELIGENTE (IA Local)") ai_text = data.get("ai_analysis")