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)