e27933b1cb
Thin passthrough over reporting.fn_inc_search (fleettickets migration 13):
ad-hoc ticket lookup by ticket_id / owner / cluster / status / state / time for
the FleetOps ticket explorer. Mirrors /webhook/inc-dashboard (read-only conn,
jsonb passthrough); validates state in {open,closed,all} and ISO-8601 from/to.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1001 lines
46 KiB
Python
1001 lines
46 KiB
Python
"""
|
|
dashboard_api_rev.py — Fireside Communications · Map Dashboard Read API
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
Stable replacement for the n8n webhooks that fed the Live Position and Fleet
|
|
Trips map dashboards. n8n was acting only as a thin HTTP→SQL proxy; this
|
|
service does the same job directly against the proven reporting.* functions,
|
|
removing n8n's credential-management / reload / version-drift fragility from
|
|
the live-data path.
|
|
|
|
It REUSES the existing stack: ts_shared_rev's psycopg2 pool and DATABASE_URL,
|
|
the same Docker image, the same Coolify deploy. The reporting.* functions
|
|
(already verified to return correct GeoJSON) are the single source of truth.
|
|
|
|
Endpoints mirror the original n8n webhook paths so the only frontend change
|
|
is the base URL (the `N8N_BASE` constant in each dashboard SPA):
|
|
|
|
GET /webhook/live-positions?cost_centre=&acc_status=
|
|
→ { summary, geojson } (reporting.fn_live_positions)
|
|
GET /webhook/live-positions/track?vehicle_number=&hours=
|
|
(alias: /webhook/vehicle-track)
|
|
→ GeoJSON Feature (reporting.fn_vehicle_track)
|
|
GET /webhook/fleet-dashboard
|
|
→ { drivers, cost_centres, cities, vehicles } (filter options)
|
|
POST /webhook/fleet-dashboard body: {period, vehicle_numbers, driver,
|
|
cost_centre, assigned_city,
|
|
start_date, end_date}
|
|
→ trips payload (reporting.fn_trips_for_map)
|
|
GET /webhook/tickets?service_type=&status=&open_only=
|
|
→ { summary, geojson } (reporting.fn_tickets_for_map) — INC/CRQ map layer
|
|
GET /health
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from urllib.parse import parse_qs
|
|
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
from fastapi import FastAPI, Query, Request
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from ts_shared_rev import close_pool, get_conn, get_logger
|
|
|
|
log = get_logger("dashboard_api")
|
|
|
|
# Comma-separated list of allowed browser origins (the dashboard domains).
|
|
_ALLOWED_ORIGINS = [
|
|
o.strip()
|
|
for o in os.getenv(
|
|
"DASHBOARD_CORS_ORIGINS",
|
|
"https://liveposition.rahamafresh.com,https://fleetintelligence.rahamafresh.com,https://fleetnow.rahamafresh.com",
|
|
).split(",")
|
|
if o.strip()
|
|
]
|
|
|
|
|
|
# ── v_trips materialized-view refresher ─────────────────────────────────────
|
|
# The Fleet Trips dashboard reads reporting.v_trips (a materialized view). Its
|
|
# refresh used to be a scheduled n8n workflow; when n8n was retired the matview
|
|
# went stale (data froze). We now keep it fresh in-process: a background loop
|
|
# refreshes it on an interval. A Postgres advisory lock makes this safe across
|
|
# uvicorn workers (only one worker refreshes per tick); the work runs in a
|
|
# thread so the async event loop never blocks on the ~9s REFRESH.
|
|
_DATABASE_URL = os.environ["DATABASE_URL"]
|
|
# The request pool (get_conn / DATABASE_URL) can be a READ-ONLY role
|
|
# (dashboard_ro) — least privilege for serving. The v_trips refresher needs write
|
|
# perms (it owns the REFRESH), so it connects via a SEPARATE privileged URL:
|
|
# REFRESH_DATABASE_URL if set, else DATABASE_URL (single-role / legacy deploys).
|
|
# So a prod bridge runs DATABASE_URL=dashboard_ro + REFRESH_DATABASE_URL=<app role>.
|
|
_REFRESH_DB_URL = os.getenv("REFRESH_DATABASE_URL") or _DATABASE_URL
|
|
# VTRIPS_REFRESH_INTERVAL_S <= 0 disables the in-process refresher entirely.
|
|
# Staging sets it to 0: it connects read-only and prod owns the refresh, so a
|
|
# staging instance must never attempt REFRESH (it would only log permission
|
|
# errors). Prod keeps the 300s default.
|
|
_REFRESH_INTERVAL_S = int(os.getenv("VTRIPS_REFRESH_INTERVAL_S", "300"))
|
|
_REFRESH_LOCK_KEY = 920_145 # arbitrary, stable advisory-lock key for this job
|
|
|
|
|
|
def _refresh_v_trips_once() -> str:
|
|
"""Refresh reporting.v_trips. Blocking — call via asyncio.to_thread.
|
|
|
|
Uses a dedicated autocommit connection: REFRESH ... CONCURRENTLY cannot run
|
|
inside a transaction block (so the pooled get_conn, which wraps a txn, won't
|
|
do). Connects via _REFRESH_DB_URL (REFRESH_DATABASE_URL or DATABASE_URL) — a
|
|
privileged role that may REFRESH the matview, distinct from a read-only
|
|
request pool.
|
|
"""
|
|
conn = psycopg2.connect(_REFRESH_DB_URL, connect_timeout=10)
|
|
try:
|
|
conn.autocommit = True
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT pg_try_advisory_lock(%s)", (_REFRESH_LOCK_KEY,))
|
|
if not cur.fetchone()[0]:
|
|
return "skipped (another worker holds the lock)"
|
|
try:
|
|
t0 = time.monotonic()
|
|
cur.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY reporting.v_trips")
|
|
dur_ms = int((time.monotonic() - t0) * 1000)
|
|
cur.execute(
|
|
"INSERT INTO reporting.refresh_log"
|
|
"(refreshed_at, source, duration_ms, row_count, notes) "
|
|
"VALUES (now(), 'dashboard_api', %s,"
|
|
" (SELECT count(*) FROM reporting.v_trips), 'scheduled')",
|
|
(dur_ms,),
|
|
)
|
|
return f"refreshed in {dur_ms}ms"
|
|
finally:
|
|
cur.execute("SELECT pg_advisory_unlock(%s)", (_REFRESH_LOCK_KEY,))
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
async def _refresh_loop():
|
|
# Brief startup delay so the first refresh doesn't race container init.
|
|
await asyncio.sleep(15)
|
|
while True:
|
|
try:
|
|
result = await asyncio.to_thread(_refresh_v_trips_once)
|
|
log.info("v_trips refresh: %s", result)
|
|
except Exception:
|
|
log.exception("v_trips refresh failed (will retry next interval)")
|
|
await asyncio.sleep(_REFRESH_INTERVAL_S)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
refresher = None
|
|
if _REFRESH_INTERVAL_S > 0:
|
|
log.info(
|
|
"Dashboard API starting (v1.2). Origins=%s. v_trips refresh every %ss.",
|
|
_ALLOWED_ORIGINS, _REFRESH_INTERVAL_S,
|
|
)
|
|
refresher = asyncio.create_task(_refresh_loop())
|
|
else:
|
|
log.info(
|
|
"Dashboard API starting (v1.2). Origins=%s. v_trips refresher DISABLED "
|
|
"(VTRIPS_REFRESH_INTERVAL_S<=0) — read-only / staging mode.",
|
|
_ALLOWED_ORIGINS,
|
|
)
|
|
yield
|
|
if refresher is not None:
|
|
refresher.cancel()
|
|
try:
|
|
await refresher
|
|
except asyncio.CancelledError:
|
|
pass
|
|
close_pool()
|
|
|
|
|
|
app = FastAPI(title="Fireside Map Dashboard API", lifespan=lifespan)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_ALLOWED_ORIGINS,
|
|
allow_methods=["GET", "POST", "OPTIONS"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
_EMPTY_GEOJSON = {"type": "FeatureCollection", "features": []}
|
|
|
|
|
|
def _clean(v):
|
|
"""Treat missing / blank / sentinel values as None (= SQL wildcard)."""
|
|
if v is None:
|
|
return None
|
|
s = str(v).strip()
|
|
return s if s and s.lower() not in ("null", "undefined") else None
|
|
|
|
|
|
# ── Health ────────────────────────────────────────────────────────────────────
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
# ── Ingest pipeline freshness ────────────────────────────────────────────────
|
|
# Replaces the Grafana pipeline-health panels (Grafana removed 2026-06-10).
|
|
# Reads reporting.v_ingest_health (migration 19) — one row per ingest endpoint
|
|
# with last-run age + freshness verdict (ok|stale|error). Lets FleetOps show
|
|
# whether the ingest_worker pollers are alive without a separate dashboard product.
|
|
@app.get("/health/ingest")
|
|
def ingest_health():
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute("SELECT * FROM reporting.v_ingest_health")
|
|
rows = cur.fetchall()
|
|
worst = (
|
|
"error" if any(r["freshness"] == "error" for r in rows)
|
|
else "stale" if any(r["freshness"] == "stale" for r in rows)
|
|
else "ok"
|
|
) if rows else "unknown"
|
|
return JSONResponse({"overall": worst, "endpoints": rows})
|
|
except Exception:
|
|
log.exception("ingest-health failed")
|
|
return JSONResponse(
|
|
{"overall": "unknown", "endpoints": [],
|
|
"error": {"type": "unknown",
|
|
"message": "Ingest-health feed is unavailable. Try again in a few seconds."}}
|
|
)
|
|
|
|
|
|
# ── Live positions (#004) ───────────────────────────────────────────────────
|
|
|
|
@app.get("/webhook/live-positions")
|
|
def live_positions(cost_centre: str | None = None, acc_status: str | None = None):
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT reporting.fn_live_positions(%s, %s)",
|
|
(_clean(cost_centre), _clean(acc_status)),
|
|
)
|
|
payload = cur.fetchone()[0] or {}
|
|
return JSONResponse(
|
|
{
|
|
"summary": payload.get("summary") or {},
|
|
"geojson": payload.get("geojson") or _EMPTY_GEOJSON,
|
|
}
|
|
)
|
|
except Exception:
|
|
log.exception("live-positions failed")
|
|
return JSONResponse(
|
|
{
|
|
"error": {
|
|
"type": "unknown",
|
|
"message": "Live-position feed is unavailable. Try again in a few seconds.",
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
# `/webhook/live-positions/track` is the path the Live Positions SPA actually
|
|
# calls; `/webhook/vehicle-track` is kept as an alias. Both hit the same handler
|
|
# so the only frontend change is the base URL (N8N_BASE).
|
|
@app.get("/webhook/live-positions/track")
|
|
@app.get("/webhook/vehicle-track")
|
|
def vehicle_track(vehicle_number: str | None = None, hours: int = 1):
|
|
veh = _clean(vehicle_number)
|
|
if not veh:
|
|
return JSONResponse({"error": "vehicle_number is required"})
|
|
hours = max(1, min(24, hours or 1))
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT reporting.fn_vehicle_track(%s, %s::int)", (veh, hours)
|
|
)
|
|
feature = cur.fetchone()[0]
|
|
return JSONResponse(
|
|
feature
|
|
or {"type": "Feature", "geometry": {"type": "LineString", "coordinates": []}, "properties": {}}
|
|
)
|
|
except Exception:
|
|
log.exception("vehicle-track failed for %s", veh)
|
|
return JSONResponse({"error": "vehicle-track unavailable"})
|
|
|
|
|
|
# ── Tickets (FleetOps Tickets map) ───────────────────────────────────────────
|
|
# INC (incident / customer fault) + CRQ (new-installation) tickets as a GeoJSON
|
|
# FeatureCollection for the FleetOps Tickets tab (FleetNow-style map). Backed by
|
|
# reporting.fn_tickets_for_map over tickets.inc / tickets.crq. The schema, ingest,
|
|
# and that read function are owned by the separate `fleettickets` repo
|
|
# (repo.rahamafresh.com/kianiadee/fleettickets.git); this endpoint just calls it.
|
|
# Only geocoded rows are mapped; open_only (default true) restricts to actionable.
|
|
@app.get("/webhook/tickets")
|
|
def tickets(
|
|
service_type: str | None = None, # 'inc' | 'crq' | None (both)
|
|
status: str | None = None, # normalized_status exact match
|
|
open_only: bool = True, # actionable tickets only (the map default)
|
|
):
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT reporting.fn_tickets_for_map(%s, %s, %s)",
|
|
(_clean(service_type), _clean(status), open_only),
|
|
)
|
|
payload = cur.fetchone()[0] or {}
|
|
return JSONResponse(
|
|
{
|
|
"summary": payload.get("summary") or {},
|
|
"geojson": payload.get("geojson") or _EMPTY_GEOJSON,
|
|
}
|
|
)
|
|
except Exception:
|
|
log.exception("tickets failed")
|
|
return JSONResponse(
|
|
{
|
|
"error": {
|
|
"type": "unknown",
|
|
"message": "Ticket feed is unavailable. Try again in a few seconds.",
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
# ── INC operations dashboard ──────────────────────────────────────────────────
|
|
# Thin passthrough over reporting.fn_inc_dashboard (fleettickets migration 09):
|
|
# returns { window, open: GeoJSON, closed: GeoJSON, metrics, freshness } for the
|
|
# FleetOps live INC map. open = all currently-open INC tickets (not time-filtered);
|
|
# closed = closures within the selected window; metrics react to the selection.
|
|
# Vehicle positions/routes are overlaid by the SPA (FleetNow), not this endpoint.
|
|
_INC_WINDOWS = {"today", "week", "month", "custom"}
|
|
|
|
|
|
def _bad_request(msg):
|
|
return JSONResponse({"error": {"type": "bad_request", "message": msg}}, status_code=400)
|
|
|
|
|
|
@app.get("/webhook/inc-dashboard")
|
|
def inc_dashboard(
|
|
cluster: str | None = None, # exact tickets.inc.cluster (UPPERCASE), blank = all
|
|
status: str | None = None, # exact normalized_status, blank = all
|
|
window: str = "today", # today | week | month | custom (calendar EAT)
|
|
from_: str | None = Query(None, alias="from"), # custom start (inclusive), ISO-8601
|
|
to: str | None = None, # custom end (exclusive), ISO-8601
|
|
):
|
|
# Validation (mirrors the contract). The SQL treats any from/to as a custom
|
|
# window; presets resolve to EAT calendar bounds inside the function.
|
|
if window not in _INC_WINDOWS:
|
|
return _bad_request("window must be one of today|week|month|custom")
|
|
f, t = _clean(from_), _clean(to)
|
|
if window == "custom" and not f and not t:
|
|
return _bad_request("custom window requires from and/or to")
|
|
|
|
def _parse(v):
|
|
if not v:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(v)
|
|
except ValueError:
|
|
return False
|
|
|
|
pf, pt = _parse(f), _parse(t)
|
|
if pf is False or pt is False:
|
|
return _bad_request("from/to must be ISO-8601 timestamps with an offset/Z")
|
|
if pf and pt and pf >= pt:
|
|
return _bad_request("from must be earlier than to")
|
|
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT reporting.fn_inc_dashboard(%s, %s, %s, %s, %s)",
|
|
(_clean(cluster), _clean(status), window, f, t),
|
|
)
|
|
payload = cur.fetchone()[0] or {}
|
|
return JSONResponse(payload) # jsonb body returned unchanged
|
|
except Exception:
|
|
log.exception("inc-dashboard failed")
|
|
return JSONResponse(
|
|
{
|
|
"error": {
|
|
"type": "unknown",
|
|
"message": "INC dashboard is unavailable. Try again in a few seconds.",
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
# ── INC ticket explorer (search) ──────────────────────────────────────────────
|
|
# Thin passthrough over reporting.fn_inc_search (fleettickets migration 13): ad-hoc
|
|
# ticket lookup by id / engineer / cluster / status / state / time, for historical +
|
|
# current tracking. Returns { count, truncated, limit, state, rows }.
|
|
_INC_STATES = {"open", "closed", "all"}
|
|
|
|
|
|
@app.get("/webhook/inc-search")
|
|
def inc_search(
|
|
ticket_id: str | None = None, # substring match on ticket_id
|
|
owner: str | None = None, # engineer — case-insensitive substring on owner
|
|
cluster: str | None = None, # exact tickets.inc.cluster
|
|
status: str | None = None, # exact normalized_status
|
|
state: str = "closed", # closed | open | all
|
|
from_: str | None = Query(None, alias="from"), # closed-at range start (ISO-8601)
|
|
to: str | None = None, # closed-at range end (exclusive, ISO-8601)
|
|
):
|
|
state = (state or "closed").lower()
|
|
if state not in _INC_STATES:
|
|
return _bad_request("state must be one of open|closed|all")
|
|
f, t = _clean(from_), _clean(to)
|
|
|
|
def _parse(v):
|
|
if not v:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(v)
|
|
except ValueError:
|
|
return False
|
|
|
|
pf, pt = _parse(f), _parse(t)
|
|
if pf is False or pt is False:
|
|
return _bad_request("from/to must be ISO-8601 timestamps with an offset/Z")
|
|
if pf and pt and pf >= pt:
|
|
return _bad_request("from must be earlier than to")
|
|
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT reporting.fn_inc_search(%s, %s, %s, %s, %s, %s, %s)",
|
|
(_clean(ticket_id), _clean(owner), _clean(cluster), _clean(status), state, f, t),
|
|
)
|
|
payload = cur.fetchone()[0] or {}
|
|
return JSONResponse(payload) # jsonb body returned unchanged
|
|
except Exception:
|
|
log.exception("inc-search failed")
|
|
return JSONResponse(
|
|
{
|
|
"error": {
|
|
"type": "unknown",
|
|
"message": "Ticket search is unavailable. Try again in a few seconds.",
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
# ── Fleet trips (#002) ───────────────────────────────────────────────────────
|
|
|
|
_FILTER_OPTIONS_SQL = """
|
|
SELECT
|
|
(SELECT array_agg(driver ORDER BY driver) FROM reporting.v_filter_drivers) AS drivers,
|
|
(SELECT array_agg(cost_centre ORDER BY cost_centre) FROM reporting.v_filter_cost_centres) AS cost_centres,
|
|
(SELECT array_agg(assigned_city ORDER BY assigned_city) FROM reporting.v_filter_cities) AS cities,
|
|
(SELECT jsonb_agg(jsonb_build_object(
|
|
'vehicle_number', vehicle_number, 'drivers', drivers,
|
|
'cost_centre', cost_centre, 'assigned_city', assigned_city)
|
|
ORDER BY vehicle_number) FROM reporting.v_filter_vehicles) AS vehicles
|
|
"""
|
|
|
|
|
|
@app.get("/webhook/fleet-dashboard")
|
|
def fleet_filter_options():
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(_FILTER_OPTIONS_SQL)
|
|
row = cur.fetchone() or {}
|
|
return JSONResponse(
|
|
{
|
|
"drivers": row.get("drivers") or [],
|
|
"cost_centres": row.get("cost_centres") or [],
|
|
"cities": row.get("cities") or [],
|
|
"vehicles": row.get("vehicles") or [],
|
|
}
|
|
)
|
|
except Exception:
|
|
log.exception("fleet-dashboard filter options failed")
|
|
return JSONResponse({"drivers": [], "cost_centres": [], "cities": [], "vehicles": []})
|
|
|
|
|
|
def _preset_to_range(period: str | None, start_date, end_date):
|
|
"""Mirror of the n8n preset_to_range node."""
|
|
today = datetime.now(timezone.utc).date()
|
|
p = (period or "").strip()
|
|
if p == "today":
|
|
return today, today
|
|
if p == "30d":
|
|
return today - timedelta(days=29), today
|
|
if p == "custom":
|
|
def _d(v, default):
|
|
v = _clean(v)
|
|
if not v:
|
|
return default
|
|
try:
|
|
return date.fromisoformat(v)
|
|
except ValueError:
|
|
return default
|
|
return _d(start_date, today), _d(end_date, today)
|
|
# generic 'Nd' window (e.g. 7d / 30d / 90d / 365d) — used by the Fuel Log tab
|
|
if p.endswith("d") and p[:-1].isdigit() and int(p[:-1]) > 0:
|
|
return today - timedelta(days=int(p[:-1]) - 1), today
|
|
# default + '7d'
|
|
return today - timedelta(days=6), today
|
|
|
|
|
|
@app.post("/webhook/fleet-dashboard")
|
|
async def fleet_trips(request: Request):
|
|
# The dashboard SPA posts application/x-www-form-urlencoded (not JSON), so
|
|
# parse by content-type. Reading the raw body + parse_qs avoids pulling in
|
|
# python-multipart. JSON is still accepted defensively (n8n-compat callers).
|
|
body: dict = {}
|
|
ctype = request.headers.get("content-type", "").lower()
|
|
try:
|
|
raw = await request.body()
|
|
if "application/json" in ctype:
|
|
parsed = json.loads(raw or b"{}")
|
|
body = parsed if isinstance(parsed, dict) else {}
|
|
else:
|
|
# x-www-form-urlencoded — parse_qs yields lists; keep the last value.
|
|
body = {k: v[-1] for k, v in parse_qs(raw.decode("utf-8", "replace")).items()}
|
|
except Exception:
|
|
body = {}
|
|
if isinstance(body.get("body"), dict):
|
|
body = body["body"]
|
|
|
|
start, end = _preset_to_range(
|
|
body.get("period"), body.get("start_date"), body.get("end_date")
|
|
)
|
|
veh = _clean(body.get("vehicle_numbers")) # comma-separated string or None
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT reporting.fn_trips_for_map(
|
|
CASE WHEN %(veh)s IS NULL THEN NULL
|
|
ELSE string_to_array(%(veh)s, ',') END,
|
|
%(driver)s, %(cc)s, %(city)s, %(start)s::date, %(end)s::date
|
|
)
|
|
""",
|
|
{
|
|
"veh": veh,
|
|
"driver": _clean(body.get("driver")),
|
|
"cc": _clean(body.get("cost_centre")),
|
|
"city": _clean(body.get("assigned_city")),
|
|
"start": start,
|
|
"end": end,
|
|
},
|
|
)
|
|
payload = cur.fetchone()[0]
|
|
return JSONResponse(payload if payload is not None else {})
|
|
except Exception:
|
|
log.exception("fleet-dashboard trips failed")
|
|
return JSONResponse(
|
|
{"error": {"type": "unknown", "message": "Fleet feed is unavailable. Try again in a few seconds."}}
|
|
)
|
|
|
|
|
|
# ── FleetOps analytics (#15) ─────────────────────────────────────────────────
|
|
# Read-only roll-ups powering the FleetOps SPA (fleetops.rahamafresh.com):
|
|
# utilisation, distance, driver behaviour and fuel. Every query SELECTs the
|
|
# indexed reporting.* / tracksolid.v_* views and never writes — so the staging
|
|
# instance serves them against the prod DB as a read-only role. Numeric/date
|
|
# values come back as Decimal/date from psycopg2, so responses pass through
|
|
# jsonable_encoder (Decimal→float, date→ISO) before JSONResponse.
|
|
#
|
|
# GET /analytics/fleet-summary per-vehicle + per-cost-centre roll-up
|
|
# GET /analytics/utilisation per-vehicle utilisation + daily fleet trend
|
|
# GET /analytics/driver-behaviour per-driver speeding / harsh index
|
|
# GET /analytics/fuel actual vs estimated litres (data-gated)
|
|
# GET /analytics/filters dropdown options (alias of GET /webhook/fleet-dashboard)
|
|
#
|
|
# Shared query params: period (today|7d|30d|custom, default 30d), start_date,
|
|
# end_date, and optional dims cost_centre / assigned_city / vehicle_number /
|
|
# driver.
|
|
|
|
def _json(obj):
|
|
"""Serialise dicts that may carry Decimal / date values from psycopg2."""
|
|
return JSONResponse(jsonable_encoder(obj))
|
|
|
|
|
|
def _analytics_window(period, start_date, end_date):
|
|
"""Date range for analytics — defaults to a 30-day window (vs the 7d trips default)."""
|
|
return _preset_to_range(period or "30d", start_date, end_date)
|
|
|
|
|
|
def _dim_filters(cost_centre=None, assigned_city=None, vehicle_number=None, driver=None):
|
|
"""Optional WHERE fragments shared by the reporting.* analytics views.
|
|
|
|
Column names are fixed literals (not user input); only the values are
|
|
parameterised, so interpolating the fragments into the query is injection-safe.
|
|
"""
|
|
clauses, params = [], {}
|
|
for col, val in (
|
|
("cost_centre", cost_centre),
|
|
("assigned_city", assigned_city),
|
|
("vehicle_number", vehicle_number),
|
|
("assigned_driver", driver),
|
|
):
|
|
v = _clean(val)
|
|
if v is not None:
|
|
clauses.append(f"{col} = %({col})s")
|
|
params[col] = v
|
|
return clauses, params
|
|
|
|
|
|
def _analytics_error(name):
|
|
log.exception("%s failed", name)
|
|
return JSONResponse(
|
|
{"error": {"type": "unknown",
|
|
"message": "Analytics feed is unavailable. Try again in a few seconds."}}
|
|
)
|
|
|
|
|
|
@app.get("/analytics/fleet-summary")
|
|
def analytics_fleet_summary(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
cost_centre: str | None = None, assigned_city: str | None = None,
|
|
vehicle_number: str | None = None, driver: str | None = None,
|
|
):
|
|
start, end = _analytics_window(period, start_date, end_date)
|
|
clauses, params = _dim_filters(cost_centre, assigned_city, vehicle_number, driver)
|
|
params |= {"start": start, "end": end}
|
|
where = " AND ".join(["trip_date BETWEEN %(start)s AND %(end)s"] + clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT vehicle_number, cost_centre, assigned_city, assigned_driver,
|
|
count(DISTINCT trip_date) AS active_days,
|
|
sum(trip_count) AS trips,
|
|
round(sum(total_km), 1) AS total_km,
|
|
round(sum(driving_hours), 1) AS driving_hours,
|
|
round(sum(idle_hours), 1) AS idle_hours,
|
|
round(100.0 * sum(idle_hours)
|
|
/ NULLIF(sum(idle_hours + driving_hours), 0), 1) AS idle_pct,
|
|
round(max(max_speed_kmh)) AS max_speed_kmh
|
|
FROM reporting.v_daily_summary
|
|
WHERE {where}
|
|
GROUP BY vehicle_number, cost_centre, assigned_city, assigned_driver
|
|
ORDER BY total_km DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.execute(
|
|
f"""
|
|
SELECT cost_centre,
|
|
count(DISTINCT vehicle_number) AS vehicles,
|
|
sum(trip_count) AS trips,
|
|
round(sum(total_km), 1) AS total_km,
|
|
round(sum(driving_hours), 1) AS driving_hours,
|
|
round(sum(idle_hours), 1) AS idle_hours,
|
|
round(100.0 * sum(idle_hours)
|
|
/ NULLIF(sum(idle_hours + driving_hours), 0), 1) AS idle_pct
|
|
FROM reporting.v_daily_summary
|
|
WHERE {where}
|
|
GROUP BY cost_centre
|
|
ORDER BY total_km DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
by_cc = cur.fetchall()
|
|
totals = {
|
|
"vehicles": len({r["vehicle_number"] for r in rows}),
|
|
"trips": sum(int(r["trips"] or 0) for r in rows),
|
|
"total_km": round(sum(float(r["total_km"] or 0) for r in rows), 1),
|
|
"driving_hours": round(sum(float(r["driving_hours"] or 0) for r in rows), 1),
|
|
"idle_hours": round(sum(float(r["idle_hours"] or 0) for r in rows), 1),
|
|
}
|
|
return _json({"window": {"start": str(start), "end": str(end)},
|
|
"totals": totals, "rows": rows, "by_cost_centre": by_cc})
|
|
except Exception:
|
|
return _analytics_error("analytics/fleet-summary")
|
|
|
|
|
|
@app.get("/analytics/utilisation")
|
|
def analytics_utilisation(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
cost_centre: str | None = None, assigned_city: str | None = None,
|
|
vehicle_number: str | None = None, driver: str | None = None,
|
|
):
|
|
start, end = _analytics_window(period, start_date, end_date)
|
|
clauses, params = _dim_filters(cost_centre, assigned_city, vehicle_number, driver)
|
|
params |= {"start": start, "end": end}
|
|
where = " AND ".join(["trip_date BETWEEN %(start)s AND %(end)s"] + clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT vehicle_number, cost_centre, assigned_city,
|
|
count(DISTINCT trip_date) AS active_days,
|
|
round(sum(total_km), 1) AS total_km,
|
|
round(sum(total_km)
|
|
/ NULLIF(count(DISTINCT trip_date), 0), 1) AS km_per_active_day,
|
|
round(sum(driving_hours), 1) AS driving_hours,
|
|
round(sum(idle_hours), 1) AS idle_hours,
|
|
round(100.0 * sum(idle_hours)
|
|
/ NULLIF(sum(idle_hours + driving_hours), 0), 1) AS idle_pct
|
|
FROM reporting.v_daily_summary
|
|
WHERE {where}
|
|
GROUP BY vehicle_number, cost_centre, assigned_city
|
|
ORDER BY total_km DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
by_vehicle = cur.fetchall()
|
|
cur.execute(
|
|
f"""
|
|
SELECT trip_date,
|
|
count(DISTINCT vehicle_number) AS active_vehicles,
|
|
round(sum(total_km), 1) AS total_km,
|
|
round(sum(driving_hours), 1) AS driving_hours,
|
|
round(sum(idle_hours), 1) AS idle_hours,
|
|
round(100.0 * sum(idle_hours)
|
|
/ NULLIF(sum(idle_hours + driving_hours), 0), 1) AS idle_pct
|
|
FROM reporting.v_daily_summary
|
|
WHERE {where}
|
|
GROUP BY trip_date
|
|
ORDER BY trip_date
|
|
""",
|
|
params,
|
|
)
|
|
daily_trend = cur.fetchall()
|
|
return _json({"window": {"start": str(start), "end": str(end)},
|
|
"by_vehicle": by_vehicle, "daily_trend": daily_trend})
|
|
except Exception:
|
|
return _analytics_error("analytics/utilisation")
|
|
|
|
|
|
@app.get("/analytics/driver-behaviour")
|
|
def analytics_driver_behaviour(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
assigned_city: str | None = None, driver: str | None = None,
|
|
):
|
|
start, end = _analytics_window(period, start_date, end_date)
|
|
clauses = ["day BETWEEN %(start)s AND %(end)s", "driver_name IS NOT NULL"]
|
|
params = {"start": start, "end": end}
|
|
city, drv = _clean(assigned_city), _clean(driver)
|
|
if city is not None:
|
|
clauses.append("assigned_city = %(city)s")
|
|
params["city"] = city
|
|
if drv is not None:
|
|
clauses.append("driver_name = %(drv)s")
|
|
params["drv"] = drv
|
|
where = " AND ".join(clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT driver_name, assigned_city,
|
|
count(DISTINCT day) AS active_days,
|
|
round(sum(km), 1) AS total_km,
|
|
sum(trips) AS trips,
|
|
sum(events_80) AS events_80,
|
|
sum(events_100) AS events_100,
|
|
sum(events_120) AS events_120,
|
|
sum(harsh_events) AS harsh_events,
|
|
round(sum(events_80)::numeric
|
|
/ NULLIF(sum(km), 0) * 100, 2) AS speeding_per_100km,
|
|
round(sum(harsh_events)::numeric
|
|
/ NULLIF(sum(km), 0) * 100, 2) AS harsh_per_100km
|
|
FROM tracksolid.v_driver_aggregates_daily
|
|
WHERE {where}
|
|
GROUP BY driver_name, assigned_city
|
|
ORDER BY speeding_per_100km DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
# driver_name may be sparsely populated (enrichment via tools/import_drivers_csv
|
|
# or at source), so this can legitimately return []; it fills in as drivers land.
|
|
return _json({"window": {"start": str(start), "end": str(end)},
|
|
"drivers_populated": bool(rows), "rows": rows})
|
|
except Exception:
|
|
return _analytics_error("analytics/driver-behaviour")
|
|
|
|
|
|
@app.get("/analytics/fuel")
|
|
def analytics_fuel(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
cost_centre: str | None = None, assigned_city: str | None = None,
|
|
vehicle_number: str | None = None, driver: str | None = None,
|
|
):
|
|
start, end = _analytics_window(period, start_date, end_date)
|
|
clauses, params = _dim_filters(cost_centre, assigned_city, vehicle_number, driver)
|
|
params |= {"start": start, "end": end}
|
|
where = " AND ".join(["trip_date BETWEEN %(start)s AND %(end)s"] + clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT vehicle_number, cost_centre, assigned_city,
|
|
round(sum(distance_km), 1) AS total_km,
|
|
round(sum(actual_fuel_l), 2) AS actual_fuel_l,
|
|
round(sum(estimated_fuel_l), 2) AS estimated_fuel_l,
|
|
count(*) FILTER (WHERE actual_fuel_l IS NOT NULL) AS trips_with_actual,
|
|
count(*) AS trips
|
|
FROM reporting.v_fuel_daily
|
|
WHERE {where}
|
|
GROUP BY vehicle_number, cost_centre, assigned_city
|
|
ORDER BY total_km DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.execute(
|
|
f"""
|
|
SELECT bool_or(actual_fuel_l IS NOT NULL) AS actual_available,
|
|
bool_or(estimated_fuel_l IS NOT NULL) AS estimated_available
|
|
FROM reporting.v_fuel_daily
|
|
WHERE {where}
|
|
""",
|
|
params,
|
|
)
|
|
flags = cur.fetchone() or {}
|
|
data_status = {
|
|
"actual_fuel_available": bool(flags.get("actual_available")),
|
|
"estimated_fuel_available": bool(flags.get("estimated_available")),
|
|
"notes": [
|
|
"actual_fuel_l comes from trips.fuel_consumed_l (/pushtripreport webhook).",
|
|
"estimated_fuel_l needs devices.fuel_100km set per vehicle "
|
|
"(currently NULL fleet-wide — see CLAUDE.md Open Items).",
|
|
"Fuel-cost monetisation is unavailable: ops.cost_rates was purged 2026-06-05.",
|
|
],
|
|
}
|
|
return _json({"window": {"start": str(start), "end": str(end)},
|
|
"data_status": data_status, "rows": rows})
|
|
except Exception:
|
|
return _analytics_error("analytics/fuel")
|
|
|
|
|
|
# ── Fuel Log (#fuelfuel) — actual fills from the WhatsApp feed ───────────────
|
|
# Backed by reporting.v_fuel_fills / v_fuel_efficiency (owned by the `fleetfuel`
|
|
# repo, which ingests the rustfs `fuel` bucket). Separate from /analytics/fuel
|
|
# above (that one is the trip-derived estimate); this is real litres + KES spend.
|
|
def _fuel_filters(cost_centre, assigned_city, vehicle_number, driver, department, fuel_type):
|
|
"""Shared dims (_dim_filters) plus the fuel-native department / fuel_type."""
|
|
clauses, params = _dim_filters(cost_centre, assigned_city, vehicle_number, driver)
|
|
for col, val in (("department", department), ("fuel_type", fuel_type)):
|
|
v = _clean(val)
|
|
if v is not None:
|
|
clauses.append(f"{col} = %({col})s")
|
|
params[col] = v
|
|
return clauses, params
|
|
|
|
|
|
@app.get("/analytics/fuel-fills")
|
|
def analytics_fuel_fills(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
cost_centre: str | None = None, assigned_city: str | None = None,
|
|
vehicle_number: str | None = None, driver: str | None = None,
|
|
department: str | None = None, fuel_type: str | None = None,
|
|
):
|
|
start, end = _analytics_window(period or "90d", start_date, end_date)
|
|
clauses, params = _fuel_filters(cost_centre, assigned_city, vehicle_number, driver,
|
|
department, fuel_type)
|
|
params |= {"start": start, "end": end}
|
|
where = " AND ".join(["fuel_date BETWEEN %(start)s AND %(end)s"] + clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT round(sum(liters), 1) AS litres,
|
|
round(sum(amount), 0) AS spend_kes,
|
|
count(*) AS fills,
|
|
round(sum(amount) / NULLIF(sum(liters), 0), 1) AS avg_price_per_litre,
|
|
count(DISTINCT plate) AS vehicles_fuelled,
|
|
count(*) FILTER (WHERE vehicle_number IS NULL) AS unmatched_fills
|
|
FROM reporting.v_fuel_fills
|
|
WHERE {where}
|
|
""",
|
|
params,
|
|
)
|
|
totals = cur.fetchone() or {}
|
|
cur.execute(
|
|
f"""
|
|
SELECT f.plate, f.vehicle_number, f.cost_centre, f.assigned_city,
|
|
round(sum(f.liters), 1) AS litres,
|
|
round(sum(f.amount), 0) AS spend_kes,
|
|
count(*) AS fills,
|
|
max(f.odometer) AS last_odometer,
|
|
round(sum(f.amount) / NULLIF(sum(f.liters), 0), 1) AS avg_price_per_litre,
|
|
eff.km_per_litre
|
|
FROM reporting.v_fuel_fills f
|
|
LEFT JOIN (
|
|
SELECT plate, round(avg(km_per_litre), 2) AS km_per_litre
|
|
FROM reporting.v_fuel_efficiency
|
|
WHERE fuel_date BETWEEN %(start)s AND %(end)s
|
|
AND km_per_litre IS NOT NULL
|
|
GROUP BY plate
|
|
) eff ON eff.plate = f.plate
|
|
WHERE {where}
|
|
GROUP BY f.plate, f.vehicle_number, f.cost_centre, f.assigned_city, eff.km_per_litre
|
|
ORDER BY spend_kes DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.execute(
|
|
f"""
|
|
SELECT coalesce(department, '(unspecified)') AS department,
|
|
round(sum(liters), 1) AS litres,
|
|
round(sum(amount), 0) AS spend_kes,
|
|
count(*) AS fills
|
|
FROM reporting.v_fuel_fills
|
|
WHERE {where}
|
|
GROUP BY department
|
|
ORDER BY spend_kes DESC NULLS LAST
|
|
""",
|
|
params,
|
|
)
|
|
by_department = cur.fetchall()
|
|
cur.execute(
|
|
f"""
|
|
SELECT fuel_date,
|
|
round(sum(liters), 1) AS litres,
|
|
round(sum(amount), 0) AS spend_kes,
|
|
count(*) AS fills
|
|
FROM reporting.v_fuel_fills
|
|
WHERE {where}
|
|
GROUP BY fuel_date
|
|
ORDER BY fuel_date
|
|
""",
|
|
params,
|
|
)
|
|
trend = cur.fetchall()
|
|
data_status = {
|
|
"matched_to_fleet": (totals.get("fills") or 0) - (totals.get("unmatched_fills") or 0),
|
|
"unmatched_fills": totals.get("unmatched_fills") or 0,
|
|
"notes": [
|
|
"Fills are real WhatsApp fuel-update records (litres + KES amount).",
|
|
"unmatched_fills are records whose plate didn't match a known vehicle "
|
|
"in tracksolid.devices — they still count in totals.",
|
|
"km_per_litre is derived from consecutive odometer readings; sparse where "
|
|
"odometer is missing or implausible.",
|
|
],
|
|
}
|
|
return _json({"window": {"start": str(start), "end": str(end)},
|
|
"data_status": data_status, "totals": totals, "rows": rows,
|
|
"by_department": by_department, "trend": trend})
|
|
except Exception:
|
|
return _analytics_error("analytics/fuel-fills")
|
|
|
|
|
|
@app.get("/analytics/fuel-fills/recent")
|
|
def analytics_fuel_fills_recent(
|
|
period: str | None = None, start_date: str | None = None, end_date: str | None = None,
|
|
cost_centre: str | None = None, assigned_city: str | None = None,
|
|
vehicle_number: str | None = None, driver: str | None = None,
|
|
department: str | None = None, fuel_type: str | None = None,
|
|
limit: int = 50,
|
|
):
|
|
start, end = _analytics_window(period or "90d", start_date, end_date)
|
|
clauses, params = _fuel_filters(cost_centre, assigned_city, vehicle_number, driver,
|
|
department, fuel_type)
|
|
params |= {"start": start, "end": end, "lim": max(1, min(limit, 500))}
|
|
where = " AND ".join(["fuel_date BETWEEN %(start)s AND %(end)s"] + clauses)
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(
|
|
f"""
|
|
SELECT record_datetime, plate, vehicle_number, cost_centre, assigned_city,
|
|
department, driver, liters, amount, fuel_type, odometer
|
|
FROM reporting.v_fuel_fills
|
|
WHERE {where}
|
|
ORDER BY record_datetime DESC NULLS LAST
|
|
LIMIT %(lim)s
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
return _json({"window": {"start": str(start), "end": str(end)}, "rows": rows})
|
|
except Exception:
|
|
return _analytics_error("analytics/fuel-fills/recent")
|
|
|
|
|
|
@app.get("/analytics/filters")
|
|
def analytics_filters():
|
|
# Trips dropdowns (drivers / cost_centres / cities / vehicles) plus the fuel
|
|
# dropdowns (departments / fuel_types), so FleetOps has a single /analytics/*
|
|
# filter surface for every tab including Fuel Log.
|
|
try:
|
|
with get_conn() as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(_FILTER_OPTIONS_SQL)
|
|
row = cur.fetchone() or {}
|
|
# Fuel dims are best-effort: a missing reporting.v_fuel_fills (fleetfuel
|
|
# migration not yet applied) must NOT break the trips dropdowns, so query
|
|
# it in its own savepoint and degrade to empty lists if it isn't there.
|
|
fuel = {}
|
|
try:
|
|
cur.execute("SAVEPOINT fuel_dims")
|
|
cur.execute(
|
|
"SELECT array_agg(DISTINCT department) FILTER (WHERE department IS NOT NULL) AS departments,"
|
|
" array_agg(DISTINCT fuel_type) FILTER (WHERE fuel_type IS NOT NULL) AS fuel_types"
|
|
" FROM reporting.v_fuel_fills"
|
|
)
|
|
fuel = cur.fetchone() or {}
|
|
except Exception:
|
|
cur.execute("ROLLBACK TO SAVEPOINT fuel_dims")
|
|
log.warning("fuel filter dims unavailable (reporting.v_fuel_fills missing?)")
|
|
return JSONResponse({
|
|
"drivers": row.get("drivers") or [],
|
|
"cost_centres": row.get("cost_centres") or [],
|
|
"cities": row.get("cities") or [],
|
|
"vehicles": row.get("vehicles") or [],
|
|
"departments": sorted(fuel.get("departments") or []),
|
|
"fuel_types": sorted(fuel.get("fuel_types") or []),
|
|
})
|
|
except Exception:
|
|
log.exception("analytics/filters failed")
|
|
return JSONResponse({"drivers": [], "cost_centres": [], "cities": [],
|
|
"vehicles": [], "departments": [], "fuel_types": []})
|