Асинхронне читання та відправка повідомлень

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

• Обробляти кілька вхідних і вихідних повідомлень одночасно.
• Швидко реагувати на події в реальному часі (наприклад, нові повідомлення в чаті).
• Підтримувати високу продуктивність, оскільки процес не блокується довгими задачами.

Якщо ви будете використовувати синхронний підхід, то кожна операція (відправка або отримання повідомлення) буде "заморожувати" виконання програми, поки вона не завершиться. Асинхронність — ваш квиток у світ швидших і ефективніших застосунків.

Асинхронне читання повідомлень

1. Робота з асинхронними задачами

   В бібліотеці Telethon вся взаємодія з Telegram API побудована на асинхронних методах. Наше завдання — використовувати async def і ключове слово await, щоб взаємодіяти з функціями бібліотеки.

   Почнемо з прикладу, де ми витягуємо останні повідомлення з чату. Припустимо, нам потрібно отримати 10 останніх повідомлень з конкретного каналу.

   
   from telethon import TelegramClient
   
   # Конфігурація: замініть на свої значення
   api_id = 'YOUR_API_ID'
   api_hash = 'YOUR_API_HASH'
   
   async def read_messages():
       # Створюємо клієнта
       async with TelegramClient('session_name', api_id, api_hash) as client:
           # Вказуємо чат або канал: username або ID
           chat = 'some_channel_or_chat'
           
           # Отримуємо останні 10 повідомлень
           async for message in client.iter_messages(chat, limit=10):
               print(f"Повідомлення від {message.sender_id}: {message.text}")
   
   # Запуск асинхронної функції
   import asyncio
   asyncio.run(read_messages())
       

   Як це працює?

   1. Ми використовуємо async with TelegramClient, щоб автоматично підключатися та відключатися від API.
   2. Метод iter_messages повертає повідомлення як асинхронний генератор. Це дозволяє отримувати повідомлення по одному, не завантажуючи все одразу в пам'ять.
   3. message.sender_id – ідентифікатор відправника повідомлення. message.text – це текст повідомлення.
2. Обробка нових повідомлень

   Тепер давайте навчимося відслідковувати нові повідомлення в реальному часі. Для цього ми будемо використовувати подійно-орієнтований API Telethon.

   
   from telethon import TelegramClient, events
   
   # Конфігурація: замініть на свої значення
   api_id = 'YOUR_API_ID'
   api_hash = 'YOUR_API_HASH'
   
   async def main():
       async with TelegramClient('session_name', api_id, api_hash) as client:
           @client.on(events.NewMessage(chats='some_channel_or_chat'))
           async def handler(event):
               print(f"Нове повідомлення: {event.raw_text}")
   
           print("Чекаємо нові повідомлення...")
           await client.run_until_disconnected()
   
   import asyncio
   asyncio.run(main())
       

   Що тут відбувається?

   1. Ми використовуємо @client.on(events.NewMessage) для реєстрації обробника нових повідомлень.
   2. Аргумент chats='some_channel_or_chat' вказує, що ми хочемо відстежувати повідомлення тільки з конкретного чату/каналу. Можна залишити порожнім, щоб ловити повідомлення з усіх чатів.
   3. event.raw_text містить текст нового повідомлення.
   4. Метод run_until_disconnected дозволяє нашому клієнту залишатися активним, щоб він міг отримувати події.

Асинхронна відправка повідомлень

Відправлення повідомлень через Telegram API так само просте, як і їх читання. Розглянемо кілька сценаріїв.

1. Відправка простого повідомлення

   
   from telethon import TelegramClient
   
   # Конфігурація: замініть на свої значення
   api_id = 'YOUR_API_ID'
   api_hash = 'YOUR_API_HASH'
   
   async def send_message():
       async with TelegramClient('session_name', api_id, api_hash) as client:
           chat = 'some_channel_or_chat'
           await client.send_message(chat, 'Привіт, це моя перша відправка повідомлення через Telethon!')
   
   import asyncio
   asyncio.run(send_message())
       

   Як це працює?

   • Ми використовуємо метод send_message, щоб відправити текстове повідомлення в чат/канал.
   • Перший аргумент – вказівка на чат/канал (може бути username, ID або навіть об'єкт, отриманий раніше).
   • Другий аргумент – текст повідомлення.

Відправка мультимедіа

У Telegram можна відправляти не лише текст, але й зображення, відео, аудіофайли та документи. Ось як виглядає відправка зображення:

from telethon import TelegramClient

# Конфігурація: замініть на свої значення
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

async def send_photo():
    async with TelegramClient('session_name', api_id, api_hash) as client:
        chat = 'some_channel_or_chat'
        await client.send_file(chat, 'path/to/your/photo.jpg', caption='Ось моє фото!')

import asyncio
asyncio.run(send_photo())

Як це працює?

• Метод send_file відправляє файл у чат/канал.
• Опціональний аргумент caption додає підпис до файлу.

Обробка повідомлень: читаємо й відправляємо одночасно

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

from telethon import TelegramClient, events

# Конфігурація: замініть на свої значення
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

async def main():
    async with TelegramClient('session_name', api_id, api_hash) as client:
        @client.on(events.NewMessage(chats='some_channel_or_chat'))
        async def handler(event):
            incoming_message = event.raw_text
            print(f"Отримано повідомлення: {incoming_message}")

            # Відповімо на повідомлення
            await event.reply(f"Ви сказали: {incoming_message}")

        print("Бот запущено, чекаємо повідомлень...")
        await client.run_until_disconnected()

import asyncio
asyncio.run(main())

Типові помилки та поради

Працюючи з асинхронним програмуванням у Telethon, ви можете натрапити на кілька підводних каменів:

• Помилка "Event loop is closed": це пов'язано з тим, як asyncio працює в деяких середовищах (наприклад, Windows). Виправлення — додати asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) перед запуском скрипту.
• Блокування операцій: якщо ви виконуєте важкі задачі всередині асинхронної функції, це може блокувати процес. Використовуйте asyncio.to_thread, щоб виконувати синхронні задачі в окремому потоці.
• Обмеження Telegram API: Telegram може обмежити відправку повідомлень (спам-фільтр). Переконайтеся, що ви дотримуєтеся їхню документацію.

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

3

Опитування

Введення в Telethon, рівень 19, лекція 4

Недоступний

Введення в Telethon

Введення в Telethon

Відкрити