ecff77ef66
Город больше не захардкожен в коде. Переопределяется через
WEATHER_CITY (по умолчанию "Магнитогорск"). Шаблон изменён на
"Погода: {city}:" для обхода проблемы падежей в русском языке.
Файлы: utils/pogoda.py, .env.example, tests/test_morning_runner.py
344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""Погода через Яндекс Погоду API.
|
||
|
||
Яндекс Погода: актуальные данные для городов России и мира,
|
||
текущие условия + прогноз. Требуется API-ключ в YANDEX_WEATHER_API_KEY.
|
||
|
||
https://yandex.ru/dev/weather/
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
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__)
|
||
|
||
# Город для погоды (переопределяется WEATHER_CITY)
|
||
_WEATHER_CITY: str = os.getenv("WEATHER_CITY", "Магнитогорск")
|
||
|
||
# Координаты города
|
||
_LATITUDE: float = 53.40716
|
||
_LONGITUDE: float = 58.980289
|
||
|
||
_session = requests.Session()
|
||
|
||
# Кэш погоды: (data, timestamp) или (None, timestamp) при провале запроса
|
||
_weather_cache: tuple[Optional[dict], float] = (None, 0.0)
|
||
_weather_cache_lock = asyncio.Lock()
|
||
# TTL в секундах — по умолчанию 1 час, переопределяется WEATHER_CACHE_TTL
|
||
_WEATHER_CACHE_TTL: float = float(os.getenv("WEATHER_CACHE_TTL", "3600"))
|
||
|
||
|
||
def _get_api_key() -> str:
|
||
"""Получить API-ключ Яндекс Погоды из переменных окружения."""
|
||
key = os.getenv("YANDEX_WEATHER_API_KEY")
|
||
if not key:
|
||
raise EnvironmentError(
|
||
"YANDEX_WEATHER_API_KEY не найден в переменных окружения"
|
||
)
|
||
return key
|
||
|
||
|
||
def clear_weather_cache() -> None:
|
||
"""Очистить кэш погоды. Используется в тестах."""
|
||
global _weather_cache
|
||
_weather_cache = (None, 0.0)
|
||
|
||
|
||
async def fetch_weather(
|
||
lat: float = _LATITUDE,
|
||
lon: float = _LONGITUDE,
|
||
timeout: int = 10,
|
||
max_retries: int = 3,
|
||
bypass_cache: bool = False,
|
||
) -> Optional[dict]:
|
||
"""Получить текущую погоду через Яндекс Погоду API.
|
||
|
||
Результат кэшируется на _WEATHER_CACHE_TTL секунд (по умолчанию 1 ч).
|
||
Повторные вызовы в течение TTL возвращают кэшированные данные без запроса к API.
|
||
|
||
:param bypass_cache: Пропустить кэш и сделать свежий запрос.
|
||
:return: dict в унифицированном формате или None при ошибке.
|
||
"""
|
||
global _weather_cache
|
||
|
||
# Проверка кэша
|
||
if not bypass_cache:
|
||
async with _weather_cache_lock:
|
||
cached_data, cached_time = _weather_cache
|
||
if time.monotonic() - cached_time < _WEATHER_CACHE_TTL:
|
||
logger.debug(
|
||
"Погода: возвращены данные из кэша (TTL %.0f сек)",
|
||
_WEATHER_CACHE_TTL,
|
||
)
|
||
return cached_data
|
||
|
||
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", {})
|
||
|
||
result = {
|
||
"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"),
|
||
}
|
||
]
|
||
}
|
||
async with _weather_cache_lock:
|
||
_weather_cache = (result, time.monotonic())
|
||
return result
|
||
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("Все попытки Яндекс Погоды не удались")
|
||
# Сохраняем провал в кэш, чтобы не долбить API повторно
|
||
async with _weather_cache_lock:
|
||
_weather_cache = (None, time.monotonic())
|
||
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 f"Погода: {_WEATHER_CITY}:\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, "Неизвестно")
|