🤖 Про телеграм боти.

Телеграм бот це програма яка отримує події/оновлення (Updates) від телеграм сервера і реагує на них, або ігнорує.

Боти можна розділяти по способу отримання оновлень (Update):

1️⃣ Long Polling
2️⃣ Webhook

Long Polling — бот періодично запитує сервер Телеграм, чи є нові оновлення. Цей варіант підходить для більшості ботів.

Webhook — бот отримує оновлення відразу, як тільки вони надходять, бо Телеграм відправляє їх на заданий Вами URL. Цей варіант для високонавантажених ботів.

На відміну від Long polling, для Webhook потрібен публічний IP або домен з SSL, щоб Телеграм міг надсилати запити безпосередньо на твій сервер.

#tg_bots

🤖 Телеграм боти. Update.

Оновлення (Update) - важливе поняття у телеграм ботах, важливо розуміти які вони бувають.

Повний список оновлень та їх опис можна знайти у оф.документації телеграму.

Для прикладу на скріншоті наведено три типи оновлень:
▪️ message - відбувається коли користувач відправляє текстове повідомлення.
▪️ edit_message - відбувається коли користувач його редагує
▪️ callback_query - відбувається коли користувач натискає на inline-кнопку.

Завжди є нюанси, наприклад натиснення на replay-кнопку теж рахується як message.

Фізично дані оновлення це json-контент який присилає нам сервер. Бібліотека aiogram3 перетворює json у об'єкти класів (вик. pydantic моделі).

#tg_bots

🤖 Телеграм боти. Long Polling

Long Polling ніщо інше як HTTP-запити на телеграм сервіс.

Нижче простий приклад отримання оновлень за допомогою requests.
Робимо запит і отримуємо оновлення у вигляді json.

import requests

TOKEN = 'Ваш токен від телеграм боту'
URL = f'https://api.telegram.org/bot{TOKEN}/getUpdates'

def get_updates(session: requests.Session, offset: int | None = None):
    params = {'timeout': 30, 'offset': offset}
    try:
        r = session.get(URL, params=params, timeout=35)
        r.raise_for_status()
        data = r.json()
        if not data.get('ok'):
            raise RuntimeError("Telegram API response is not OK")
        return data['result']
    except requests.exceptions.Timeout:
        return []

def long_polling():
    offset = None
    with requests.Session() as session:
        while True:
            if updates := get_updates(session, offset):
                for update in updates:
                    print(update)
                    offset = update['update_id'] + 1
            else:
                print("No updates")

long_polling()

Приклад виводу у коментарях.

#tg_bots

🤖 Приклад телеграм бота на aiogram3

Повернемось до теми Telegram ботів.
Будем розглядати приклад ехо-бота.

Файл .env

BOT_TOKEN=Ваш_токен

Файл main.py

import asyncio
import os

# pip install aiogram
from aiogram import Bot, Dispatcher, Router, html
from aiogram.filters import Command
from aiogram.types import Message
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
# pip install python-dotenv
from dotenv import load_dotenv

router = Router()

async def startup(bot: Bot) -> None:
    await bot.delete_webhook(drop_pending_updates=True)
    print("Bot started.")

async def shutdown(bot: Bot) -> None:
    print("Bot stopped.")

@router.message(Command("start"))
async def start_handler(message: Message):
    await message.answer(
        f"<i>Hello</i> {html.unparse(message.from_user.first_name)}!", 
        parse_mode=ParseMode.HTML
    ) 
    
@router.message()
async def echo_handler(message: Message):
    await message.answer(
         f"Echo: {message.html_text}", 
         parse_mode=ParseMode.HTML
    )

async def main():
    load_dotenv()
    token = os.environ.get("BOT_TOKEN")
    assert token, "BOT_TOKEN is missing!"
    
    async with Bot(token=token) as bot:
        storage = MemoryStorage()
        dp = Dispatcher(storage=storage)
        dp.startup.register(startup)
        dp.shutdown.register(shutdown)
        dp.include_router(router)
        await dp.start_polling(bot)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Stopped by user.")

Для написання телеграм ботів по мінімум потрібні знання:
1️⃣ Асинхронності
2️⃣ ООП
3️⃣ Декораторів
4️⃣ Бази python, наприклад виключень та менеджерів контексту.

І це тільки для початку, бо для побудови більш менш нормального потрібна база даних, то треба ще знання SQL та ORM (SqlAlchemy)

Без цих основ немає сенсу братись за телеграм боти.

#tg_bots

🟣 Опис роботи echo-бота. Основні класи.

✅ Токен бота беремо з змінних оточення або .env-файлу якщо він є.

✅ Чисто з естетичних причин ми перехоплюємо виключення KeyboardInterrupt.
Це виключення виникає при натисненні юзером Ctrl+C у терміналі і в таких випадках мабуть ми не хочемо бачити stack trace помилок.

✅ Для автоматичного закриття сесії бота використовуємо асинхронний контекстний менеджер.

✅ Класи Dispatcher, Router дозволяють реєструвати обробники оновлень(подій).
Загалом система диспетчеризації оновлень по обробникам може мати складну деревовидну структуру.

Dispatcher - це головний об'єкт який може включати в собі Router-и, водночас Router може включати в себе інші Router-и. Звучить складно, але загалом практичний сенс цього полягає у тому що ми групуємо обробники, по якомусь принципу у логічну одиницю -- Router, а Router-и додаємо у Dispatcher. Користь Router-ів у тому що ми можемо легко винести обробники оновлень у окремі файли - модулі. Крім того кожен роутер може мати свій Filter і/та Middleware, що дуже зручно (про це можливо пізніше)

Розглянемо приклад структури проєкту:

project/
├── main.py              (містить dp: Dispatcher)
├── keyboards.py
├── middleware.py
└── handlers/
    ├── __init__.py
    ├── admins.py        (містить admin_router: Router)
    ├── private.py       (містить private_router: Router)
    ├── сhannels.py      (містить сhannels_router: Router)
    └── group_users.py   (містить group_router: Router)

main.py

from handles.admins import admin_router
from handles.private import private_router
# ...
dp = Dispatcher(storage=storage)
dp.include_router(admin_router)
dp.include_router(private_router)
# ...
dp.include_router(group_router)

Логіка розділення:

▪️Очевидно, що адміністратори бота і/або чату можуть мати особливі права і їх дії треба обробляти окремо.
▪️Окремо також варто оброблювати повідомлення у особисте спілкування з ботом.
▪️ Окремо оброблювати події в каналах та групах.

⚠️ Це лише приклад, структура-розділення має залежати від суті бота.
Головне мати логічну вибудовану структуру, а не всі обробники у одному файлі.

✅ MemoryStorage - це клас сховища станів. Воно потрібне щоб тимчасово зберігати стан заповнення форми, а також дані полів які користувач вже заповнив перед тим як ці дані застосовувати(наприклад записати у БД)

#tg_bots

🟤 Опис роботи echo-бота. Lifespan events обробники

async def startup(bot: Bot) -> None:
    await bot.delete_webhook(drop_pending_updates=True)
    print("Bot started.")

async def shutdown(bot: Bot) -> None:
    print("Bot stopped.")

# ...
dp = Dispatcher(storage=storage)
dp.startup.register(startup)
dp.shutdown.register(shutdown)

Коли бот запущен і готовий приймати оновлення викликається startup-обробник.
У ньому можна відкинути всі оновлення які накопичились поки бот не працював.

Коли бот зупиняється викликається shutdown-обробник

#tg_bots

🔴 Опис роботи echo-бота. Підписка на події/оновлення.

Подивимось код реалізації класу Router у aiogram3:

aiogram/dispatcher/router.py 🔗

class Router:
    def __init__(self, *, name: str | None = None) -> None:
        # ...
        self.message = TelegramEventObserver(
            router=self, 
            event_name="message",
        )

aiogram/dispatcher/event/telegram.py 🔗

class TelegramEventObserver:
    # ...
    def __call__(self, *filters: CallbackType, flags, **kwargs):      
        
        def wrapper(callback: CallbackType) -> CallbackType:
            self.register(callback, *filters, flags=flags, **kwargs)
            return callback

        return wrapper

З цих фрагментів коду бачимо що:

▪️ message це об'єкт класу TelegramEventObserver. Назва класа каже сама за себе.
▪️ У класі реалізований магічний метод __call__, тобто об'єкти цього класу можна викликати як функції/методи.
▪️ В тілі методу __call__ бачимо типову реалізацію декоратора.

Отже правильна відповідь на питання у опитуванні: message — це:

✅ Об'єкт класу, клас якого реалізує патерн Observer (реалізує підписку на події/оновлення)
✅ Об'єкт класу (клас-декоратор)
✅ Декоратор (клас-декоратор)

Cтаття про патерн Observer у контексті aiogram3 🔗

Що варто підкреслити:
🔺 У класі Router є атрибути назви яких відповідають типам подій(update) Telegram
🔺 Кожен з цих атрибутів дозволяє реєструвати асинхронну функцію обробник на відповідний тип події(update). Для цього використовується синтаксис декораторів.
🔺 Об'єкт клас-декоратора параметром може приймати фільтри. Функція обробник зберігаєтьься разом з фільтрами.
🔺 Асинхронна функція приймає першим параметром json від Telegram серверу десеріалізований у об'єкт події у нашему випадку це message: Message.

#tg_bots

🟢 Опис роботи echo-бота. Обробники подій. Фільтри

Трохи теорії по Telegram:

👉 Команда - це текстове повідомлення яке починається на символ / (або інший символ)

👉 /start - команда з якої починається робота у особисті з ботом

router = Router()

@router.message(Command("start"))
async def start_handler(message: Message):
    ... # код обробника
    
@router.message()
async def echo_handler(message: Message):
    ... # код обробника

У router реєструються дві функції-обробника при чому підписка йде на одну ту саму подію message.

Різниця лише у тому що start_handler завдяки встановленому фільтру Command("start") буде реагувати тільки команду /start, а на всі інші повідомлення буде реагувати echo_handler.

👉 Фільтр - це функція, спеціальний об'єкт або "магічний" вираз який перевіряє чи відповідають певні дані події/оновлення якимось критеріям.

#tg_bots

🔗 Update propagation - схема наглядно показує як відбувається вибір обробника з урахуванням фільтрів та порядку оголошення.

⚠️ Завжди треба враховувати порядок оголошення обробників, і як наслідок порядок включення роутерів у диспетчер.

#tg_bots

🟣 Опис роботи echo-бота. Альтернативний спосіб реєстрації обробників.

Якщо повернутись до коду TelegramEventObserver у методі __call__ викликається метод register. Цей метод можна використовувати напряму. Не те що б це мало практичне застоcування, скоріше це пояснює як воно працює. У aiogram2 це мало сенс — це дозволяло виносити обробники у окремі модулі без використання Router-ів (бо їх ще не було)

async def start_handler(message: Message):
    ...  # код обробника

async def echo_handler(message: Message):
    ...  # код обробника

# ...
router = Router()
router.message.register(start_handler, Command("start"))
router.message.register(echo_handler)

#tg_bots

Відео з гарним поясненням про те як працює телеграм бот.
(Але стосовно коду, особисто в мене є питання)
https://www.youtube.com/watch?v=fOQh61IoS4M

#tg_bots

🔵 Telegram боти. Сигнатура обробника оновлень.

Як я вже зазначав першим параметром обробника оновлень обов'язково має бути об'єкт напряму асоційований з оновленням, у нього може бути будь-яке ім'я але він буде відповідного класу.

@router.message(Command("start"))
async def start_handler(message: Message):
    ...   

В даному випадку це клас Message

Крім першого обов'язкового параметру є ще необов'язкові параметри.
Наприклад ми можемо просто додати параметр bot: Bot у функцію обробник:

@router.message(Command("start"))
async def start_handler(message: Message, bot: Bot):
    ... # код використання bot  

І це буде прауювати. Параметр bot: Bot буде автоматично переданий при виклику обробника. Тут важлива точна назва параметра так як він передається по "ключу". Аiogram3 для реалізації цього моменту під капотом використовує introspection (модуль inspect)

Серед необов'язкових параметрів також може передаватись наш власний об'єкт через механізм 🔗 Dependency injection .Це може бути якійсь налаштування, сесія БД, якийсь кеш.

#tg_bots

🟠 Telegram боти. Приблизний список додаткових параметрів в обробниках.

bot: Bot
command: CommandObject
fsm_storage: MemoryStorage
state: FSMContext
dispatcher: Dispatcher
bots: tuple
raw_state: str | None
handler: HandlerObject
event_update: Update
event_router: Router
event_context: EventContext
event_from_user: User
event_chat: Chat

#tg_bots

🟡 Telegram боти. Приклад використання додаткових параметрів у обробнику

Команди телеграм бота можуть мати аргументи.
Уявімо ситуацію коли ми хочемо реалізувати команду яка рахує суму чисел:

/sum 1 2 3.5

У відповідь будемо очікувати:

Сума: 6.5

До обробника додамо параметр command: CommandObject 🔗 , який полегшує розбір аргументів команди.
Атрибут command.args це рядок з аргументами, якщо вони є, а якщо немає він дорівнює None.

from aiogram.filters import CommandObject

@router.message(Command("sum"))
async def sum_handler(message: Message, command: CommandObject):
    success = False
    if args := command.args and command.args.split():
        try:
            result = sum(map(float, args))
            success = True
            await message.answer(f"Сума: {result}")
        except ValueError:
            pass

    if not success:
        await message.answer(
           "Щось пішло не так, можливо передані неправильні аргументи /sum"
        )

#tg_bots

🔴 Опис роботи echo-бота. Відправка повідомлень.

Розглянемо код тіла функції-обробника.

Є два способи відправляти повідомлення:
▪️ через об'єкт Message
▪️ через об'єкт Bot

from aiogram import html

@router.message(Command("start"))
async def start_handler(message: Message):
    await message.answer(
        f"<i>Hello</i> {html.unparse(message.from_user.first_name)}!", 
        parse_mode=ParseMode.HTML
    )

Перевага відправки через bot: Bot у тому що можна відсилати повідомлення у інші чати і гілки чатів.

from aiogram import html

@router.message(Command("start"))
async def start_handler(message: Message, bot: Bot):
    await bot.send_message(
        chat_id=message.chat.id,
        text=f"<i>Hello</i> {html.unparse(message.from_user.first_name)}!",
        parse_mode=ParseMode.HTML
    )

Telegram дозволяє надсилати форматований текст у форматах HTML і Markdown
(насправді — у їхніх дуже спрощених версіях).
🔗 Telegram Bot API: Markdown v2
🔗 Telegram Bot API: HTML

⚠️ Якщо передати некоректний HTML, виникне виняток типу:

TelegramBadRequest('Telegram server says - Bad Request: can't parse entities ...')

Тому завжди потрібно зважати на текст, який передається у вигляді змінних — чи не може він зламати HTML-розмітку. У нашому прикладі message.from_user.first_name — це ім’я користувача, яке може містити символи < або > що зламає розмітку.

👉 Щоб уникнути таких помилок, використовуйте екранування за допомогою html.unparse() та інші функції для форматування у aiogram3

Корисні посилання по темі:
🔗 Aiogram 3 FAQ : Bad Request: can't parse entities
🔗 aiogram3 documentation: Formatting

#tg_bots

🟤 Телеграм боти. Використання магічних фільтрів.

🔗 Message.from_user
🔗 User

class User(TelegramObject):
   id: int
   first_name: str
   last_name: str | None
   username: str | None
   # ...

Як бачимо:
▪️ username і last_name можуть бути None
▪️ cам атрибут from_user може бути None

⚠️ Об'єкти у aiogram3 часто можуть бути None — варто зважати на це.

Те що ми використовуємо first_name у коді обробника не спричиняє проблем.А ось те що не перевіряєм from_user на None це потенційна проблема. Чесно кажучи я не можу назвати випадок коли повідомлення може приходити без відправника, але мабуть може ... і мабуть ми хотіли би уникнути проблем пов'язаних з цим. Крім того тайп чекери типу mypy при перевірці коду будут вказувати на цю потенційну проблему.

👉 Рішенням може бути використання магічних фільтрів:

from aiogram  F
from aiogram.types import  User

@router.message(Command("start"), F.from_user.as_("from_user"))
async def start_handler(message: Message, from_user: User):
    await message.answer(
        f"<i>Hello</i> {html.unparse(from_user.first_name)}!",
        parse_mode=ParseMode.HTML
    )

Фільтр бере об'єкт message і перевіряє атрибут from_user, якщо from_user is None обробник просто не буде викликатись, якщо не None то from_user буде переданий як додатковий параметр.

Виклик методу .as_() як раз відповідає за передачу додаткового параметру у обробник, у ньому вказується ім'я цього параметру.

#tg_bots

✳️ Телеграм боти. Dependency Injection

Доопрацюємо нашого бота: зробимо різні привітання для звичайних користувачів та адміністраторів.
ID адміністраторів у нас міститься в об’єкті Settings, і виникає проблема — як його отримати у функції обробника.

Це можна зробити кількома способами:

1️⃣ Створити об’єкт Settings як глобальну змінну.
2️⃣ Використати Middleware для Dependency Injection.
3️⃣ Використати Dispatcher для Dependency Injection.

Dependency Injection (DI) — це підхід у програмуванні, який допомагає відокремити створення сутності від її використання.

Глобальні змінні — це не дуже добре. Використання Middleware — більш складний варіант. Тому давайте зосередимося на останньому варіанті.

🔗 aiogram documentation: Dependency Injection

async def main():
    settings = Settings(
        _env_file=".env",
        _env_nested_delimiter='__'
    )
    async with Bot(token=settings.bot.token.get_secret_value()) as bot:
        storage = MemoryStorage()
        dp = Dispatcher(storage=storage)
        dp.startup.register(startup)
        dp.include_router(router)

        # Use Dependency Injection   
        dp["settings"] = settings

        await dp.start_polling(bot)

У Dispatcher можна, як у словник, додати власні об'єкти-залежності, щоб потім отримати їх у обробнику як додаткові параметри:

@router.message(Command("start"), F.from_user.as_("from_user"))
async def start_handler(message: Message, from_user: User, settings: Settings):
    if from_user.id in settings.bot.admins:
        text = f"<i>Hello admin</i> {html.unparse(from_user.first_name)}!"
    else:
        text = f"<i>Hello</i> {html.unparse(from_user.first_name)}!"

    await message.answer(text=text, parse_mode=ParseMode.HTML)

На жаль, Router такої можливості не має. Якщо потрібно ізолювати дані одного роутера від іншого, варто використати Middleware.

#tg_bots

❇️ Телеграм-боти. Власний клас-фільтр.

Ми має код обробника у якому є if-else для визначення логіки звичайний користувач/адміністратор.
Але це не дуже добре:

1️⃣ Якщо в нас таких обробників буде багато, то if-else буде дублюватися і ускладнювати читання коду.
2️⃣ Ми не зможемо розділити логіку звич.користувачів і адміністраторів на різні файли-модулі.

Написання власного фільтра дозволить розбити логіку на два обробника.
Ми вже ознайомились з використанням готового бібліотечного фільтра Command та магічних фільтрів,
давайте напишемо тепер свій клас-фільтр:

from aiogram.filters import BaseFilter

class BotAdminFilter(BaseFilter):
    async def __call__(
       self, 
       message: Message, 
       settings: Settings
    ) -> bool | dict[str, Any]:
        if message.from_user and message.from_user.id in settings.bot.admins:
            return {"from_user": message.from_user}
        return False

Як бачимо фільтр це клас який містить метод __call__ , отже об'єкти цього класу можуть викликатись як функції.

⚠️ Тип першого параметру __call__ буде залежати від типу події на яку підписуємось.

При виклику фільтр вертає:
▪️ вертає False якщо при цьому оновленні не треба викликати обробник — воно не підходить
▪️ вертає dict якщо треба викликати обробник і підставити у нього додатковий аргумент from_user

🔗 aiogram documentation: Writing own filters

@router.message(Command("start"), BotAdminFilter())
async def start_admin_handler(message: Message, from_user: User):
    await message.answer(
        f"<i>Hello admin</i> {html.unparse(from_user.first_name)}!",
        parse_mode=ParseMode.HTML
    )

@router.message(Command("start"), F.from_user.as_("from_user"))
async def start_handler(message: Message, from_user: User):
    await message.answer(
        f"<i>Hello</i> {html.unparse(from_user.first_name)}!",
        parse_mode=ParseMode.HTML
    )

👉Фільтр у aiogram:
1️⃣ Визначає чи буде викликаний вказаний обробник подій
2️⃣ Може додавати параметри у обробник, вертаючи їх у словнику

#tg_bots

🔰 Телеграм-боти. Функція як фільтр.

Класи, об’єкти яких можна викликати, — це добре: можна передавати додаткові налаштування через їхній ініціалізатор.
Але в нашому випадку це не потрібно — список ID адміністраторів ми отримуємо через механізм DI.
Тому ми можемо спростити код, написавши просто функцію, яка буде фільтром:

async def is_bot_admin(message: Message, settings: Settings) -> bool | dict[str, Any]:
    if message.from_user and message.from_user.id in settings.bot.admins:
        return {"from_user": message.from_user}
    return False

@router.message(Command("start"), is_bot_admin)
async def start_admin_handler(message: Message, from_user: User):
     ... # код обробника

#tg_bots

♻️ Телеграм боти. Використання консольних аргументів.

Іноді зручно мати кілька .env-файлів із різними налаштуваннями та мати змогу запускати проєкт, указуючи, з яким файлом його запускати. Ми можемо мати, наприклад, два файли — .env.dev та .env.prod, тобто два профілі налаштувань: для розробки та для роботи на сервері.

import argparse

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--env-file",
        type=str,
        default=None,
        help="Path to the dotenv-file",
    )
    args = parser.parse_args()
    
    settings = Settings(
        _env_file=args.env_file,
        _env_nested_delimiter='__',
    )
    async with Bot(token=settings.bot.token.get_secret_value()) as bot:
        storage = MemoryStorage()
        dp = Dispatcher(storage=storage)
        dp["settings"] = settings
        dp.startup.register(startup)
        dp.include_router(router)
        await dp.start_polling(bot)

Ми можемо запукати проєкт вказуючи шлях/ім'я до файлу опцією --env-file
python main.py --env-file .env.dev
Або
python main.py --env-file .env.prod

Якщо запускати без зазначення опції — значення братимуться лише з системного оточення; якщо їх немає, виникне помилка валідації.

#tg_bots