bot-backend
14f3ef2019
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
365 lines
19 KiB
Python
365 lines
19 KiB
Python
"""TTL-кэш для Open-Meteo (PR #1130 Phase A+B — Site Finder analyze latency).
|
||
|
||
ПРОБЛЕМА: `analyze_parcel` в hot-path делал ТРИ внешних HTTP-вызова к Open-Meteo на
|
||
каждый запрос (forecast + climate normals + air-quality), без кэша. На проде backend в
|
||
private network с restricted egress — DNS до `*.open-meteo.com` фейлится, каждый replay
|
||
висит полный timeout (5s + 15s + 5s = до 25с лишних на analyze). Профиль cProfile показал
|
||
~0.26с air-quality в успешном hit и +5с при DNS-fail (тот же антипаттерн, что был у
|
||
forecast/climate до Phase A).
|
||
|
||
РЕШЕНИЕ: in-process TTL-кэш по округлённым до ~1 км координатам.
|
||
|
||
• Forecast (`get_weather_cached`) — 7-day погода, мало меняется в пределах часов.
|
||
TTL hit: 6 часов; TTL negative (failure): 5 минут — чтобы DNS-fail не повторял
|
||
`timeout` на каждый analyze, но восстановление подхватывалось через ~5 минут.
|
||
• Climate normals (`get_seasonal_weather_cached`) — 30-летние сезонные нормали,
|
||
обновляются ~раз в год. TTL hit: 7 суток; TTL negative: 5 минут.
|
||
• Air quality (`get_air_quality_cached`) — текущий час pm2_5/pm10/no2.
|
||
TTL hit: 1 час (фронт-API API hourly bucket, обновлять чаще forecast);
|
||
TTL negative: 5 минут.
|
||
|
||
Ключ — `(round(lat, 2), round(lon, 2))`: округление до сотых ≈ 1 км, ЕКБ-wide один cad
|
||
≈ один ключ. Soft-cap 256 записей per-cache (eviction первого попавшегося) — защита от
|
||
утечки при потенциальном дрейфе ключей; для одного города этот лимит не достигается.
|
||
|
||
КОРРЕКТНОСТЬ: контракт для caller'а не меняется — None по-прежнему допустим (caller в
|
||
`parcels.py:_build_environmental_score` обрабатывает `bool(weather)` как env_ok-флаг).
|
||
Все исключения httpx/json/KeyError ловятся → `logger.warning` + negative-cache + None.
|
||
|
||
THREAD-SAFETY: analyze_parcel — sync def, под Uvicorn идёт в threadpool. Каждый кэш
|
||
защищён своим `threading.Lock`, который удерживается ТОЛЬКО на время check/store — сам
|
||
сетевой httpx-вызов идёт ВНЕ lock'а (#1370), иначе все analyze одного cache-типа
|
||
сериализуются на время вызова (lock — per-cache-type, не per-key). Trade-off: конкурентный
|
||
cold-start на ОДИН ключ может породить несколько параллельных запросов (last-write wins),
|
||
что приемлемо — вызовы идемпотентны, а TTL длинный.
|
||
|
||
time.monotonic используется вместо time.time — устойчиво к скачкам системного времени
|
||
(NTP sync, ручной выставление). Тесты могут подменять `weather_cache._now` для проталки
|
||
времени за expires_at без манки самого time.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import math
|
||
import threading
|
||
import time
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# TTL в секундах (см. модульный docstring).
|
||
_WEATHER_TTL_S: float = 6 * 3600 # forecast — 6 часов
|
||
_SEASONAL_TTL_S: float = 7 * 86400 # climate normals — 7 суток
|
||
_AIR_TTL_S: float = 1 * 3600 # air quality — 1 час (текущий час, обновлять чаще forecast)
|
||
_NEGATIVE_TTL_S: float = 300 # failed/timeout/exception — 5 минут все кэши
|
||
|
||
# Тайм-ауты httpx сокращены против оригинальных (5s/15s/5s) — open-meteo обычно отвечает
|
||
# за 200-600ms; при DNS-fail лучше быстрее упасть в negative-cache, чем держать worker
|
||
# 15с в hot-path.
|
||
_WEATHER_TIMEOUT_S: float = 2.0
|
||
_SEASONAL_TIMEOUT_S: float = 3.0
|
||
_AIR_TIMEOUT_S: float = 2.0
|
||
|
||
# Soft-cap на размер каждой таблицы (округлённых координат для одного города мало).
|
||
_MAX_ENTRIES: int = 256
|
||
|
||
# (value | None, expires_at_monotonic_seconds). None в value = negative-cache.
|
||
_CacheEntry = tuple[dict[str, Any] | None, float]
|
||
|
||
_FORECAST_CACHE: dict[tuple[float, float], _CacheEntry] = {}
|
||
_FORECAST_LOCK = threading.Lock()
|
||
|
||
_CLIMATE_CACHE: dict[tuple[float, float], _CacheEntry] = {}
|
||
_CLIMATE_LOCK = threading.Lock()
|
||
|
||
_AIR_CACHE: dict[tuple[float, float], _CacheEntry] = {}
|
||
_AIR_LOCK = threading.Lock()
|
||
|
||
|
||
def _now() -> float:
|
||
"""Монотонный таймер (тесты подменяют этот helper, а не сам `time`)."""
|
||
return time.monotonic()
|
||
|
||
|
||
def _round_key(lat: float, lon: float) -> tuple[float, float]:
|
||
"""Ключ кэша — координаты, округлённые до 0.01° (~1 км). См. docstring модуля."""
|
||
return (round(lat, 2), round(lon, 2))
|
||
|
||
|
||
def _evict_one_if_full(cache: dict[tuple[float, float], _CacheEntry]) -> None:
|
||
"""Soft-cap eviction: при превышении `_MAX_ENTRIES` выбросить старейшую по вставке запись.
|
||
|
||
Фактически FIFO благодаря insertion-order dict в CPython 3.7+: `next(iter(cache))` —
|
||
первый по порядку вставки ключ, т.е. самый старый (с большой вероятностью stale).
|
||
Не LRU — достаточно простой защиты от утечки на случай дрейфа ключей; LRU потребовал
|
||
бы `OrderedDict.move_to_end` на каждый hit (over-engineering при 256 lim + 6h TTL).
|
||
Под lock'ом вызывающей стороны.
|
||
"""
|
||
if len(cache) >= _MAX_ENTRIES:
|
||
victim = next(iter(cache))
|
||
cache.pop(victim, None)
|
||
|
||
|
||
def _fetch_weather_remote(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Forecast API (7-day) — собственно сетевой вызов и парсинг.
|
||
|
||
Любое исключение (httpx, json, KeyError) → logger.warning + None (caller обернёт
|
||
None в negative-cache на короткий TTL).
|
||
"""
|
||
try:
|
||
with httpx.Client(timeout=_WEATHER_TIMEOUT_S) as c:
|
||
r = c.get(
|
||
"https://api.open-meteo.com/v1/forecast",
|
||
params={
|
||
"latitude": lat,
|
||
"longitude": lon,
|
||
"daily": (
|
||
"temperature_2m_max,temperature_2m_min,"
|
||
"precipitation_sum,uv_index_max,"
|
||
"winddirection_10m_dominant,windspeed_10m_max"
|
||
),
|
||
"timezone": "Europe/Moscow",
|
||
"forecast_days": 7,
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
daily = r.json().get("daily", {})
|
||
if not daily.get("time"):
|
||
return None
|
||
|
||
# Open-Meteo штатно возвращает null для отдельных дней (uv_index_max и др.)
|
||
# при непустом daily.time — отфильтровываем None ПЕРЕД min/max/sum, иначе
|
||
# TypeError в Python 3.12 уронит весь 7-day forecast в negative-cache (#1577).
|
||
t_max = [v for v in (daily.get("temperature_2m_max") or []) if v is not None]
|
||
t_min = [v for v in (daily.get("temperature_2m_min") or []) if v is not None]
|
||
precip = [v for v in (daily.get("precipitation_sum") or []) if v is not None]
|
||
uv = [v for v in (daily.get("uv_index_max") or []) if v is not None]
|
||
wind_d = daily.get("winddirection_10m_dominant") or []
|
||
wind_s = [v for v in (daily.get("windspeed_10m_max") or []) if v is not None]
|
||
|
||
# Circular mean направления ветра (vector sum) — избегает jump 359→1
|
||
x = sum(math.cos(math.radians(d)) for d in wind_d if d is not None)
|
||
y = sum(math.sin(math.radians(d)) for d in wind_d if d is not None)
|
||
dominant = (math.degrees(math.atan2(y, x)) + 360) % 360 if wind_d else 0.0
|
||
rose = ["Север", "С-В", "Восток", "Ю-В", "Юг", "Ю-З", "Запад", "С-З"]
|
||
wind_label = rose[round(dominant / 45) % 8]
|
||
|
||
return {
|
||
"forecast_days": len(daily.get("time", [])),
|
||
"temperature": {
|
||
"min_c": round(min(t_min), 1) if t_min else None,
|
||
"max_c": round(max(t_max), 1) if t_max else None,
|
||
"avg_max_c": round(sum(t_max) / len(t_max), 1) if t_max else None,
|
||
"avg_min_c": round(sum(t_min) / len(t_min), 1) if t_min else None,
|
||
},
|
||
"precipitation_total_mm": round(sum(precip), 1) if precip else 0,
|
||
"precipitation_days": sum(1 for p in precip if p and p > 0.5),
|
||
"uv_index_max": round(max(uv), 1) if uv else None,
|
||
"wind": {
|
||
"dominant_direction_deg": round(dominant),
|
||
"dominant_direction_label": wind_label,
|
||
"max_speed_m_s": round(max(wind_s), 1) if wind_s else None,
|
||
},
|
||
"source": "open-meteo",
|
||
"note": (
|
||
"7-day forecast. Для исторических норм и B2B-данных — "
|
||
"Yandex Business / Gismeteo (платно)."
|
||
),
|
||
}
|
||
except Exception as e:
|
||
logger.warning("weather fetch failed: %s", e)
|
||
return None
|
||
|
||
|
||
def _fetch_seasonal_remote(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Climate API — 30-летние сезонные нормали (MRI-AGCM3.2-S).
|
||
|
||
Группирует 1995-2024 по четырём сезонам. Любое исключение → logger.warning + None
|
||
(caller обернёт None в negative-cache на короткий TTL).
|
||
"""
|
||
try:
|
||
with httpx.Client(timeout=_SEASONAL_TIMEOUT_S) as c:
|
||
r = c.get(
|
||
"https://climate-api.open-meteo.com/v1/climate",
|
||
params={
|
||
"latitude": lat,
|
||
"longitude": lon,
|
||
"start_date": "1995-01-01",
|
||
"end_date": "2024-12-31",
|
||
"models": "MRI_AGCM3_2_S",
|
||
"daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
daily = data.get("daily", {})
|
||
times = daily.get("time") or []
|
||
t_max = daily.get("temperature_2m_max") or []
|
||
t_min = daily.get("temperature_2m_min") or []
|
||
precip = daily.get("precipitation_sum") or []
|
||
if not times:
|
||
return None
|
||
|
||
seasons_months = {
|
||
"winter": [12, 1, 2],
|
||
"spring": [3, 4, 5],
|
||
"summer": [6, 7, 8],
|
||
"autumn": [9, 10, 11],
|
||
}
|
||
buckets: dict[str, dict[str, list[float]]] = {
|
||
k: {"t_max": [], "t_min": [], "precip": []} for k in seasons_months
|
||
}
|
||
for i, t in enumerate(times):
|
||
month = int(t[5:7]) # 'YYYY-MM-DD'
|
||
for season, months in seasons_months.items():
|
||
if month in months:
|
||
if i < len(t_max) and t_max[i] is not None:
|
||
buckets[season]["t_max"].append(t_max[i])
|
||
if i < len(t_min) and t_min[i] is not None:
|
||
buckets[season]["t_min"].append(t_min[i])
|
||
if i < len(precip) and precip[i] is not None:
|
||
buckets[season]["precip"].append(precip[i])
|
||
break
|
||
|
||
seasons: dict[str, Any] = {}
|
||
for season, vals in buckets.items():
|
||
if not vals["t_max"]:
|
||
seasons[season] = None
|
||
continue
|
||
# t_min/precip накапливаются НЕЗАВИСИМО от t_max (раздельные None-guard'ы
|
||
# выше) — при непустом t_max и all-null t_min/precip эти списки пусты, и
|
||
# sum()/len() даст ZeroDivisionError, min([]) — ValueError (#1578). Метрики
|
||
# по пустому списку → None вместо падения всего сезонного ответа.
|
||
t_min = vals["t_min"]
|
||
precip = vals["precip"]
|
||
seasons[season] = {
|
||
"avg_t_max_c": round(sum(vals["t_max"]) / len(vals["t_max"]), 1),
|
||
"avg_t_min_c": round(sum(t_min) / len(t_min), 1) if t_min else None,
|
||
"max_t_c": round(max(vals["t_max"]), 1),
|
||
"min_t_c": round(min(t_min), 1) if t_min else None,
|
||
"avg_precip_per_day_mm": (
|
||
round(sum(precip) / len(precip), 1) if precip else None
|
||
),
|
||
"total_precip_mm": round(sum(precip), 0) if precip else 0,
|
||
"days_observed": len(vals["t_max"]),
|
||
}
|
||
return {
|
||
"seasons": seasons,
|
||
"period": "1995-2024 (30 лет)",
|
||
"model": "MRI-AGCM3-2-S",
|
||
"source": "open-meteo-climate",
|
||
"note": ("Климатические нормали. Текущая погода — отдельный API."),
|
||
}
|
||
except Exception as e:
|
||
logger.warning("seasonal weather fetch failed: %s", e)
|
||
return None
|
||
|
||
|
||
def get_weather_cached(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Forecast (7-day) с TTL-кэшем.
|
||
|
||
Hit (успех): кэш 6 часов. Negative (None при ошибке): кэш 5 минут. Ключ —
|
||
координаты, округлённые до 0.01° (~1 км). На любой Exception от httpx/json
|
||
возвращает None и кэширует None под коротким TTL — контракт для caller'а не
|
||
меняется (None по-прежнему допустим).
|
||
"""
|
||
key = _round_key(lat, lon)
|
||
now = _now()
|
||
with _FORECAST_LOCK:
|
||
entry = _FORECAST_CACHE.get(key)
|
||
if entry is not None and entry[1] > now:
|
||
return entry[0]
|
||
# MISS или истёк → сетевой вызов ВНЕ lock'а, иначе все analyze одного cache-типа
|
||
# (даже для разных координат) сериализуются на время httpx-вызова (#1370). Ценой —
|
||
# cold-start на ОДИН ключ может породить несколько параллельных запросов (last-write
|
||
# wins при store ниже), что приемлемо: вызовы идемпотентны, TTL длинный.
|
||
value = _fetch_weather_remote(lat, lon)
|
||
ttl = _WEATHER_TTL_S if value is not None else _NEGATIVE_TTL_S
|
||
with _FORECAST_LOCK:
|
||
_evict_one_if_full(_FORECAST_CACHE)
|
||
_FORECAST_CACHE[key] = (value, _now() + ttl)
|
||
return value
|
||
|
||
|
||
def get_seasonal_weather_cached(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Climate API (30-летние сезонные нормали) с TTL-кэшем.
|
||
|
||
Hit (успех): кэш 7 суток (нормали меняются ~раз в год). Negative (None при
|
||
ошибке): кэш 5 минут. Ключ — координаты, округлённые до 0.01° (~1 км). На любой
|
||
Exception от httpx/json возвращает None и кэширует None под коротким TTL.
|
||
"""
|
||
key = _round_key(lat, lon)
|
||
now = _now()
|
||
with _CLIMATE_LOCK:
|
||
entry = _CLIMATE_CACHE.get(key)
|
||
if entry is not None and entry[1] > now:
|
||
return entry[0]
|
||
# Сетевой вызов ВНЕ lock'а — не сериализуем разные ключи (#1370). См. get_weather_cached.
|
||
value = _fetch_seasonal_remote(lat, lon)
|
||
ttl = _SEASONAL_TTL_S if value is not None else _NEGATIVE_TTL_S
|
||
with _CLIMATE_LOCK:
|
||
_evict_one_if_full(_CLIMATE_CACHE)
|
||
_CLIMATE_CACHE[key] = (value, _now() + ttl)
|
||
return value
|
||
|
||
|
||
def _fetch_air_quality_remote(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Air Quality API — pm2_5/pm10/no2 текущего часа.
|
||
|
||
Использует `current=...` (bucket текущего часа), а не `hourly[0]` (=00:00 forecast-дня) —
|
||
docstring/UX обещают «качество воздуха сейчас» (#1377). Формат возвращаемого dict
|
||
(ключи pm2_5/pm10/no2/ts/source) сохранён — фронт зависит. Любое исключение →
|
||
logger.warning + None (caller обернёт None в negative-cache на короткий TTL).
|
||
"""
|
||
try:
|
||
with httpx.Client(timeout=_AIR_TIMEOUT_S) as c:
|
||
r = c.get(
|
||
"https://air-quality-api.open-meteo.com/v1/air-quality",
|
||
params={
|
||
"latitude": lat,
|
||
"longitude": lon,
|
||
# `current` отдаёт bucket текущего часа (а не hourly[0]=00:00) —
|
||
# docstring/UX обещают «качество воздуха сейчас». См. #1377.
|
||
"current": "pm2_5,pm10,nitrogen_dioxide",
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
current = data.get("current", {})
|
||
if not current.get("time"):
|
||
return None
|
||
return {
|
||
"pm2_5": current.get("pm2_5"),
|
||
"pm10": current.get("pm10"),
|
||
"no2": current.get("nitrogen_dioxide"),
|
||
"ts": current["time"],
|
||
"source": "open-meteo",
|
||
}
|
||
except Exception as e:
|
||
logger.warning("air quality fetch failed: %s", e)
|
||
return None
|
||
|
||
|
||
def get_air_quality_cached(lat: float, lon: float) -> dict[str, Any] | None:
|
||
"""Open-Meteo Air Quality API (pm2_5/pm10/no2 текущего часа) с TTL-кэшем.
|
||
|
||
Hit (успех): кэш 1 час (hourly bucket API — обновлять чаще forecast). Negative
|
||
(None при ошибке): кэш 5 минут. Ключ — координаты, округлённые до 0.01° (~1 км).
|
||
На любой Exception от httpx/json возвращает None и кэширует None под коротким TTL.
|
||
"""
|
||
key = _round_key(lat, lon)
|
||
now = _now()
|
||
with _AIR_LOCK:
|
||
entry = _AIR_CACHE.get(key)
|
||
if entry is not None and entry[1] > now:
|
||
return entry[0]
|
||
# Сетевой вызов ВНЕ lock'а — не сериализуем разные ключи (#1370). См. get_weather_cached.
|
||
value = _fetch_air_quality_remote(lat, lon)
|
||
ttl = _AIR_TTL_S if value is not None else _NEGATIVE_TTL_S
|
||
with _AIR_LOCK:
|
||
_evict_one_if_full(_AIR_CACHE)
|
||
_AIR_CACHE[key] = (value, _now() + ttl)
|
||
return value
|