webhook_receiver_rev.py

@@ -41,7 +41,7 @@ from typing import Optional
 # monopolising a worker or blowing the DB pool. Jimi normally sends ≤ 200.
 MAX_ITEMS_PER_POST = int(os.getenv("WEBHOOK_MAX_ITEMS", "5000"))
 
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import FastAPI, Form, HTTPException
 from fastapi.responses import JSONResponse
 from psycopg2.extras import execute_values
 
@@ -86,7 +86,7 @@ def _validate_token(token: str) -> None:
 
 
 def _parse_data_list(raw: str) -> list[dict]:
-    """Parse a JSON string into a list of dicts. raw may be a JSON array or single object."""
+    """Parse the JSON string from Jimi's data_list form field."""
     try:
         parsed = json.loads(raw)
         items = parsed if isinstance(parsed, list) else [parsed]
@@ -96,59 +96,10 @@ def _parse_data_list(raw: str) -> list[dict]:
             items = items[:MAX_ITEMS_PER_POST]
         return items
     except (json.JSONDecodeError, TypeError):
+        log.warning("Failed to parse data_list: %.200s", raw)
         return []
 
 
-async def _parse_request(request: Request) -> tuple[str, list[dict]]:
-    """Extract token + data_list from either a JSON body or form-encoded body.
-
-    Jimi's integration push API sends:
-      Content-Type: application/json
-      Body: {"token": "...", "data_list": [{...}, ...]}
-
-    Some older/configured endpoints may still use form-encoded. This helper
-    handles both so each endpoint doesn't need to know which format arrived.
-    """
-    content_type = request.headers.get("content-type", "")
-    body = await request.body()
-
-    if not body:
-        log.debug("push: empty request body")
-        return "", []
-
-    # ── Try JSON body first (integration push format) ──────────────────────────
-    if "application/json" in content_type or body.lstrip()[:1] == b"{":
-        try:
-            payload = json.loads(body)
-            token = str(payload.get("token", ""))
-            raw_dl = payload.get("data_list", [])
-            if isinstance(raw_dl, list):
-                items = raw_dl[:MAX_ITEMS_PER_POST]
-            elif isinstance(raw_dl, str):
-                items = _parse_data_list(raw_dl)
-            else:
-                items = []
-            log.debug("push: parsed JSON body — %d items", len(items))
-            return token, items
-        except (json.JSONDecodeError, TypeError):
-            log.warning("push: JSON body parse failed — body: %.200s",
-                        body.decode("utf-8", errors="replace"))
-
-    # ── Fall back to form-encoded ───────────────────────────────────────────────
-    try:
-        form = await request.form()
-        token = str(form.get("token", ""))
-        raw_dl = str(form.get("data_list", ""))
-        items = _parse_data_list(raw_dl) if raw_dl else []
-        log.debug("push: parsed form body — %d items", len(items))
-        return token, items
-    except Exception:
-        log.warning("push: form body parse failed — body: %.200s",
-                    body.decode("utf-8", errors="replace"), exc_info=True)
-
-    return "", []
-
-
 def unix_to_ts(v) -> Optional[str]:
     """Convert Unix timestamp (seconds or milliseconds) to ISO string."""
     if v is None:
@@ -194,9 +145,9 @@ def health():
 # ── 1. OBD Diagnostics (Priority 1) ──────────────────────────────────────────
 
 @app.post("/pushobd")
-async def push_obd(request: Request):
+def push_obd(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -265,9 +216,9 @@ async def push_obd(request: Request):
 # ── 2. DTC Fault Codes (Priority 1) ──────────────────────────────────────────
 
 @app.post("/pushfaultinfo")
-async def push_fault_info(request: Request):
+def push_fault_info(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -335,9 +286,9 @@ async def push_fault_info(request: Request):
 # ── 3. Alarm Events (Priority 2) ─────────────────────────────────────────────
 
 @app.post("/pushalarm")
-async def push_alarm(request: Request):
+def push_alarm(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -392,9 +343,9 @@ async def push_alarm(request: Request):
 # ── 4. GPS Positions (Priority 2) ────────────────────────────────────────────
 
 @app.post("/pushgps")
-async def push_gps(request: Request):
+def push_gps(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -456,9 +407,9 @@ async def push_gps(request: Request):
 # ── 5. Device Heartbeats (Priority 2) ────────────────────────────────────────
 
 @app.post("/pushhb")
-async def push_heartbeat(request: Request):
+def push_heartbeat(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -505,9 +456,9 @@ async def push_heartbeat(request: Request):
 # ── 6. Trip Reports (Priority 2) ─────────────────────────────────────────────
 
 @app.post("/pushtripreport")
-async def push_trip_report(request: Request):
+def push_trip_report(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)
 
@@ -580,9 +531,9 @@ async def push_trip_report(request: Request):
 # ── 7. Device Events (LOGIN / LOGOUT) ────────────────────────────────────────
 
 @app.post("/pushevent")
-async def push_event(request: Request):
+def push_event(token: str = Form(""), data_list: str = Form("")):
-    token, items = await _parse_request(request)
     _validate_token(token)
+    items = _parse_data_list(data_list)
     if not items:
         return JSONResponse(content=SUCCESS)