consultoria-as
7b2a904498
- Migrate from SQLite to PostgreSQL with normalized schema - Add 11 lookup tables (fuel_type, body_type, drivetrain, transmission, materials, position_part, manufacture_type, quality_tier, countries, reference_type, shapes) - Rewrite dashboard/server.py (76 routes) using SQLAlchemy text() queries - Rewrite console/db.py (27 methods) using SQLAlchemy ORM - Add models.py with 27 SQLAlchemy model definitions - Add config.py for centralized DB_URL configuration - Add migrate_to_postgres.py migration script - Add docs/METABASE_GUIDE.md with complete data entry guide - Rebrand from "AUTOPARTS DB" to "NEXUS AUTOPARTS" - Fill vehicle data gaps via NHTSA API + heuristics: engines (cylinders, power, torque), brands (country, founded_year), models (body_type, production years), MYE (drivetrain, transmission, trim) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
"""
|
|
NHTSA VIN Decoder API client for the NEXUS AUTOPARTS console application.
|
|
|
|
Wraps the National Highway Traffic Safety Administration (NHTSA) Vehicle
|
|
Product Information Catalog (vPIC) DecodeVin endpoint to retrieve vehicle
|
|
specifications from a 17-character VIN.
|
|
"""
|
|
|
|
import requests
|
|
|
|
from console.config import NHTSA_API_URL
|
|
|
|
|
|
# NHTSA result variables we care about, mapped to our internal keys.
|
|
_FIELD_MAP = {
|
|
"Make": "make",
|
|
"Model": "model",
|
|
"Model Year": "year",
|
|
"Body Class": "body_class",
|
|
"Drive Type": "drive_type",
|
|
"Displacement (L)": "displacement_l",
|
|
"Engine Number of Cylinders": "cylinders",
|
|
"Fuel Type - Primary": "fuel_type",
|
|
"Engine Brake (hp) From": "power_hp",
|
|
}
|
|
|
|
|
|
def decode_vin_nhtsa(vin: str) -> dict:
|
|
"""Decode a VIN using the NHTSA vPIC API.
|
|
|
|
Parameters
|
|
----------
|
|
vin : str
|
|
A 17-character Vehicle Identification Number.
|
|
|
|
Returns
|
|
-------
|
|
dict
|
|
On success::
|
|
|
|
{
|
|
"make": "TOYOTA",
|
|
"model": "Corolla",
|
|
"year": "2020",
|
|
"body_class": "Sedan/Saloon",
|
|
"drive_type": "FWD",
|
|
"engine_info": {
|
|
"displacement_l": "2.0",
|
|
"cylinders": "4",
|
|
"fuel_type": "Gasoline",
|
|
"power_hp": "169",
|
|
"raw": { ... full variable->value mapping ... },
|
|
},
|
|
}
|
|
|
|
On error::
|
|
|
|
{"error": "<description>"}
|
|
"""
|
|
try:
|
|
url = f"{NHTSA_API_URL}/{vin}"
|
|
response = requests.get(url, params={"format": "json"}, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
results = data.get("Results", [])
|
|
|
|
# Build a flat lookup: variable name -> value (skip empty/None)
|
|
raw: dict[str, str] = {}
|
|
for item in results:
|
|
var = item.get("Variable", "")
|
|
val = item.get("Value")
|
|
if val and str(val).strip():
|
|
raw[var] = str(val).strip()
|
|
|
|
# Extract top-level vehicle fields
|
|
vehicle: dict = {}
|
|
engine_info: dict = {"raw": raw}
|
|
|
|
engine_keys = {"displacement_l", "cylinders", "fuel_type", "power_hp"}
|
|
|
|
for nhtsa_var, our_key in _FIELD_MAP.items():
|
|
value = raw.get(nhtsa_var, "")
|
|
if our_key in engine_keys:
|
|
engine_info[our_key] = value
|
|
else:
|
|
vehicle[our_key] = value
|
|
|
|
vehicle["engine_info"] = engine_info
|
|
return vehicle
|
|
|
|
except Exception as e:
|
|
return {"error": str(e)}
|