gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Telegram 高频推送内存优化：LRU 缓存限制、TTL 驱逐、速率控制、连接复用

Browse Source 
core.py: _cache 改用 LRUDict(maxsize=256) + threading.Lock 线程安全读写
analysis_service.py: _CACHE_LOCK 保护读写，_SUMMARY_CACHE 改用 LRUDict(maxsize=128)
weather_sources.py: 新增 _maybe_trim_caches() 定期清理 11 个过期缓存条目，防止无限增长
telegram_push.py: requests.Session 连接复用替代裸 requests.get，ThreadPoolExecutor 跨周期复用，
新增 _rate_limited_send 消息发送速率控制防止 API 限流堆积

Tested: ruff check 通过, 182 pytest 通过
This commit is contained in:
2569718930@qq.com
2026-05-25 02:20:15 +08:00
parent 3534f759d0
commit ad936bfd91
4 changed files with 180 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_collection/weather_sources.py
+55
View File
@@ -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)
src/utils/telegram_push.py
+74 -26
View File
@@ -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