Пишем Telegram бот текущей погоды по IP адресу на Python aiogram
- вторник, 23 августа 2022 г. в 00:54:54
Штош. В этой статье я расскажу вам, как создать Telegram бота, который получает текущую погоду по IP адресу. Мы будем использовать язык Python и асинхронную библиотеку для взаимодействия с Telegram Bot API - aiogram.
Итак, как же вы можете создать такого бота?
Склонируйте репозиторий shtosh-weather-bot и пройдите по инструкции в README.
Данные о текущей погоде нам нужно откуда-то брать. Еще желательно, чтобы это было бесплатно. У сайта OpenWeatherMap есть нужный нам API текущей погоды. Бесплатно можно посылать 1000 запросов в день.
https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={API key}
Кстати, если вы ищете какой-то application user interface для своего проекта, рекомендую репозиторий public-apis.
Итак, для запроса нужны координаты и специальный ключ, который можно получить, зарегистрировав аккаунт. Ну это вообще не проблема, можно зарегать на временную почту. Конечно, если вы собираетесь серьезно использовать API и покупать больше 1000 запросов в день, лучше регистрировать аккаунт на свою почту. Капитан очевидность.
Заходим в My API keys и видим здесь тот самый ключ. Можете взять мой, мне не жалко.
Итак, давайте сформируем запрос. Я выбрал координаты Нью-Йорка, просто потому что хочу и могу.
{
"coord": {
"lon": -74.006,
"lat": 40.7143
},
"weather": [
{
"id": 501,
"main": "Rain",
"description": "moderate rain",
"icon": "10d"
}
],
"base": "stations",
"main": {
"temp": 297.11,
"feels_like": 297.75,
"temp_min": 295.11,
"temp_max": 298.79,
"pressure": 1013,
"humidity": 84
},
"visibility": 10000,
"wind": {
"speed": 5.81,
"deg": 123,
"gust": 6.71
},
"rain": {
"1h": 2.83
},
"clouds": {
"all": 100
},
"dt": 1661183439,
"sys": {
"type": 2,
"id": 2039034,
"country": "US",
"sunrise": 1661163203,
"sunset": 1661211890
},
"timezone": -14400,
"id": 5128581,
"name": "New York",
"cod": 200
}
Создайте Telegram бота с помощью BotFather и возьмите его токен.
Из названия видео вы могли догадаться, что мы будем использовать язык Python и библиотеку aiogram. Я надеюсь, с установкой Python у вас не возникнет проблем. С aiogram тоже.
pip install aiogram
Я много позаимствовал у проекта Алексея Голобурдина - автора YouTube канала "Диджитализируй!" Проблема в том, что его проект предназначен только для macOS устройств, потому что координаты берутся с помощью инструмента командной строки whereami. Пример вывода:
Latitude: 45.424807,
Longitude: -75.699234
Accuracy (m): 65.000000
Timestamp: 2019-09-28, 12:40:20 PM EDT
Также его скрипт просто выводит всю форматированную информацию в терминал, хотелось бы иметь интерфейс поприятнее и удобнее.
Я решил, что можно доработать идею и охватить максимальное количество пользователей, пренебрегая точностью информации.
Итак, файл config.py
содержит константы:
Токен бота BOT_API_TOKEN
Ключ OpenWeather WEATHER_API_KEY
Запрос текущей погоды CURRENT_WEATHER_API_CALL
BOT_API_TOKEN = ''
WEATHER_API_KEY = ''
CURRENT_WEATHER_API_CALL = (
'https://api.openweathermap.org/data/2.5/weather?'
'lat={latitude}&lon={longitude}&'
'appid=' + WEATHER_API_KEY + '&units=metric'
)
Конечно, такие данные, как токены и ключи нужно хранить в переменных окружения, но это пет-проект, деплоить я его не буду, поэтому особо не заморачиваюсь.
Для получения координат я создал отдельный модуль. Датакласс Coordinates
содержит широту и долготу с типами float.
from dataclasses import dataclass
@dataclass(slots=True, frozen=True)
class Coordinates:
latitude: float
longitude: float
По IP адресу их можно найти с помощью ipinfo.io/json. Получается вот такой ответ.
{
"ip": "228.228.228.228",
"city": "Moscow",
"region": "Moscow",
"country": "RU",
"loc": "55.7522,37.6156",
"org": "Starlink",
"postal": "101000",
"timezone": "Europe/Moscow",
"readme": "https://ipinfo.io/missingauth"
}
Нас интересует ключ "loc"
сокращенно от location. Опять капитан очевидность. Делаем запрос с помощью функции urlopen
модуля request
библиотеки urllib.
Возвращаем словарь с помощью json.load()
from urllib.request import urlopen
import json
def _get_ip_data() -> dict:
url = 'http://ipinfo.io/json'
response = urlopen(url)
return json.load(response)
В функции получения координат парсим этот словарь и возвращаем датакласс координат.
def get_coordinates() -> Coordinates:
"""Returns current coordinates using IP address"""
data = _get_ip_data()
latitude = data['loc'].split(',')[0]
longitude = data['loc'].split(',')[1]
return Coordinates(latitude=latitude, longitude=longitude)
from urllib.request import urlopen
from dataclasses import dataclass
import json
@dataclass(slots=True, frozen=True)
class Coordinates:
latitude: float
longitude: float
def get_coordinates() -> Coordinates:
"""Returns current coordinates using IP address"""
data = _get_ip_data()
latitude = data['loc'].split(',')[0]
longitude = data['loc'].split(',')[1]
return Coordinates(latitude=latitude, longitude=longitude)
def _get_ip_data() -> dict:
url = 'http://ipinfo.io/json'
response = urlopen(url)
return json.load(response)
Далее рассмотрим модуль api_service.
В нем происходит вся суета с погодой. Температура измеряется в градусах Цельсия, чему соответствует псевдоним float
числа.
from typing import TypeAlias
Celsius: TypeAlias = float
Как известно, градусы Фаренгейта были созданы только для того, чтобы Рэй Брэдбери смог красиво назвать свою антиутопию.
В ответе API направление ветра дается в градусах. Я решил привести их в более удобный формат. Для этого я создал перечисление основных направлений ветра.
from enum import IntEnum
class WindDirection(IntEnum):
North = 0
Northeast = 45
East = 90
Southeast = 135
South = 180
Southwest = 225
West = 270
Northwest = 315
В функции парсинга округление по 45 градусов выглядит таким образом: делим градусы на 45, округляем и умножаем обратно на 45. Результат может округлиться до 360 градусов, поэтому обрабатываем этот случай.
def _parse_wind_direction(openweather_dict: dict) -> str:
degrees = openweather_dict['wind']['deg']
degrees = round(degrees / 45) * 45
if degrees == 360:
degrees = 0
return WindDirection(degrees).name
Все данные о погоде будут храниться в датаклассе. При желании вы можете добавить сюда остальную информацию из ответа OpenWeather, например атмосферное давление, часовой пояс, минимальную и максимальную зафиксированную в данный момент температуру.
@dataclass(slots=True, frozen=True)
class Weather:
location: str
temperature: Celsius
temperature_feeling: Celsius
description: str
wind_speed: float
wind_direction: str
sunrise: datetime
sunset: datetime
В остальном ничего интересного в модуле не происходит, просто парсинг json.
from typing import Literal, TypeAlias
from urllib.request import urlopen
from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum
import json
from coordinates import Coordinates
import config
Celsius: TypeAlias = float
class WindDirection(IntEnum):
North = 0
Northeast = 45
East = 90
Southeast = 135
South = 180
Southwest = 225
West = 270
Northwest = 315
@dataclass(slots=True, frozen=True)
class Weather:
location: str
temperature: Celsius
temperature_feeling: Celsius
description: str
wind_speed: float
wind_direction: str
sunrise: datetime
sunset: datetime
def get_weather(coordinates=Coordinates) -> Weather:
"""Requests the weather in OpenWeather API and returns it"""
openweather_response = _get_openweather_response(
longitude=coordinates.longitude, latitude=coordinates.latitude
)
weather = _parse_openweather_response(openweather_response)
return weather
def _get_openweather_response(latitude: float, longitude: float) -> str:
url = config.CURRENT_WEATHER_API_CALL.format(latitude=latitude, longitude=longitude)
return urlopen(url).read()
def _parse_openweather_response(openweather_response: str) -> Weather:
openweather_dict = json.loads(openweather_response)
return Weather(
location=_parse_location(openweather_dict),
temperature=_parse_temperature(openweather_dict),
temperature_feeling=_parse_temperature_feeling(openweather_dict),
description=_parse_description(openweather_dict),
sunrise=_parse_sun_time(openweather_dict, 'sunrise'),
sunset=_parse_sun_time(openweather_dict, 'sunset'),
wind_speed=_parse_wind_speed(openweather_dict),
wind_direction=_parse_wind_direction(openweather_dict)
)
def _parse_location(openweather_dict: dict) -> str:
return openweather_dict['name']
def _parse_temperature(openweather_dict: dict) -> Celsius:
return openweather_dict['main']['temp']
def _parse_temperature_feeling(openweather_dict: dict) -> Celsius:
return openweather_dict['main']['feels_like']
def _parse_description(openweather_dict) -> str:
return str(openweather_dict['weather'][0]['description']).capitalize()
def _parse_sun_time(openweather_dict: dict, time: Literal["sunrise", "sunset"]) -> datetime:
return datetime.fromtimestamp(openweather_dict['sys'][time])
def _parse_wind_speed(openweather_dict: dict) -> float:
return openweather_dict['wind']['speed']
def _parse_wind_direction(openweather_dict: dict) -> str:
degrees = openweather_dict['wind']['deg']
degrees = round(degrees / 45) * 45
if degrees == 360:
degrees = 0
return WindDirection(degrees).name
В модуле messages собраны сообщения для бота по командам. Сообщение о погоде /weather
содержит локацию, описание погоды, температуру и ее ощущение.
from coordinates import get_coordinates
from api_service import get_weather
def weather() -> str:
"""Returns a message about the temperature and weather description"""
wthr = get_weather(get_coordinates())
return f'{wthr.location}, {wthr.description}\n' \
f'Temperature is {wthr.temperature}°C, feels like {wthr.temperature_feeling}°C'
Сообщение о ветре /wind
показывает его направление и скорость в метрах в секунду.
def wind() -> str:
"""Returns a message about wind direction and speed"""
wthr = get_weather(get_coordinates())
return f'{wthr.wind_direction} wind {wthr.wind_speed} m/s'
Ну и сообщение о времени восхода и заката солнца /sun_time
. Здесь datetime объект форматируется в часы и минуты, остальное в данном случае неважно.
def sun_time() -> str:
"""Returns a message about the time of sunrise and sunset"""
wthr = get_weather(get_coordinates())
return f'Sunrise: {wthr.sunrise.strftime("%H:%M")}\n' \
f'Sunset: {wthr.sunset.strftime("%H:%M")}\n'
Нужно заметить, что при каждом вызове функции создается новый API запрос. Почему это нужно заметить? Потому что сначала я сделал бота с одним запросом и недоумевал, почему информация не меняется через время. Потому что в идеале делать один запрос в 5 или 10 минут, за это время погода не особо меняется, да и данные OpenWeather тоже не обновляются каждую секунду.
from coordinates import get_coordinates
from api_service import get_weather
def weather() -> str:
"""Returns a message about the temperature and weather description"""
wthr = get_weather(get_coordinates())
return f'{wthr.location}, {wthr.description}\n' \
f'Temperature is {wthr.temperature}°C, feels like {wthr.temperature_feeling}°C'
def wind() -> str:
"""Returns a message about wind direction and speed"""
wthr = get_weather(get_coordinates())
return f'{wthr.wind_direction} wind {wthr.wind_speed} m/s'
def sun_time() -> str:
"""Returns a message about the time of sunrise and sunset"""
wthr = get_weather(get_coordinates())
return f'Sunrise: {wthr.sunrise.strftime("%H:%M")}\n' \
f'Sunset: {wthr.sunset.strftime("%H:%M")}\n'
Можно было сделать reply клавиатуру, но мне больше по душе Inline. 3 кнопки для 3 команд.
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
BTN_WEATHER = InlineKeyboardButton('Weather', callback_data='weather')
BTN_WIND = InlineKeyboardButton('Wind', callback_data='wind')
BTN_SUN_TIME = InlineKeyboardButton('Sunrise and sunset', callback_data='sun_time')
4 клавиатуры для 4 команд, добавляется команда помощи. В чем суть? После сообщения погоды нам не нужно показывать ее кнопку. Такая же логика для всех других команд, кроме помощи. Для нее выводятся кнопки всех 3 команд.
WEATHER = InlineKeyboardMarkup().add(BTN_WIND, BTN_SUN_TIME)
WIND = InlineKeyboardMarkup().add(BTN_WEATHER).add(BTN_SUN_TIME)
SUN_TIME = InlineKeyboardMarkup().add(BTN_WEATHER, BTN_WIND)
HELP = InlineKeyboardMarkup().add(BTN_WEATHER, BTN_WIND).add(BTN_SUN_TIME)
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
BTN_WEATHER = InlineKeyboardButton('Weather', callback_data='weather')
BTN_WIND = InlineKeyboardButton('Wind', callback_data='wind')
BTN_SUN_TIME = InlineKeyboardButton('Sunrise and sunset',
callback_data='sun_time')
WEATHER = InlineKeyboardMarkup().add(BTN_WIND, BTN_SUN_TIME)
WIND = InlineKeyboardMarkup().add(BTN_WEATHER).add(BTN_SUN_TIME)
SUN_TIME = InlineKeyboardMarkup().add(BTN_WEATHER, BTN_WIND)
HELP = InlineKeyboardMarkup().add(BTN_WEATHER, BTN_WIND).add(BTN_SUN_TIME)
Ну и в главном модуле бота присутствует стандартная настройка, хэндлеры сообщений и коллбэков для inline кнопок, ничего сверхъестественного.
Нужно хоть что-нибудь рассказать. Под стандартной настройкой aiogram подразумевается следующий блок кода:
import logging
from aiogram import Bot, Dispatcher, executor, types
import config
logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.BOT_API_TOKEN)
dp = Dispatcher(bot)
Хэндлер для сообщений /start
и /weather
выглядит следующим образом. Все работает с помощью магии декораторов aiogram.
@dp.message_handler(commands=['start', 'weather'])
async def show_weather(message: types.Message):
await message.answer(text=messages.weather(),
reply_markup=inline_keyboard.WEATHER)
Хэндлер коллбэка для инлайн-кнопки погоды:
@dp.callback_query_handler(text='weather')
async def process_callback_weather(callback_query: types.CallbackQuery):
await bot.answer_callback_query(callback_query.id)
await bot.send_message(
callback_query.from_user.id,
text=messages.weather(),
reply_markup=inline_keyboard.WEATHER)
Запускаем скрипт с помощью такой конструкции:
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)
import logging
from aiogram import Bot, Dispatcher, executor, types
import inline_keyboard
import messages
import config
logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.BOT_API_TOKEN)
dp = Dispatcher(bot)
@dp.message_handler(commands=['start', 'weather'])
async def show_weather(message: types.Message):
await message.answer(text=messages.weather(),
reply_markup=inline_keyboard.WEATHER)
@dp.message_handler(commands='help')
async def show_help_message(message: types.Message):
await message.answer(
text=f'This bot can get the current weather from your IP address.',
reply_markup=inline_keyboard.HELP)
@dp.message_handler(commands='wind')
async def show_wind(message: types.Message):
await message.answer(text=messages.wind(),
reply_markup=inline_keyboard.WIND)
@dp.message_handler(commands='sun_time')
async def show_sun_time(message: types.Message):
await message.answer(text=messages.sun_time(),
reply_markup=inline_keyboard.SUN_TIME)
@dp.callback_query_handler(text='weather')
async def process_callback_weather(callback_query: types.CallbackQuery):
await bot.answer_callback_query(callback_query.id)
await bot.send_message(
callback_query.from_user.id,
text=messages.weather(),
reply_markup=inline_keyboard.WEATHER
)
@dp.callback_query_handler(text='wind')
async def process_callback_wind(callback_query: types.CallbackQuery):
await bot.answer_callback_query(callback_query.id)
await bot.send_message(
callback_query.from_user.id,
text=messages.wind(),
reply_markup=inline_keyboard.WIND
)
@dp.callback_query_handler(text='sun_time')
async def process_callback_sun_time(callback_query: types.CallbackQuery):
await bot.answer_callback_query(callback_query.id)
await bot.send_message(
callback_query.from_user.id,
text=messages.sun_time(),
reply_markup=inline_keyboard.SUN_TIME
)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)
Смотрим логирование, вы должны увидеть 3 сообщения:
INFO:aiogram:Bot: superultramegaweatherbot [@superultramegaweatherbot]
WARNING:aiogram:Updates were skipped successfully.
INFO:aiogram.dispatcher.dispatcher:Start polling.
Пока что все работает, давайте посмотрим по IP из Германии.
Бывают такие случаи, когда запрос долго обрабатывается. Я не обрабатывал ошибки и не делал для них сообщений, бот просто ничего не делает в таких случаях. Я посчитал, что уже и так хорошо. Как говорится:
Лучшее - враг хорошего
Работает - не трогай
Еще сотня фраз для оправдания лени
Еще тысяча успокаивающих фраз для перфекционистов
Также можно реализовать получение координат через отправление геолокации боту, тогда получится в разы точнее.
Штош. Спасибо за прочтение. Надеюсь на отзывы, комментарии и критику.