vehicle-map-0803/server.py

"""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()