diff --git a/station-navigation/README.md b/station-navigation/README.md index 3257a16f..b07b6dbb 100644 --- a/station-navigation/README.md +++ b/station-navigation/README.md @@ -2,7 +2,7 @@ 面向运维和司机调度的独立公共站点导航服务，默认本地端口为 `20804`。 -它只调用一次上游加氢站目录接口，并通过正向字段白名单对外提供：名称、位置、省市区、地址与合作状态。车辆数据、加氢量及其他运营字段不会被加载或从这个服务的 API 返回。 +它复用运营地图已经按 GPS 逆地理解析、落盘缓存的站点目录，并只对外提供：名称、位置、省市区、地址与合作状态。车辆数据、加氢量及其他运营字段不会被加载或从这个服务的 API 返回。 ## 本地运行 @@ -19,7 +19,7 @@ export OPEN_PLATFORM_APP_KEY='<32位AppKey>' - `GET /api/health`：进程健康状态； - 其他 `/api/*` 路由均返回 404。 -站点目录在服务端缓存 2 分钟，并在服务启动后后台预热；因此正常页面打开直接返回缓存数据，最多延迟 2 分钟更新一次。 +站点目录在服务端缓存 2 分钟，并在服务启动后后台预热；因此正常页面打开直接返回缓存数据，最多延迟 2 分钟更新一次。省、市、区由运营地图根据 GCJ-02 坐标调用高德逆地理服务生成，结果按坐标缓存 7 天；详细地址仍直接展示资产原始字段。 ## ECS 部署 diff --git a/station-navigation/server.py b/station-navigation/server.py index 2f681ec5..db96862a 100644 --- a/station-navigation/server.py +++ b/station-navigation/server.py @@ -28,6 +28,7 @@ OPEN_PLATFORM_BASE_URL = os.getenv("OPEN_PLATFORM_BASE_URL", "https://open.d.lno 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")) +VEHICLE_MAP_INTERNAL_BASE_URL = os.getenv("VEHICLE_MAP_INTERNAL_BASE_URL", "http://127.0.0.1:20800").rstrip("/") _cache_lock = threading.Lock() _cache = {} @@ -91,27 +92,24 @@ def _cached(key, ttl_seconds, loader): 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, - } + """Use the operations map's GPS-resolved public station directory. + + The map service owns reverse-geocoding and its persistent cache, keeping + the public navigation view and the operations view on identical boundaries. + """ + request = Request( + VEHICLE_MAP_INTERNAL_BASE_URL + "/api/stations", + headers={"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, URLError, ValueError) as exc: + raise RuntimeError("vehicle map station directory unavailable: {}".format(exc)) from exc + if payload.get("status") != "ok" or not isinstance(payload.get("stations"), list): + raise RuntimeError("vehicle map returned an invalid station directory") + return payload def _prewarm_station_directory_cache(): diff --git a/station-navigation/tests/test_server.py b/station-navigation/tests/test_server.py index c4c4b013..22385fab 100644 --- a/station-navigation/tests/test_server.py +++ b/station-navigation/tests/test_server.py @@ -1,4 +1,6 @@ import importlib.util +import io +import json from pathlib import Path import unittest from unittest.mock import patch @@ -11,21 +13,24 @@ SPEC.loader.exec_module(server) class StationNavigationTest(unittest.TestCase): def test_directory_is_a_positive_allowlist_without_hydrogen_data(self): - source = [{ + source = { + "status": "ok", + "asOf": "2026-08-12 12:00:00", + "summary": {"totalStations": 1, "cooperativeStations": 1}, + "stations": [{ "id": "GD-1", "name": "广州合作站", "province": "广东省", "city": "广州市", "district": "黄埔区", "address": "开源大道1号", "longitude": 113.2, "latitude": 23.1, - "cooperative": True, "monthlyHydrogenKg": 88.8, "totalHydrogenKg": 999.9, - "vehicleCount": 20, - }] - with patch.object(server, "_post_open_platform", return_value=source): + "cooperative": True, + }], + } + with patch.object(server, "urlopen", return_value=io.BytesIO(json.dumps(source).encode("utf-8"))): result = server._load_station_directory() self.assertEqual(result["summary"], {"totalStations": 1, "cooperativeStations": 1}) station = result["stations"][0] self.assertEqual(station["name"], "广州合作站") - self.assertNotIn("monthlyHydrogenKg", station) - self.assertNotIn("totalHydrogenKg", station) - self.assertNotIn("vehicleCount", station) + self.assertEqual(station["province"], "广东省") + self.assertEqual(server.VEHICLE_MAP_INTERNAL_BASE_URL, "http://127.0.0.1:20800") def test_static_assets_are_fingerprinted(self): self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$") diff --git a/vehicle-map/.env.example b/vehicle-map/.env.example index 4969dfa3..99393ca3 100644 --- a/vehicle-map/.env.example +++ b/vehicle-map/.env.example @@ -5,3 +5,8 @@ OPEN_PLATFORM_APP_KEY=replace-with-32-character-app-key UPSTREAM_TIMEOUT_SECONDS=15 DASHBOARD_CACHE_SECONDS=5 STATION_CACHE_SECONDS=3600 +# 高德 Web 服务 Key；用于以 GCJ-02 坐标逆地理得到省、市、区。 +AMAP_REGEOCODE_KEY=replace-with-amap-web-service-key +# 行政区结果按坐标落盘缓存 7 天，避免每次刷新重复调用高德。 +STATION_GEOCODE_CACHE_SECONDS=604800 +STATION_GEOCODE_CACHE_PATH=/opt/lingniu-vehicle-map/cache/station-geocode.json diff --git a/vehicle-map/README.md b/vehicle-map/README.md index f8a00c7d..6b366f48 100644 --- a/vehicle-map/README.md +++ b/vehicle-map/README.md @@ -32,12 +32,13 @@ python3 server.py - `POST /api/v1/vehicles/mileage/query`：全部授权车辆当日里程； - `POST /api/v1/hydrogen-stations/query`：资产库只读加氢站地图点位。 -以上三个接口均使用开放平台的 `Authorization: Bearer ` 鉴权。生产 AppKey 仅保存在 ECS 的 `/opt/lingniu-vehicle-map/env/vehicle-map.env`，文件权限为 `root:vehicle-map 0640`。 +以上三个接口均使用开放平台的 `Authorization: Bearer ` 鉴权。生产 AppKey 仅保存在 ECS 的 `/opt/lingniu-vehicle-map/env/vehicle-map.env`，文件权限为 `root:vehicle-map 0640`。同一配置文件中的 `AMAP_REGEOCODE_KEY` 仅在服务端使用：站点省、市、区按 GCJ-02 坐标调用高德逆地理生成，不能使用资产表的行政区字段；结果以坐标为键落盘缓存 7 天。详细地址仍直接保留资产原始字段。 本服务提供： - `GET /api/health`：进程及配置状态； - `GET /api/dashboard`：面向前端的公开聚合数据，默认缓存 2 分钟；加氢站目录同样缓存 2 分钟。服务启动后会后台预热缓存，避免首位访问者等待上游数据请求。 +- `GET /api/stations`：已按 GPS 解析行政区的公开站点目录，供独立加氢站导航服务复用；不含加氢量。 全国视图按车辆最新 GPS 坐标落入省级行政区，不按车牌归属地推断；没有有效实时坐标的车辆会单独计入“无实时位置”，不会伪造省份归属。页面每 15 秒请求一次，但服务端最多每 2 分钟刷新一次上游快照，以换取更快、更稳定的首屏加载。 diff --git a/vehicle-map/deploy/install-release.sh b/vehicle-map/deploy/install-release.sh index eaf4d015..cc0f8365 100644 --- a/vehicle-map/deploy/install-release.sh +++ b/vehicle-map/deploy/install-release.sh @@ -15,7 +15,7 @@ test -f "$archive" if ! id vehicle-map >/dev/null 2>&1; then useradd --system --home-dir "$root" --shell /sbin/nologin vehicle-map fi -mkdir -p "$root/releases" "$root/env" +mkdir -p "$root/releases" "$root/env" "$root/cache" next="$root/releases/$release_id" test ! -e "$next" mkdir -p "$next" @@ -32,6 +32,8 @@ chown -R root:vehicle-map "$next" chmod -R u=rwX,g=rX,o= "$next" chown root:vehicle-map "$root/env/vehicle-map.env" chmod 0640 "$root/env/vehicle-map.env" +chown vehicle-map:vehicle-map "$root/cache" +chmod 0750 "$root/cache" ln -s "$next" "$root/current.next" python3 -c 'import os,sys; os.replace(sys.argv[1],sys.argv[2])' "$root/current.next" "$root/current" diff --git a/vehicle-map/server.py b/vehicle-map/server.py index 5af61cd9..6d545162 100644 --- a/vehicle-map/server.py +++ b/vehicle-map/server.py @@ -10,7 +10,7 @@ import threading import time from socketserver import ThreadingMixIn from urllib.error import HTTPError, URLError -from urllib.parse import urlparse +from urllib.parse import urlencode, urlparse from urllib.request import Request, urlopen @@ -33,8 +33,17 @@ 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")) +AMAP_REGEOCODE_KEY = os.getenv("AMAP_REGEOCODE_KEY", "").strip() +STATION_GEOCODE_CACHE_SECONDS = int(os.getenv("STATION_GEOCODE_CACHE_SECONDS", "604800")) +STATION_GEOCODE_CACHE_PATH = Path(os.getenv( + "STATION_GEOCODE_CACHE_PATH", str(ROOT / ".station-geocode-cache.json") +)) _cache_lock = threading.Lock() _cache = {} +_cache_load_locks = {} +_station_geocode_cache_lock = threading.Lock() +_station_geocode_cache = None +_station_geocode_cache_dirty = False class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): @@ -79,20 +88,153 @@ def _cached(key, ttl_seconds, loader): 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 + load_lock = _cache_load_locks.setdefault(key, threading.Lock()) + # Startup prewarming and the first browser request can arrive together. + # Coalesce them so a cold geocode cache is filled only once. + with load_lock: + 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 _station_coordinate_key(station): + try: + return "{:.6f},{:.6f}".format(float(station.get("longitude")), float(station.get("latitude"))) + except (TypeError, ValueError): + return "" + + +def _component_value(value): + if isinstance(value, list): + return str(value[0]).strip() if value else "" + return str(value or "").strip() + + +def _load_station_geocode_cache(): + global _station_geocode_cache + with _station_geocode_cache_lock: + if _station_geocode_cache is not None: + return _station_geocode_cache + try: + with STATION_GEOCODE_CACHE_PATH.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + _station_geocode_cache = payload if isinstance(payload, dict) else {} + except (OSError, ValueError, TypeError): + _station_geocode_cache = {} + return _station_geocode_cache + + +def _persist_station_geocode_cache(): + try: + STATION_GEOCODE_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) + temporary = STATION_GEOCODE_CACHE_PATH.with_suffix(STATION_GEOCODE_CACHE_PATH.suffix + ".tmp") + with temporary.open("w", encoding="utf-8") as handle: + json.dump(_station_geocode_cache, handle, ensure_ascii=False, separators=(",", ":")) + os.replace(str(temporary), str(STATION_GEOCODE_CACHE_PATH)) + except OSError as exc: + print("station geocode cache write deferred: {}".format(exc)) + + +def _reverse_geocode_station(longitude, latitude): + if not AMAP_REGEOCODE_KEY: + raise RuntimeError("AMAP_REGEOCODE_KEY is not configured") + query = urlencode({ + "key": AMAP_REGEOCODE_KEY, + "location": "{:.6f},{:.6f}".format(longitude, latitude), + "extensions": "base", + "radius": "1000", + "batch": "false", + }) + request = Request( + "https://restapi.amap.com/v3/geocode/regeo?" + query, + headers={"Accept": "application/json", "User-Agent": "lingniu-station-geocoder/1.0"}, + ) + try: + with urlopen(request, timeout=UPSTREAM_TIMEOUT_SECONDS) as response: + payload = json.load(response) + except (HTTPError, URLError, ValueError) as exc: + raise RuntimeError("AMap reverse geocode unavailable: {}".format(exc)) from exc + if payload.get("status") != "1": + raise RuntimeError("AMap reverse geocode rejected request: {} {}".format(payload.get("infocode"), payload.get("info"))) + component = (payload.get("regeocode") or {}).get("addressComponent") or {} + province = _component_value(component.get("province")) + city = _component_value(component.get("city")) or province + district = _component_value(component.get("district")) + if not province: + raise RuntimeError("AMap reverse geocode returned no province") + return {"province": province, "city": city, "district": district, "adcode": _component_value(component.get("adcode"))} + + +def _station_region_from_gps(station, now=None): + global _station_geocode_cache_dirty + coordinate_key = _station_coordinate_key(station) + if not coordinate_key: + return {"province": "", "city": "", "district": ""} + now = time.time() if now is None else now + cache = _load_station_geocode_cache() + cached = cache.get(coordinate_key) or {} + if now - float(cached.get("updatedAt") or 0) < STATION_GEOCODE_CACHE_SECONDS: + return cached.get("region") or {"province": "", "city": "", "district": ""} + try: + longitude, latitude = (float(value) for value in coordinate_key.split(",")) + region = _reverse_geocode_station(longitude, latitude) + except Exception as exc: + print("station reverse geocode deferred for {}: {}".format(coordinate_key, exc)) + return cached.get("region") or {"province": "", "city": "", "district": ""} + with _station_geocode_cache_lock: + cache[coordinate_key] = {"updatedAt": now, "region": region} + _station_geocode_cache_dirty = True + return region + + +def _stations_with_gps_regions(stations): + global _station_geocode_cache_dirty + def enrich(station): + item = dict(station) + region = _station_region_from_gps(item) + # Never fall back to data-table administrative fields. The raw address + # remains untouched and is the only directly displayed location text. + item.update({key: region.get(key, "") for key in ("province", "city", "district")}) + return item + + with ThreadPoolExecutor(max_workers=8) as executor: + enriched = list(executor.map(enrich, stations)) + with _station_geocode_cache_lock: + if _station_geocode_cache_dirty: + _persist_station_geocode_cache() + _station_geocode_cache_dirty = False + return enriched def _load_stations(): return _cached( "stations", STATION_CACHE_SECONDS, - lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}), + lambda: _stations_with_gps_regions(_post_open_platform("/api/v1/hydrogen-stations/query", {})), ) +def _public_station_directory(): + stations = _load_stations() + fields = ("id", "name", "shortName", "province", "city", "district", "address", "longitude", "latitude", "cooperative") + directory = [{field: station.get(field) for field in fields if station.get(field) is not None} 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 _load_dashboard(): today = time.strftime("%Y-%m-%d", time.localtime()) with ThreadPoolExecutor(max_workers=3) as executor: @@ -188,6 +330,17 @@ class VehicleMapHandler(SimpleHTTPRequestHandler): "message": "地图数据暂时不可用，请稍后重试", }) return + if path == "/api/stations": + try: + self._write_json(200, _public_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 super().do_GET() def _serve_index(self): diff --git a/vehicle-map/tests/test_server.py b/vehicle-map/tests/test_server.py index 421eb558..ac8ba4ad 100644 --- a/vehicle-map/tests/test_server.py +++ b/vehicle-map/tests/test_server.py @@ -1,5 +1,6 @@ import importlib.util from pathlib import Path +import tempfile import unittest from unittest.mock import patch @@ -59,5 +60,30 @@ class DashboardTest(unittest.TestCase): self.assertEqual(calls, ["load"]) self.assertEqual(first, second) + def test_station_region_comes_from_gps_reverse_geocode_and_persists_by_coordinate(self): + station = { + "longitude": 113.200001, + "latitude": 23.100001, + "province": "错误省份", + "city": "错误城市", + "district": "错误区县", + } + with tempfile.TemporaryDirectory() as directory: + cache_path = Path(directory) / "station-geocode.json" + with patch.object(server, "STATION_GEOCODE_CACHE_PATH", cache_path), \ + patch.object(server, "_station_geocode_cache", None), \ + patch.object(server, "_station_geocode_cache_dirty", False), \ + patch.object(server, "_reverse_geocode_station", return_value={ + "province": "广东省", "city": "广州市", "district": "黄埔区", "adcode": "440112" + }) as reverse: + first = server._stations_with_gps_regions([station])[0] + second = server._stations_with_gps_regions([dict(station, province="仍然错误")])[0] + self.assertEqual(first["province"], "广东省") + self.assertEqual(first["city"], "广州市") + self.assertEqual(first["district"], "黄埔区") + self.assertEqual(second["province"], "广东省") + self.assertEqual(reverse.call_count, 1) + self.assertTrue(cache_path.exists()) + if __name__ == "__main__": unittest.main()