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")