261 lines
8.8 KiB
Python
261 lines
8.8 KiB
Python
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
|