import logging import aiohttp import random from datetime import datetime, timedelta from config import get_proxy_url logger = logging.getLogger(__name__) # Координаты Ульяновска LAT = 54.3188 LON = 48.3978 TIMEZONE = "Europe/Ulyanovsk" # WMO Weather Codes to Emoji & Description WEATHER_CODES = { 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: ("⛈️", "Сильная гроза с градом"), } CYCLING_QUOTES = [ "«Жизнь — как вождение велосипеда. Чтобы сохранить равновесие, ты должен двигаться». — Альберт Эйнштейн 🚲", "«Ничто не сравнится с простым удовольствием от поездки на велосипеде». — Джон Ф. Кеннеди 🚴", "«Когда мне тяжело, я просто сажусь на велосипед и кручу педали». — Эдди Меркс 🏁", "Велосипед — это самый быстрый способ добраться из унылой рутины в прекрасное приключение! 🌲🚲", "Крути педали, пока молод и пока светит солнце (или хотя бы нет ливня)! ☀️🚴", "Ветер в лицо — это не препятствие, это признак того, что ты летишь! 💨🚴", "«Велосипед спасет мир». — Эмилио Эмилиани 🌍🚲", "Хорошая покатушка очищает разум лучше любого психолога. Только ты, дорога и педали! 🚴✨", "«Велосипед — это часть меня». — Марко Пантани 🇮🇹", "Каждый раз, когда я вижу взрослого на велосипеде, я перестаю отчаиваться за будущее человечества. — Герберт Уэллс 🚲", ] class WeatherService: @staticmethod def get_emoji(code): return WEATHER_CODES.get(code, ("🌦️", "Переменная погода"))[0] @staticmethod def get_desc(code): return WEATHER_CODES.get(code, ("🌦️", "Переменная погода"))[1] @classmethod async def _fetch_raw(cls, url): proxy = get_proxy_url() async with aiohttp.ClientSession() as session: try: async with session.get(url, proxy=proxy, timeout=10) as resp: if resp.status == 200: return await resp.json() else: logger.error(f"Weather API status: {resp.status}") return None except Exception as e: logger.error(f"Weather API connection error: {e}") return None @classmethod async def get_today_forecast(cls) -> str: """Подробный прогноз на сегодня (по часам) + вероятность дождя сегодня/завтра""" url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={LAT}&longitude={LON}&" f"hourly=temperature_2m,precipitation_probability,weathercode,windspeed_10m&" f"daily=weathercode,temperature_2m_max,temperature_2m_min,precipitation_probability_max&" f"timezone={TIMEZONE}&forecast_days=2" ) data = await cls._fetch_raw(url) if not data: return "❌ Не удалось получить прогноз погоды. Попробуйте позже." temp_max_today = data["daily"]["temperature_2m_max"][0] temp_min_today = data["daily"]["temperature_2m_min"][0] rain_prob_today = data["daily"]["precipitation_probability_max"][0] rain_prob_tomorrow = data["daily"]["precipitation_probability_max"][1] hourly = data["hourly"] times = hourly["time"] idx_morning = next((i for i, t in enumerate(times) if t.endswith("T09:00")), 9) idx_afternoon = next((i for i, t in enumerate(times) if t.endswith("T15:00")), 15) idx_evening = next((i for i, t in enumerate(times) if t.endswith("T21:00")), 21) idx_night = next((i for i, t in enumerate(times) if t.endswith("T03:00") and i > 20), 27) def fmt_hour(idx, label): temp = hourly["temperature_2m"][idx] prob = hourly["precipitation_probability"][idx] w_code = hourly["weathercode"][idx] emoji = cls.get_emoji(w_code) return f"▪️ {label}: {emoji} {temp:+.1f}°C (дождь: {prob}%)" weather_code_today = data["daily"]["weathercode"][0] general_emoji = cls.get_emoji(weather_code_today) general_desc = cls.get_desc(weather_code_today) msg = [ f"🌦️ Прогноз погоды в Ульяновске на сегодня:", f"Погода: {general_emoji} {general_desc}", f"Температура: от {temp_min_today:+.1f}°C до {temp_max_today:+.1f}°C", "", "🕒 Детали по часам на сегодня:", fmt_hour(idx_morning, "Утро (09:00)"), fmt_hour(idx_afternoon, "День (15:00)"), fmt_hour(idx_evening, "Вечер (21:00)"), fmt_hour(idx_night, "Ночь (03:00)"), "", f"☔ Макс. вероятность осадков:", f"• Сегодня: {rain_prob_today}%", f"• Завтра: {rain_prob_tomorrow}%" ] return "\n".join(msg) @classmethod async def get_3day_forecast(cls) -> str: """Прогноз на 3 дня (для понедельника или по запросу)""" url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={LAT}&longitude={LON}&" f"daily=weathercode,temperature_2m_max,temperature_2m_min,precipitation_probability_max&" f"timezone={TIMEZONE}&forecast_days=3" ) data = await cls._fetch_raw(url) if not data: return "❌ Не удалось получить прогноз погоды. Попробуйте позже." daily = data["daily"] msg = ["🌦️ Прогноз погоды на ближайшие 3 дня:"] for i in range(3): date_str = daily["time"][i] date_obj = datetime.strptime(date_str, "%Y-%m-%d") ru_days = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"] day_name = ru_days[date_obj.weekday()] formatted_date = f"{date_obj.strftime('%d.%m')} ({day_name})" w_code = daily["weathercode"][i] emoji = cls.get_emoji(w_code) desc = cls.get_desc(w_code) t_max = daily["temperature_2m_max"][i] t_min = daily["temperature_2m_min"][i] prob = daily["precipitation_probability_max"][i] msg.append( f"📅 {formatted_date}: {emoji} {desc}\n" f" 🌡️ от {t_min:+.1f}°C до {t_max:+.1f}°C | ☔ Дождь: {prob}%" ) return "\n\n".join(msg) @classmethod async def get_7day_forecast(cls) -> str: """Прогноз на неделю""" url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={LAT}&longitude={LON}&" f"daily=weathercode,temperature_2m_max,temperature_2m_min,precipitation_probability_max&" f"timezone={TIMEZONE}&forecast_days=7" ) data = await cls._fetch_raw(url) if not data: return "❌ Не удалось получить прогноз погоды. Попробуйте позже." daily = data["daily"] msg = ["📅 Прогноз погоды на неделю:"] for i in range(7): date_str = daily["time"][i] date_obj = datetime.strptime(date_str, "%Y-%m-%d") ru_days = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"] day_name = ru_days[date_obj.weekday()] formatted_date = f"{date_obj.strftime('%d.%m')} ({day_name})" w_code = daily["weathercode"][i] emoji = cls.get_emoji(w_code) t_max = daily["temperature_2m_max"][i] t_min = daily["temperature_2m_min"][i] prob = daily["precipitation_probability_max"][i] msg.append( f"• {formatted_date}: {emoji} {t_min:+.0f}°..{t_max:+.0f}°C | ☔ {prob}%" ) return "\n".join(msg) @classmethod async def get_weekend_forecast_cycling(cls) -> str: """Прогноз на выходные (Сб-Вс) с фокусом на велосипедные поездки""" url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={LAT}&longitude={LON}&" f"daily=weathercode,temperature_2m_max,temperature_2m_min,precipitation_probability_max,windspeed_10m_max&" f"timezone={TIMEZONE}&forecast_days=7" ) data = await cls._fetch_raw(url) if not data: return "❌ Не удалось получить прогноз погоды для велоквеста." daily = data["daily"] times = daily["time"] idx_sat = None idx_sun = None for i, t_str in enumerate(times): dt = datetime.strptime(t_str, "%Y-%m-%d") if dt.weekday() == 5: # Суббота idx_sat = i elif dt.weekday() == 6: # Воскресенье idx_sun = i if idx_sat is None or idx_sun is None: return "❌ Не удалось определить выходные дни в прогнозе." quote = random.choice(CYCLING_QUOTES) def fmt_day(idx, label): w_code = daily["weathercode"][idx] emoji = cls.get_emoji(w_code) desc = cls.get_desc(w_code) t_max = daily["temperature_2m_max"][idx] t_min = daily["temperature_2m_min"][idx] prob = daily["precipitation_probability_max"][idx] wind = daily["windspeed_10m_max"][idx] if prob > 50 or w_code in [65, 82, 95]: cycling_verdict = "⚠️ Неблагоприятно (высокий риск ливня)" elif wind > 25: cycling_verdict = "💨 Сильный ветер! Будет тяжело крутить" elif t_max < 10: cycling_verdict = "🥶 Холодновато, одевайтесь теплее" else: cycling_verdict = "✅ Отличная погода для катания!" return ( f"📅 {label}: {emoji} {desc}\n" f" 🌡️ Температура: {t_min:+.1f}°C..{t_max:+.1f}°C\n" f" ☔ Вероятность осадков: {prob}% | 💨 Ветер до: {wind:.1f} км/ч\n" f" 👉 Вердикт: {cycling_verdict}" ) msg = [ "🚴 Прогноз на выходные (для покатушек на велике):", fmt_day(idx_sat, f"Суббота ({datetime.strptime(times[idx_sat], '%Y-%m-%d').strftime('%d.%m')})"), "", fmt_day(idx_sun, f"Воскресенье ({datetime.strptime(times[idx_sun], '%Y-%m-%d').strftime('%d.%m')})"), "", "📣 Мотивация дня:", f"{quote}" ] return "\n".join(msg) @classmethod async def get_thursday_weekend_broadcast(cls) -> str: """Прогноз на четверг, пятницу и выходные (для еженедельной рассылки по четвергам)""" url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={LAT}&longitude={LON}&" f"daily=weathercode,temperature_2m_max,temperature_2m_min,precipitation_probability_max&" f"timezone={TIMEZONE}&forecast_days=4" # Чт, Пт, Сб, Вс ) data = await cls._fetch_raw(url) if not data: return "❌ Не удалось получить прогноз погоды на выходные." daily = data["daily"] msg = ["🌦️ Прогноз погоды на остаток недели и выходные (Чт - Вс):"] for i in range(4): date_str = daily["time"][i] date_obj = datetime.strptime(date_str, "%Y-%m-%d") ru_days = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"] day_name = ru_days[date_obj.weekday()] formatted_date = f"{date_obj.strftime('%d.%m')} ({day_name})" w_code = daily["weathercode"][i] emoji = cls.get_emoji(w_code) desc = cls.get_desc(w_code) t_max = daily["temperature_2m_max"][i] t_min = daily["temperature_2m_min"][i] prob = daily["precipitation_probability_max"][i] msg.append( f"📅 {formatted_date}: {emoji} {desc}\n" f" 🌡️ от {t_min:+.1f}°C до {t_max:+.1f}°C | ☔ Дождь: {prob}%" ) return "\n\n".join(msg)