Files
lingniu-vehicle-ingest/station-navigation/server.py
T

234 lines
8.7 KiB
Python

"""Public hydrogen station navigation service.
This process deliberately has a smaller API boundary than the operations map:
it retrieves only station directory fields and never exposes vehicle records or
hydrogen-volume fields.
"""
import hashlib
from http.server import HTTPServer, SimpleHTTPRequestHandler
import json
import os
from pathlib import Path
from socketserver import ThreadingMixIn
import threading
import time
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen
ROOT = Path(__file__).resolve().parent
# Production deliberately reuses the operations map's visual assets. Keeping
# this configurable lets the standalone release live under its own directory.
VEHICLE_MAP_ROOT = Path(os.getenv("VEHICLE_MAP_ASSET_ROOT", str(ROOT.parent / "vehicle-map")))
HOST = os.getenv("STATION_NAVIGATION_HOST", "0.0.0.0")
PORT = int(os.getenv("STATION_NAVIGATION_PORT", "20804"))
OPEN_PLATFORM_BASE_URL = os.getenv("OPEN_PLATFORM_BASE_URL", "https://open.d.lnoneos.com").rstrip("/")
OPEN_PLATFORM_APP_KEY = os.getenv("OPEN_PLATFORM_APP_KEY", "").strip()
UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15"))
STATION_CACHE_SECONDS = int(os.getenv("STATION_NAVIGATION_CACHE_SECONDS", "120"))
_cache_lock = threading.Lock()
_cache = {}
def _static_asset_version():
digest = hashlib.sha256()
for filename in ("app.js", "styles.css"):
digest.update((ROOT / filename).read_bytes())
digest.update((VEHICLE_MAP_ROOT / "styles.css").read_bytes())
return digest.hexdigest()[:16]
STATIC_ASSET_VERSION = _static_asset_version()
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
allow_reuse_address = True
def _json_bytes(value):
return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def _post_open_platform(path, body):
if not OPEN_PLATFORM_APP_KEY:
raise RuntimeError("OPEN_PLATFORM_APP_KEY is not configured")
request = Request(
OPEN_PLATFORM_BASE_URL + path,
data=_json_bytes(body),
method="POST",
headers={
"Authorization": "Bearer " + OPEN_PLATFORM_APP_KEY,
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "lingniu-station-navigation/1.0",
},
)
try:
with urlopen(request, timeout=UPSTREAM_TIMEOUT_SECONDS) as response:
payload = json.load(response)
except HTTPError as exc:
detail = exc.read(2048).decode("utf-8", errors="replace")
raise RuntimeError(f"open platform returned HTTP {exc.code}: {detail}") from exc
except URLError as exc:
raise RuntimeError(f"open platform unavailable: {exc.reason}") from exc
if payload.get("code") != "SUCCESS":
raise RuntimeError(f"open platform rejected request: {payload.get('code')} {payload.get('message')}")
return payload.get("data") or []
def _cached(key, ttl_seconds, loader):
now = time.time()
with _cache_lock:
cached = _cache.get(key)
if cached and now - cached[0] < ttl_seconds:
return cached[1]
value = loader()
with _cache_lock:
_cache[key] = (now, value)
return value
def _public_station(station):
"""Positive allowlist: do not add operational or hydrogen fields here."""
fields = (
"id", "name", "shortName", "province", "city", "district", "address",
"longitude", "latitude", "cooperative",
)
return {field: station.get(field) for field in fields if station.get(field) is not None}
def _load_station_directory():
stations = _post_open_platform("/api/v1/hydrogen-stations/query", {})
directory = [_public_station(station) for station in stations]
return {
"status": "ok",
"asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
"summary": {
"totalStations": len(directory),
"cooperativeStations": sum(1 for station in directory if station.get("cooperative")),
},
"stations": directory,
}
def _prewarm_station_directory_cache():
"""Fetch the public directory after startup so first navigation is warm."""
try:
_cached("station-directory", STATION_CACHE_SECONDS, _load_station_directory)
except Exception as exc: # The request path will retry without blocking startup.
print(f"station navigation cache warmup deferred: {exc}")
class StationNavigationHandler(SimpleHTTPRequestHandler):
server_version = "LingniuStationNavigation/1.0"
def do_GET(self):
path = urlparse(self.path).path
if path in {"/", "/index.html"}:
self._serve_index()
return
if path == "/api/health":
self._write_json(200, {
"status": "ok",
"service": "station-navigation",
"openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY),
})
return
if path == "/api/stations":
try:
self._write_json(200, _cached("station-directory", STATION_CACHE_SECONDS, _load_station_directory))
except Exception as exc:
self.log_error("station directory refresh failed: %s", exc)
self._write_json(502, {
"status": "error",
"code": "UPSTREAM_UNAVAILABLE",
"message": "加氢站数据暂时不可用,请稍后重试",
})
return
if path == "/assets/logo_light.svg":
self._serve_logo()
return
if path == "/assets/vehicle-map.css":
self._serve_vehicle_map_styles()
return
if path.startswith("/api/"):
self._write_json(404, {"status": "error", "code": "NOT_FOUND"})
return
super().do_GET()
def _serve_index(self):
try:
template = (ROOT / "index.html").read_text(encoding="utf-8")
body = template.replace("__ASSET_VERSION__", STATIC_ASSET_VERSION).encode("utf-8")
except OSError as exc:
self.log_error("index template unavailable: %s", exc)
self.send_error(500, "Station navigation is temporarily unavailable")
return
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(body)
def _serve_logo(self):
try:
body = (VEHICLE_MAP_ROOT / "logo_light.svg").read_bytes()
except OSError:
self.send_error(404)
return
self.send_response(200)
self.send_header("Content-Type", "image/svg+xml")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "public, max-age=86400")
self.end_headers()
self.wfile.write(body)
def _serve_vehicle_map_styles(self):
try:
body = (VEHICLE_MAP_ROOT / "styles.css").read_bytes()
except OSError:
self.send_error(404)
return
self.send_response(200)
self.send_header("Content-Type", "text/css; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(body)
def end_headers(self):
if self.path == "/" or self.path.split("?", 1)[0].endswith((".html", ".css", ".js", ".svg")):
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "strict-origin-when-cross-origin")
self.send_header("X-Frame-Options", "SAMEORIGIN")
super().end_headers()
def _write_json(self, status, payload):
body = _json_bytes(payload)
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
# Python 3.6 on ECS predates SimpleHTTPRequestHandler(directory=...).
# Resolve static assets from this release directory before serving.
os.chdir(str(ROOT))
server = ThreadingHTTPServer((HOST, PORT), StationNavigationHandler)
threading.Thread(target=_prewarm_station_directory_cache, daemon=True).start()
print(f"Lingniu Station Navigation listening on http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()