Пишем Telegram бот текущей погоды по IP адресу на Python aiogram

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

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

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

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

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

{
  "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)

Парсим ответ 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.

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'

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)

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

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

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

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

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

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

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

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

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