alro65/AR-VINchecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: VIN-specific data — recall check, safety ratings, full VPIC specs

Browse Source
This commit is contained in:
Alvaro Enrique Romero Donado
2026-07-03 13:12:11 -04:00
parent 3eee5b0acf
commit 078128f7b1
4 changed files with 315 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+17 -5
View File
@@ -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,
src/ai/ollama_analyzer.py
+43 -1
View File
@@ -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).
@@ -62,6 +95,9 @@ DATOS DEL VEHÍCULO
═══════════════════════════════════════════
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}
src/api/nhtsa.py
+67 -9
View File
@@ -3,29 +3,30 @@ import httpx
_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",
"Engine Model": "EngineModel",
"Series": "Series",
}
_SKIP_VALS = {"Not Applicable", "null", "None", "0", ""}
async def decode_vin(vin: str) -> dict:
url = f"{_VPIC_BASE}/decodevin/{vin}?format=json"
@@ -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}
src/report/pdf.py
+167 -10
View File
@@ -145,12 +145,35 @@ 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", [])
vin_recalls = data.get("vin_recalls", [])
safety_ratings = data.get("safety_ratings", {})
epa = data.get("epa", {})
risk = data.get("risk", {})
@@ -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: <b>{len(complaints)}</b>",
@@ -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: <b>{vdesc}</b>", 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")