"""DHL Shipment Tracking – Unified API. Kostenloser Key: https://developer.dhl.com (API "Shipment Tracking - Unified") """ import httpx API = "https://api-eu.dhl.com/track/shipments" STATUS = { "pre-transit": "inforeceived", "transit": "intransit", "delivered": "delivered", "failure": "failed", "unknown": "unknown", } def _place(loc: dict | None) -> str | None: addr = (loc or {}).get("address") if not addr: return None return ", ".join(filter(None, [addr.get("addressLocality"), addr.get("countryCode")])) or None async def track_dhl(number: str, key: str) -> dict | None: async with httpx.AsyncClient(timeout=15) as client: res = await client.get( API, params={"trackingNumber": number, "language": "de"}, headers={"DHL-API-Key": key, "Accept": "application/json"}, ) if res.status_code == 404: return None # keine DHL-Sendung -> nächsten Anbieter probieren res.raise_for_status() shipments = res.json().get("shipments") or [] if not shipments: return None s = shipments[0] status = s.get("status") or {} events = [ {"time": e.get("timestamp"), "description": e.get("description") or e.get("status"), "location": _place(e.get("location"))} for e in s.get("events") or [] ] eta = None if s.get("estimatedTimeOfDelivery"): eta = {"from": s["estimatedTimeOfDelivery"], "to": None} elif s.get("estimatedDeliveryTimeFrame"): frame = s["estimatedDeliveryTimeFrame"] eta = {"from": frame.get("estimatedFrom"), "to": frame.get("estimatedThrough")} return { "source": "DHL API", "carrier": "DHL Express" if s.get("service") == "express" else "DHL", "status": STATUS.get(status.get("statusCode"), "unknown"), "statusDetail": status.get("description") or status.get("status"), "eta": eta, "deliveredAt": status.get("timestamp") if status.get("statusCode") == "delivered" else None, "origin": _place(s.get("origin")), "destination": _place(s.get("destination")), "events": events, }