b8b1801f23
- JSONDecodeError теперь ловится в fetch_weather (был краш при невалидном JSON от API) - Порывы ветра больше не теряются при отсутствии base wind (gust fallback) - Устранено дублирование 'м/с' при порывах + отсутствии направления ветра - Переименован windspeedKmph → wind_speed_mps (ключ хранил м/с, а не км/ч) - Обновлены тесты, добавлены 6 новых (gust-only, no-duplication, JSONDecodeError)
299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""Погода через Яндекс Погоду API.
|
||
|
||
Яндекс Погода: актуальные данные для городов России и мира,
|
||
текущие условия + прогноз. Требуется API-ключ в YANDEX_WEATHER_API_KEY.
|
||
|
||
https://yandex.ru/dev/weather/
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
from typing import Optional
|
||
|
||
import requests
|
||
from requests.exceptions import ConnectionError, Timeout, SSLError
|
||
|
||
from utils.rate_limiter import yandex_weather_limiter
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Координаты Магнитогорска
|
||
_LATITUDE: float = 53.40716
|
||
_LONGITUDE: float = 58.980289
|
||
|
||
_session = requests.Session()
|
||
|
||
|
||
def _get_api_key() -> str:
|
||
"""Получить API-ключ Яндекс Погоды из переменных окружения."""
|
||
key = os.getenv("YANDEX_WEATHER_API_KEY")
|
||
if not key:
|
||
raise EnvironmentError(
|
||
"YANDEX_WEATHER_API_KEY не найден в переменных окружения"
|
||
)
|
||
return key
|
||
|
||
|
||
async def fetch_weather(
|
||
lat: float = _LATITUDE,
|
||
lon: float = _LONGITUDE,
|
||
timeout: int = 10,
|
||
max_retries: int = 3,
|
||
) -> Optional[dict]:
|
||
"""Получить текущую погоду через Яндекс Погоду API.
|
||
|
||
Возвращает dict в унифицированном формате для форматирования:
|
||
{"current_condition": [{"temp_C": ..., "weatherDesc": ..., ...}]}
|
||
"""
|
||
await yandex_weather_limiter.acquire()
|
||
url = f"https://api.weather.yandex.ru/v1/informers?lat={lat}&lon={lon}"
|
||
headers = {"X-Yandex-API-Key": _get_api_key()}
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = await asyncio.to_thread(
|
||
_session.get, url, headers=headers, timeout=timeout
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
fact = data.get("fact", {})
|
||
|
||
return {
|
||
"current_condition": [
|
||
{
|
||
"temp_C": fact.get("temp"),
|
||
"FeelsLikeC": fact.get("feels_like"),
|
||
"weatherDesc": [{"value": yandex_condition_to_russian(fact.get("condition"))}],
|
||
"humidity": fact.get("humidity"),
|
||
"wind_speed_mps": fact.get("wind_speed"),
|
||
"wind_gust": fact.get("wind_gust"),
|
||
"wind_dir": fact.get("wind_dir"),
|
||
"pressure": fact.get("pressure_mm"),
|
||
"pressure_pa": fact.get("pressure_pa"),
|
||
}
|
||
]
|
||
}
|
||
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, json.JSONDecodeError, ValueError) as e:
|
||
logger.error("Ошибка Яндекс Погоды: %s", e)
|
||
break
|
||
|
||
logger.warning("Все попытки Яндекс Погоды не удались")
|
||
return None
|
||
|
||
|
||
# Яндекс condition → русский перевод
|
||
_YANDEX_CONDITION_MAPPING: dict[str, str] = {
|
||
"clear": "Ясно",
|
||
"partly_cloudy": "Переменная облачность",
|
||
"cloudy": "Облачно",
|
||
"overcast": "Пасмурно",
|
||
"light_rain": "Небольшой дождь",
|
||
"rain": "Дождь",
|
||
"heavy_rain": "Сильный дождь",
|
||
"drizzle": "Морось",
|
||
"heavy_showers": "Сильные осадки",
|
||
"thunderstorm": "Гроза",
|
||
"thunderstorm_with_rain": "Гроза с дождём",
|
||
"thunderstorm_with_heavy_rain": "Сильная гроза с дождём",
|
||
"thunderstorm_with_hail": "Гроза с градом",
|
||
"snow_showers": "Снежные осадки",
|
||
"light_snow": "Небольшой снег",
|
||
"snow": "Снег",
|
||
"heavy_snow": "Сильный снег",
|
||
"snowstorm": "Метель",
|
||
"blizzard": "Буран",
|
||
"fog": "Туман",
|
||
}
|
||
|
||
|
||
def yandex_condition_to_russian(condition: Optional[str]) -> str:
|
||
"""Перевод Яндекс condition в русский."""
|
||
if condition is None:
|
||
return "Неизвестно"
|
||
return _YANDEX_CONDITION_MAPPING.get(condition, "Неизвестно")
|
||
|
||
|
||
def get_weather_description(current: dict) -> str:
|
||
"""Извлечь описание погоды из weatherDesc."""
|
||
return (current.get("weatherDesc") or [{}])[0].get("value") or "—"
|
||
|
||
|
||
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 or not current_condition_list[0]:
|
||
return None
|
||
current = current_condition_list[0]
|
||
|
||
temp = current.get("temp_C")
|
||
if temp is None:
|
||
temp = "—"
|
||
feels_like = current.get("FeelsLikeC")
|
||
if feels_like is None:
|
||
feels_like = "—"
|
||
description = get_weather_description(current)
|
||
humidity = current.get("humidity")
|
||
if humidity is None:
|
||
humidity = "—"
|
||
|
||
# Скорость ветра (м/с)
|
||
wind_mps = current.get("wind_speed_mps")
|
||
wind_value = None
|
||
if wind_mps is not None:
|
||
try:
|
||
wind_value = round(float(wind_mps), 1)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# Собираем компоненты ветра
|
||
wind_parts = []
|
||
|
||
if wind_value is not None:
|
||
wind_parts.append(str(wind_value))
|
||
elif wind_value is None:
|
||
# Базовая скорость отсутствует — покажем порывы, если есть
|
||
wind_gust = current.get("wind_gust")
|
||
if wind_gust is not None:
|
||
try:
|
||
wind_parts.append(f"порывы {round(float(wind_gust), 1)}")
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# Порывы ветра (м/с) — добавляем к базовой скорости
|
||
wind_gust = current.get("wind_gust")
|
||
if wind_gust is not None and wind_value is not None:
|
||
try:
|
||
wind_parts[-1] = f"{wind_parts[-1]} (порывы {round(float(wind_gust), 1)})"
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# Направление ветра
|
||
wind_dir = current.get("wind_dir")
|
||
if wind_dir is not None:
|
||
wind_dir_ru = _wind_dir_to_russian(wind_dir)
|
||
wind_parts.append(wind_dir_ru)
|
||
|
||
if wind_parts:
|
||
wind = ", ".join(wind_parts) + " м/с"
|
||
else:
|
||
wind = "— м/с"
|
||
|
||
# Давление — Яндекс уже возвращает в мм рт. ст.
|
||
pressure_mm = current.get("pressure")
|
||
if pressure_mm is None:
|
||
pressure_mm = "—"
|
||
elif pressure_mm != "—":
|
||
try:
|
||
pressure_mm = round(float(pressure_mm), 1)
|
||
except (ValueError, TypeError):
|
||
pressure_mm = "—"
|
||
|
||
lines = [
|
||
f"Температура: {temp}°C (ощущается как {feels_like}°C)",
|
||
f"Описание: {description}",
|
||
f"Влажность: {humidity}%",
|
||
f"Ветер: {wind}",
|
||
f"Давление: {pressure_mm} мм рт. ст.",
|
||
]
|
||
|
||
return lines
|
||
|
||
|
||
def _wind_dir_to_russian(direction: str) -> str:
|
||
"""Перевод направления ветра из кода в русский."""
|
||
mapping = {
|
||
"n": "северный",
|
||
"ne": "северо-восточный",
|
||
"e": "восточный",
|
||
"se": "юго-восточный",
|
||
"s": "южный",
|
||
"sw": "юго-западный",
|
||
"w": "западный",
|
||
"nw": "северо-западный",
|
||
}
|
||
return mapping.get(direction, direction)
|
||
|
||
|
||
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 "—"
|
||
|
||
|
||
# WMO mapping оставлен для обратной совместимости (используется в тестах)
|
||
_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, "Неизвестно")
|