From 535b2b0446dc57279dd2a884a7aa596e904bec9e Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 11 Mar 2026 04:39:20 +0800 Subject: [PATCH] feat: add disk-based cache persistence for Open-Meteo to eliminate cold-start empty cache gaps --- src/data_collection/weather_sources.py | 95 ++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 6ade1c9d..7ef6e0c6 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -96,6 +96,13 @@ class WeatherDataCollector: self._metar_cache: Dict[str, Dict] = {} self._metar_cache_lock = threading.Lock() + # 磁盘持久化缓存：重启后即可加载上次的预报数据，避免冷启动请求爆发 + self._disk_cache_path = os.getenv( + "OPEN_METEO_DISK_CACHE_PATH", "/app/.cache/open_meteo_cache.json" + ) + self._disk_cache_lock = threading.Lock() + self._load_open_meteo_disk_cache() + # 设置代理 proxy = config.get("proxy") if proxy: @@ -106,6 +113,65 @@ class WeatherDataCollector: logger.info("天气数据采集器初始化完成。") + def _load_open_meteo_disk_cache(self) -> None: + """启动时从磁盘加载 Open-Meteo 三类缓存，避免重启后冷启动打爆 API""" + import json as _json + try: + if not os.path.exists(self._disk_cache_path): + return + with open(self._disk_cache_path, "r", encoding="utf-8") as f: + saved = _json.load(f) + now = time.time() + # 最多允许加载 2× TTL 的旧数据（TTL 内视为新鲜，超出则过期但仍可作为 stale fallback） + max_age = max( + self.open_meteo_cache_ttl_sec, + self.open_meteo_ensemble_cache_ttl_sec, + self.open_meteo_multi_model_cache_ttl_sec, + ) * 2 + loaded = 0 + for key, entry in saved.get("forecast", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + self._open_meteo_cache[key] = entry + loaded += 1 + for key, entry in saved.get("ensemble", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + self._ensemble_cache[key] = entry + loaded += 1 + for key, entry in saved.get("multi_model", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + self._multi_model_cache[key] = entry + loaded += 1 + if loaded: + logger.info(f"✅ 从磁盘加载 Open-Meteo 缓存 {loaded} 条 ({self._disk_cache_path})") + except Exception as e: + logger.warning(f"磁盘缓存加载失败（首次启动不影响运行）: {e}") + + def _flush_open_meteo_disk_cache(self) -> None: + """将三类 Open-Meteo 内存缓存持久化到磁盘""" + import json as _json + try: + os.makedirs(os.path.dirname(self._disk_cache_path), exist_ok=True) + with self._open_meteo_cache_lock: + forecast_snapshot = dict(self._open_meteo_cache) + with self._ensemble_cache_lock: + ensemble_snapshot = dict(self._ensemble_cache) + with self._multi_model_cache_lock: + multi_model_snapshot = dict(self._multi_model_cache) + payload = { + "forecast": forecast_snapshot, + "ensemble": ensemble_snapshot, + "multi_model": multi_model_snapshot, + "saved_at": time.time(), + } + with self._disk_cache_lock: + tmp_path = self._disk_cache_path + ".tmp" + with open(tmp_path, "w", encoding="utf-8") as f: + _json.dump(payload, f) + os.replace(tmp_path, self._disk_cache_path) # 原子替换，防止写入一半时被读到 + except Exception as e: + logger.warning(f"磁盘缓存写入失败: {e}") + + def fetch_from_openweather(self, city: str, country: str = None) -> Optional[Dict]: """ Fetch current weather and forecast from OpenWeatherMap @@ -1083,6 +1149,7 @@ class WeatherDataCollector: "t": time.time(), "data": dict(result), } + self._flush_open_meteo_disk_cache() return result except Exception as e: status_code = getattr(getattr(e, "response", None), "status_code", None) @@ -1119,6 +1186,17 @@ class WeatherDataCollector: f"{'f' if use_fahrenheit else 'c'}" ) now_ts = time.time() + # ── 429 冷却期检查（所有 Open-Meteo 端点共享）───────────────── + with self._open_meteo_rl_lock: + if now_ts < self._open_meteo_rate_limit_until: + remaining = int(self._open_meteo_rate_limit_until - now_ts) + logger.debug(f"Open-Meteo Ensemble 冷却期中，跳过请求，还需 {remaining}s") + with self._ensemble_cache_lock: + stale = self._ensemble_cache.get(cache_key) + if stale and isinstance(stale.get("data"), dict): + return dict(stale["data"]) + return None + with self._ensemble_cache_lock: cached = self._ensemble_cache.get(cache_key) if ( @@ -1200,6 +1278,7 @@ class WeatherDataCollector: "t": time.time(), "data": dict(result), } + self._flush_open_meteo_disk_cache() return result except Exception as e: status_code = getattr(getattr(e, "response", None), "status_code", None) @@ -1207,6 +1286,8 @@ class WeatherDataCollector: logger.warning( f"Ensemble API rate limited (429), fallback to cache if available: lat={lat}, lon={lon}" ) + with self._open_meteo_rl_lock: + self._open_meteo_rate_limit_until = time.time() + self._open_meteo_rl_cooldown else: logger.warning(f"Ensemble API 请求失败: {e}") with self._ensemble_cache_lock: @@ -1241,6 +1322,17 @@ class WeatherDataCollector: f"{'f' if use_fahrenheit else 'c'}" ) now_ts = time.time() + # ── 429 冷却期检查（所有 Open-Meteo 端点共享）───────────────── + with self._open_meteo_rl_lock: + if now_ts < self._open_meteo_rate_limit_until: + remaining = int(self._open_meteo_rate_limit_until - now_ts) + logger.debug(f"Open-Meteo Multi-model 冷却期中，跳过请求，还需 {remaining}s") + with self._multi_model_cache_lock: + stale = self._multi_model_cache.get(cache_key) + if stale and isinstance(stale.get("data"), dict): + return dict(stale["data"]) + return None + with self._multi_model_cache_lock: cached = self._multi_model_cache.get(cache_key) if ( @@ -1321,6 +1413,7 @@ class WeatherDataCollector: "t": time.time(), "data": dict(result), } + self._flush_open_meteo_disk_cache() return result except Exception as e: status_code = getattr(getattr(e, "response", None), "status_code", None) @@ -1328,6 +1421,8 @@ class WeatherDataCollector: logger.warning( f"Multi-model API rate limited (429), fallback to cache if available: lat={lat}, lon={lon}" ) + with self._open_meteo_rl_lock: + self._open_meteo_rate_limit_until = time.time() + self._open_meteo_rl_cooldown else: logger.warning(f"Multi-model API 请求失败: {e}") with self._multi_model_cache_lock: