135 lines
6.2 KiB
Python
135 lines
6.2 KiB
Python
import aiohttp
|
||
from datetime import datetime
|
||
import asyncio
|
||
import logging
|
||
|
||
async def get_coords_by_city(city_name: str):
|
||
"""Ищет координаты по названию города."""
|
||
url = f"https://geocoding-api.open-meteo.com/v1/search?name={city_name}&count=1&language=ru&format=json"
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
try:
|
||
async with session.get(url, timeout=5) as response:
|
||
if response.status != 200:
|
||
return None, None, None, None
|
||
|
||
data = await response.json()
|
||
results = data.get("results")
|
||
|
||
if not results:
|
||
return None, None, None, None
|
||
|
||
city_data = results[0]
|
||
return (
|
||
city_data.get("latitude"),
|
||
city_data.get("longitude"),
|
||
city_data.get("name", city_name),
|
||
city_data.get("timezone")
|
||
)
|
||
except Exception:
|
||
return None, None, None, None
|
||
|
||
async def get_timezone_by_coords(lat: float, lon: float) -> str | None:
|
||
"""Определяет часовой пояс по координатам, делая запрос к API погоды."""
|
||
# Делаем минимальный запрос к API, чтобы получить часовой пояс
|
||
data = await fetch_weather_data(lat, lon, 1)
|
||
if data and "timezone" in data:
|
||
return data["timezone"]
|
||
return None
|
||
|
||
async def fetch_weather_data(lat: float, lon: float, days: int):
|
||
"""Базовая функция для запроса данных о погоде на N дней."""
|
||
url = (
|
||
f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}"
|
||
f"¤t=temperature_2m"
|
||
f"&daily=temperature_2m_max,temperature_2m_min,precipitation_probability_max"
|
||
f"&timezone=auto&forecast_days={days}"
|
||
)
|
||
# Логика с повторными попытками для надежности
|
||
for attempt in range(3):
|
||
async with aiohttp.ClientSession() as session:
|
||
try:
|
||
async with session.get(url, timeout=10) as response:
|
||
if response.status == 200:
|
||
return await response.json()
|
||
logging.warning(f"Попытка {attempt+1}: API погоды вернуло статус {response.status}")
|
||
except Exception as e:
|
||
logging.warning(f"Попытка {attempt+1}: Ошибка при запросе к API погоды: {e}")
|
||
|
||
if attempt < 2: # Не спим после последней попытки
|
||
await asyncio.sleep(attempt + 2) # Спим 2, 3 секунды
|
||
return None
|
||
|
||
async def get_daily_weather_summary(lat: float, lon: float) -> str:
|
||
"""Прогноз на сегодня."""
|
||
data = await fetch_weather_data(lat, lon, 1)
|
||
if not data or "daily" not in data or "current" not in data:
|
||
return "Не удалось получить данные о погоде от сервера."
|
||
|
||
current_temp = data["current"].get("temperature_2m", "N/A")
|
||
temp_max = data["daily"]["temperature_2m_max"][0]
|
||
temp_min = data["daily"]["temperature_2m_min"][0]
|
||
precip_prob = data["daily"]["precipitation_probability_max"][0]
|
||
|
||
temp_diff = round(temp_max - temp_min, 1)
|
||
|
||
return (
|
||
f"🌤 **Погода на сегодня:**\n\n"
|
||
f"🌡 **Сейчас:** {current_temp}°C\n"
|
||
f"📊 **Дневная норма:** от {temp_min}°C до {temp_max}°C\n"
|
||
f"📉 **Перепад температур за день:** {temp_diff}°C\n"
|
||
f"🌧 **Макс. вероятность осадков:** {precip_prob}%\n\n"
|
||
f"Одевайтесь по погоде и хорошего дня!"
|
||
)
|
||
|
||
async def get_tomorrow_weather_summary(lat: float, lon: float) -> str:
|
||
"""Прогноз на завтра (берем индекс [1] из массивов)."""
|
||
data = await fetch_weather_data(lat, lon, 2)
|
||
if not data or "daily" not in data:
|
||
return "Не удалось получить данные о погоде от сервера."
|
||
|
||
temp_max = data["daily"]["temperature_2m_max"][1]
|
||
temp_min = data["daily"]["temperature_2m_min"][1]
|
||
precip_prob = data["daily"]["precipitation_probability_max"][1]
|
||
|
||
temp_diff = round(temp_max - temp_min, 1)
|
||
|
||
return (
|
||
f"📅 **Погода на завтра:**\n\n"
|
||
f"📊 **Дневная норма:** от {temp_min}°C до {temp_max}°C\n"
|
||
f"📉 **Перепад температур:** {temp_diff}°C\n"
|
||
f"🌧 **Макс. вероятность осадков:** {precip_prob}%\n\n"
|
||
f"Запланируйте свой день с учетом прогноза!"
|
||
)
|
||
|
||
async def get_weekly_weather_summary(lat: float, lon: float) -> str:
|
||
"""Прогноз на 7 дней."""
|
||
data = await fetch_weather_data(lat, lon, 7)
|
||
if not data or "daily" not in data:
|
||
return "Не удалось получить данные о погоде от сервера."
|
||
|
||
dates = data["daily"]["time"]
|
||
max_temps = data["daily"]["temperature_2m_max"]
|
||
min_temps = data["daily"]["temperature_2m_min"]
|
||
precips = data["daily"]["precipitation_probability_max"]
|
||
|
||
week_days = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"]
|
||
|
||
summary = "📆 **Прогноз на неделю:**\n\n"
|
||
|
||
for i in range(7):
|
||
# Преобразуем дату "YYYY-MM-DD" в красивый формат "DD.MM (День недели)"
|
||
date_obj = datetime.strptime(dates[i], "%Y-%m-%d")
|
||
day_name = week_days[date_obj.weekday()]
|
||
formatted_date = date_obj.strftime("%d.%m")
|
||
|
||
t_min = min_temps[i]
|
||
t_max = max_temps[i]
|
||
prob = precips[i]
|
||
|
||
# Эмодзи для осадков, чтобы легче читалось
|
||
precip_emoji = "☔️" if prob > 40 else ("☁️" if prob > 10 else "☀️")
|
||
|
||
summary += f"🔹 **{formatted_date} ({day_name}):** от {t_min}°C до {t_max}°C | {precip_emoji} {prob}%\n"
|
||
|
||
return summary |