"""Vehicle Map static server and server-side proxy for the vehicle open platform.""" from concurrent.futures import ThreadPoolExecutor 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.request import Request, urlopen ROOT = Path(__file__).resolve().parent 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", "5")) STATION_CACHE_SECONDS = int(os.getenv("STATION_CACHE_SECONDS", "3600")) _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_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( _cached, "stations", STATION_CACHE_SECONDS, lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}), ) 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" } driving = sum(1 for item in vehicles if item.get("motionStatus") == "driving") idle = sum(1 for item in vehicles if item.get("motionStatus") == "idle") offline = sum(1 for item in vehicles if item.get("motionStatus") == "offline") daily_mileage = round( sum(float(row.get("dailyMileageKm") or 0) for row in mileage_rows), 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": driving + idle, "drivingVehicles": driving, "idleVehicles": idle, "offlineVehicles": offline, "todayMileageKm": daily_mileage, "totalStations": len(stations), "cooperativeStations": sum(1 for item in stations if item.get("cooperative")), }, "vehicles": vehicles, "stations": stations, } class VehicleMapHandler(SimpleHTTPRequestHandler): server_version = "LingniuVehicleMap/1.0" def do_GET(self): if self.path == "/api/health": self._write_json(200, { "status": "ok", "service": "vehicle-map", "openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY), }) return if self.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 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__": server = ThreadingHTTPServer((HOST, PORT), VehicleMapHandler) print(f"Lingniu Vehicle Map listening on http://{HOST}:{PORT}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()