AR-VINchecker/src/api/nhtsa.py

import httpx

_VPIC_BASE  = "https://vpic.nhtsa.dot.gov/api/vehicles"
_NHTSA_BASE = "https://api.nhtsa.gov"

_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",
}

_SKIP_VALS = {"Not Applicable", "null", "None", "0", ""}


async def decode_vin(vin: str) -> dict:
    url = f"{_VPIC_BASE}/decodevin/{vin}?format=json"
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url)
            resp.raise_for_status()
            data = resp.json()
    except Exception:
        return {}

    result: 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 _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}
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
            return resp.json().get("results", [])
    except Exception:
        return []


async def fetch_complaints(make: str, model: str, year: str) -> list:
    url = f"{_NHTSA_BASE}/complaints/complaintsByVehicle"
    params = {"make": make, "model": model, "modelYear": year}
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
            return resp.json().get("results", [])
    except Exception:
        return []


def check_odometer_complaints(complaints: list) -> dict:
    """Filter already-fetched complaints for odometer/speedometer fraud mentions."""
    _ODO_KW = ("odometer", "speedometer", "mileage", "rollback", "rolled back", "fraud")
    hits = []
    for c in complaints:
        comp = (c.get("Component") or c.get("components") or c.get("component") or "").lower()
        desc = (c.get("summary") or c.get("Summary") or c.get("description") or "").lower()
        if any(kw in comp or kw in desc for kw in _ODO_KW):
            hits.append(c)
    return {"count": len(hits), "samples": hits[:3]}


async def fetch_investigations(make: str, model: str, year: str) -> list:
    url = f"{_NHTSA_BASE}/investigations/investigationsByVehicle"
    params = {"make": make, "model": model, "modelYear": year}
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
            return resp.json().get("results", [])
    except Exception:
        return []