3bea7abaea
Точное совпадение (case-insensitive) теперь имеет приоритет над substring-matching, что исключает ложные совпадения вида 'Heavy rain' внутри 'Heavy rain shower'.
260 lines
9.9 KiB
Python
260 lines
9.9 KiB
Python
import asyncio
|
||
import logging
|
||
from typing import Optional
|
||
|
||
import requests
|
||
from requests.exceptions import ConnectionError, Timeout, SSLError
|
||
|
||
from utils.rate_limiter import weather_limiter, open_meteo_limiter
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
API_URL_WEATHER = "https://wttr.in/Magnitogorsk?format=j1&lang=ru"
|
||
|
||
_session = requests.Session()
|
||
|
||
|
||
async def fetch_weather(
|
||
api_url: str, timeout: int = 10, max_retries: int = 3
|
||
) -> Optional[dict]:
|
||
"""Получить данные о погоде с retry."""
|
||
await weather_limiter.acquire()
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = await asyncio.to_thread(_session.get, api_url, timeout=timeout)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
except (SSLError, ConnectionError, Timeout):
|
||
if attempt < max_retries - 1:
|
||
delay = 2**attempt
|
||
logger.warning(
|
||
"Попытка %d не удалась. Повтор через %d сек...", attempt + 1, delay
|
||
)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
break
|
||
except requests.RequestException as e:
|
||
logger.error("Ошибка при получении данных: %s", e)
|
||
break
|
||
|
||
logger.warning("Все попытки wttr.in не удались, переход на Open-Meteo")
|
||
return await fetch_open_meteo()
|
||
|
||
|
||
async def fetch_open_meteo(
|
||
lat: float = 53.4069, lon: float = 58.9797, timeout: int = 10, max_retries: int = 3
|
||
) -> Optional[dict]:
|
||
"""Fallback на Open-Meteo API."""
|
||
await open_meteo_limiter.acquire()
|
||
url = (
|
||
f"https://api.open-meteo.com/v1/forecast?"
|
||
f"latitude={lat}&longitude={lon}¤t=temperature,"
|
||
f"apparent_temperature,weather_code,wind_speed_10m,"
|
||
f"relative_humidity_2m,pressure_msl&timezone=Asia/Chelyabinsk"
|
||
)
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = await asyncio.to_thread(_session.get, url, timeout=timeout)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
current = data.get("current", {})
|
||
# weather_code WMO код -> перевод (https://open-meteo.com/en/docs)
|
||
weather_code = current.get("weather_code", None)
|
||
desc = wmo_to_russian(weather_code)
|
||
return {
|
||
"current_condition": [
|
||
{
|
||
"temp_C": current.get("temperature", "—"),
|
||
"FeelsLikeC": current.get("apparent_temperature", "—"),
|
||
"weatherDesc": [{"value": desc}],
|
||
"humidity": current.get("relative_humidity_2m", "—"),
|
||
"windspeedKmph": current.get("wind_speed_10m", "—"),
|
||
"pressure": current.get("pressure_msl", "—"),
|
||
}
|
||
]
|
||
}
|
||
except (SSLError, ConnectionError, Timeout):
|
||
if attempt < max_retries - 1:
|
||
delay = 2**attempt
|
||
logger.warning(
|
||
"Попытка %d не удалась. Повтор через %d сек...", attempt + 1, delay
|
||
)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
break
|
||
except requests.RequestException as e:
|
||
logger.error("Ошибка при получении данных: %s", e)
|
||
return None
|
||
|
||
logger.warning("Все попытки Open-Meteo не удались")
|
||
return None
|
||
|
||
|
||
_WMO_MAPPING: dict[int, str] = {
|
||
0: "Ясно",
|
||
1: "Ясно",
|
||
2: "Переменная облачность",
|
||
3: "Пасмурно",
|
||
45: "Туман",
|
||
48: "Туман",
|
||
51: "Лёгкая морось",
|
||
53: "Морось",
|
||
55: "Сильная морось",
|
||
56: "Ледяная морось",
|
||
57: "Сильная ледяная морось",
|
||
61: "Небольшой дождь",
|
||
63: "Дождь",
|
||
65: "Сильный дождь",
|
||
66: "Ледяной дождь",
|
||
67: "Сильный ледяной дождь",
|
||
71: "Небольшой снег",
|
||
73: "Снег",
|
||
75: "Сильный снег",
|
||
77: "Снежная крупа",
|
||
80: "Небольшой ливень",
|
||
81: "Ливень",
|
||
82: "Сильный ливень",
|
||
85: "Снежный ливень",
|
||
86: "Сильный снежный ливень",
|
||
95: "Гроза",
|
||
96: "Гроза с градом",
|
||
99: "Сильная гроза с градом",
|
||
}
|
||
|
||
|
||
def wmo_to_russian(code: Optional[int]) -> str:
|
||
"""Перевод WMO weather code в русский."""
|
||
return _WMO_MAPPING.get(code, "Неизвестно")
|
||
|
||
|
||
_WEATHER_MAPPING = [
|
||
# Отсортировано по убыванию длины ключа для корректного substring-matching:
|
||
# более длинные фразы проверяются первыми, чтобы "Light rain" не совпал
|
||
# раньше "Light rain shower".
|
||
("Moderate or heavy freezing rain at a distance", "Ледяной дождь"),
|
||
("Moderate or heavy freezing rain in area", "Ледяной дождь"),
|
||
("Moderate or heavy sleet at a distance", "Слякоть"),
|
||
("Moderate or heavy sleet in area", "Слякоть"),
|
||
("Moderate or heavy rain at times", "Дождь"),
|
||
("Moderate or heavy snow at times", "Снег"),
|
||
("Moderate or heavy snow in area", "Снег"),
|
||
("Moderate or heavy rain in area", "Дождь"),
|
||
("Thundery outbreaks in nearby", "Гроза вблизи"),
|
||
("Moderate or light sleet", "Слякоть"),
|
||
("Moderate rain at times", "Умеренный дождь"),
|
||
("Patchy light drizzle", "Местами лёгкая морось"),
|
||
("Heavy freezing rain", "Сильный ледяной дождь"),
|
||
("Light freezing rain", "Лёгкий ледяной дождь"),
|
||
("Patchy sleet nearby", "Местами слякоть"),
|
||
("Heavy rain at times", "Сильный дождь"),
|
||
("Patchy rain nearby", "Местами дождь"),
|
||
("Patchy snow nearby", "Местами снег"),
|
||
("Patchy light snow", "Местами лёгкий снег"),
|
||
("Light rain shower", "Небольшой дождь"),
|
||
("Heavy rain shower", "Сильный дождь"),
|
||
("Moderate rain", "Умеренный дождь"),
|
||
("Moderate snow", "Умеренный снег"),
|
||
("Partly cloudy", "Переменная облачность"),
|
||
("Blowing snow", "Метель"),
|
||
("Light rain", "Небольшой дождь"),
|
||
("Heavy rain", "Сильный дождь"),
|
||
("Light snow", "Небольшой снег"),
|
||
("Heavy snow", "Сильный снег"),
|
||
("Overcast", "Пасмурно"),
|
||
("Cloudy", "Облачно"),
|
||
("Foggy", "Туманно"),
|
||
("Clear", "Ясно"),
|
||
("Sunny", "Ясно"),
|
||
("Mist", "Туман"),
|
||
("Haze", "Дымка"),
|
||
("Fog", "Туман"),
|
||
]
|
||
|
||
|
||
def translate_weather(en: Optional[str]) -> str:
|
||
if not en or not en.strip():
|
||
return "—"
|
||
en_stripped = en.strip()
|
||
en_lower = en_stripped.lower()
|
||
|
||
# Сначала проверяем точное совпадение (приоритет над substring)
|
||
for key, value in _WEATHER_MAPPING:
|
||
if key.lower() == en_lower:
|
||
return value
|
||
|
||
# Fallback: substring-matching (отсортировано по убыванию длины ключа)
|
||
for key, value in _WEATHER_MAPPING:
|
||
if key.lower() in en_lower:
|
||
return value
|
||
|
||
return en
|
||
|
||
|
||
def format_weather_data_for_console(data: Optional[dict]) -> Optional[list[str]]:
|
||
"""
|
||
Форматировать погодные данные для консольного вывода.
|
||
|
||
:param data: Ответ от API (dict)
|
||
:return: Строки с отформатированной погодой
|
||
"""
|
||
if data is None:
|
||
return None
|
||
current_condition_list = data.get("current_condition", [])
|
||
if not current_condition_list:
|
||
return None
|
||
current = current_condition_list[0]
|
||
if not current:
|
||
return None
|
||
|
||
# dict.get() возвращает None если ключ существует со значением null;
|
||
# используем проверку на None для корректного fallback
|
||
temp = current.get("temp_C")
|
||
if temp is None:
|
||
temp = "—"
|
||
feels_like = current.get("FeelsLikeC")
|
||
if feels_like is None:
|
||
feels_like = "—"
|
||
weather_desc = current.get("weatherDesc", [{}])[0].get("value")
|
||
if weather_desc is None:
|
||
weather_desc = "—"
|
||
description = translate_weather(weather_desc)
|
||
humidity = current.get("humidity")
|
||
if humidity is None:
|
||
humidity = "—"
|
||
wind_kmh = current.get("windspeedKmph")
|
||
if wind_kmh is None:
|
||
wind_kmh = "—"
|
||
try:
|
||
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "—" else "—"
|
||
except (ValueError, TypeError):
|
||
wind = "—"
|
||
pressure_mb = current.get("pressure", "—")
|
||
pressure_mm = pressure_to_mmhg(pressure_mb)
|
||
|
||
return [
|
||
f"Температура: {temp}°C (ощущается как {feels_like}°C)",
|
||
f"Описание: {description}",
|
||
f"Влажность: {humidity}%",
|
||
f"Ветер: {wind} м/с",
|
||
f"Давление: {pressure_mm} мм рт. ст.",
|
||
]
|
||
|
||
|
||
def format_weather_for_message(data: Optional[dict]) -> Optional[str]:
|
||
"""Форматировать погоду для plain text сообщения (с заголовком)."""
|
||
if data is None:
|
||
return None
|
||
lines = format_weather_data_for_console(data)
|
||
if not lines:
|
||
return None
|
||
return "Погода в Магнитогорске:\n" + "\n".join(lines)
|
||
|
||
|
||
def pressure_to_mmhg(mb: float | int | str | None) -> float | str:
|
||
if mb == "—" or mb is None or mb == "":
|
||
return "—"
|
||
try:
|
||
return round(float(mb) * 0.750062, 1)
|
||
except (ValueError, TypeError):
|
||
return "—"
|