Мониторинг Telegram-чата — это процесс автоматической проверки и обработки новых сообщений в реальном времени. Он может быть полезен для решения различных задач, таких как:
- Отслеживание ключевых слов в чате (например, вопросов от пользователей или упоминаний);
- Сбор статистики активности (количество сообщений, наиболее активные участники);
- Реализация системы автоматических уведомлений;
- Контроль соблюдения правил в группе (например, мгновенное удаление сообщений со спамом).
Для реализации мониторинга мы будем использовать библиотеку Telethon и её функциональность для обработки новых событий. Наш пример будет включать:
- Настройку проекта.
- Подключение к Telegram-чату.
- Реализацию логики мониторинга.
- Интеграцию с уведомлениями.
Настройка проекта
Создадим структуру проекта. У нас будет два файла:
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("Уведомление отправлено!")
Направления дальнейшего развития
- Обработка изображений и файлов: добавьте обработчики для медиафайлов, чтобы мониторить прикрепления.
- Фильтрация по участникам: отслеживайте только сообщения от конкретных пользователей.
- Интеграция с FastAPI: создайте REST API, чтобы управлять ботом через HTTP-запросы.
На этом этапе у вас уже есть полноценная система мониторинга Telegram-чата, которая не только отслеживает сообщения, но и отправляет уведомления. Такая система может быть полезной для администрирования больших групп, автоматизации задач и борьбы со спамом.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ