diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 8449f2e3..a6bccbb2 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -256,6 +256,10 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour logger.info( f"Open-Meteo 磁盘缓存路径: {self._disk_cache_path} (max_age={self._disk_cache_max_age_sec}s)" ) + self._cache_trim_counter = 0 + self._CACHE_TRIM_EVERY_N_WRITES = int( + os.getenv("POLYWEATHER_CACHE_TRIM_INTERVAL", "200") + ) # 设置代理 proxy = config.get("proxy") @@ -273,6 +277,56 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour logger.info("天气数据采集器初始化完成。") + @staticmethod + def _trim_inmemory_cache( + cache: Dict[str, Dict], + lock: threading.Lock, + ttl_sec: float, + ) -> None: + now = time.time() + with lock: + stale = [ + key + for key, entry in cache.items() + if now - float(entry.get("t", 0)) > ttl_sec + ] + for key in stale: + del cache[key] + + def _maybe_trim_caches(self) -> None: + self._cache_trim_counter += 1 + if self._cache_trim_counter < self._CACHE_TRIM_EVERY_N_WRITES: + return + self._cache_trim_counter = 0 + now = time.time() + # Trim each cache with 2x its TTL as the hard eviction window + for cache, lock, ttl in [ + (self._open_meteo_cache, self._open_meteo_cache_lock, 3600.0), + (self._ensemble_cache, self._ensemble_cache_lock, 7200.0), + (self._multi_model_cache, self._multi_model_cache_lock, 7200.0), + (self._metar_cache, self._metar_cache_lock, float(self.metar_cache_ttl_sec * 2)), + (self._taf_cache, self._taf_cache_lock, float(self.taf_cache_ttl_sec * 2)), + (self._jma_cache, self._jma_cache_lock, float(self.jma_cache_ttl_sec * 2)), + (self._settlement_cache, self._settlement_cache_lock, float(self.settlement_cache_ttl_sec * 2)), + (self._fmi_cache, self._fmi_cache_lock, float(self.fmi_cache_ttl_sec * 2)), + (self._knmi_cache, self._knmi_cache_lock, float(self.knmi_cache_ttl_sec * 2)), + (self._hko_obs_cache, self._hko_obs_cache_lock, float(self.hko_obs_cache_ttl_sec * 2)), + ]: + stale = [ + key + for key, entry in cache.items() + if now - float(entry.get("t", 0)) > ttl + ] + if stale: + with lock: + for key in stale: + cache.pop(key, None) + # MADIS cache is a single list, not a keyed dict — expire on age + with self._madis_cache_lock: + if self._madis_cache is not None and now - self._madis_cache_ts > self.madis_cache_ttl_sec * 2: + self._madis_cache = None + self._madis_cache_ts = 0.0 + @staticmethod def _is_retryable_status(status_code: int) -> bool: return status_code in {408, 500, 502, 503, 504} @@ -1333,6 +1387,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour """ Fetch weather data from all available sources """ + self._maybe_trim_caches() results = {} city_lower = city.lower().strip() use_fahrenheit = self._uses_fahrenheit(city_lower) diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 85e24691..438fbc1b 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -8,6 +8,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Set, Tuple +import requests as requests_lib from loguru import logger from src.database.db_manager import DBManager @@ -28,6 +29,54 @@ _CITY_THREAD_IDS_PATH = os.path.join( _FORUM_CHAT_ID = "-1003927451869" _city_thread_ids: dict = {} +# Shared HTTP session for AROME and auxiliary queries (connection reuse) +_HTTP_SESSION: Optional[requests_lib.Session] = None +_HTTP_SESSION_LOCK = threading.Lock() + +# Bot send_message rate limiter: max N messages per second across all threads +_SEND_MSG_LOCK = threading.Lock() +_SEND_MSG_LAST_TS: float = 0.0 +_SEND_MSG_MIN_INTERVAL_SEC = float(os.getenv("TELEGRAM_SEND_RATE_LIMIT_SEC", "0.05")) + + +def _get_http_session() -> requests_lib.Session: + global _HTTP_SESSION + if _HTTP_SESSION is None: + with _HTTP_SESSION_LOCK: + if _HTTP_SESSION is None: + _HTTP_SESSION = requests_lib.Session() + return _HTTP_SESSION + + +# Reusable executor for airport push cycles (avoids thread pool churn) +_AIRPORT_EXECUTOR: Optional[ThreadPoolExecutor] = None +_AIRPORT_EXECUTOR_LOCK = threading.Lock() +_AIRPORT_EXECUTOR_MAX_WORKERS: int = 0 + + +def _get_airport_executor(max_workers: int) -> ThreadPoolExecutor: + global _AIRPORT_EXECUTOR, _AIRPORT_EXECUTOR_MAX_WORKERS + if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: + with _AIRPORT_EXECUTOR_LOCK: + if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: + if _AIRPORT_EXECUTOR is not None: + _AIRPORT_EXECUTOR.shutdown(wait=False) + _AIRPORT_EXECUTOR = ThreadPoolExecutor(max_workers=max_workers) + _AIRPORT_EXECUTOR_MAX_WORKERS = max_workers + return _AIRPORT_EXECUTOR + + +def _rate_limited_send(bot: Any, chat_id: str, message: str, **kwargs: Any) -> None: + """Throttle bot.send_message calls to avoid hitting Telegram rate limits.""" + global _SEND_MSG_LAST_TS + with _SEND_MSG_LOCK: + now = time.time() + wait = _SEND_MSG_MIN_INTERVAL_SEC - (now - _SEND_MSG_LAST_TS) + if wait > 0: + time.sleep(wait) + _SEND_MSG_LAST_TS = time.time() + bot.send_message(chat_id, message, **kwargs) + def _load_city_thread_ids() -> dict: global _city_thread_ids @@ -814,7 +863,6 @@ def _fetch_arome_temp() -> Optional[float]: if cached is not None and (now - cached_at) < _AROME_CACHE_TTL_SEC: return cached try: - import requests url = ( "https://api.open-meteo.com/v1/forecast?" "latitude=48.9673&longitude=2.4277" @@ -823,7 +871,7 @@ def _fetch_arome_temp() -> Optional[float]: "&timezone=Europe/Paris" "&forecast_minutely_15=2" ) - resp = requests.get(url, timeout=8) + resp = _get_http_session().get(url, timeout=8) data = resp.json() temps = (data.get("minutely_15") or {}).get("temperature_2m") or [] result = float(temps[-1]) if temps else None @@ -1314,7 +1362,7 @@ def _process_airport_city( thread_id = _resolve_thread_id(chat_id, city) if thread_id: kwargs["message_thread_id"] = thread_id - bot.send_message(chat_id, message, **kwargs) + _rate_limited_send(bot, chat_id, message, **kwargs) sent = True except Exception as exc: logger.warning("airport push failed city={} chat_id={}: {}", city, chat_id, exc) @@ -1340,29 +1388,29 @@ def _run_high_freq_airport_cycle( logger.info("airport cycle tick cities={} max_workers={}", len(HIGH_FREQ_AIRPORT_CITIES), max_workers) cities = sorted(HIGH_FREQ_AIRPORT_CITIES) - with ThreadPoolExecutor(max_workers=max_workers) as pool: - futures = { - pool.submit( - _process_airport_city, - city, - now_ts, - last_by_city.get(city) or {}, - chat_ids, - bot, - ): city - for city in cities - } - for future in as_completed(futures): - try: - result = future.result() - except Exception: - logger.exception("airport city task crashed city={}", futures[future]) - continue - if result is None: - continue - city, entry = result - last_by_city[city] = entry - state_dirty = True + pool = _get_airport_executor(max_workers) + futures = { + pool.submit( + _process_airport_city, + city, + now_ts, + last_by_city.get(city) or {}, + chat_ids, + bot, + ): city + for city in cities + } + for future in as_completed(futures): + try: + result = future.result() + except Exception: + logger.exception("airport city task crashed city={}", futures[future]) + continue + if result is None: + continue + city, entry = result + last_by_city[city] = entry + state_dirty = True return state_dirty diff --git a/web/analysis_service.py b/web/analysis_service.py index ad099ea5..102a2a90 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -11,7 +11,9 @@ from fastapi import HTTPException from loguru import logger from web.core import ( + LRUDict, _cache, + _CACHE_LOCK, CACHE_TTL, CACHE_TTL_ANKARA, CACHE_TTL_KOREAN_AMOS, @@ -89,7 +91,8 @@ _ANALYSIS_CACHE_STATS: Dict[str, Any] = { "last_city": None, } _SUMMARY_CACHE_LOCK = threading.Lock() -_SUMMARY_CACHE: Dict[str, Dict[str, Any]] = {} +_SUMMARY_CACHE_MAXSIZE = 128 +_SUMMARY_CACHE = LRUDict(maxsize=_SUMMARY_CACHE_MAXSIZE) def _dedupe_forecast_daily(rows: Any) -> list[Dict[str, Any]]: if not isinstance(rows, list): return [] @@ -413,20 +416,21 @@ def _get_cached_analysis( now_ts = _time.time() freshest_payload: Optional[Dict[str, Any]] = None freshest_ts = 0.0 - for detail_mode in detail_modes: - cached = _cache.get(_analysis_cache_key(city, detail_mode)) - if not cached: - continue - cached_ts = float(cached.get("t", 0)) - payload = cached.get("d") - if ( - cached_ts - and now_ts - cached_ts < ttl - and isinstance(payload, dict) - and cached_ts >= freshest_ts - ): - freshest_payload = payload - freshest_ts = cached_ts + with _CACHE_LOCK: + for detail_mode in detail_modes: + cached = _cache.get(_analysis_cache_key(city, detail_mode)) + if not cached: + continue + cached_ts = float(cached.get("t", 0)) + payload = cached.get("d") + if ( + cached_ts + and now_ts - cached_ts < ttl + and isinstance(payload, dict) + and cached_ts >= freshest_ts + ): + freshest_payload = payload + freshest_ts = cached_ts return freshest_payload @@ -1882,7 +1886,8 @@ def _analyze( result, ) - _cache[cache_key] = {"t": _time.time(), "d": result} + with _CACHE_LOCK: + _cache[cache_key] = {"t": _time.time(), "d": result} return result diff --git a/web/core.py b/web/core.py index 3e886f36..94ad2f73 100644 --- a/web/core.py +++ b/web/core.py @@ -5,9 +5,11 @@ PolyWeather Web Core Context import json import os import sqlite3 +import threading import time +from collections import OrderedDict from datetime import datetime, timezone -from typing import Dict, Any, Optional +from typing import Any, Dict, Optional from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware @@ -77,7 +79,33 @@ SETTLEMENT_SOURCE_LABELS: Dict[str, str] = { "wunderground": "Wunderground", } -_cache: Dict[str, Dict] = {} +class LRUDict: + """Size-bounded ordered dict that evicts oldest entries on overflow.""" + + def __init__(self, maxsize: int = 256) -> None: + self._maxsize = max(1, int(maxsize)) + self._data: OrderedDict[str, Dict] = OrderedDict() + + def get(self, key: str) -> Optional[Dict]: + return self._data.get(key) + + def __setitem__(self, key: str, value: Dict) -> None: + if key in self._data: + self._data.move_to_end(key) + self._data[key] = value + while len(self._data) > self._maxsize: + self._data.popitem(last=False) + + def __contains__(self, key: str) -> bool: + return key in self._data + + def __len__(self) -> int: + return len(self._data) + + +_CACHE_MAXSIZE = int(os.getenv("POLYWEATHER_ANALYSIS_CACHE_MAXSIZE", "256")) +_cache: LRUDict = LRUDict(maxsize=_CACHE_MAXSIZE) +_CACHE_LOCK = threading.Lock() CACHE_TTL = 300 CACHE_TTL_ANKARA = 60 CACHE_TTL_KOREAN_AMOS = 60