功能：完善车辆地图合作站信息与缓存刷新

vehicle-map/server.py

@@ -42,11 +42,15 @@ AMAP_REGEOCODE_MIN_INTERVAL_SECONDS = float(os.getenv("AMAP_REGEOCODE_MIN_INTERV
 _cache_lock = threading.Lock()
 _cache = {}
 _cache_load_locks = {}
+_cache_refreshing = set()
 _station_geocode_cache_lock = threading.Lock()
 _station_geocode_cache = None
 _station_geocode_cache_dirty = False
 _station_geocode_rate_lock = threading.Lock()
 _station_geocode_next_request_at = 0.0
+_station_geocode_refresh_lock = threading.Lock()
+_station_geocode_refresh_pending = {}
+_station_geocode_refresh_running = False
 
 
 class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
@@ -85,24 +89,54 @@ def _post_open_platform(path, body):
     return payload.get("data") or []
 
 
+def _refresh_cached_value(key, loader):
+    try:
+        value = loader()
+    except Exception as exc:
+        print("{} cache background refresh deferred: {}".format(key, exc))
+    else:
+        with _cache_lock:
+            _cache[key] = (time.time(), value)
+    finally:
+        with _cache_lock:
+            _cache_refreshing.discard(key)
+
+
+def _start_cache_refresh(key, loader):
+    thread = threading.Thread(target=_refresh_cached_value, args=(key, loader), daemon=True)
+    thread.start()
+
+
 def _cached(key, ttl_seconds, loader):
     now = time.time()
+    refresh_in_background = False
     with _cache_lock:
         cached = _cache.get(key)
         if cached and now - cached[0] < ttl_seconds:
             return cached[1]
+        if cached:
+            if key not in _cache_refreshing:
+                _cache_refreshing.add(key)
+                refresh_in_background = True
+            stale_value = cached[1]
+        else:
+            stale_value = None
         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.
+    if cached:
+        if refresh_in_background:
+            _start_cache_refresh(key, loader)
+        return stale_value
+    # Cold starts still need one synchronous load. Coalesce startup prewarming
+    # and the first browser request so only one upstream request is issued.
     with load_lock:
         now = time.time()
         with _cache_lock:
             cached = _cache.get(key)
-            if cached and now - cached[0] < ttl_seconds:
-                return cached[1]
+        if cached:
+            return _cached(key, ttl_seconds, loader)
         value = loader()
         with _cache_lock:
-            _cache[key] = (now, value)
+            _cache[key] = (time.time(), value)
         return value
 
 
@@ -181,7 +215,7 @@ def _reverse_geocode_station(longitude, latitude):
     return {"province": province, "city": city, "district": district, "adcode": _component_value(component.get("adcode"))}
 
 
-def _station_region_from_gps(station, now=None):
+def _station_region_from_gps(station, now=None, refresh_queue=None):
     global _station_geocode_cache_dirty
     coordinate_key = _station_coordinate_key(station)
     if not coordinate_key:
@@ -191,6 +225,11 @@ def _station_region_from_gps(station, now=None):
     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": ""}
+    cached_region = cached.get("region")
+    if isinstance(cached_region, dict):
+        if refresh_queue is not None:
+            refresh_queue.append(station)
+        return cached_region
     try:
         longitude, latitude = (float(value) for value in coordinate_key.split(","))
         region = _reverse_geocode_station(longitude, latitude)
@@ -203,22 +242,92 @@ def _station_region_from_gps(station, now=None):
     return region
 
 
+def _refresh_station_region(station):
+    coordinate_key = _station_coordinate_key(station)
+    if not coordinate_key:
+        return False
+    now = time.time()
+    cache = _load_station_geocode_cache()
+    with _station_geocode_cache_lock:
+        cached = cache.get(coordinate_key) or {}
+        if now - float(cached.get("updatedAt") or 0) < STATION_GEOCODE_CACHE_SECONDS:
+            return False
+    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 refresh deferred for {}: {}".format(coordinate_key, exc))
+        return False
+    with _station_geocode_cache_lock:
+        cache[coordinate_key] = {"updatedAt": now, "region": region}
+    return True
+
+
+def _refresh_station_regions(stations):
+    with ThreadPoolExecutor(max_workers=8) as executor:
+        changed = any(list(executor.map(_refresh_station_region, stations)))
+    if changed:
+        with _station_geocode_cache_lock:
+            _persist_station_geocode_cache()
+
+
+def _drain_station_region_refreshes():
+    global _station_geocode_refresh_running
+    restart = False
+    try:
+        while True:
+            with _station_geocode_refresh_lock:
+                batch = list(_station_geocode_refresh_pending.values())
+                _station_geocode_refresh_pending.clear()
+            if not batch:
+                return
+            _refresh_station_regions(batch)
+    except Exception as exc:
+        print("station reverse geocode background refresh deferred: {}".format(exc))
+    finally:
+        with _station_geocode_refresh_lock:
+            _station_geocode_refresh_running = False
+            restart = bool(_station_geocode_refresh_pending)
+        if restart:
+            _schedule_station_region_refresh([])
+
+
+def _schedule_station_region_refresh(stations):
+    global _station_geocode_refresh_running
+    start_worker = False
+    with _station_geocode_refresh_lock:
+        for station in stations:
+            coordinate_key = _station_coordinate_key(station)
+            if coordinate_key:
+                _station_geocode_refresh_pending[coordinate_key] = station
+        if _station_geocode_refresh_pending and not _station_geocode_refresh_running:
+            _station_geocode_refresh_running = True
+            start_worker = True
+    if start_worker:
+        threading.Thread(target=_drain_station_region_refreshes, daemon=True).start()
+
+
 def _stations_with_gps_regions(stations):
     global _station_geocode_cache_dirty
     def enrich(station):
         item = dict(station)
-        region = _station_region_from_gps(item)
+        refresh_queue = []
+        region = _station_region_from_gps(item, refresh_queue=refresh_queue)
         # 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
+        return item, refresh_queue[0] if refresh_queue else None
 
     with ThreadPoolExecutor(max_workers=8) as executor:
-        enriched = list(executor.map(enrich, stations))
+        results = list(executor.map(enrich, stations))
+    enriched = [item for item, _ in results]
+    stale_stations = [station for _, station in results if station is not None]
     with _station_geocode_cache_lock:
         if _station_geocode_cache_dirty:
             _persist_station_geocode_cache()
             _station_geocode_cache_dirty = False
+    if stale_stations:
+        _schedule_station_region_refresh(stale_stations)
     return enriched
 
 
@@ -233,7 +342,13 @@ def _load_stations():
 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]
+    partner_fields = ("contactPerson", "contactPhone", "unitPrice")
+    directory = []
+    for station in stations:
+        item = {field: station.get(field) for field in fields if station.get(field) is not None}
+        if item.get("cooperative"):
+            item.update({field: station.get(field) for field in partner_fields if station.get(field) not in (None, "", 0)})
+        directory.append(item)
     return {
         "status": "ok",
         "asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
@@ -331,7 +446,6 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
         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, {
@@ -339,10 +453,12 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
                     "code": "UPSTREAM_UNAVAILABLE",
                     "message": "地图数据暂时不可用，请稍后重试",
                 })
+                return
+            self._write_json(200, payload)
             return
         if path == "/api/stations":
             try:
-                self._write_json(200, _public_station_directory())
+                payload = _public_station_directory()
             except Exception as exc:
                 self.log_error("station directory refresh failed: %s", exc)
                 self._write_json(502, {
@@ -350,6 +466,8 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
                     "code": "UPSTREAM_UNAVAILABLE",
                     "message": "加氢站数据暂时不可用，请稍后重试",
                 })
+                return
+            self._write_json(200, payload)
             return
         super().do_GET()
 
@@ -385,7 +503,10 @@ class VehicleMapHandler(SimpleHTTPRequestHandler):
         for name, value in (headers or {}).items():
             self.send_header(name, value)
         self.end_headers()
-        self.wfile.write(body)
+        try:
+            self.wfile.write(body)
+        except (BrokenPipeError, ConnectionResetError):
+            self.log_message("client disconnected before JSON response completed")
 
 
 if __name__ == "__main__":