Приклад моніторингу Telegram-чату через API

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

• Відстеження ключових слів у чаті (наприклад, запитів від користувачів або згадувань);
• Збір статистики активності (кількість повідомлень, найбільш активні учасники);
• Реалізація системи автоматичних сповіщень;
• Контроль дотримання правил у групі (наприклад, миттєве видалення повідомлень зі спамом).

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

1. Налаштування проєкту.
2. Підключення до Telegram-чату.
3. Реалізацію логіки моніторингу.
4. Інтеграцію зі сповіщеннями.

Налаштування проєкту

Створимо структуру проєкту. У нас буде два файли:

• config.py: для зберігання чутливих даних, таких як API ID і API Hash;
• monitor.py: для основної логіки моніторингу.

Заповніть файл config.py такими даними:

# config.py

API_ID = "ваш_telegram_api_id"
API_HASH = "ваш_telegram_api_hash"
SESSION_NAME = "monitor_session"

Не забудьте зареєструвати свій застосунок на my.telegram.org і отримати API ID та API Hash, якщо ви ще цього не зробили.

Підключення до Telegram-чату

Тепер підключимо нашого клієнта до Telegram і налаштуємо його на моніторинг чатів. Це буде відбуватися у файлі monitor.py.

# monitor.py

from telethon import TelegramClient, events
import config

# Створюємо клієнта
client = TelegramClient(config.SESSION_NAME, config.API_ID, config.API_HASH)

async def main():
    # Підключення клієнта
    print("Підключаємося до Telegram...")
    await client.start()
    print("Клієнт підключений!")
    
    # Отримання інформації про користувача
    me = await client.get_me()
    print(f"Ви увійшли як {me.username or me.first_name}")

with client:
    client.loop.run_until_complete(main())

Запустіть цей скрипт, щоб перевірити підключення. Якщо все налаштовано правильно, ви побачите повідомлення про підключення і дані вашого користувача.

Реалізація логіки моніторингу

Тепер ми налаштуємо обробник подій для моніторингу повідомлень у чаті. Ми будемо використовувати Telethon для відстеження нових повідомлень у реальному часі. Додайте наступний код у monitor.py:

# Додаємо подію для відстеження нових повідомлень
@client.on(events.NewMessage)
async def new_message_listener(event):
    # Отримуємо текст нового повідомлення
    message = event.message.message
    sender = await event.get_sender()
    sender_name = sender.username or sender.first_name

    print(f"Нове повідомлення від {sender_name}: {message}")

    # Налаштування фільтрів
    if "питання" in message.lower():  # Приклад фільтра за ключовим словом
        await event.reply("Привіт! Яке у тебе питання?")

Запустіть скрипт знову. Тепер, якщо ви надішлете повідомлення з текстом, що містить слово "питання", бот автоматично відповість.

Налаштування моніторингу конкретного чату

Якщо потрібно відстежувати повідомлення тільки з певної групи або каналу, можна налаштувати фільтри за chat ID. Наприклад:

CHAT_ID = -1001234567890  # Замініть на ID вашого чату

@client.on(events.NewMessage(chats=CHAT_ID))  # Обмеження на конкретний чат
async def new_message_listener(event):
    message = event.message.message
    sender = await event.get_sender()
    sender_name = sender.username or sender.first_name

    print(f"Повідомлення з чату: {message} (від {sender_name})")

Щоб дізнатися ID потрібного чату, ви можете використати метод get_entity:

chat = await client.get_entity("ім'я_чату_або_каналу")
print(chat.id)

Збереження та аналіз даних

Для аналізу чату ми можемо зберігати повідомлення у файл або в базу даних, а потім обробляти їх. Приклад запису повідомлень у файл:

import datetime

LOG_FILE = "chat_log.txt"

@client.on(events.NewMessage(chats=CHAT_ID))
async def new_message_listener(event):
    message = event.message.message
    sender = await event.get_sender()
    sender_name = sender.username or sender.first_name
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    log_entry = f"[{timestamp}] {sender_name}: {message}\n"
    
    with open(LOG_FILE, "a") as f:
        f.write(log_entry)
    
    print(f"Лог збережено: {log_entry}")

Реалізація системи сповіщень

Щоб сповіщати вас про важливі події (наприклад, спам у чаті), можна використовувати e-mail або Telegram. Як приклад надішлемо повідомлення у ваш Telegram акаунт:

ALERT_USER_ID = "ваш_user_id"  # Замініть вашим ID

@client.on(events.NewMessage(chats=CHAT_ID))
async def alert_system(event):
    message = event.message.message
    if "спам" in message.lower():
        await client.send_message(ALERT_USER_ID, f"Виявлено спам: {message}")
        print("Сповіщення відправлено!")

Напрямки подальшого розвитку

1. Обробка зображень і файлів: додайте обробники для медіафайлів, щоб моніторити вкладення.
2. Фільтрація за учасниками: відстежуйте тільки повідомлення від конкретних користувачів.
3. Інтеграція з FastAPI: створіть REST API, щоб керувати ботом через HTTP-запити.

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