Руководство по aiogram

Время на прочтение
11 мин

Количество просмотров 47K

Штош. В этой статье я расскажу вам, как создать Telegram бота, который получает текущую погоду по IP адресу. Мы будем использовать язык Python и асинхронную библиотеку для взаимодействия с Telegram Bot API — aiogram.

Итак, как же вы можете создать такого бота?

TL;DR

Склонируйте репозиторий shtosh-weather-bot и пройдите по инструкции в README.

Выбираем погодный сервис с бесплатным API

Данные о текущей погоде нам нужно откуда-то брать. Еще желательно, чтобы это было бесплатно. У сайта 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 и видим здесь тот самый ключ. Можете взять мой, мне не жалко.

Итак, давайте сформируем запрос. Я выбрал координаты Нью-Йорка, просто потому что хочу и могу. Не забудем добавить флаг units со значением metric, чтобы температура показывалась в градусах Цельсия. По умолчанию дается в Кельвинах.

https://api.openweathermap.org/data/2.5/weather?lat=40,7143&lon=-74,006&appid=8537d9ef6386cb97156fd47d832f479c&units=metric

Вот такой json мы получаем.

{
  "coord": {
    "lon": -74.006,
    "lat": 40.7128
  },
  "weather": [
    {
      "id": 802,
      "main": "Clouds",
      "description": "scattered clouds",
      "icon": "03d"
    }
  ],
  "base": "stations",
  "main": {
    "temp": 31.36,
    "feels_like": 31.23,
    "temp_min": 28.86,
    "temp_max": 33.94,
    "pressure": 1013,
    "humidity": 39
  },
  "visibility": 10000,
  "wind": {
    "speed": 5.14,
    "deg": 290
  },
  "clouds": {
    "all": 40
  },
  "dt": 1661375748,
  "sys": {
    "type": 2,
    "id": 2039034,
    "country": "US",
    "sunrise": 1661336121,
    "sunset": 1661384510
  },
  "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

config.py

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)

coordinates.py

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)

Парсим ответ OpenWeather API

Далее рассмотрим модуль 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.

api_service.py

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 тоже не обновляются каждую секунду.

messages.py

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'

Inline клавиатура

Можно было сделать 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)

inline_keyboard.py

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)

bot.py

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 из Германии.

Бывают такие случаи, когда запрос долго обрабатывается. Я не обрабатывал ошибки и не делал для них сообщений, бот просто ничего не делает в таких случаях. Я посчитал, что уже и так хорошо. Как говорится:

• Лучшее — враг хорошего

• Работает — не трогай

• Еще сотня фраз для оправдания лени

• Еще тысяча успокаивающих фраз для перфекционистов

Также можно реализовать получение координат через отправление геолокации боту, тогда получится в разы точнее.

Штош. Спасибо за прочтение. Надеюсь на отзывы, комментарии и критику.

GitHub репозиторий shtosh-weather-bot

python --version

sudo pip3 install virtualenv

mkdir telegram_bot
cd telegram_bot

virtualenv venv -p python3

source venv/bin/activate

pip install aiogram

from aiogram import Bot, Dispatcher, executor, types

API_TOKEN = '5602787567:AAGYv7NrSjwyW7qPs_yvu70C060zrcfZDbQ' #В одинарных кавычках размещаем токен, полученный от @BotFather.

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start']) #Явно указываем в декораторе, на какую команду реагируем. 
async def send_welcome(message: types.Message):
   await message.reply("Привет!nЯ Эхо-бот от Skillbox!nОтправь мне любое сообщение, а я тебе обязательно отвечу.") #Так как код работает асинхронно, то обязательно пишем await.

@dp.message_handler() #Создаём новое событие, которое запускается в ответ на любой текст, введённый пользователем.
async def echo(message: types.Message): #Создаём функцию с простой задачей — отправить обратно тот же текст, что ввёл пользователь.
   await message.answer(message.text)

if __name__ == '__main__':
   executor.start_polling(dp, skip_updates=True)

from aiogram import Bot, Dispatcher, executor, types
 
API_TOKEN = '5602787567:AAGYv7NrSjwyW7qPs_yvu70C060zrcfZDbQ'
 
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
 
@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
   await message.reply("Привет!nЯ Эхо-бот от Skillbox!nОтправь мне любое сообщение, а я тебе обязательно отвечу.")
 
@dp.message_handler()
async def echo(message: types.Message):
   await message.answer(message.text)
 
if __name__ == '__main__':
   executor.start_polling(dp, skip_updates=True)

python3 main.py

Научитесь: Профессия Python-разработчик
Узнать больше