В этой серии статей мы напишем телеграм бота на python. Он работает с внешним API, запрашивает результаты футбольных матчей и выводить их в сообщении.
Когда локальная версия будет готова, разместим бота на сервере. Вместо Heroku, я выбрал отдельную виртуальную машину, что бы бот не засыпал. Это ближе к реальности.
Вся разработка разбита на этапы:
- Локальная установка библиотек и Redis.
- Регистрация и получение токена.
- Настройка , подключение к базам данных.
- Написание основной функциональности бота.
- Регистрации, выбор и настройка внешнего апи футбольных матчей.
- Добавление сбора результатов матчей и интеграция в бота.
- Деплой, публикация на сервере:
- Регистрация дешевого или бесплатного VPS.
- Запуск Редис-клиента.
- Запуск и настройка бота на сервере.
Рабочая версия бота запущена в телеграме до конца февраля @FonlineBOT. Бот отключен.
Вводные данные
Материал рассчитан на уровень Начинающий+, нужно понимать как работают классы и функции, знать основы базы данных и async/await. Если знаний мало, крайне желательно писать код в Pycharm, бесплатная версия подходит.
Используйте указанные версии библиотек, что бы проект работал без изменений. При установке иных версий вы можете получать ошибки, связанные с совместимостью.
Версия Python - 3.8+
aiogram==2.11.2
emoji==1.1.0
redis==3.5.3
ujson==4.0.1
uvloop==0.14.0 # не работает и не требуется на WindowsРепозиторий с кодом этой для этой части бота:
https://gitlab.com/PythonRu/fonlinebot/-/tree/master/first_step
Локальная установка библиотек для бота и Redis
Для начала нужно создать проект «fonlinebot» с виртуальным окружение. В Pycharm это делается так:
Затем установить библиотеки в виртуальном окружении. Сразу понадобятся 4: для бота, работы с redis, ускорения и emoji в сообщениях.
pip install aiogram==2.11.2 redis==3.5.3 ujson==4.0.1 emoji==1.1.0Установка Redis локально
Redis — это резидентная база данных (такая, которая хранит записи прямо в оперативной памяти) в виде пар ключ-значение. Чтение и запись в память происходит намного быстрее, чем в случае с дисками, поэтому такой подход отлично подходит для хранения второстепенных данных.
Из недавней статьи — Redis для приложений на Python
Для установки Redis на Linux/Mac следуйте этим инструкциям: https://redis.io/download#from-source-code. Для запуска достаточно ввести src/redis-server.
Что бы установить на Windows скачайте и распакуйте архив отсюда. Для запуска откройте «redis-server.exe».
Теперь нужно убедиться, что все работает. Создайте файл «main.py» в корне проекта и выполните этот код:
# fonlinebot/main.py
import redis
r = redis.StrictRedis()
print(r.ping())
Вывод будет True, в другом случае ошибка.
Регистрация бота и получение токена
Для регистрации напишем https://t.me/botfather команду /newbot. Далее он просит ввести имя и адрес бота. Если данные корректны, выдает токен. Учтите, что адрес должен быть уникальным, нельзя использовать «fonlinebot» снова.
На время разработки сохраним токен в файл. Создайте «config.py» в папке проекта для хранения настроек и запишите токен TOKEN = "ВАШ ТОКЕН"
Настройка бота
Теперь нужно связать бота с redis и базой данных, проверить работоспособность.
Создадим необходимые модули и файлы. В папке «fonlinebot» к созданным ранее «main.py» и «config.py» добавим: «database.py», «requirements.txt» и папку «app». В папку «app» добавьте: «bot.py», «dialogs.py», «service.py». Вот такая структура получится:
Разделив бот на модули, его удобнее поддерживать и дорабатывать.
- «main.py» — для запуска бота.
- «config.py» — хранит настройки, ключи доступов и другую статическую информацию.
- «database.py» — для работы с базой данных и кешем(redis).
- «requirements.txt» — хранит зависимости проекта, для запуска на сервере.
- «app» — папка самого бота.
- «bot.py» — для взаимодействия бота с юзерами, ответы на сообщения.
- «dialogs.py» — все текстовые ответы бота.
- «service.py» — бизнес логика, получение и обработка данных о матчах.
Пришло время перейти к программированию. Запишем в «requirements.txt» наши зависимости:
aiogram==2.11.2
emoji==1.1.0
redis==3.5.3
ujson==4.0.1
uvloop==0.14.0Так как большая часть программирует на Windows, uvloop мы не устанавливали локально. Установим его на сервере.
В «config.py» к токену добавим данные бота и подключения к redis.
# fonlinebot/config.py
import ujson
import logging
logging.basicConfig(level=logging.INFO)
TOKEN = "здесь должен быть токен"
BOT_VERSION = 0.1
# База данных хранит выбранные юзером лиги
BOT_DB_NAME = "users_leagues"
# Тестовые данные поддерживаемых лиг
BOT_LEAGUES = {
"1": "Бундеслига",
"2": "Серия А",
"3": "Ла Лига",
"4": "Турецкая Суперлига",
"5": "Чемпионат Нидерландов",
"6": "Про-лига Бельгии",
"7": "Английская Премьер-лига",
"8": "Лига 1",
}
# Флаги для сообщений, emoji-код
BOT_LEAGUE_FLAGS = {
"1": ":Germany:",
"2": ":Italy:",
"3": ":Spain:",
"4": ":Turkey:",
"5": ":Netherlands:",
"6": ":Belgium:",
"7": ":England:",
"8": ":France:",
}
# Данные redis-клиента
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
# По умолчанию пароля нет. Он будет на сервере
REDIS_PASSWORD = None
Информацию о лигах в будущем можно будет вынести в отдельный json файл. Эта версия бота будет поддерживать не более 10 вариантов, я явно их записал.
Добавление базы данных
Теперь добавим классы для работы с базой данных sqlite и redis. База данных нужна для сохранения предпочтений по лигам юзеров.
Юзер будет выбирать 3 чемпионата для отслеживания, бот сохранит их в БД и использует для запроса результатов.
Кеш(redis) будет сохранять результаты матчей, что бы уменьшить количество запросов к API и ускорить время ответов. Как правило, бесплатные API лимитирует запросы.
# fonlinebot/database.py
import os
import logging
import sqlite3
import redis
import ujson
import config
# класс наследуется от redis.StrictRedis
class Cache(redis.StrictRedis):
def __init__(self, host, port, password,
charset="utf-8",
decode_responses=True):
super(Cache, self).__init__(host, port,
password=password,
charset=charset,
decode_responses=decode_responses)
logging.info("Redis start")
def jset(self, name, value, ex=0):
"""функция конвертирует python-объект в Json и сохранит"""
r = self.get(name)
if r is None:
return r
return ujson.loads(r)
def jget(self, name):
"""функция возвращает Json и конвертирует в python-объект"""
return ujson.loads(self.get(name))
Класс Cache наследуется от StrictRedis. Мы добавляем 2 метода jset, jget для сохранения списков и словарей python в хранилище redis. Изначально он не работает с ними.
Теперь добавим класс, который будет создавать базы данных и выполнять функции CRUD.
# fonlinebot/database.py
#...
class Database:
""" Класс работы с базой данных """
def __init__(self, name):
self.name = name
self._conn = self.connection()
logging.info("Database connection established")
def create_db(self):
connection = sqlite3.connect(f"{self.name}.db")
logging.info("Database created")
cursor = connection.cursor()
cursor.execute('''CREATE TABLE users
(id INTEGER PRIMARY KEY,
leagues VARCHAR NOT NULL);''')
connection.commit()
cursor.close()
def connection(self):
db_path = os.path.join(os.getcwd(), f"{self.name}.db")
if not os.path.exists(db_path):
self.create_db()
return sqlite3.connect(f"{self.name}.db")
def _execute_query(self, query, select=False):
cursor = self._conn.cursor()
cursor.execute(query)
if select:
records = cursor.fetchone()
cursor.close()
return records
else:
self._conn.commit()
cursor.close()
async def insert_users(self, user_id: int, leagues: str):
insert_query = f"""INSERT INTO users (id, leagues)
VALUES ({user_id}, "{leagues}")"""
self._execute_query(insert_query)
logging.info(f"Leagues for user {user_id} added")
async def select_users(self, user_id: int):
select_query = f"""SELECT leagues from leagues
where id = {user_id}"""
record = self._execute_query(select_query, select=True)
return record
async def update_users(self, user_id: int, leagues: str):
update_query = f"""Update leagues
set leagues = "{leagues}" where id = {user_id}"""
self._execute_query(update_query)
logging.info(f"Leagues for user {user_id} updated")
async def delete_users(self, user_id: int):
delete_query = f"""DELETE FROM users WHERE id = {user_id}"""
self._execute_query(delete_query)
logging.info(f"User {user_id} deleted")
Sqlite подходит для тестовых проектов. В будущем потребуется переход на внешнюю базу данных и асинхронная работа. Что бы не переписывать всю логику работы с базой, я сразу добавил асинхронный синтаксис.
Файл базы данных будет создаваться один раз, автоматически. Теперь нужно создать экземпляры классов:
# fonlinebot/database.py
#...
# создание объектов cache и database
cache = Cache(
host=config.REDIS_HOST,
port=config.REDIS_PORT,
password=config.REDIS_PASSWORD
)
database = Database(config.BOT_DB_NAME)
Добавление текстовых сообщений
Для шаблонов сообщений создадим неизменяемый dataclass. Здесь будут все текстовые ответы. А dataclass удобно использовать при вызове аргументов.
# fonlinebot/app/dialogs.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Messages:
test: str = "Привет {name}. Работаю..."
msg = Messages()
Создание бота
В файле «bot.py» создадим бота. Импортируем зависимости, создадим объект бота и первое сообщение.
# fonlinebot/app/bot.py
from aiogram import Bot, types
from aiogram.contrib.middlewares.logging import LoggingMiddleware
from aiogram.dispatcher import Dispatcher
from config import TOKEN
from app.dialogs import msg
from database import database as db
# стандартный код создания бота
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
dp.middleware.setup(LoggingMiddleware())
@dp.message_handler()
async def test_message(message: types.Message):
# имя юзера из настроек Телеграма
user_name = message.from_user.first_name
await message.answer(msg.test.format(name=user_name))
Функция test_message принимает любое сообщение и отвечает на него. Кроме этого, нужно закрывать соединение с базой данных перед завершением работы. Добавим в конец файла on_shutdown.
# fonlinebot/app/bot.py
# ...
async def on_shutdown(dp):
logging.warning('Shutting down..')
# закрытие соединения с БД
db._conn.close()
logging.warning("DB Connection closed")
Первый запуск бота
Для запуска используем файл «main.py». Импортируем бота и настроим пулинг:
# fonlinebot/main.py
from aiogram import executor
from app import bot
executor.start_polling(bot.dp,
skip_updates=True,
on_shutdown=bot.on_shutdown)
Теперь запустим проект. Если вы в Pycharm, откройте вкладку «Terminal» и введите python main.py. Что бы запустить бота без Pycharm:
- перейдите в папку проекта,
- активируйте виртуальное окружение,
- запустите скрипт (
python main.py).
Теперь откроем телеграм и проверим.
Отлично, работает! Теперь самое время написать несколько базовых тестов, что бы избежать проблем в будущем.
Тестирование бота
В процессе написания проекта код будем изменяться. Что бы избежать появления новых ошибок после размещения на сервере, нужно сразу написать несколько тестов.
Добавьте модуль «test.py» в папку fonlinebot. На этом этапе достаточно 4 теста, вы можете добавить свои.
# fonlinebot/test.py
import unittest
import aiohttp
from unittest import IsolatedAsyncioTestCase
from database import cache, database
from app import bot
class TestDatabase(IsolatedAsyncioTestCase):
async def test_crud(self):
await database.insert_users(1111, "1 2 3")
self.assertEqual(await database.select_users(1111), ('1 2 3',))
await database.delete_users(1111)
self.assertEqual(await database.select_users(1111), None)
class TestCache(unittest.TestCase):
def test_connection(self):
self.assertTrue(cache.ping())
def test_response_type(self):
cache.setex("test_type", 10, "Hello")
response = cache.get("test_type")
self.assertEqual(type(response), str)
class TestBot(IsolatedAsyncioTestCase):
async def test_bot_auth(self):
bot.bot._session = aiohttp.ClientSession()
bot_info = await bot.bot.get_me()
await bot.bot._session.close()
self.assertEqual(bot_info["username"], "FonlineBOT")
if __name__ == '__main__':
unittest.main()
Мы проверим CRUD функции базы данных, запишем тестовые данные и удалим. Проверим соединение с redis и ботом.
Запуск теста такой же как и бота, изменяется только имя файла python main.py.
На скриншоте видно уведомление об ошибке после завершения тестирования. Это известная проблема aiohttp на windows, можно игнорировать.
Ошибки, которые можно встретить
aiogram.utils.exceptions.Unauthorized: Unauthorized— неверный токен бота. Токен нужно сохранить как строку, его структура «цифры:буквы-и-цифры», проверьте.redis.exceptions.ConnectionError: Error 10061 connecting to ...— redis-server не запущен.sqlite3.IntegrityError: UNIQUE constraint failed: ...— вы пытаетесь добавить значение в базу данных, которое уже существует.
На этом подготовка проекта окончена. Переходите ко второй части: Написание ядра бота.
