ByteByteGo
Top 3 Things You Should Know About Webhooks!
1 hour ago
https://www.youtube.com/watch?v=x_jjhcDrISk

🤖 Телеграм боти. 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

▶️ Корисні речі, які варто знати про процес.

Коли ви запускаєте файл своєї програми, вона завантажується в оперативну пам’ять і починає виконуватися як процес.

Є кілька важливих аспектів процесу, які варто знати, щоб ефективніше використовувати їх у своїх проєктах:

1️⃣ Змінні оточення (Environment Variables) [1] [2]
2️⃣ Консольні аргументи (Command-line Arguments) [1] [2] [3]
3️⃣ Код завершення процесу (Exit Code) [1]
4️⃣ Потоки вводу/виводу
(Input/Output Streams, stdin/stdout/stderr) [1] [2]

Планую розглянути ці речі поступово детальніше.

🟡 Змінні оточення (Environment Variables)

Змінні оточення — це іменовані пари ключ-значення, які операційна система передає програмам під час їх запуску.
Вони містять інформацію про середовище виконання, конфігурацію системи та користувача інше.

Є загальноприйняті системні змінні оточення, які діють у всій системі, а також ті,
що встановлюються іншими програмами або shell-скриптами під час запуску.

Для роботи зі змінними оточення у Python використовується словник os.environ.

Приклад використання системних змінних:

import os
import sys

def get_home_dir() -> str | None:
    if sys.platform == "win32": # Windows
        home_drive = os.environ.get("HOMEDRIVE", "")
        home_path = os.environ.get("HOMEPATH", "")
        if home_drive and home_path:
            return os.path.join(home_drive, home_path)
    else: # Linux подібні
        return os.environ.get("HOME")

    return None


print(get_home_dir())

Цей код виводить розташування каталогу користувача в системі.

Як бачимо під різними системами різні назви змінних, крім того під Windows треба отримати дві зміних.

Аналогічним чином можна отримувати шлях до каталогу з тимчасовим файлами (TEMP/TMPDIR)

Відео по темі: Your .env File Is Lying to You

🟢 Змінні оточення. Власні змінні.

Розглянемо приклад використання власної змінної.

Файл main.py

import os
import sys

name = "USER_VAR"
value = os.environ.get(name)

print(value)

Якщо ми запустимо цей код стандартним способом отримуємо вивід None що логічно бо такої змінної немає.

Але якщо ми запустимо цей код у терміналі командою:

Під Windows

set USER_VAR="42" && python main.py

Або під Linux

export USER_VAR="42" && python main.py

Отримаємо очікувані "42"

👉 Таки чином ми можемо передавати у програму якісь налаштування:
▪️ виставляючи змінні глобально у системі
▪️ за допомогою терміналу і sh/bat - скриптів
▪️ запускаючи іншими програмами

P/S: У PyСharm є налаштування запуску куди можна додавати власні змінні оточення.

🔵 Dot env files

У минулому пості я перелічив три способи, як можна передавати змінні оточення у програму, але ці способи не дуже зручні саме під час розробки програмного забезпечення. У вас може бути багато проєктів, і незручно встановлювати все в системні змінні.
Писати командні скрипти можна, але потрібно створювати окремий скрипт для кожної операційної системи.

Тому під час розробки часто використовують .env файл.
Цей звичайний текстовий файл який містить змінні оточення та їх значення розділені знаком рівності.

⚠️ Якщо ваш .env файл містить або може містити в майбутньому секретні ключі, паролі тощо — його обов’язково треба додати у .gitignore, щоб не залити у GitHub репозиторій.

Для роботи з такими файлами у python є стороння бібліотека python-dotenv

Файл .env

BOT_TOKEN="Ваш токен"

Файл main.py

from dotenv import load_dotenv # pip install python-dotenv
import os

load_dotenv() 
bot_token = os.environ.get("BOT_TOKEN")

print(bot_token)

Що робить ця бібліотека?
1️⃣ Вона шукає у теці проєкту файл .env (іншу назву файлу можна вказати параметром).
2️⃣ Якщо файл є, вона його завантажує, парсить і додає змінні з файлу у словник os.environ.
3️⃣ Якщо файлу немає, os.environ залишається незмінним.

У відео йдеться про різні цікаві можливості python-dotenv

https://www.youtube.com/watch?v=8dlQ_nDE7dQ

🤖 Приклад телеграм бота на 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