feat(map): derive station regions from GPS

This commit is contained in:
lingniu
2026-08-12 12:03:09 +08:00
parent f707e6ad68
commit 4c547757a7
8 changed files with 228 additions and 38 deletions
+2 -2
View File
@@ -2,7 +2,7 @@
面向运维和司机调度的独立公共站点导航服务,默认本地端口为 `20804` 面向运维和司机调度的独立公共站点导航服务,默认本地端口为 `20804`
只调用一次上游加氢站目录接口,并通过正向字段白名单对外提供:名称、位置、省市区、地址与合作状态。车辆数据、加氢量及其他运营字段不会被加载或从这个服务的 API 返回。 复用运营地图已经按 GPS 逆地理解析、落盘缓存的站点目录,并只对外提供:名称、位置、省市区、地址与合作状态。车辆数据、加氢量及其他运营字段不会被加载或从这个服务的 API 返回。
## 本地运行 ## 本地运行
@@ -19,7 +19,7 @@ export OPEN_PLATFORM_APP_KEY='<32位AppKey>'
- `GET /api/health`:进程健康状态; - `GET /api/health`:进程健康状态;
- 其他 `/api/*` 路由均返回 404。 - 其他 `/api/*` 路由均返回 404。
站点目录在服务端缓存 2 分钟,并在服务启动后后台预热;因此正常页面打开直接返回缓存数据,最多延迟 2 分钟更新一次。 站点目录在服务端缓存 2 分钟,并在服务启动后后台预热;因此正常页面打开直接返回缓存数据,最多延迟 2 分钟更新一次。省、市、区由运营地图根据 GCJ-02 坐标调用高德逆地理服务生成,结果按坐标缓存 7 天;详细地址仍直接展示资产原始字段。
## ECS 部署 ## ECS 部署
+18 -20
View File
@@ -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() OPEN_PLATFORM_APP_KEY = os.getenv("OPEN_PLATFORM_APP_KEY", "").strip()
UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15")) UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15"))
STATION_CACHE_SECONDS = int(os.getenv("STATION_NAVIGATION_CACHE_SECONDS", "120")) 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_lock = threading.Lock()
_cache = {} _cache = {}
@@ -91,27 +92,24 @@ def _cached(key, ttl_seconds, loader):
return 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(): def _load_station_directory():
stations = _post_open_platform("/api/v1/hydrogen-stations/query", {}) """Use the operations map's GPS-resolved public station directory.
directory = [_public_station(station) for station in stations]
return { The map service owns reverse-geocoding and its persistent cache, keeping
"status": "ok", the public navigation view and the operations view on identical boundaries.
"asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), """
"summary": { request = Request(
"totalStations": len(directory), VEHICLE_MAP_INTERNAL_BASE_URL + "/api/stations",
"cooperativeStations": sum(1 for station in directory if station.get("cooperative")), headers={"Accept": "application/json", "User-Agent": "lingniu-station-navigation/1.0"},
}, )
"stations": directory, 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(): def _prewarm_station_directory_cache():
+13 -8
View File
@@ -1,4 +1,6 @@
import importlib.util import importlib.util
import io
import json
from pathlib import Path from pathlib import Path
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
@@ -11,21 +13,24 @@ SPEC.loader.exec_module(server)
class StationNavigationTest(unittest.TestCase): class StationNavigationTest(unittest.TestCase):
def test_directory_is_a_positive_allowlist_without_hydrogen_data(self): 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": "广州市", "id": "GD-1", "name": "广州合作站", "province": "广东省", "city": "广州市",
"district": "黄埔区", "address": "开源大道1号", "longitude": 113.2, "latitude": 23.1, "district": "黄埔区", "address": "开源大道1号", "longitude": 113.2, "latitude": 23.1,
"cooperative": True, "monthlyHydrogenKg": 88.8, "totalHydrogenKg": 999.9, "cooperative": True,
"vehicleCount": 20, }],
}] }
with patch.object(server, "_post_open_platform", return_value=source): with patch.object(server, "urlopen", return_value=io.BytesIO(json.dumps(source).encode("utf-8"))):
result = server._load_station_directory() result = server._load_station_directory()
self.assertEqual(result["summary"], {"totalStations": 1, "cooperativeStations": 1}) self.assertEqual(result["summary"], {"totalStations": 1, "cooperativeStations": 1})
station = result["stations"][0] station = result["stations"][0]
self.assertEqual(station["name"], "广州合作站") self.assertEqual(station["name"], "广州合作站")
self.assertNotIn("monthlyHydrogenKg", station) self.assertEqual(station["province"], "广东省")
self.assertNotIn("totalHydrogenKg", station) self.assertEqual(server.VEHICLE_MAP_INTERNAL_BASE_URL, "http://127.0.0.1:20800")
self.assertNotIn("vehicleCount", station)
def test_static_assets_are_fingerprinted(self): def test_static_assets_are_fingerprinted(self):
self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$") self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$")
+5
View File
@@ -5,3 +5,8 @@ OPEN_PLATFORM_APP_KEY=replace-with-32-character-app-key
UPSTREAM_TIMEOUT_SECONDS=15 UPSTREAM_TIMEOUT_SECONDS=15
DASHBOARD_CACHE_SECONDS=5 DASHBOARD_CACHE_SECONDS=5
STATION_CACHE_SECONDS=3600 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
+2 -1
View File
@@ -32,12 +32,13 @@ python3 server.py
- `POST /api/v1/vehicles/mileage/query`:全部授权车辆当日里程; - `POST /api/v1/vehicles/mileage/query`:全部授权车辆当日里程;
- `POST /api/v1/hydrogen-stations/query`:资产库只读加氢站地图点位。 - `POST /api/v1/hydrogen-stations/query`:资产库只读加氢站地图点位。
以上三个接口均使用开放平台的 `Authorization: Bearer <AppKey>` 鉴权。生产 AppKey 仅保存在 ECS 的 `/opt/lingniu-vehicle-map/env/vehicle-map.env`,文件权限为 `root:vehicle-map 0640` 以上三个接口均使用开放平台的 `Authorization: Bearer <AppKey>` 鉴权。生产 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/health`:进程及配置状态;
- `GET /api/dashboard`:面向前端的公开聚合数据,默认缓存 2 分钟;加氢站目录同样缓存 2 分钟。服务启动后会后台预热缓存,避免首位访问者等待上游数据请求。 - `GET /api/dashboard`:面向前端的公开聚合数据,默认缓存 2 分钟;加氢站目录同样缓存 2 分钟。服务启动后会后台预热缓存,避免首位访问者等待上游数据请求。
- `GET /api/stations`:已按 GPS 解析行政区的公开站点目录,供独立加氢站导航服务复用;不含加氢量。
全国视图按车辆最新 GPS 坐标落入省级行政区,不按车牌归属地推断;没有有效实时坐标的车辆会单独计入“无实时位置”,不会伪造省份归属。页面每 15 秒请求一次,但服务端最多每 2 分钟刷新一次上游快照,以换取更快、更稳定的首屏加载。 全国视图按车辆最新 GPS 坐标落入省级行政区,不按车牌归属地推断;没有有效实时坐标的车辆会单独计入“无实时位置”,不会伪造省份归属。页面每 15 秒请求一次,但服务端最多每 2 分钟刷新一次上游快照,以换取更快、更稳定的首屏加载。
+3 -1
View File
@@ -15,7 +15,7 @@ test -f "$archive"
if ! id vehicle-map >/dev/null 2>&1; then if ! id vehicle-map >/dev/null 2>&1; then
useradd --system --home-dir "$root" --shell /sbin/nologin vehicle-map useradd --system --home-dir "$root" --shell /sbin/nologin vehicle-map
fi fi
mkdir -p "$root/releases" "$root/env" mkdir -p "$root/releases" "$root/env" "$root/cache"
next="$root/releases/$release_id" next="$root/releases/$release_id"
test ! -e "$next" test ! -e "$next"
mkdir -p "$next" mkdir -p "$next"
@@ -32,6 +32,8 @@ chown -R root:vehicle-map "$next"
chmod -R u=rwX,g=rX,o= "$next" chmod -R u=rwX,g=rX,o= "$next"
chown root:vehicle-map "$root/env/vehicle-map.env" chown root:vehicle-map "$root/env/vehicle-map.env"
chmod 0640 "$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" ln -s "$next" "$root/current.next"
python3 -c 'import os,sys; os.replace(sys.argv[1],sys.argv[2])' "$root/current.next" "$root/current" python3 -c 'import os,sys; os.replace(sys.argv[1],sys.argv[2])' "$root/current.next" "$root/current"
+155 -2
View File
@@ -10,7 +10,7 @@ import threading
import time import time
from socketserver import ThreadingMixIn from socketserver import ThreadingMixIn
from urllib.error import HTTPError, URLError from urllib.error import HTTPError, URLError
from urllib.parse import urlparse from urllib.parse import urlencode, urlparse
from urllib.request import Request, urlopen 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")) UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15"))
DASHBOARD_CACHE_SECONDS = int(os.getenv("DASHBOARD_CACHE_SECONDS", "120")) DASHBOARD_CACHE_SECONDS = int(os.getenv("DASHBOARD_CACHE_SECONDS", "120"))
STATION_CACHE_SECONDS = int(os.getenv("STATION_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_lock = threading.Lock()
_cache = {} _cache = {}
_cache_load_locks = {}
_station_geocode_cache_lock = threading.Lock()
_station_geocode_cache = None
_station_geocode_cache_dirty = False
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
@@ -74,6 +83,15 @@ def _post_open_platform(path, body):
def _cached(key, ttl_seconds, loader): 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]
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() now = time.time()
with _cache_lock: with _cache_lock:
cached = _cache.get(key) cached = _cache.get(key)
@@ -85,14 +103,138 @@ def _cached(key, ttl_seconds, loader):
return 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(): def _load_stations():
return _cached( return _cached(
"stations", "stations",
STATION_CACHE_SECONDS, 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(): def _load_dashboard():
today = time.strftime("%Y-%m-%d", time.localtime()) today = time.strftime("%Y-%m-%d", time.localtime())
with ThreadPoolExecutor(max_workers=3) as executor: with ThreadPoolExecutor(max_workers=3) as executor:
@@ -188,6 +330,17 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
"message": "地图数据暂时不可用,请稍后重试", "message": "地图数据暂时不可用,请稍后重试",
}) })
return 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() super().do_GET()
def _serve_index(self): def _serve_index(self):
+26
View File
@@ -1,5 +1,6 @@
import importlib.util import importlib.util
from pathlib import Path from pathlib import Path
import tempfile
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
@@ -59,5 +60,30 @@ class DashboardTest(unittest.TestCase):
self.assertEqual(calls, ["load"]) self.assertEqual(calls, ["load"])
self.assertEqual(first, second) 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__": if __name__ == "__main__":
unittest.main() unittest.main()