747 lines
30 KiB
Python
747 lines
30 KiB
Python
# FuelBoard Relay API — keyless proxy for the official UK Fuel Finder API
|
||
#
|
||
# Holds the GOV.UK Fuel Finder OAuth 2.0 client credentials server-side so the
|
||
# iOS app / widget never embeds a secret. The app calls this relay with no auth
|
||
# and gets stations + prices as simple JSON.
|
||
#
|
||
# This is the API-source relay (sibling of fuelboard-relay, which reads the
|
||
# public CSV mirror). It talks to the OFFICIAL Fuel Finder API:
|
||
#
|
||
# FuelBoard app → GET http://<relay>:8789/api/v1/stations
|
||
# relay → OAuth token (cached ~1h) → GET /api/v1/pfs (station info)
|
||
# GET /api/v1/pfs/fuel-prices (prices)
|
||
# relay joins on node_id, normalises to the SAME output shape as the CSV
|
||
# relay, caches the snapshot (default 5 min) → returns JSON to the app
|
||
#
|
||
# The /api/v1/stations contract is IDENTICAL to fuelboard-relay, so the app,
|
||
# widget and alerts work unchanged against either relay — switching sources is
|
||
# just pointing the app at a different host:port.
|
||
#
|
||
# Endpoints:
|
||
# GET /health — cache status + last sync timestamp
|
||
# GET /stats — admin telemetry (localhost only): client hits by
|
||
# kind/IP/route over a rolling 24 h window
|
||
# GET /api/v1/stations — full station list with current prices (keyless)
|
||
# GET /api/v1/stations?fuel=e10&lat=53.72&lng=-1.85&radius=10
|
||
# — optional server-side filter/sort (cheapest first)
|
||
#
|
||
# Data shape notes (from developer.fuel-finder.service.gov.uk):
|
||
# - Token: POST /api/v1/oauth/generate_access_token with client_id/client_secret
|
||
# → { "data": { "access_token": ..., "expires_in": 3600 } }
|
||
# - Stations: GET /api/v1/pfs → array, batched 500/page via ?batch-number=N
|
||
# - Prices: GET /api/v1/pfs/fuel-prices → array, same batching
|
||
# - Price values are DECIMAL STRINGS in pence ("0120.0000" = 120.0p); some
|
||
# stations report pounds (< 2.0) — multiply by 100.
|
||
# - lat/lng are STRINGS inside location: { latitude, longitude }.
|
||
# - Fuel types: E10, E5, B7_STANDARD (diesel), B10, HVO, SDV, ...
|
||
|
||
from contextlib import asynccontextmanager
|
||
from collections import deque
|
||
import os
|
||
import time
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
from fastapi import FastAPI, HTTPException, Query, Request
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Config (all via environment; never hardcode credentials)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
CLIENT_ID = os.environ.get("FUEL_API_CLIENT_ID", "")
|
||
CLIENT_SECRET = os.environ.get("FUEL_API_CLIENT_SECRET", "")
|
||
# Base host for the official Fuel Finder API (note: NOT api.fuelfinder... —
|
||
# the real host is www.fuel-finder.service.gov.uk).
|
||
API_BASE = os.environ.get(
|
||
"FUEL_API_BASE",
|
||
"https://www.fuel-finder.service.gov.uk",
|
||
)
|
||
TOKEN_URL = os.environ.get("FUEL_API_TOKEN_URL", f"{API_BASE}/api/v1/oauth/generate_access_token")
|
||
# Official Fuel Finder endpoints (batched, 500 records/page).
|
||
PFS_URL = os.environ.get("FUEL_API_PFS_URL", f"{API_BASE}/api/v1/pfs")
|
||
PRICES_URL = os.environ.get(
|
||
"FUEL_API_PRICES_URL",
|
||
f"{API_BASE}/api/v1/pfs/fuel-prices",
|
||
)
|
||
PAGE_SIZE = int(os.environ.get("FUEL_API_PAGE_SIZE", "500"))
|
||
CACHE_TTL = int(os.environ.get("FUEL_CACHE_TTL", "300")) # seconds (5 min)
|
||
TOKEN_EXPIRY_BUFFER = int(os.environ.get("FUEL_TOKEN_BUFFER", "60")) # seconds
|
||
# Fuel Finder rate limit: 30 rpm per client, sequential only (429 on overlap).
|
||
# uff.py's default of 4.0s between batch pages ≈ 15 rpm — stay inside.
|
||
BATCH_SLEEP = float(os.environ.get("FUEL_API_BATCH_SLEEP", "4.0"))
|
||
|
||
# Public full-UK mirror of the Fuel Finder dataset (GitHub Actions, hourly),
|
||
# used as the fallback source when OAuth credentials are not configured so the
|
||
# relay is testable end-to-end before GOV.UK One Login creds exist.
|
||
# Licence: Open Government Licence v3.0. No auth required.
|
||
CSV_URL = os.environ.get(
|
||
"FUEL_CSV_URL",
|
||
"https://raw.githubusercontent.com/matthewgall/fuelfinder-archive/main/data.csv",
|
||
)
|
||
CSV_REFRESH_TTL = int(os.environ.get("FUEL_CSV_TTL", "3600")) # seconds (1h)
|
||
|
||
# Source selection: "auto" (API if creds, CSV fallback), "csv", "api"
|
||
SOURCE = os.environ.get("FUEL_SOURCE", "auto").lower()
|
||
|
||
DEMO_MODE = not (CLIENT_ID and CLIENT_SECRET) and SOURCE != "csv"
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# In-memory cache state
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_state = {
|
||
"stations": None, # normalised station list [{...}, ...]
|
||
"stations_fetched_at": 0.0,
|
||
"token": None,
|
||
"token_expires_at": 0.0,
|
||
"last_error": None,
|
||
"sync_attempts": 0,
|
||
"sync_failures": 0,
|
||
"retry_attempt": 0, # consecutive failures (backoff exponent)
|
||
"retry_active": False, # a backoff sleep is in flight
|
||
"retry_in": None, # seconds until the next retry (None when idle)
|
||
"next_retry_at": 0.0, # epoch of the next scheduled retry
|
||
"source_used": None,
|
||
"data_updated": None, # freshest govUK price_last_updated across dataset
|
||
}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Admin telemetry (inbound connections only — purely observational, no effect
|
||
# on responses and no GOV.UK traffic). Aggregates who connects, how often and
|
||
# from where so the developer can monitor the app/widget poll model. Toggle:
|
||
# FUEL_STATS_ENABLED=0 disables collection entirely.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_stats_enabled = os.environ.get("FUEL_STATS_ENABLED", "1") == "1"
|
||
_stats_started = time.time()
|
||
_stats = {
|
||
"hits": 0,
|
||
"by_client": {}, # client kind ("app" | "widget" | "siri" | "admin" | "unknown") -> count
|
||
"by_route": {}, # request path -> count
|
||
"by_ip": {}, # client IP -> {"hits": n, "last_seen": epoch}
|
||
"recent": deque(maxlen=2000), # (epoch, ip, client, path) — rolling 24 h window
|
||
}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Normalisation helpers (shared by CSV fallback and API path)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Map official Fuel Finder fuel types → the app's grades (E5/E10/DIESEL).
|
||
# The app tracks exactly these three; everything else (HVO, SDV, B10, ...) is
|
||
# dropped so it can't poison the cheapest-price reference.
|
||
FUEL_MAP = {
|
||
"E5": "E5",
|
||
"E10": "E10",
|
||
"B7": "DIESEL",
|
||
"B7_STANDARD": "DIESEL",
|
||
"B7S": "DIESEL",
|
||
"B7P": "DIESEL",
|
||
"B7_PREMIUM": "DIESEL",
|
||
"B10": "DIESEL",
|
||
}
|
||
|
||
|
||
def _parse_price(value) -> Optional[float]:
|
||
"""Parse a Fuel Finder price value into pence/litre, or None.
|
||
|
||
The API sends DECIMAL STRINGS in pence ("0120.0000" = 120.0p). Some
|
||
stations report pounds (values below 2.0) — detect and multiply by 100.
|
||
Returns None for junk (non-numeric, or outside the sane 50–500p band so a
|
||
pounds-denominated column can't poison the cheapest reference).
|
||
"""
|
||
if value is None:
|
||
return None
|
||
try:
|
||
raw = float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if raw < 2.0:
|
||
# Pounds-denominated ("1.299" → 129.9p). Guard against a near-zero
|
||
# empty/placeholder value being inflated to something plausible.
|
||
raw *= 100
|
||
if raw < 50 or raw > 500:
|
||
return None
|
||
return raw
|
||
|
||
|
||
def _normalise_station(station_id, name, brand, address, postcode, lat, lng,
|
||
prices, price_updated=None, is_motorway=False) -> Optional[dict]:
|
||
"""Build a station dict in the relay's canonical output shape.
|
||
|
||
Returns None if the station has no tracked fuel prices (sells nothing the
|
||
app monitors) or its location is unusable.
|
||
"""
|
||
if not prices:
|
||
return None
|
||
if lat is None or lng is None:
|
||
return None
|
||
return {
|
||
"id": station_id,
|
||
"name": name,
|
||
"brand": brand or "",
|
||
"address": address or "",
|
||
"postcode": postcode or "",
|
||
"lat": lat,
|
||
"lng": lng,
|
||
"prices": prices, # {"E5": 144.9, "E10": 137.9, "DIESEL": ...}
|
||
"price_updated": price_updated,
|
||
"is_motorway": bool(is_motorway),
|
||
}
|
||
|
||
|
||
def _normalise_api_payload(stations_info: list, fuel_prices: list) -> list:
|
||
"""Join the official API's two endpoints on node_id → canonical stations.
|
||
|
||
stations_info: records from GET /api/v1/pfs (site details + location).
|
||
fuel_prices: records from GET /api/v1/pfs/fuel-prices (prices array).
|
||
"""
|
||
info_by_id = {}
|
||
for rec in stations_info:
|
||
node_id = rec.get("node_id")
|
||
if not node_id:
|
||
continue
|
||
loc = rec.get("location") or {}
|
||
lat = _parse_lat_lng(loc.get("latitude"))
|
||
lng = _parse_lat_lng(loc.get("longitude"))
|
||
info_by_id[node_id] = {
|
||
"name": rec.get("trading_name"),
|
||
"brand": rec.get("brand_name"),
|
||
"address": _join_address(loc),
|
||
"postcode": loc.get("postcode"),
|
||
"lat": lat,
|
||
"lng": lng,
|
||
"is_motorway": rec.get("is_motorway_service_station", False),
|
||
}
|
||
|
||
stations = []
|
||
for rec in fuel_prices:
|
||
node_id = rec.get("node_id")
|
||
info = info_by_id.get(node_id)
|
||
if info is None:
|
||
continue
|
||
prices = {}
|
||
latest_updated = None
|
||
for entry in rec.get("fuel_prices") or []:
|
||
fuel_type = (entry.get("fuel_type") or "").upper()
|
||
grade = FUEL_MAP.get(fuel_type)
|
||
if grade is None:
|
||
continue
|
||
price = _parse_price(entry.get("price"))
|
||
if price is None:
|
||
continue
|
||
# Multiple diesel variants (B7_STANDARD, B7_PREMIUM, B10) map to
|
||
# one DIESEL grade — take the MINIMUM so the cheapest-diesel
|
||
# reference is never inflated by a premium price.
|
||
if grade == "DIESEL":
|
||
prices[grade] = min(prices.get(grade, float("inf")), price)
|
||
else:
|
||
prices[grade] = price
|
||
updated = entry.get("price_last_updated")
|
||
if updated and (latest_updated is None or updated > latest_updated):
|
||
latest_updated = updated
|
||
station = _normalise_station(
|
||
station_id=node_id,
|
||
name=info["name"],
|
||
brand=info["brand"],
|
||
address=info["address"],
|
||
postcode=info["postcode"],
|
||
lat=info["lat"],
|
||
lng=info["lng"],
|
||
prices=prices,
|
||
price_updated=latest_updated,
|
||
is_motorway=info["is_motorway"],
|
||
)
|
||
if station is not None:
|
||
stations.append(station)
|
||
return stations
|
||
|
||
|
||
def _parse_lat_lng(value) -> Optional[float]:
|
||
"""API sends lat/lng as STRINGS — parse to float, tolerate garbage."""
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _join_address(loc: dict) -> str:
|
||
parts = [loc.get("address_line_1"), loc.get("address_line_2")]
|
||
return " ".join(p for p in parts if p) or ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CSV fallback (public mirror — same parser as fuelboard-relay)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _parse_csv_to_stations(text: str) -> list:
|
||
"""Parse the Fuel Finder archive CSV into a station list.
|
||
|
||
Header uses dotted keys: forecourts.trading_name, forecourts.location.latitude,
|
||
forecourts.fuel_price.E10, forecourts.price_change_effective_timestamp.E10, etc.
|
||
"""
|
||
import csv
|
||
import io
|
||
|
||
reader = csv.DictReader(io.StringIO(text))
|
||
stations = []
|
||
for row in reader:
|
||
def num(key):
|
||
try:
|
||
val = row.get(key)
|
||
return float(val) if val not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
lat = num("forecourts.location.latitude")
|
||
lng = num("forecourts.location.longitude")
|
||
prices = {}
|
||
for grade, label in (("E5", "E5"), ("E10", "E10"), ("B7S", "DIESEL"), ("B7P", "DIESEL")):
|
||
p = num(f"forecourts.fuel_price.{grade}")
|
||
if p is not None and grade not in prices:
|
||
prices[label] = p
|
||
station = _normalise_station(
|
||
station_id=row.get("forecourts.node_id"),
|
||
name=row.get("forecourts.trading_name"),
|
||
brand=row.get("forecourts.brand_name"),
|
||
address=_join_address({
|
||
"address_line_1": row.get("forecourts.location.address_line_1"),
|
||
"address_line_2": row.get("forecourts.location.address_line_2"),
|
||
}),
|
||
postcode=row.get("forecourts.location.postcode"),
|
||
lat=lat,
|
||
lng=lng,
|
||
prices=prices,
|
||
price_updated=row.get("forecourts.price_change_effective_timestamp.E10")
|
||
or row.get("forecourts.price_change_effective_timestamp.E5"),
|
||
is_motorway=row.get("forecourts.is_motorway_service_station") == "true",
|
||
)
|
||
if station is not None:
|
||
stations.append(station)
|
||
return stations
|
||
|
||
|
||
def _fetch_csv() -> list:
|
||
"""Download the hourly full-UK CSV mirror."""
|
||
import csv
|
||
import io
|
||
|
||
with httpx.Client(timeout=120) as client:
|
||
resp = client.get(CSV_URL)
|
||
resp.raise_for_status()
|
||
stations = _parse_csv_to_stations(resp.text)
|
||
if not stations:
|
||
raise RuntimeError("CSV parse returned zero stations")
|
||
# Freshest price-change timestamp in the CSV dump (govUK-side update
|
||
# time for the mirror's source data).
|
||
latest = None
|
||
for row in csv.DictReader(io.StringIO(resp.text)):
|
||
for key, val in row.items():
|
||
if key.startswith("forecourts.price_change_effective_timestamp."):
|
||
if val and (latest is None or val > latest):
|
||
latest = val
|
||
_state["data_updated"] = latest
|
||
return stations
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Official API fetch (OAuth + batched paging + node_id join)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _get_token(client: httpx.Client) -> str:
|
||
"""Return a valid bearer token, refreshing via client-credentials grant.
|
||
|
||
The Fuel Finder token endpoint returns the token under "data":
|
||
{ "data": { "access_token": "...", "expires_in": 3600 }, ... }
|
||
Tolerates a flat { "access_token": ... } shape too (older docs).
|
||
"""
|
||
if _state["token"] and time.time() < _state["token_expires_at"] - TOKEN_EXPIRY_BUFFER:
|
||
return _state["token"]
|
||
|
||
resp = client.post(
|
||
TOKEN_URL,
|
||
json={"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
|
||
)
|
||
resp.raise_for_status()
|
||
payload = resp.json()
|
||
data = payload.get("data") if isinstance(payload.get("data"), dict) else payload
|
||
token = data.get("access_token") or payload.get("access_token")
|
||
if not token:
|
||
raise RuntimeError(f"Token response missing access_token: {list(payload.keys())}")
|
||
expires_in = int(data.get("expires_in", payload.get("expires_in", 3600)))
|
||
_state["token"] = token
|
||
_state["token_expires_at"] = time.time() + expires_in
|
||
return token
|
||
|
||
|
||
def _fetch_all_pages(client: httpx.Client, url: str) -> list:
|
||
"""Fetch every page of a batched Fuel Finder endpoint (500/page).
|
||
|
||
Fuel Finder is strictly sequential (30 rpm per client): a request sent
|
||
before the previous one completes gets a 429. We therefore sleep
|
||
BATCH_SLEEP between pages and never issue a second in-flight request.
|
||
"""
|
||
results = []
|
||
batch = 1
|
||
while True:
|
||
resp = client.get(
|
||
url,
|
||
params={"batch-number": batch},
|
||
headers={"Authorization": f"Bearer {_get_token(client)}"},
|
||
)
|
||
resp.raise_for_status()
|
||
page = resp.json()
|
||
if not isinstance(page, list) or not page:
|
||
break
|
||
results.extend(page)
|
||
if len(page) < PAGE_SIZE:
|
||
break
|
||
batch += 1
|
||
if batch > 200: # safety cap (~100k records)
|
||
break
|
||
time.sleep(BATCH_SLEEP)
|
||
return results
|
||
|
||
|
||
def _fetch_api() -> list:
|
||
"""Fetch the full dataset from the official API and normalise it.
|
||
|
||
Two batched endpoints joined on node_id:
|
||
1. GET /api/v1/pfs → site details (trading_name, location, ...)
|
||
2. GET /api/v1/pfs/fuel-prices → per-station fuel_prices array
|
||
"""
|
||
with httpx.Client(timeout=60) as client:
|
||
# Warm the token once so both loops reuse it (token endpoint is the
|
||
# most rate-sensitive; docs say reuse until near expiry).
|
||
_get_token(client)
|
||
stations_info = _fetch_all_pages(client, PFS_URL)
|
||
fuel_prices = _fetch_all_pages(client, PRICES_URL)
|
||
stations = _normalise_api_payload(stations_info, fuel_prices)
|
||
if not stations:
|
||
raise RuntimeError("API fetch returned zero stations")
|
||
# Freshest price timestamp the GOV.UK server itself reports — this is
|
||
# the dataset's own update time, independent of when we synced.
|
||
latest = None
|
||
for rec in fuel_prices:
|
||
for entry in rec.get("fuel_prices") or []:
|
||
ts = entry.get("price_last_updated")
|
||
if ts and (latest is None or ts > latest):
|
||
latest = ts
|
||
_state["data_updated"] = latest
|
||
return stations
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Cache refresh
|
||
# ---------------------------------------------------------------------------
|
||
|
||
import threading
|
||
|
||
_refresh_lock = threading.Lock()
|
||
|
||
_RETRY_BASE_S = 15 # first retry 15s after a failure
|
||
_RETRY_MAX_S = 900 # exponential backoff capped at 15 minutes
|
||
|
||
|
||
def _refresh_cache() -> None:
|
||
"""Refresh the station cache (called synchronously or from a thread)."""
|
||
with _refresh_lock:
|
||
_refresh_cache_locked()
|
||
|
||
|
||
def _refresh_cache_locked() -> None:
|
||
"""Refresh the station cache if stale (or on first run).
|
||
|
||
Callers hold _refresh_lock. Full API sync takes ~1-2 min (17+ pages × 2
|
||
endpoints × BATCH_SLEEP), so this must NEVER run inline in a request path
|
||
— see _ensure_fresh_cache().
|
||
"""
|
||
_state["sync_attempts"] += 1
|
||
try:
|
||
if SOURCE == "csv":
|
||
_state["stations"] = _fetch_csv()
|
||
_state["source_used"] = "csv"
|
||
elif SOURCE == "api":
|
||
_state["stations"] = _fetch_api()
|
||
_state["source_used"] = "api"
|
||
else: # auto: official API if creds present, otherwise public CSV
|
||
if not DEMO_MODE:
|
||
_state["stations"] = _fetch_api()
|
||
_state["source_used"] = "api"
|
||
else:
|
||
_state["stations"] = _fetch_csv()
|
||
_state["source_used"] = "csv"
|
||
_state["stations_fetched_at"] = time.time()
|
||
_state["last_error"] = None
|
||
_state["retry_attempt"] = 0
|
||
_state["retry_in"] = None
|
||
_state["next_retry_at"] = 0.0
|
||
except Exception as exc: # keep serving stale cache on failure
|
||
_state["sync_failures"] += 1
|
||
_state["last_error"] = f"{type(exc).__name__}: {exc}"
|
||
_state["retry_attempt"] += 1
|
||
delay = min(_RETRY_BASE_S * (2 ** (_state["retry_attempt"] - 1)), _RETRY_MAX_S)
|
||
_state["retry_in"] = delay
|
||
_state["next_retry_at"] = time.time() + delay
|
||
_schedule_retry(delay)
|
||
|
||
|
||
def _schedule_retry(delay: float) -> None:
|
||
"""Retry a failed sync on a daemon thread after `delay` seconds.
|
||
|
||
Only one retry sleeps at a time; the worker clears the flag before
|
||
refreshing, so a repeated failure can chain the next (longer) retry.
|
||
"""
|
||
if _state["retry_active"]:
|
||
return
|
||
_state["retry_active"] = True
|
||
|
||
def worker() -> None:
|
||
time.sleep(delay)
|
||
_state["retry_active"] = False
|
||
_refresh_cache()
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
|
||
def _ensure_fresh_cache() -> None:
|
||
"""Non-blocking staleness check: serve stale data, refresh in background.
|
||
|
||
A full API sync takes minutes (rate-limited batched paging); blocking a
|
||
request on it would blow past the app/widget's 5-second timeout. We return
|
||
whatever we have immediately and let the sync catch up on a daemon thread.
|
||
"""
|
||
if _state["stations"] is None:
|
||
# First run: block (usually the lifespan warm-up, but guards a race
|
||
# where a request beats it). CSV fallback is fast; API takes a while.
|
||
_refresh_cache()
|
||
return
|
||
if time.time() - _state["stations_fetched_at"] > CACHE_TTL:
|
||
threading.Thread(target=_refresh_cache, daemon=True).start()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(_: FastAPI):
|
||
# Always try to warm the cache. In auto mode with no credentials this
|
||
# fetches the public full-UK CSV; with credentials it uses the OAuth API.
|
||
_refresh_cache()
|
||
yield
|
||
|
||
|
||
app = FastAPI(title="FuelBoard Relay API", version="1.0.0", lifespan=lifespan)
|
||
|
||
|
||
@app.middleware("http")
|
||
async def _record_stats(request: Request, call_next):
|
||
"""Admin telemetry — tag every request with client kind + IP. The X-Client
|
||
header is set by the app ("app") and widget extension ("widget"); Siri
|
||
answers from the cached dataset and never fetches. Untagged requests fall
|
||
back to path inference (widget-diag beacon) or "unknown". Purely
|
||
observational — no effect on responses (see FUEL_STATS_ENABLED toggle)."""
|
||
response = await call_next(request)
|
||
if _stats_enabled:
|
||
path = request.url.path
|
||
client = (request.headers.get("x-client") or "").lower()
|
||
if not client:
|
||
if path == "/api/v1/widget-diag":
|
||
client = "widget"
|
||
elif path in ("/stats", "/health"):
|
||
client = "admin"
|
||
else:
|
||
client = "unknown"
|
||
ip = request.client.host if request.client else "?"
|
||
now = time.time()
|
||
_stats["hits"] += 1
|
||
_stats["by_client"][client] = _stats["by_client"].get(client, 0) + 1
|
||
_stats["by_route"][path] = _stats["by_route"].get(path, 0) + 1
|
||
ipstat = _stats["by_ip"].setdefault(ip, {"hits": 0, "last_seen": 0.0})
|
||
ipstat["hits"] += 1
|
||
ipstat["last_seen"] = now
|
||
_stats["recent"].append((now, ip, client, path))
|
||
return response
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _stations_list() -> list:
|
||
if _state["stations"] is None:
|
||
return []
|
||
data = _state["stations"]
|
||
if isinstance(data, list):
|
||
return data
|
||
if isinstance(data, dict):
|
||
for key in ("stations", "prices", "data"):
|
||
if isinstance(data.get(key), list):
|
||
return data[key]
|
||
return []
|
||
|
||
|
||
def _distance_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
|
||
import math
|
||
r = 6371.0
|
||
dlat = math.radians(lat2 - lat1)
|
||
dlng = math.radians(lng2 - lng1)
|
||
a = math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlng / 2) ** 2
|
||
return r * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
||
|
||
|
||
def _fuel_price(station: dict, fuel: str) -> Optional[float]:
|
||
"""Extract a price for the fuel grade from a normalised station dict."""
|
||
prices = station.get("prices") or {}
|
||
raw = prices.get(fuel.upper())
|
||
if isinstance(raw, (int, float)) and 50 <= raw <= 500:
|
||
return float(raw)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Routes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/health")
|
||
def health():
|
||
return {
|
||
"status": "ok" if not DEMO_MODE and _state["stations"] is not None else "degraded",
|
||
"demo_mode": DEMO_MODE,
|
||
"source": _state["source_used"],
|
||
"stations_cached": _state["stations"] is not None,
|
||
"stations_count": len(_stations_list()),
|
||
"data_updated": _state["data_updated"],
|
||
"last_sync": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_state["stations_fetched_at"])) if _state["stations_fetched_at"] else None,
|
||
"sync_attempts": _state["sync_attempts"],
|
||
"sync_failures": _state["sync_failures"],
|
||
"retry_active": _state["retry_active"],
|
||
"retry_in": _state["retry_in"],
|
||
"next_retry_at": (time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_state["next_retry_at"]))
|
||
if _state["next_retry_at"] else None),
|
||
"last_error": _state["last_error"],
|
||
}
|
||
|
||
|
||
@app.get("/api/v1/stations")
|
||
def stations(
|
||
fuel: str = Query("e10", description="Fuel grade: e10, e5, diesel"),
|
||
lat: Optional[float] = Query(None, description="User latitude"),
|
||
lng: Optional[float] = Query(None, description="User longitude"),
|
||
radius: Optional[float] = Query(None, description="Search radius in km (optional; omit for the full dataset)"),
|
||
limit: int = Query(10000, ge=1, le=10000),
|
||
):
|
||
if _state["stations"] is None:
|
||
raise HTTPException(status_code=502, detail="No data cached yet and upstream fetch failed")
|
||
|
||
# Refresh in the background if stale; never block a request on a
|
||
# multi-minute rate-limited API sync.
|
||
_ensure_fresh_cache()
|
||
|
||
rows = []
|
||
for station in _stations_list():
|
||
s_lat = station.get("lat")
|
||
s_lng = station.get("lng")
|
||
# All known fuel grades for this station (E5/E10/DIESEL) so clients can
|
||
# switch fuel tabs without another request. Every station is returned
|
||
# regardless of the requested fuel — clients filter on-device.
|
||
prices = {}
|
||
for grade in ("E5", "E10", "DIESEL"):
|
||
p = _fuel_price(station, grade)
|
||
if p is not None:
|
||
prices[grade] = p
|
||
if not prices:
|
||
continue # station sells nothing we track — skip
|
||
item = {
|
||
"id": station.get("id"),
|
||
"name": station.get("name"),
|
||
"brand": station.get("brand") or "",
|
||
"address": station.get("address") or "",
|
||
"postcode": station.get("postcode") or "",
|
||
"lat": s_lat,
|
||
"lng": s_lng,
|
||
"price": prices.get(fuel.upper()),
|
||
"prices": prices,
|
||
"price_updated": station.get("price_updated"),
|
||
}
|
||
if lat is not None and lng is not None and isinstance(s_lat, (int, float)) and isinstance(s_lng, (int, float)):
|
||
item["distance_km"] = round(_distance_km(lat, lng, s_lat, s_lng), 2)
|
||
rows.append(item)
|
||
|
||
if lat is not None and lng is not None:
|
||
# Optional server-side radius filter — used by the alert path for a
|
||
# focused fresh fetch. The main app fetch omits radius and gets all.
|
||
if radius is not None:
|
||
rows = [r for r in rows if r.get("distance_km") is not None and r["distance_km"] <= radius]
|
||
rows.sort(key=lambda r: r.get("distance_km") if r.get("distance_km") is not None else float("inf"))
|
||
else:
|
||
rows.sort(key=lambda r: r["price"] if r["price"] is not None else float("inf"))
|
||
|
||
return {
|
||
"fuel": fuel,
|
||
"count": len(rows[:limit]),
|
||
"stations_count": len(rows),
|
||
"source": _state["source_used"],
|
||
"data_updated": _state["data_updated"],
|
||
"stations": rows[:limit],
|
||
}
|
||
|
||
|
||
@app.get("/api/v1/widget-diag")
|
||
async def widget_diag(kind: str = "", source: str = "", n: str = "", fuel: str = "",
|
||
sort: str = "", fav: str = "", station: str = "", intent: str = ""):
|
||
"""Widget diagnostics beacon — the widget extension fire-and-forgets a GET
|
||
here at the end of every makeEntry so the relay log records whether a
|
||
widget timeline actually ran in the extension and what it produced.
|
||
Logged via the access log; returns 204 (no body)."""
|
||
from fastapi import Response
|
||
return Response(status_code=204)
|
||
|
||
|
||
@app.get("/stats")
|
||
def stats(request: Request):
|
||
"""Admin telemetry — LOCALHOST ONLY (404 for LAN/Tailscale clients so the
|
||
endpoint is indistinguishable from a non-existent route to probes).
|
||
|
||
Aggregates inbound client traffic over a rolling 24 h window (the last
|
||
2000 requests): total hits, hits by client kind (app/widget/siri/admin/
|
||
unknown), by route, and per client IP with last-seen. Purely observational
|
||
— see the _record_stats middleware and FUEL_STATS_ENABLED toggle."""
|
||
host = request.client.host if request.client else ""
|
||
if host not in ("127.0.0.1", "::1"):
|
||
raise HTTPException(status_code=404, detail="Not found")
|
||
now = time.time()
|
||
recent = [r for r in _stats["recent"] if now - r[0] <= 86400]
|
||
by_client: dict = {}
|
||
by_route: dict = {}
|
||
by_ip: dict = {}
|
||
for ts, ip, client, path in recent:
|
||
by_client[client] = by_client.get(client, 0) + 1
|
||
by_route[path] = by_route.get(path, 0) + 1
|
||
ipstat = by_ip.setdefault(ip, {"hits": 0, "last_seen": 0.0})
|
||
ipstat["hits"] += 1
|
||
ipstat["last_seen"] = max(ipstat["last_seen"], ts)
|
||
|
||
def _fmt_ts(epoch: float) -> str:
|
||
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))
|
||
|
||
return {
|
||
"enabled": _stats_enabled,
|
||
"started_at": _fmt_ts(_stats_started),
|
||
"uptime_seconds": int(now - _stats_started),
|
||
"total_hits": _stats["hits"],
|
||
"hits_24h": len(recent),
|
||
"window": "rolling 24h of last 2000 requests",
|
||
"by_client": {k: by_client[k] for k in sorted(by_client, key=lambda k: -by_client[k])},
|
||
"by_route": {k: by_route[k] for k in sorted(by_route, key=lambda k: -by_route[k])},
|
||
"by_ip": {
|
||
ip: {"hits": s["hits"], "last_seen": _fmt_ts(s["last_seen"])}
|
||
for ip, s in sorted(by_ip.items(), key=lambda kv: -kv[1]["hits"])
|
||
},
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
|
||
port = int(os.environ.get("FUEL_RELAY_PORT", "8789"))
|
||
uvicorn.run(app, host="0.0.0.0", port=port)
|