From fd59cd018da64287f061f6fb01836e66357b2918 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 18 May 2026 00:13:51 +0800 Subject: [PATCH] Harden Open-Meteo rate limiting --- .env.example | 21 +++--- src/data_collection/nws_open_meteo_sources.py | 21 +++--- src/data_collection/open_meteo_cache.py | 25 +++++++ src/data_collection/weather_sources.py | 6 +- src/database/runtime_state.py | 47 +++++++++++++ tests/test_multi_model_sources.py | 66 +++++++++++++++++++ tests/test_runtime_state_storage.py | 11 ++++ web/app.py | 4 +- 8 files changed, 178 insertions(+), 23 deletions(-) diff --git a/.env.example b/.env.example index 18c43884..8258eadf 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,7 @@ POLYWEATHER_MAP_URL=https://polyweather-pro.vercel.app/ POLYWEATHER_RUNTIME_DATA_DIR=/var/lib/polyweather POLYWEATHER_DB_PATH=/var/lib/polyweather/polyweather.db OPEN_METEO_DISK_CACHE_PATH=/var/lib/polyweather/open_meteo_cache.json +UVICORN_WORKERS=1 # Optional: host user/group mapping for Docker on Linux. # Windows / macOS can usually keep the defaults. UID=1000 @@ -40,13 +41,16 @@ TELEGRAM_AIRPORT_PUSH_MAX_WORKERS=1 ######################################## # 3) Weather + cache ######################################## -OPEN_METEO_CACHE_TTL_SEC=7200 -OPEN_METEO_ENSEMBLE_CACHE_TTL_SEC=7200 -OPEN_METEO_MULTI_MODEL_CACHE_TTL_SEC=7200 +OPEN_METEO_CACHE_TTL_SEC=21600 +OPEN_METEO_ENSEMBLE_CACHE_TTL_SEC=21600 +OPEN_METEO_MULTI_MODEL_CACHE_TTL_SEC=21600 OPEN_METEO_MULTI_MODEL_CACHE_VERSION=v2 -OPEN_METEO_RATE_LIMIT_COOLDOWN_SEC=900 +OPEN_METEO_RATE_LIMIT_COOLDOWN_SEC=3600 OPEN_METEO_RATE_CACHE_TTL_SEC=3600 -OPEN_METEO_MIN_CALL_INTERVAL_SEC=1 +OPEN_METEO_MIN_CALL_INTERVAL_SEC=5 +POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS=1 +POLYWEATHER_SCAN_TERMINAL_PAYLOAD_TTL_SEC=600 +POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC=45 POLYWEATHER_HTTP_TIMEOUT_SEC=8 POLYWEATHER_HTTP_RETRY_COUNT=0 POLYWEATHER_HTTP_RETRY_BACKOFF_SEC=0.2 @@ -113,13 +117,6 @@ NEXT_PUBLIC_POLYWEATHER_API_BASE_URL= # 7) Optional modules ######################################## -# Optional Groq commentary rewrite for intraday structure cards -POLYWEATHER_GROQ_COMMENTARY_ENABLED=false -GROQ_API_KEY= -POLYWEATHER_GROQ_COMMENTARY_MODEL=openai/gpt-oss-20b -POLYWEATHER_GROQ_COMMENTARY_TIMEOUT_SEC=8 -POLYWEATHER_GROQ_COMMENTARY_CACHE_TTL_SEC=1800 - # Optional OpenAI-compatible market scan review for Pro users # Temporary default provider: MiMo via https://token-plan-cn.xiaomimimo.com/v1. POLYWEATHER_SCAN_AI_ENABLED=false diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index 30c59c64..957970bc 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -351,8 +351,9 @@ class NwsOpenMeteoSourceMixin: 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) + cooldown_until = self._refresh_open_meteo_rate_limit_until() + if now_ts < cooldown_until: + remaining = int(cooldown_until - now_ts) logger.debug(f"Open-Meteo 冷却期中，跳过请求，还需 {remaining}s") with self._open_meteo_cache_lock: stale = self._open_meteo_cache.get(cache_key) @@ -494,7 +495,7 @@ class NwsOpenMeteoSourceMixin: ) # 设置全局冷却期，避免短时内重复触发 429 with self._open_meteo_rl_lock: - self._open_meteo_rate_limit_until = time.time() + cooldown_to_use + self._set_open_meteo_rate_limit_until(time.time() + cooldown_to_use, reason="forecast_429") logger.warning(f"Open-Meteo 触发限流，设置 {cooldown_to_use}s 冷却期") else: logger.error(f"Open-Meteo forecast failed: {e}") @@ -527,8 +528,9 @@ class NwsOpenMeteoSourceMixin: 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) + cooldown_until = self._refresh_open_meteo_rate_limit_until() + if now_ts < cooldown_until: + remaining = int(cooldown_until - now_ts) logger.debug(f"Open-Meteo Ensemble 冷却期中，跳过请求，还需 {remaining}s") with self._ensemble_cache_lock: stale = self._ensemble_cache.get(cache_key) @@ -647,7 +649,7 @@ class NwsOpenMeteoSourceMixin: 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() + cooldown_to_use + self._set_open_meteo_rate_limit_until(time.time() + cooldown_to_use, reason="ensemble_429") else: logger.warning(f"Ensemble API 请求失败: {e}") with self._ensemble_cache_lock: @@ -690,8 +692,9 @@ class NwsOpenMeteoSourceMixin: 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) + cooldown_until = self._refresh_open_meteo_rate_limit_until() + if now_ts < cooldown_until: + remaining = int(cooldown_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) @@ -808,7 +811,7 @@ class NwsOpenMeteoSourceMixin: 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() + cooldown_to_use + self._set_open_meteo_rate_limit_until(time.time() + cooldown_to_use, reason="multi_model_429") else: logger.warning(f"Multi-model API 请求失败: {e}") with self._multi_model_cache_lock: diff --git a/src/data_collection/open_meteo_cache.py b/src/data_collection/open_meteo_cache.py index 643e2ee5..5330d2ad 100644 --- a/src/data_collection/open_meteo_cache.py +++ b/src/data_collection/open_meteo_cache.py @@ -7,11 +7,13 @@ import time from loguru import logger from src.database.runtime_state import ( OpenMeteoCacheRepository, + OpenMeteoRateLimitRepository, STATE_STORAGE_SQLITE, get_state_storage_mode, ) _open_meteo_cache_repo = OpenMeteoCacheRepository() +_open_meteo_rate_limit_repo = OpenMeteoRateLimitRepository() class OpenMeteoCacheMixin: @@ -182,3 +184,26 @@ class OpenMeteoCacheMixin: time.sleep(wait_for) now_ts = time.time() self._open_meteo_last_call_ts = now_ts + + def _refresh_open_meteo_rate_limit_until(self) -> float: + """Load the cross-process Open-Meteo cooldown marker from runtime state.""" + if get_state_storage_mode() != STATE_STORAGE_SQLITE: + return getattr(self, "_open_meteo_rate_limit_until", 0.0) + try: + persisted_until = _open_meteo_rate_limit_repo.load_until() + except Exception as exc: + logger.debug(f"Open-Meteo cooldown load skipped: {exc}") + return getattr(self, "_open_meteo_rate_limit_until", 0.0) + if persisted_until > getattr(self, "_open_meteo_rate_limit_until", 0.0): + self._open_meteo_rate_limit_until = persisted_until + return self._open_meteo_rate_limit_until + + def _set_open_meteo_rate_limit_until(self, rate_limit_until: float, *, reason: str = "") -> None: + """Persist the shared Open-Meteo cooldown marker for all workers.""" + self._open_meteo_rate_limit_until = float(rate_limit_until) + if get_state_storage_mode() != STATE_STORAGE_SQLITE: + return + try: + _open_meteo_rate_limit_repo.set_until(rate_limit_until, reason=reason) + except Exception as exc: + logger.debug(f"Open-Meteo cooldown persist skipped: {exc}") diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 722bc5bd..218979c5 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -1316,7 +1316,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour lat=lat, lon=lon, use_fahrenheit=use_fahrenheit, - keep_model_caches=force_refresh_observations_only, + # Force-refresh is usually a UI/API freshness request for live + # observations. Do not evict longer-lived Open-Meteo forecast + # caches by default: one accidental all-city refresh can + # otherwise cold-start the VPS into Open-Meteo 429s. + keep_model_caches=True, ) self._log_temperature_unit(city, use_fahrenheit) self._attach_settlement_sources(results, city_lower) diff --git a/src/database/runtime_state.py b/src/database/runtime_state.py index a7109374..7b9fa64a 100644 --- a/src/database/runtime_state.py +++ b/src/database/runtime_state.py @@ -193,6 +193,16 @@ class RuntimeStateDB: conn.execute( "CREATE INDEX IF NOT EXISTS idx_open_meteo_cache_expires ON open_meteo_cache_store(source_kind, expires_at)" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS open_meteo_rate_limit_state ( + state_key TEXT PRIMARY KEY, + rate_limit_until REAL NOT NULL, + reason TEXT, + updated_at REAL NOT NULL + ) + """ + ) conn.execute( """ CREATE TABLE IF NOT EXISTS official_intraday_observations_store ( @@ -1001,6 +1011,43 @@ class OpenMeteoCacheRepository: return 0.0 +class OpenMeteoRateLimitRepository: + def __init__(self, db: Optional[RuntimeStateDB] = None): + self.db = db or RuntimeStateDB.instance() + + def set_until(self, rate_limit_until: float, *, reason: str = "") -> None: + with self.db.connect() as conn: + conn.execute( + """ + INSERT INTO open_meteo_rate_limit_state ( + state_key, rate_limit_until, reason, updated_at + ) VALUES ('global', ?, ?, ?) + ON CONFLICT(state_key) DO UPDATE SET + rate_limit_until = excluded.rate_limit_until, + reason = excluded.reason, + updated_at = excluded.updated_at + """, + (float(rate_limit_until), str(reason or ""), time.time()), + ) + conn.commit() + + def load_until(self) -> float: + with self.db.connect() as conn: + row = conn.execute( + """ + SELECT rate_limit_until + FROM open_meteo_rate_limit_state + WHERE state_key = 'global' + """ + ).fetchone() + if not row: + return 0.0 + try: + return float(row["rate_limit_until"] or 0.0) + except Exception: + return 0.0 + + class OfficialIntradayObservationRepository: def __init__(self, db: Optional[RuntimeStateDB] = None): self.db = db or RuntimeStateDB.instance() diff --git a/tests/test_multi_model_sources.py b/tests/test_multi_model_sources.py index 5445b16f..200f744c 100644 --- a/tests/test_multi_model_sources.py +++ b/tests/test_multi_model_sources.py @@ -2,7 +2,9 @@ from src.data_collection.nws_open_meteo_sources import ( OPEN_METEO_MULTI_MODEL_ORDER, _parse_open_meteo_multi_model_daily, ) +import src.data_collection.open_meteo_cache as open_meteo_cache_module from src.data_collection.weather_sources import WeatherDataCollector +from src.database.runtime_state import OpenMeteoRateLimitRepository, RuntimeStateDB def test_multi_model_parser_exposes_open_recommended_models(): @@ -94,3 +96,67 @@ def test_fetch_all_sources_prioritizes_multi_model_before_forecast(monkeypatch, assert calls[:2] == ["multi_model", "open_meteo"] assert result["multi_model"]["forecasts"]["ECMWF"] == 24.0 + + +def test_force_refresh_preserves_open_meteo_model_caches_by_default(monkeypatch, tmp_path): + monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) + collector = WeatherDataCollector({}) + captured = {} + + def fake_evict(*args, **kwargs): + captured["keep_model_caches"] = kwargs.get("keep_model_caches") + + monkeypatch.setattr(collector, "_evict_city_caches", fake_evict) + monkeypatch.setattr(collector, "_log_temperature_unit", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_settlement_sources", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_supports_aviationweather", lambda city: False) + monkeypatch.setattr(collector, "fetch_multi_model", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "fetch_from_open_meteo", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_nws_and_models", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_turkish_mgm_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_korean_amos_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_china_amsc_awos_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_madis_hfmetar_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_singapore_mss_data", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_china_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_japan_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_fmi_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_knmi_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_hko_obs_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_cwa_settlement_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_russia_official_nearby", lambda *args, **kwargs: None) + monkeypatch.setattr(collector, "_attach_global_nearby_cluster", lambda *args, **kwargs: None) + + collector.fetch_all_sources( + "ankara", + lat=40.1281, + lon=32.9951, + force_refresh=True, + include_ensemble=False, + include_nearby=False, + include_taf=False, + include_mgm=False, + ) + + assert captured["keep_model_caches"] is True + + +def test_persisted_open_meteo_cooldown_skips_outbound_request(monkeypatch, tmp_path): + monkeypatch.setenv("POLYWEATHER_STATE_STORAGE_MODE", "sqlite") + monkeypatch.setenv("POLYWEATHER_DB_PATH", str(tmp_path / "polyweather.db")) + monkeypatch.setenv("OPEN_METEO_DISK_CACHE_PATH", str(tmp_path / "om-cache.json")) + db = RuntimeStateDB(str(tmp_path / "polyweather.db")) + rate_repo = OpenMeteoRateLimitRepository(db) + rate_repo.set_until(9999999999.0, reason="test_429") + monkeypatch.setattr(open_meteo_cache_module, "_open_meteo_rate_limit_repo", rate_repo) + + collector = WeatherDataCollector({}) + + def fail_http(*args, **kwargs): + raise AssertionError("Open-Meteo HTTP request should be skipped during persisted cooldown") + + monkeypatch.setattr(collector, "_http_get", fail_http) + + result = collector.fetch_multi_model(40.1281, 32.9951, city="ankara") + + assert result is None diff --git a/tests/test_runtime_state_storage.py b/tests/test_runtime_state_storage.py index 16b13ee6..343fb81b 100644 --- a/tests/test_runtime_state_storage.py +++ b/tests/test_runtime_state_storage.py @@ -4,6 +4,7 @@ import time from src.database.runtime_state import ( DailyRecordRepository, OpenMeteoCacheRepository, + OpenMeteoRateLimitRepository, ProbabilitySnapshotRepository, RuntimeStateDB, TelegramAlertStateRepository, @@ -77,6 +78,16 @@ def test_open_meteo_cache_repository_roundtrip(tmp_path, monkeypatch): assert loaded['ensemble']['ankara']['spread'] == 1.5 +def test_open_meteo_rate_limit_repository_roundtrip(tmp_path, monkeypatch): + monkeypatch.setenv('POLYWEATHER_DB_PATH', str(tmp_path / 'polyweather.db')) + repo = OpenMeteoRateLimitRepository(RuntimeStateDB(str(tmp_path / 'polyweather.db'))) + + repo.set_until(12345.0, reason='429') + + loaded = repo.load_until() + assert loaded == 12345.0 + + def test_truth_record_repository_tracks_revisions(tmp_path, monkeypatch): monkeypatch.setenv('POLYWEATHER_DB_PATH', str(tmp_path / 'polyweather.db')) db = RuntimeStateDB(str(tmp_path / 'polyweather.db')) diff --git a/web/app.py b/web/app.py index dbdacc31..f5055683 100644 --- a/web/app.py +++ b/web/app.py @@ -39,5 +39,7 @@ if __name__ == "__main__": "web.app:app", host="0.0.0.0", port=8000, - workers=int(os.getenv("UVICORN_WORKERS", "4")), + # Default to a single worker so per-process weather-source guards do + # not multiply outbound Open-Meteo traffic on small VPS deployments. + workers=int(os.getenv("UVICORN_WORKERS", "1")), )