Reject stale Open-Meteo forecast cache

src/data_collection/forecast_source_bundle.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 from typing import Any, Dict
 
-from src.data_collection.multi_model_freshness import multi_model_has_current_window
+from src.data_collection.multi_model_freshness import (
+    multi_model_has_current_window,
+    open_meteo_forecast_has_current_window,
+)
 
 
 def _open_meteo_cache_key(
@@ -48,8 +51,9 @@ def _read_open_meteo_bundle_from_cache(
     with collector._open_meteo_cache_lock:
         om_cached = collector._open_meteo_cache.get(om_key)
 
-    if om_cached and isinstance(om_cached.get("data"), dict):
-        results["open-meteo"] = dict(om_cached["data"])
+    om_data = om_cached.get("data") if isinstance(om_cached, dict) else None
+    if isinstance(om_data, dict) and open_meteo_forecast_has_current_window(om_data):
+        results["open-meteo"] = dict(om_data)
     if include_multi_model:
         mm_key = _multi_model_cache_key(
             collector,

src/data_collection/multi_model_freshness.py

@@ -63,6 +63,26 @@ def multi_model_has_current_window(multi_model: Any, *, today: Optional[date] =
     return max(parsed_dates) >= today
 
 
+def open_meteo_forecast_has_current_window(
+    forecast: Any,
+    *,
+    today: Optional[date] = None,
+) -> bool:
+    """Return False when cached Open-Meteo forecast data is definitely older than today."""
+    if not isinstance(forecast, dict):
+        return False
+    today = today or datetime.now(timezone.utc).date()
+    raw_date_values = []
+    daily = forecast.get("daily") if isinstance(forecast.get("daily"), dict) else {}
+    raw_date_values.extend(daily.get("time") or [])
+    hourly = forecast.get("hourly") if isinstance(forecast.get("hourly"), dict) else {}
+    raw_date_values.extend(hourly.get("time") or [])
+    parsed_dates = [parsed for raw in raw_date_values for parsed in [_parse_date(raw)] if parsed]
+    if not parsed_dates:
+        return True
+    return max(parsed_dates) >= today
+
+
 def multi_model_covers_local_date(multi_model: Any, local_date: str) -> bool:
     if not isinstance(multi_model, dict):
         return False

src/data_collection/nws_open_meteo_sources.py

@@ -6,7 +6,10 @@ from typing import Any, Dict, Optional
 
 from loguru import logger
 
-from src.data_collection.multi_model_freshness import multi_model_has_current_window
+from src.data_collection.multi_model_freshness import (
+    multi_model_has_current_window,
+    open_meteo_forecast_has_current_window,
+)
 from src.utils.metrics import record_source_call
 
 
@@ -528,16 +531,18 @@ class NwsOpenMeteoSourceMixin:
                 logger.debug(f"Open-Meteo 冷却期中，跳过请求，还需 {remaining}s")
                 with self._open_meteo_cache_lock:
                         stale = self._open_meteo_cache.get(cache_key)
-                        if stale and isinstance(stale.get("data"), dict):
+                        stale_data = stale.get("data") if isinstance(stale, dict) else None
+                        if isinstance(stale_data, dict) and open_meteo_forecast_has_current_window(stale_data):
                             record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0)
-                            return dict(stale["data"])
+                            return dict(stale_data)
                 # Memory miss: force-reload from disk and retry once
                 self._load_open_meteo_disk_cache()
                 with self._open_meteo_cache_lock:
                     stale2 = self._open_meteo_cache.get(cache_key)
-                    if stale2 and isinstance(stale2.get("data"), dict):
+                    stale2_data = stale2.get("data") if isinstance(stale2, dict) else None
+                    if isinstance(stale2_data, dict) and open_meteo_forecast_has_current_window(stale2_data):
                         record_source_call("open_meteo", "forecast", "disk_fallback", (time.perf_counter() - started) * 1000.0)
-                        return dict(stale2["data"])
+                        return dict(stale2_data)
                 record_source_call("open_meteo", "forecast", "cooldown_skip", (time.perf_counter() - started) * 1000.0)
                 return None
         with self._open_meteo_cache_lock:
@@ -547,6 +552,11 @@ class NwsOpenMeteoSourceMixin:
                 and now_ts - float(cached.get("t", 0)) < self.open_meteo_cache_ttl_sec
             ):
                 cached_data = cached.get("data")
+                if isinstance(cached_data, dict):
+                    if not open_meteo_forecast_has_current_window(cached_data):
+                        self._open_meteo_cache.pop(cache_key, None)
+                        record_source_call("open_meteo", "forecast", "expired_cache_skip", (time.perf_counter() - started) * 1000.0)
+                        cached_data = None
                 if isinstance(cached_data, dict):
                     record_source_call("open_meteo", "forecast", "cache_hit", (time.perf_counter() - started) * 1000.0)
                     return dict(cached_data)
@@ -672,8 +682,9 @@ class NwsOpenMeteoSourceMixin:
                 logger.error(f"Open-Meteo forecast failed: {e}")
             with self._open_meteo_cache_lock:
                 stale = self._open_meteo_cache.get(cache_key)
-                if stale and isinstance(stale.get("data"), dict):
-                    fallback = dict(stale["data"])
+                stale_data = stale.get("data") if isinstance(stale, dict) else None
+                if isinstance(stale_data, dict) and open_meteo_forecast_has_current_window(stale_data):
+                    fallback = dict(stale_data)
                     fallback["stale_cache"] = True
                     record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0)
                     return fallback