guyg1474andClaude Opus 5.5
dd58b3d1f8
FastAPI-Backend und React-Frontend mit Konten, Sendungsverfolgung, Projekt-Artikeln, Vorrat & Einkauf mit Rezeptabgleich, Reise-Weltkarte, Kanban-Projekten mit Priorität, Server-Monitoring und den Designs Matrix und Blackwall. Betrieb per Docker hinter nginx (siehe DEPLOY.md). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
"""17TRACK Tracking API v2.2 – https://api.17track.net
|
||
|
||
Nummern müssen erst registriert werden, danach lassen sich die Daten abrufen.
|
||
Bei neu registrierten Nummern dauert es oft ein paar Sekunden, bis Daten da sind.
|
||
"""
|
||
import httpx
|
||
|
||
API = "https://api.17track.net/track/v2.2"
|
||
ALREADY_REGISTERED = -18019901
|
||
|
||
STATUS = {
|
||
"NotFound": "notfound",
|
||
"InfoReceived": "inforeceived",
|
||
"InTransit": "intransit",
|
||
"Expired": "expired",
|
||
"AvailableForPickup": "pickup",
|
||
"OutForDelivery": "outfordelivery",
|
||
"DeliveryFailure": "failed",
|
||
"Delivered": "delivered",
|
||
"Exception": "exception",
|
||
}
|
||
|
||
|
||
async def _call(client: httpx.AsyncClient, endpoint: str, key: str, body: list) -> dict:
|
||
res = await client.post(f"{API}/{endpoint}", headers={"17token": key}, json=body)
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"API-Code {data.get('code')}")
|
||
return data["data"]
|
||
|
||
|
||
def _location(event: dict) -> str | None:
|
||
if event.get("location"):
|
||
return event["location"]
|
||
addr = event.get("address") or {}
|
||
return ", ".join(filter(None, [addr.get("city"), addr.get("country")])) or None
|
||
|
||
|
||
async def track_17(number: str, key: str) -> dict:
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
data = await _call(client, "gettrackinfo", key, [{"number": number}])
|
||
|
||
if not data.get("accepted"):
|
||
# Noch nicht registriert -> registrieren (17TRACK erkennt den Paketdienst selbst)
|
||
reg = await _call(client, "register", key, [{"number": number}])
|
||
rejected = (reg.get("rejected") or [None])[0]
|
||
if rejected and (rejected.get("error") or {}).get("code") != ALREADY_REGISTERED:
|
||
raise RuntimeError((rejected.get("error") or {}).get("message") or "Registrierung abgelehnt")
|
||
data = await _call(client, "gettrackinfo", key, [{"number": number}])
|
||
|
||
item = (data.get("accepted") or [None])[0]
|
||
if not item or not item.get("track_info"):
|
||
return {"source": "17TRACK", "status": "pending"}
|
||
|
||
info = item["track_info"]
|
||
provider = ((info.get("tracking") or {}).get("providers") or [{}])[0]
|
||
events = [
|
||
{
|
||
"time": e.get("time_iso") or e.get("time_utc"),
|
||
"description": e.get("description"),
|
||
"location": _location(e),
|
||
}
|
||
for e in provider.get("events") or []
|
||
]
|
||
latest = info.get("latest_status") or {}
|
||
eta = (info.get("time_metrics") or {}).get("estimated_delivery_date") or {}
|
||
shipping = info.get("shipping_info") or {}
|
||
|
||
return {
|
||
"source": "17TRACK",
|
||
"carrier": (provider.get("provider") or {}).get("name"),
|
||
"status": STATUS.get(latest.get("status"), "unknown"),
|
||
"statusDetail": latest.get("sub_status"),
|
||
"eta": {"from": eta.get("from"), "to": eta.get("to")} if eta.get("from") or eta.get("to") else None,
|
||
"deliveredAt": events[0]["time"] if latest.get("status") == "Delivered" and events else None,
|
||
"origin": (shipping.get("shipper_address") or {}).get("country"),
|
||
"destination": (shipping.get("recipient_address") or {}).get("country"),
|
||
"events": events,
|
||
}
|