Парсер Telegram на Python: Bot API, сбор сообщений, фильтрация данных

Прошлой зимой мне нужно было отслеживать упоминания конкурентов в пяти отраслевых Telegram-каналах. Копировать вручную по 200 сообщений в день не вариант. За вечер я написал бота, который сам собирал сообщения, фильтровал по ключевым словам и складывал в CSV. Весь парсер уместился в 80 строк Python-кода. Парсер Telegram через Bot API это программа, которая подключается к Telegram как бот, получает сообщения из каналов и групп, и сохраняет нужные данные.

В этой статье разбираю пошагово: создание бота, подключение к группам, сбор и фильтрацию сообщений, сохранение данных и ограничения Bot API.

Что может Bot API и что нет

Bot API это HTTP-интерфейс для управления Telegram-ботами. Бот может получать входящие сообщения, отвечать на команды, отправлять файлы. Для парсинга у Bot API есть границы, которые стоит знать заранее.

Возможности

Получать все сообщения из группы, если бот добавлен как администратор
Получать посты из канала, если бот добавлен как администратор
Фильтровать сообщения по типу: текст, фото, документы, стикеры
Работать через polling (опрос) или webhooks (push-уведомления)

Ограничения

Нельзя получить историю сообщений до добавления бота
Обновления хранятся на сервере Telegram не дольше 24 часов
getUpdates возвращает максимум 100 обновлений за запрос
В режиме приватности (по умолчанию) бот в группе видит только команды и ответы на свои сообщения

Если нужен доступ к полной истории канала или чтение чужих приватных чатов, Bot API не подойдёт. Для этого существуют Telethon и Pyrogram, которые работают через Telegram Client API от имени пользователя. Но Client API сложнее в настройке и несёт риски блокировки аккаунта.

Шаг 1: создаём бота через BotFather

BotFather это официальный бот Telegram для создания и настройки других ботов.

Откройте Telegram, найдите @BotFather.
Отправьте команду /newbot.
Введите имя бота (например, "Мой Парсер").
Введите username бота (например, my_parser_2026_bot). Должен заканчиваться на bot.
BotFather пришлёт токен вида 7123456789:AAH...xyz.

Этот токен это ключ доступа к вашему боту. Не публикуйте его в открытом коде.

Отключаем режим приватности

По умолчанию бот в группе видит только команды (сообщения, начинающиеся с /), ответы на свои сообщения и служебные уведомления. Для парсинга всех сообщений нужно отключить privacy mode.

Напишите @BotFather команду /setprivacy.
Выберите своего бота.
Выберите Disable.

После этого бот будет получать все сообщения в группах, где он состоит. Для каналов бот должен быть добавлен администратором.

Шаг 2: устанавливаем библиотеку

pip install python-telegram-bot

Библиотека python-telegram-bot это обёртка над HTTP-запросами к Bot API. Версия 22+ использует асинхронный код (async/await).

Шаг 3: минимальный парсер — логируем все сообщения

import asyncio
from telegram import Update
from telegram.ext import Application, MessageHandler, filters, ContextTypes


TOKEN = '7123456789:AAH...xyz'  # ваш токен


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = update.effective_message
    if message is None:
        return

    chat_title = update.effective_chat.title or 'Личный чат'
    username = message.from_user.username if message.from_user else 'Unknown'
    text = message.text or '[медиа без текста]'
    date = message.date.strftime('%Y-%m-%d %H:%M:%S')

    print(f'[{date}] {chat_title} | @{username}: {text[:100]}')


def main():
    app = Application.builder().token(TOKEN).build()
    app.add_handler(MessageHandler(filters.ALL, handle_message))
    print('Парсер запущен. Ctrl+C для остановки.')
    app.run_polling(allowed_updates=['message', 'channel_post'])


if __name__ == '__main__':
    main()

MessageHandler с фильтром filters.ALL перехватывает каждое входящее сообщение. Параметр allowed_updates указывает, какие типы обновлений получать: message для групп и личных чатов, channel_post для каналов.

Как запустить

Добавьте бота в нужную группу (или канал как администратора).
Запустите скрипт: python parser.py.
Отправьте сообщение в группу.

Вывод в консоли:

Парсер запущен. Ctrl+C для остановки.
[2026-04-07 12:30:15] Python Чат | @ivan_dev: Кто подскажет, как работает asyncio?
[2026-04-07 12:30:42] Python Чат | @anna_py: Посмотри документацию по event loop

Шаг 4: сохраняем сообщения в CSV

import csv
import os
from datetime import datetime
from telegram import Update
from telegram.ext import Application, MessageHandler, filters, ContextTypes


TOKEN = '7123456789:AAH...xyz'
CSV_FILE = 'messages.csv'


def init_csv():
    if not os.path.exists(CSV_FILE):
        with open(CSV_FILE, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(
                ['date', 'chat_id', 'chat_title', 'user_id', 'username', 'text']
            )


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = update.effective_message
    if message is None:
        return

    row = [
        message.date.strftime('%Y-%m-%d %H:%M:%S'),
        update.effective_chat.id,
        update.effective_chat.title or '',
        message.from_user.id if message.from_user else '',
        message.from_user.username if message.from_user else '',
        message.text or '',
    ]

    with open(CSV_FILE, 'a', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(row)


def main():
    init_csv()
    app = Application.builder().token(TOKEN).build()
    app.add_handler(MessageHandler(filters.ALL, handle_message))
    print('Парсер запущен. Данные сохраняются в', CSV_FILE)
    app.run_polling(allowed_updates=['message', 'channel_post'])


if __name__ == '__main__':
    main()

Функция init_csv() создаёт файл с заголовками, если его нет. Каждое сообщение дописывается новой строкой. Режим "a" (append) не перезаписывает старые данные.

Результат в CSV

date,chat_id,chat_title,user_id,username,text
2026-04-07 12:30:15,-1001234567890,Python Чат,123456,ivan_dev,"Кто подскажет, как работает asyncio?"
2026-04-07 12:30:42,-1001234567890,Python Чат,789012,anna_py,Посмотри документацию по event loop

Шаг 5: фильтрация по ключевым словам

Часто нужны не все сообщения, а только содержащие определённые слова.

KEYWORDS = ['python', 'django', 'fastapi', 'flask']


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = update.effective_message
    if message is None or message.text is None:
        return

    text_lower = message.text.lower()

    # Проверяем наличие хотя бы одного ключевого слова
    matched = [kw for kw in KEYWORDS if kw in text_lower]
    if not matched:
        return

    row = [
        message.date.strftime('%Y-%m-%d %H:%M:%S'),
        update.effective_chat.title or '',
        message.from_user.username if message.from_user else '',
        message.text,
        ', '.join(matched),
    ]

    with open('filtered.csv', 'a', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(row)

    print(f'[Совпадение: {matched}] {message.text[:80]}')

Список matched содержит ключевые слова, найденные в сообщении. Если список пуст, сообщение пропускается. Совпавшие ключевые слова сохраняются в отдельной колонке CSV для дальнейшего анализа.

Шаг 6: фильтрация по типу контента

Библиотека python-telegram-bot предоставляет встроенные фильтры для разных типов сообщений.

from telegram.ext import MessageHandler, filters


# Только текстовые сообщения (без команд)
text_handler = MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text)

# Только фотографии
photo_handler = MessageHandler(filters.PHOTO, handle_photo)

# Только документы (файлы)
doc_handler = MessageHandler(filters.Document.ALL, handle_document)

# Только пересланные сообщения
forwarded_handler = MessageHandler(filters.FORWARDED, handle_forwarded)

# Комбинация: текст или фото
combined_handler = MessageHandler(filters.TEXT | filters.PHOTO, handle_text_or_photo)

Фильтры комбинируются через побитовые операторы: & (и), | (или), ~ (не) .

Скачивание фотографий

import os


async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = update.effective_message
    if not message.photo:
        return

    # Берём фото максимального размера (последний элемент списка)
    photo = message.photo[-1]
    file = await context.bot.get_file(photo.file_id)

    os.makedirs('photos', exist_ok=True)
    filename = (
        f'photos/{message.date.strftime('%Y%m%d_%H%M%S')}_{photo.file_id[:8]}.jpg'
    )
    await file.download_to_drive(filename)

    print(f'Сохранено фото: {filename}')

message.photo это список объектов PhotoSize разного разрешения. Последний элемент содержит максимальное качество. get_file() запрашивает ссылку на файл у Telegram, download_to_drive() скачивает его на диск.

Шаг 7: мониторинг нескольких каналов

Бот автоматически получает сообщения из всех групп и каналов, где он состоит. Фильтрация по конкретным чатам делается через chat_id.

# Список chat_id для мониторинга (отрицательные числа для групп/каналов)
MONITORED_CHATS = {
    -1001234567890: 'Python Чат',
    -1001987654321: 'Data Science RU',
    -1001555666777: 'AI News',
}


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id

    # Парсим только целевые чаты
    if chat_id not in MONITORED_CHATS:
        return

    message = update.effective_message
    if message is None or message.text is None:
        return

    print(f'[{MONITORED_CHATS[chat_id]}] {message.text[:100]}')
    # ... сохранение в CSV

Как узнать chat_id

Добавьте бота в нужную группу и отправьте туда сообщение. Временно добавьте в обработчик строку:

print(f'Chat ID: {update.effective_chat.id}, Title: {update.effective_chat.title}')

Chat ID групп и каналов начинается с -100.

Polling vs Webhooks

Два способа получения обновлений от Telegram:

Характеристика	Polling (getUpdates)	Webhooks (setWebhook)
Принцип	Бот опрашивает сервер	Telegram присылает данные боту
Нужен сервер	Нет (работает на ноутбуке)	Да (нужен HTTPS-домен)
Задержка	0.5-2 секунды	Мгновенно
Простота настройки	Одна строка кода	Нужен SSL и белый IP
Подходит для	Разработки, малых нагрузок	Продакшена

Для парсера на этапе разработки используйте polling (run_polling). Для постоянной работы на сервере переходите на webhooks.

Webhook-версия

from telegram.ext import Application


app = Application.builder().token(TOKEN).build()
app.add_handler(MessageHandler(filters.ALL, handle_message))

# Запуск с webhook
app.run_webhook(
    listen='0.0.0.0',
    port=8443,
    url_path=TOKEN,
    webhook_url=f'https://your-domain.com/{TOKEN}',
)

Telegram отправляет обновления POST-запросом на указанный URL. Использование токена в URL предотвращает получение поддельных запросов.

Сохранение в SQLite вместо CSV

Для больших объёмов данных CSV становится неудобным. SQLite не требует отдельного сервера и работает из коробки.

import sqlite3


def init_db():
    conn = sqlite3.connect('messages.db')
    cur = conn.cursor()
    cur.execute(
        '''
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            date TEXT,
            chat_id INTEGER,
            chat_title TEXT,
            user_id INTEGER,
            username TEXT,
            text TEXT
        )
    '''
    )
    conn.commit()
    conn.close()


async def save_to_db(message, chat):
    conn = sqlite3.connect('messages.db')
    cur = conn.cursor()
    cur.execute(
        'INSERT INTO messages (date, chat_id, chat_title, user_id, username, text) '
        'VALUES (?, ?, ?, ?, ?, ?)',
        (
            message.date.strftime('%Y-%m-%d %H:%M:%S'),
            chat.id,
            chat.title or '',
            message.from_user.id if message.from_user else 0,
            message.from_user.username if message.from_user else '',
            message.text or '',
        ),
    )
    conn.commit()
    conn.close()

SQLite позволяет делать SQL-запросы к собранным данным: подсчитать сообщения по дням, найти самых активных авторов, отфильтровать по дате.

conn = sqlite3.connect('messages.db')
cur = conn.cursor()
cur.execute(
    '''
    SELECT username, COUNT(*) as cnt
    FROM messages
    WHERE username != ''
    GROUP BY username
    ORDER BY cnt DESC
    LIMIT 5
'''
)
for row in cur.fetchall():
    print(f'@{row[0]}: {row[1]} сообщений')
conn.close()

Неочевидные детали

Первый факт: бот в группе с включённым privacy mode видит только команды (/start, /help), ответы на свои сообщения и системные уведомления (вход/выход участников). Для полного парсинга privacy mode надо отключить или добавить бота администратором.

Второй факт: channel_post и message это разные типы обновлений. Если в allowed_updates указать только ["message"], бот не получит посты из каналов. Для каналов нужен "channel_post".

Третий факт: обновления на сервере Telegram живут 24 часа. Если парсер упал на сутки, все сообщения за это время потеряны. Для надёжности запускайте парсер через systemd или supervisor с автоматическим перезапуском.

Четвёртый факт: message.text равен None для сообщений с медиа без подписи (фото, стикер, голосовое). Код message.text.lower() упадёт с AttributeError. Проверяйте if message.text is None: return перед обработкой.

Пятый факт: getUpdates и webhooks взаимоисключающи. Нельзя использовать оба одновременно. При вызове setWebhook Telegram перестаёт накапливать обновления для getUpdates, и наоборот.

FAQ

Можно ли парсить канал, не будучи администратором?‍

Нет. Бот должен быть добавлен в канал как администратор, чтобы получать посты. Для групп достаточно быть участником (с отключённым privacy mode).

Как получить историю сообщений за прошлый месяц?‍

Bot API не предоставляет доступ к истории сообщений. Бот получает только новые сообщения, отправленные после его добавления. Для доступа к истории используйте Telethon или Pyrogram с Client API.

Сколько сообщений в секунду может обработать бот?‍

Telegram ограничивает ботов 30 сообщениями в секунду на отправку. На приём ограничений нет, но getUpdates возвращает максимум 100 обновлений за запрос. Для высоконагруженных парсеров используйте webhooks.

Как парсить приватную группу?‍

Добавьте бота в группу. Если группа приватная, вас должен пригласить администратор. После добавления бот работает так же, как в публичной группе.

Законно ли парсить Telegram-каналы?‍

Сбор публично доступных данных для личного анализа обычно допускается. Массовый сбор персональных данных (телефоны, ID пользователей) для перепродажи нарушает условия использования Telegram и законы о персональных данных. Проконсультируйтесь с юристом, если планируете коммерческое использование.

‍

Мой совет: начните с минимального парсера из Шага 3, убедитесь, что бот получает сообщения. Потом добавьте сохранение в CSV, фильтрацию по словам и мониторинг конкретных чатов. Bot API покрывает большинство задач парсинга "с текущего момента". Если нужна именно история сообщений, переключайтесь на Telethon, но будьте готовы к более сложной авторизации и рискам ограничений.

Автор

Артём Целин

Бэкенд-разработчик