Обробка подій у реальному часі через Telethon

У цьому уроці ми зануримось у події Telegram у реальному часі, їхню обробку та налаштування за допомогою бібліотеки Telethon. Ми навчимося відстежувати події, налаштовувати обробники і створювати сценарії реагування. Також ви дізнаєтесь, як використовувати подійно-орієнтовану архітектуру для автоматизації задач.

Подійно-орієнтована архітектура (Event-Driven Architecture, EDA) — це підхід, при якому системи реагують на події в реальному часі. У контексті Telegram клієнт може отримувати сповіщення про нові повідомлення, зміни в чаті або інші дії й реагувати на них автоматично.

Можна навести приклад із життя. Уявіть, що ви на вечірці, і хтось вимовляє ваше ім'я. Ви одразу обертаєтесь, щоб знайти джерело звуку. Це і є подійно-орієнтована система: подія — вимовляння вашого імені, реакція — поворот голови.

Для реалізації такої логіки в Telethon використовується об'єкт events, який дозволяє "підписатися" на певні типи подій, наприклад, нові повідомлення, додавання учасників у чат або редагування повідомлень.

---

Налаштування обробників подій

Бібліотека Telethon надає спеціальний декоратор @client.on(events.SomeEvent), який дозволяє зв'язати певну подію з функцією-обробником. Коли подія відбувається, ця функція автоматично викликається.

Давайте почнемо з базового прикладу, де ми будемо реагувати на повідомлення, надіслані в особисті повідомлення.

from telethon import TelegramClient, events

# Вказуємо свої дані API
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

# Створюємо клієнта
client = TelegramClient('session_name', api_id, api_hash)

# Обробник на нові повідомлення
@client.on(events.NewMessage)
async def new_message_handler(event):
    sender = await event.get_sender()
    print(f"Нове повідомлення від {sender.username}: {event.text}")

# Запускаємо клієнта
client.start()
client.run_until_disconnected()

Як це працює:

1. @client.on(events.NewMessage): підписується на подію NewMessage, яка виникає, коли клієнт отримує нове повідомлення.
2. event: об'єкт події, що містить дані про те, що сталося (наприклад, текст повідомлення, відправник тощо).
3. event.get_sender(): отримує інформацію про відправника повідомлення.

Коли ви запустите скрипт і отримаєте повідомлення в Telegram, програма виведе текст повідомлення й ім'я відправника в консолі.

---

Обробка різних типів подій

Подій у Telethon багато, давайте розглянемо основні, які можуть стати в пригоді на практиці.

a) Реакція на повідомлення з певним текстом

Уявімо, що ми хочемо автоматично відповідати на повідомлення, які містять слово "Привіт".

@client.on(events.NewMessage(pattern='Привіт'))
async def greet(event):
    await event.reply('Привіт! Як справи?')

Тут використовується параметр pattern, щоб фільтрувати повідомлення. Цей обробник спрацює лише якщо в тексті повідомлення буде "Привіт".

b) Відстеження нових учасників у чаті

Якщо ви — адміністратор групи, можна автоматично вітати нових учасників.

from telethon.tl.types import User

@client.on(events.ChatAction)
async def new_user(event):
    if event.user_added:  # Перевіряємо, чи був доданий користувач
        new_user = event.user
        if isinstance(new_user, User):  # Переконуємось, що це саме користувач
            await event.reply(f"Ласкаво просимо, {new_user.first_name}!")

c) Повідомлення про видалення повідомлень

Telegram дозволяє видаляти повідомлення, і ви можете відстежувати ці події.

@client.on(events.MessageDeleted)
async def deleted_message_handler(event):
    print(f"Повідомлення з ID {event.deleted_id} було видалено.")

---

Асинхронна обробка подій

Коли ми працюємо з подіями, може виникнути ситуація, коли обробка однієї події займе багато часу, наприклад, якщо ми виконуємо запит до зовнішнього API. Щоб не блокувати інші події, обробники в Telethon працюють асинхронно.

Наводимо приклад — напишемо обробник, який реагує на повідомлення і надсилає запит до зовнішнього API.

import httpx

@client.on(events.NewMessage(pattern='погода'))
async def weather_handler(event):
    # Пишемо у відповідь, щоб повідомити користувачу, що ми обробляємо запит
    await event.reply("Секунду, шукаю дані...")

    # Запит до зовнішнього API (наприклад, OpenWeatherMap)
    response = await httpx.get('http://api.weatherapi.com/v1/current.json', params={
        'key': 'YOUR_API_KEY',
        'q': 'London'
    })
    data = response.json()
    temperature = data['current']['temp_c']

    # Відправляємо назад результат
    await event.reply(f"У Лондоні зараз {temperature}°C.")

---

Обробка помилок

Коли ми працюємо з зовнішніми системами (наприклад, API або Telegram), помилки неминучі. Щоб ваші події не пропускалися через виняток, обгортайте обробник у try-except.

Приклад обробки помилок:

@client.on(events.NewMessage)
async def safe_handler(event):
    try:
        # Ваш основний код
        await event.reply("Приклад без помилок!")
    except Exception as e:
        # Обробка винятку
        print(f"Помилка: {e}")
        await event.reply("Щось пішло не так.")

---

Приклад практичного застосування

Створимо бота, який:

1. Привітає нових учасників групи.
2. Реагує на команду /stats, відправляючи кількість учасників групи.
3. Автоматично відповідає на слово "Привіт".

from telethon import TelegramClient, events
from telethon.tl.types import User

# Дані API
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

client = TelegramClient('session_name', api_id, api_hash)

# Привітання нових учасників
@client.on(events.ChatAction)
async def welcome(event):
    if event.user_added:
        new_user = event.user
        if isinstance(new_user, User):
            await event.reply(f"Ласкаво просимо, {new_user.first_name}!")

# Команда /stats для отримання статистики групи
@client.on(events.NewMessage(pattern='/stats'))
async def stats(event):
    chat = await event.get_chat()
    participants = await client.get_participants(chat)
    await event.reply(f"Кількість учасників: {len(participants)}")

# Відповідь на "Привіт"
@client.on(events.NewMessage(pattern='Привіт'))
async def greet(event):
    await event.reply('Привіт! Як справи?')

# Запуск
client.start()
client.run_until_disconnected()

---

Цей бот демонструє, як можна поєднувати обробку подій і автоматизацію з Telethon. Тепер ви знаєте, як обробляти події в реальному часі в Telegram, використовуючи потужність асинхронного програмування та гнучкість Telethon.