import asyncio import logging from typing import Any, 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 def wmo_to_russian(code: Optional[int]) -> str: """Перевод WMO weather code в русский.""" mapping = { 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: "Сильная гроза с градом", } return 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: return "—" en_lower = en.lower() 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: Any) -> float | str: if mb == "—" or mb is None or mb == "": return "—" try: return round(float(mb) * 0.750062, 1) except (ValueError, TypeError): return "—"