"""Vehicle Map static server and server-side proxy for the vehicle open platform.""" from concurrent.futures import ThreadPoolExecutor import hashlib from http.server import HTTPServer, SimpleHTTPRequestHandler import json import os from pathlib import Path import threading import time from socketserver import ThreadingMixIn from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parent def _static_asset_version(): """Content fingerprint used to bypass upstream static-asset caches on release.""" digest = hashlib.sha256() for filename in ("app.js", "styles.css"): digest.update((ROOT / filename).read_bytes()) return digest.hexdigest()[:16] STATIC_ASSET_VERSION = _static_asset_version() HOST = os.getenv("VEHICLE_MAP_HOST", "0.0.0.0") PORT = int(os.getenv("VEHICLE_MAP_PORT", "20800")) 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")) DASHBOARD_CACHE_SECONDS = int(os.getenv("DASHBOARD_CACHE_SECONDS", "120")) STATION_CACHE_SECONDS = int(os.getenv("STATION_CACHE_SECONDS", "120")) _cache_lock = threading.Lock() _cache = {} 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-vehicle-map/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 _load_stations(): return _cached( "stations", STATION_CACHE_SECONDS, lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}), ) def _load_dashboard(): today = time.strftime("%Y-%m-%d", time.localtime()) with ThreadPoolExecutor(max_workers=3) as executor: realtime_future = executor.submit( _post_open_platform, "/api/v1/vehicles/realtime/query", {} ) mileage_future = executor.submit( _post_open_platform, "/api/v1/vehicles/mileage/query", {"date": today} ) stations_future = executor.submit(_load_stations) vehicles = realtime_future.result() mileage_rows = mileage_future.result() stations = stations_future.result() mileage_by_vin = { row.get("vin"): row for row in mileage_rows if row.get("vin") and row.get("status") == "NORMAL" } # The daily activity view places every vehicle that reported # today is placed in exactly one bucket using its latest selected speed. active_vehicles = [item for item in vehicles if item.get("activeToday")] driving = sum(1 for item in active_vehicles if float(item.get("speedKmh") or 0) > 3) idle = len(active_vehicles) - driving offline = sum(1 for item in vehicles if item.get("motionStatus") == "offline") active_today = len(active_vehicles) daily_mileage = round( sum(float(row.get("dailyMileageKm") or 0) for row in mileage_rows), 3 ) monthly_hydrogen = round( sum(float(station.get("monthlyHydrogenKg") or 0) for station in stations), 3 ) for vehicle in vehicles: mileage = mileage_by_vin.get(vehicle.get("vin"), {}) vehicle["dailyMileageKm"] = mileage.get("dailyMileageKm", 0) return { "status": "ok", "asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "date": today, "summary": { "totalVehicles": len(vehicles), "onlineVehicles": active_today, "drivingVehicles": driving, "idleVehicles": idle, "offlineVehicles": offline, "todayMileageKm": daily_mileage, "monthlyHydrogenKg": monthly_hydrogen, "totalStations": len(stations), "cooperativeStations": sum(1 for item in stations if item.get("cooperative")), }, "vehicles": vehicles, "stations": stations, } def _prewarm_dashboard_cache(): """Warm the first dashboard snapshot after a process restart. A warm snapshot lets the first visitor receive the cached response instead of waiting for the three upstream requests to complete. """ try: _cached("dashboard", DASHBOARD_CACHE_SECONDS, _load_dashboard) except Exception as exc: # A later request can retry; startup must stay available. print(f"vehicle map cache warmup deferred: {exc}") class VehicleMapHandler(SimpleHTTPRequestHandler): server_version = "LingniuVehicleMap/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": "vehicle-map", "openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY), }) return if path == "/api/dashboard": try: payload = _cached("dashboard", DASHBOARD_CACHE_SECONDS, _load_dashboard) self._write_json(200, payload) except Exception as exc: # keep upstream details server-side only self.log_error("dashboard refresh failed: %s", exc) self._write_json(502, { "status": "error", "code": "UPSTREAM_UNAVAILABLE", "message": "地图数据暂时不可用，请稍后重试", }) 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, "Map application 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 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, headers=None): 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))) for name, value in (headers or {}).items(): self.send_header(name, value) self.end_headers() self.wfile.write(body) if __name__ == "__main__": server = ThreadingHTTPServer((HOST, PORT), VehicleMapHandler) threading.Thread(target=_prewarm_dashboard_cache, daemon=True).start() print(f"Lingniu Vehicle Map listening on http://{HOST}:{PORT}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()