From 73fdb80f2935c8edb6543d65f11a5e5dc185bbd7 Mon Sep 17 00:00:00 2001 From: Matteo Giustini Date: Mon, 11 May 2026 18:20:08 +0200 Subject: [PATCH] Funzione primaria interazione API Motornet --- backend/app/integrations/motornet.py | 260 +++++++++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 backend/app/integrations/motornet.py diff --git a/backend/app/integrations/motornet.py b/backend/app/integrations/motornet.py new file mode 100644 index 0000000..68480f8 --- /dev/null +++ b/backend/app/integrations/motornet.py @@ -0,0 +1,260 @@ +from datetime import date +from dataclasses import dataclass, field +import httpx +import structlog + +from app.core.config import settings + +logger = structlog.get_logger() + + +class MotornetUnavailableError(Exception): + pass + + +@dataclass +class MotornetVersion: + motornet_code: str + version_label: str | None = None + body_type: str | None = None + doors: int | None = None + wheelbase: int | None = None + list_price: float | None = None + production_start: date | None = None + production_end: date | None = None + commercial_start: date | None = None + commercial_end: date | None = None + model_code: str | None = None + + +@dataclass +class MotornetPlateResult: + plate: str + vin: str | None = None + vehicle_type: str | None = None + registration_date: date | None = None + homologation_code: str | None = None + engine_code: str | None = None + last_revision_date: date | None = None + foreign_plate: str | None = None + foreign_registration_date: date | None = None + foreign_country: str | None = None + is_foreign_registered: bool = False + remaining_queries: int | None = None + brand_acronym: str | None = None + brand_name: str | None = None + model_code: str | None = None + model_description: str | None = None + gamma_code: str | None = None + gamma_description: str | None = None + series_code: str | None = None + series_description: str | None = None + historical_group_code: str | None = None + historical_group_desc: str | None = None + cod_desc_model_code: str | None = None + cod_desc_model_desc: str | None = None + production_start: date | None = None + production_end: date | None = None + commercial_start: date | None = None + commercial_end: date | None = None + versions: list[MotornetVersion] = field(default_factory=list) + raw_response: dict | None = None + + +def _parse_date(value: str | None) -> date | None: + if not value: + return None + try: + return date.fromisoformat(value[:10]) + except (ValueError, TypeError): + return None + + +def _parse_response(plate: str, data: dict) -> MotornetPlateResult: + brand = None + brand_name = None + if data.get("marche"): + m = data["marche"][0] + brand = m.get("acronimo") + brand_name = m.get("nome") + + model_code = None + model_description = None + gamma_code = None + gamma_description = None + series_code = None + series_description = None + historical_group_code = None + historical_group_desc = None + cod_desc_model_code = None + cod_desc_model_desc = None + prod_start = None + prod_end = None + comm_start = None + comm_end = None + + if data.get("modelli"): + mo = data["modelli"][0] + gm = mo.get("gammaModello") or {} + model_code = gm.get("codice") + model_description = gm.get("descrizione") + gs = mo.get("gruppoStorico") or {} + historical_group_code = gs.get("codice") + historical_group_desc = gs.get("descrizione") + sg = mo.get("serieGamma") or {} + series_code = sg.get("codice") + series_description = sg.get("descrizione") + cdm = mo.get("codDescModello") or {} + cod_desc_model_code = cdm.get("codice") + cod_desc_model_desc = cdm.get("descrizione") + prod_start = _parse_date(mo.get("inizioProduzione")) + prod_end = _parse_date(mo.get("fineProduzione")) + comm_start = _parse_date(mo.get("inizioCommercializzazione")) + comm_end = _parse_date(mo.get("fineCommercializzazione")) + + if data.get("gamme"): + ga = data["gamme"][0] + gamma_code = ga.get("codice") + gamma_description = ga.get("descrizione") + + versions = [] + for v in data.get("versioni") or []: + versions.append( + MotornetVersion( + motornet_code=v["codiceMotornetUnivoco"], + version_label=v.get("versione"), + body_type=v.get("tipo"), + doors=v.get("porte"), + wheelbase=v.get("passo"), + list_price=v.get("prezzoVendita"), + production_start=_parse_date(v.get("inizioProduzione")), + production_end=_parse_date(v.get("fineProduzione")), + commercial_start=_parse_date(v.get("da")), + commercial_end=_parse_date(v.get("a")), + model_code=v.get("codiceModello"), + ) + ) + + return MotornetPlateResult( + plate=plate, + vin=data.get("telaio"), + vehicle_type=data.get("tipoVeicolo"), + registration_date=_parse_date(data.get("dataImmatricolazione")), + homologation_code=data.get("codiceOmologazione"), + engine_code=data.get("codiceMotore"), + last_revision_date=_parse_date(data.get("dataUltimaRevisione")), + foreign_plate=data.get("targaEstero"), + foreign_registration_date=_parse_date(data.get("dataImmatricolazioneEstero")), + foreign_country=data.get("statoEstero"), + is_foreign_registered=data.get("immatricolazioneEstera", "no") == "si", + remaining_queries=data.get("ricercheTargaRimanenti"), + brand_acronym=brand, + brand_name=brand_name, + model_code=model_code, + model_description=model_description, + gamma_code=gamma_code, + gamma_description=gamma_description, + series_code=series_code, + series_description=series_description, + historical_group_code=historical_group_code, + historical_group_desc=historical_group_desc, + cod_desc_model_code=cod_desc_model_code, + cod_desc_model_desc=cod_desc_model_desc, + production_start=prod_start, + production_end=prod_end, + commercial_start=comm_start, + commercial_end=comm_end, + versions=versions, + raw_response=data, + ) + + +async def _get_token() -> str: + url = f"{settings.motornet_oauth_host}/token" + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.post( + url, + data={ + "grant_type": "password", + "client_id": "webservice", + "username": settings.motornet_user_id, + "password": settings.motornet_password, + }, + ) + resp.raise_for_status() + return resp.json()["access_token"] + + +async def valuate_vehicle( + motornet_code: str, + anno: int, + mese: int, + km: int | None = None, + targa: str | None = None, + telaio: str | None = None, +) -> dict: + try: + token = await _get_token() + url = f"{settings.motornet_host}/usato/auto/valutazione" + payload: dict = { + "codiceMotornetUnivoco": motornet_code, + "anno": anno, + "mese": mese, + } + if km is not None: + payload["km"] = km + if targa is not None: + payload["targa"] = targa + if telaio is not None: + payload["telaio"] = telaio + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post( + url, + json=payload, + headers={"Authorization": f"bearer {token}"}, + ) + resp.raise_for_status() + data = resp.json() + logger.info("motornet_valuation_ok", motornet_code=motornet_code, anno=anno, mese=mese) + return data + except httpx.HTTPStatusError as exc: + logger.warning( + "motornet_valuation_http_error", + motornet_code=motornet_code, + status=exc.response.status_code, + ) + raise MotornetUnavailableError( + f"MotorNet valutazione ha risposto con status {exc.response.status_code}" + ) from exc + except Exception as exc: + logger.warning("motornet_valuation_unavailable", motornet_code=motornet_code, error=str(exc)) + raise MotornetUnavailableError(str(exc)) from exc + + +async def lookup_plate(plate: str) -> MotornetPlateResult: + normalized = plate.upper().replace(" ", "") + try: + token = await _get_token() + url = f"{settings.motornet_host}/usato/generali/targa" + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.get( + url, + params={"targa": normalized}, + headers={"Authorization": f"bearer {token}"}, + ) + resp.raise_for_status() + data = resp.json() + logger.info("motornet_lookup_ok", plate=normalized) + return _parse_response(normalized, data) + except httpx.HTTPStatusError as exc: + logger.warning( + "motornet_http_error", + plate=normalized, + status=exc.response.status_code, + ) + raise MotornetUnavailableError( + f"MotorNet ha risposto con status {exc.response.status_code}" + ) from exc + except Exception as exc: + logger.warning("motornet_unavailable", plate=normalized, error=str(exc)) + raise MotornetUnavailableError(str(exc)) from exc