Soccer Telegram bot in Python (1/4): Preparing and setting up the bot

by Alex
Soccer Telegram bot in Python (2/4): Bot functionality

In this series of articles we will write a python telegram bot. It works with an external API, queries the results of soccer matches and outputs them in a message. When the local version is ready, we will put the bot on the server. Instead of Heroku, I chose a separate virtual machine to keep the bot from falling asleep. This is closer to reality. The entire development is broken down into stages:

  1. Local installation of libraries and Redis.
  2. Registration and obtaining a token.
  3. Setting up and connecting to databases.
  4. Writing basic functionality of the bot.
  5. Registrations, selecting and configuring the external api of soccer matches.
  6. Adding the collection of match results and integration into the bot.
  7. Deploy, publish to server:
    1. Registration of a cheap or free VPS.
    2. Running the Redis-client.
    3. Running and configuring the bot on the server.

A working version of the bot is up and running on Telegram by the end of February @FonlineBOT. The bot is disabled.

Introductions

This material is for Beginner+ level, you need to understand how classes and functions work, know basics of database and async/await. If you don’t know enough, it’s highly desirable to write code in Pycharm, free version will do. Use the specified versions of libraries so that the project works without changes. If you install other versions you may get compatibility errors.

Python version - 3.8+
aiogram==2.11.2
emoji==1.1.0
redis==3.5.3
ujson==4.0.1
uvloop==0.14.0 # does not work and is not required on Windows

The repository with this code for this part of the bot: https://gitlab.com/PythonRu/fonlinebot/-/tree/master/first_step

Local installation of libraries for the bot and Redis

First you need to create a project “fonlinebot” with virtual environment. In Pycharm it is done like this: pycharm new project Then install the libraries in the virtual environment. You will need 4 at once: for the bot, for working with redis, acceleration and emoji in messages.

pip install aiogram==2.11.2 redis==3.5.3 ujson==4.0.1 emoji==1.1.0

Installing Redis locally

Redis is a resident database (the kind that stores records directly in RAM) in the form of key-value pairs. Reading and writing to memory is much faster than with disks, so this approach is great for storing secondary data. From a recent article – Redis for Python applications Follow these instructions to install Redis on Linux/Mac: https://redis.io/download#from-source-code. Just type src/redis-server to start it up. To install on Windows, download and unpack the archive from here. Open “redis-server.exe” to start. Now you need to make sure that everything works. Create a file “main.py” in the root of the project and execute this code:

# fonlinebot/main.py
import redis
r = redis.StrictRedis()
print(r.ping())

The output will be True, otherwise an error.

Registering the bot and getting a token

To register, write https://t.me/botfather command /newbot. Then it asks you to enter the name and address of the bot. If the data is correct, it gives you a token. Note that the address must be unique, you can not use “fonlinebot” again. Футбольный телеграм бот на Python (1/4): Подготовка и настройка бота Save the token to a file for the time of development. Create a “config.py” in the project folder to store the settings and write the token TOKEN = "YOUR TOKEN"

Configure the bot

Now we have to connect the bot to redis and the database and check if it works. Create necessary modules and files. In the “fonlinebot” folder add to the previously created “main.py” and “config.py”: “database.py”, “requirements.txt” and the “app” folder. Add to the “app” folder: “bot.py”, “dialogs.py”, “service.py”. This is the structure you get: Футбольный телеграм бот на Python (1/4): Подготовка и настройка бота By dividing the bot into modules, it’s easier to maintain and modify it.

  • “main.py” – to start the bot.
  • “config.py” – stores settings, access keys and other static information.
  • “database.py” – for working with the database and cache (redis).
  • “requirements.txt” – stores project dependencies, to run on the server.
  • “app” is the folder of the bot itself.
    • “bot.py” – for bot interaction with users, responses to messages.
    • “dialogs.py” – all of the bot’s text responses.
    • “service.py” – business logic, receiving and processing match data.

Now it’s time to move on to programming. Let’s write our dependencies in “requirements.txt”:

aiogram==2.11.2
emoji==1.1.0
redis==3.5.3
ujson==4.0.1
uvloop==0.14.0

Since most of the programming is on Windows, we didn’t install uvloop locally. Let’s install it on the server. In “config.py” we will add the data of the bot and the redis connections to the token.

# fonlinebot/config.py
import ujson
import logging
logging.basicConfig(level=logging.INFO)
TOKEN = "there must be a token here"
BOT_VERSION = 0.1
# database stores the leagues selected by the user
BOT_DB_NAME = "users_leagues"
# Test data about supported leagues
BOT_LEAGUES = {
"1": "Bundesliga",
"2": "Serie A",
"3": "La Liga",
"4": "Turkish Super League",
"5": "Dutch Championship",
"6": "Pro League of Belgium",
"7": "English Premier League",
"8": "Ligue 1",
}
# Flags for posts, emoji-code
BOT_LEAGUE_FLAGS = {
"1": ":Germany:",
"2": ":Italy:",
"3": ":Spain:",
"4": ":Turkey:",
"5": ":Netherlands:",
"6": ":Belgium:",
"7": ":England:",
"8": ":France:",
}
# redis-client data
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
# There is no default password. It will be on the server
REDIS_PASSWORD = None

The information about the leagues in the future can be put in a separate json file. This version of the bot will support no more than 10 variants, I’ve obviously written them down.

Adding the database

Now let’s add classes to work with the sqlite and redis database. The database is needed to save the preferences for the user leagues. The user will choose 3 championships to track, the bot will save them to the database and use them to query the results. The cache(redis) will save the results of the matches to reduce the number of API queries and speed up response time. As a rule, free APIs limit queries.

# fonlinebot/database.py
import os
import logging
import sqlite3
import redis
import ujson
import config
# class inherits from 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):
"""the function will convert the python object to Json and save"""
r = self.get(name)
if r is None:
return r
return ujson.loads(r)
def jget(self, name):
"""function returns json and converts to a python object"""
return ujson.loads(self.get(name))

The Cache class is inherited from StrictRedis. We add 2 methods jset, jget to store python lists and dictionaries in the redis repository. Initially it doesn’t work with them. Now we add a class that will create databases and perform CRUD functions.

# fonlinebot/database.py
#...
class Database:
""" Database class """
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 is suitable for test projects. In the future, you will need to move to an external database and asynchronous work. So as not to rewrite all the logic of working with the database, I immediately added asynchronous syntax. The database file will be created once, automatically. Now we need to create instances of classes:

# fonlinebot/database.py
#...
# creating cache and database objects
cache = Cache(
host=config.REDIS_HOST,
port=config.REDIS_PORT,
password=config.REDIS_PASSWORD
)
database = Database(config.BOT_DB_NAME)

Adding text messages

For the message templates, let’s create an immutable dataclass. This is where all the text responses will be. And dataclass is convenient to use when calling arguments.

# fonlinebot/app/dialogs.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Messages:
test: str = "Hi {name}. working..."
msg = Messages()

Creating a bot

In the file “bot.py” let’s create a bot. Import dependencies, create bot object and first message.

# 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
# is the standard code for bot
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
dp.middleware.setup(LoggingMiddleware())
@dp.message_handler()
async def test_message(message: types.Message):
# username from Telegram settings
user_name = message.from_user.first_name
await message.answer(msg.test.format(name=user_name))

The test_message function takes any message and responds to it. In addition, we need to close the connection to the database before closing. Let’s add on_shutdown to the end of the file.

# fonlinebot/app/bot.py
# ...
async def on_shutdown(dp):
logging.warning('Shutting down...')
# shutting down the database connection
db._conn.close()
logging.warning("DB Connection closed")

First launch of the bot

To start it, use file “main.py”. Import the bot and set up the pooling:

# fonlinebot/main.py
from aiogram import executor
from app import bot
executor.start_polling(bot.dp,
skip_updates=True,
on_shutdown=bot.on_shutdown)

Now let’s run the project. If you are in Pycharm, open the “Terminal” tab and type python main.py. To run the bot without Pycharm:

  • navigate to the project folder,
  • activate the virtual environment,
  • run the script(python main.py).

Первый запуск бота Now let’s open Telegram and check it. Первое сообщение боту Great, it works! Now it’s time to write some basic tests to avoid problems in the future.

Testing the bot

In the process of writing the project, the code will change. To avoid new bugs after placing it on the server, you should write some tests at once. Add the module “test.py” to the folder fonlinebot. At this stage 4 tests are enough, you can add your own tests.

# 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()

We will check the CRUD functions of the database, write the test data and delete it. We’ll check the connection to redis and the bot. Running the test is the same as the bot, only the name of the python main.py file changes. Футбольный телеграм бот на Python (1/4): Подготовка и настройка бота In the screenshot you can see the error notification after the test is finished. This is a known problem of aiohttp on windows, you can ignore it.

Errors that can be encountered

  1. aiogram.utils.exceptions.Unauthorized: Unauthorized – invalid bot token. Token should be saved as a string, its structure is “numbers:letters-and-digits”, check.
  2. redis.exceptions.ConnectionError: Error 10061 connecting to ... – redis-server is not running.
  3. sqlite3.IntegrityError: UNIQUE constraint failed: . – You are trying to add a value to a database that already exists.

This concludes the preparation of the project. Move on to part two: Writing the bot kernel.

Related Posts

LEAVE A COMMENT