Обработка событий в реальном времени через 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) Уведомление об удалении сообщений

Телеграм позволяет удалять сообщения, и вы можете отслеживать эти события.

@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.