Во время разработки одного из проектов появилась необходимость автоматического получения постов из Telegram канала, на который пользователь подписан. Со стороны ботов это возможно только при условии участия бота в администрации канала, что не предоставлялось возможным. Обходным решением может быть создание бота на основе пользовательского аккаунта при помощи библиотеки Pyrogram.

Также, во время реализации бота на Pyrogram попалась библиотека Telethon. Она поддерживается, работает в некоторых ситуациях стабильнее, но куда хуже задокументирована с позиции типов, что может усложнить процесс разработки, особенно если вы используете аннотацию типов, как я.

¶ Создание приложения

Прежде чем начать разработку бота, потребуется создать приложение для Telegram на специальной странице. Посредством этого приложения будет получен доступ к клиентскому API Telegram

После аутентификации, перейдите в раздел API development tools. Далее, потребуется создать приложение с произвольными данными.

После создания приложения, потребуется сохранить параметры api_id и api_hash.

¶ Подготовка окружения

¶ Создание файла окружения

Создайте файл .env, в котором будут описаны все требуемые параметры:

# api_id приложения Telegram
USERBOT_APP_ID=123 
# api_hash приложения Telegram
USERBOT_APP_HASH=abc
# мобильный телефон аккаунта, который вы будете использовать. Используется только в Pyrogram
USER_PHONE=88005553535

¶ Установка зависимостей

Для обоих библиотек потребуется вспомогательная библиотека для загрузки параметров из .env, я буду использовать python-dotenv

# для pip
pip install python-dotenv==1.0.1

Далее, установите требуемую библиотеку, которую хотите использовать:

# установка pyrogram
pip install pyrogram==2.0.106
# установка telethon
pip install telethon==1.37.0

¶ Как долго живет сессия?

Зависит от настроек аккаунта. Сессия Pyrogram/telethon регистрируется как полноценная сессия в Telegram и может быть прервана досрочно или по тайм-ауту активности, который можно установить в пользовательских настройках.

¶ Как получить ID канала?

С точки зрения клиента, пост в канале - сообщение в отдельном чате, поэтому в первую очередь потребуется ID канала. Один из простых способов для этого - переслать сообщение боту @ShowJsonBot. Далее, найти поле forward_origin и взять оттуда значение id. В примере ниже - это значение -1001112223333. Для Pyrogram потребуется полное значение, для telethon уберите -100 в начале, то есть оставьте значение 1112223333

...
  "forward_origin": {
   "type": "channel",
   "chat": {
    "id": -1001112223333,
    "title": "TestChannel",
    "type": "channel"
   },
   "message_id": 145151243,
   "date": 1735919220
  }
...

Поместите ID канала в переменную CHANNEL_ID в .env файле.

¶ Разработка на Pyrogram

¶ Получение файла сессии

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

from dotenv import load_dotenv
import asyncio
from pyrogram.client import Client
import os

load_dotenv()  # загружаем .env

API_ID = os.environ["USERBOT_APP_ID"]
API_HASH = os.environ["USERBOT_APP_HASH"]
ACCOUNT_PHONE = os.environ["USER_PHONE"]


async def main():
    async with Client(
        name="userbot", api_id=API_ID, api_hash=API_HASH, phone_number=ACCOUNT_PHONE
    ) as app:
        await app.send_message("me", "It worked!")


if __name__ == "__main__":
    asyncio.run(main())

После запуска скрипта, вы получите примерно следующий вывод, где потребуется ввести код подтверждения из Telegram и облачный пароль, если он у вас настроен. Если все отработает правильно, вы увидите в Избранных сообщение It worked!

TgCrypto is missing! Pyrogram will work the same, but at a much slower speed. More info: https://docs.pyrogram.org/topics/speedups
Welcome to Pyrogram (version 2.0.106)
Pyrogram is free software and comes with ABSOLUTELY NO WARRANTY. Licensed
under the terms of the GNU Lesser General Public License v3.0 (LGPL-3.0).

The confirmation code has been sent via Telegram app
Enter confirmation code: <ВАШ КОД>
The two-step verification is enabled and a password is required
Password hint: None
Enter password (empty to recover): <ОБЛАЧНЫЙ ПАРОЛЬ>

После выполнения скрипта будет создан файл userbot.session, который будет использоваться для управления аккаунтом. Если вы держите скрипт в другой директории (например, в папке src) лучше дополнительно указать параметр workdir для Client чтобы файл сессии не сохранялся в директорию кода. Следующий пример подойдет, если вы запускаете код из одной директории

# ...

async def main():
    async with Client(
        name="userbot",
        api_id=API_ID,
        api_hash=API_HASH,
        phone_number=ACCOUNT_PHONE,
        workdir="./",
    ) as app:
        await app.send_message("me", "It worked!")

# ...

¶ Отслеживаем определенный канал

После этого, перепишем код и создадим функцию-обработчик, которая будет проверять, от требуемого чата это сообщение или нет.

from dotenv import load_dotenv
from pyrogram.types import Message
from pyrogram.client import Client
from pyrogram import filters
import os

load_dotenv()  # загружаем .env

API_ID = os.environ["USERBOT_APP_ID"]
API_HASH = os.environ["USERBOT_APP_HASH"]
ACCOUNT_PHONE = os.environ["USER_PHONE"]
CHANNEL_ID: int = int(os.environ["CHANNEL_ID"])

app = Client(
    name="userbot",
    api_id=API_ID,
    api_hash=API_HASH,
    phone_number=ACCOUNT_PHONE,
    workdir="./",
)


@app.on_message(filters.channel) # используем встроенный в библиотеку фильтр на сообщения из каналов
async def channel_handler(client: Client, message: Message):
    if message.sender_chat.id == CHANNEL_ID: # проверяем, нужный ли это канал
        if message.caption is not None: # проверяем, есть ли подпись к медиафайлу в сообщении
            print(message.caption)
        elif message.text is not None: # проверяем, есть ли у сообщения текст
            print(message.text)
        else:
            print("Сообщение без текста!")


if __name__ == "__main__":
    app.run()

Если вы хотите усложнить логику обработки, полезно будет изучить поля класса Message.

¶ Полезные ссылки

• Общая документация Pyrogram
• Документация Pyrogram по обработке событий
• Документация Pyrogram по использованию фильтров

¶ Разработка на telethon

¶ Получение файла сессии

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

from dotenv import load_dotenv
from telethon import TelegramClient
import os

load_dotenv()  # загружаем .env

API_ID = os.environ["USERBOT_APP_ID"]
API_HASH = os.environ["USERBOT_APP_HASH"]

app = TelegramClient(
    session="userbot",
    api_id=int(API_ID),
    api_hash=API_HASH,
)

if __name__ == "__main__":
    app.start()
    app.run_until_disconnected()

После запуска скрипта, вы получите примерно следующий вывод, где потребуется ввести код подтверждения из Telegram и облачный пароль, если он у вас настроен. Если все отработает правильно, вы увидите аналогичный вывод в конце с именем вашего аккаунта. Можете завершить работу скрипта при помощи <CTRL+C>

Please enter your phone (or bot token): <ВАШ НОМЕР>
Please enter the code you received: <ВАШ КОД>
Please enter your password: <ОБЛАЧНЫЙ ПАРОЛЬ>
Signed in successfully as <ИМЯ ВАШЕГО АККАУНТА>; remember to not break the ToS or you will risk an account ban!

После выполнения скрипта будет создан файл userbot.session, который будет использоваться для управления аккаунтом. Файл создается по умолчанию в рабочей директории, откуда вы запускаете скрипт.

¶ Отслеживаем определенный канал

После этого, перепишем код и создадим функцию-обработчик, которая будет проверять, от требуемого чата это сообщение или нет.

from dotenv import load_dotenv
from telethon import TelegramClient, events, types
from telethon.tl.types import PeerChannel
import os

load_dotenv()  # загружаем .env

API_ID = os.environ["USERBOT_APP_ID"]
API_HASH = os.environ["USERBOT_APP_HASH"]
CHANNEL_ID: int = int(os.environ["CHANNEL_ID"])

app = TelegramClient(
    session="userbot",
    api_id=int(API_ID),
    api_hash=API_HASH,
)


@app.on(events.NewMessage(incoming=True))
async def handler(event: events.NewMessage.Event):
    message: types.Message = event.message
    text = message.message
    peer_id = event.message.peer_id
    if not isinstance(peer_id, PeerChannel):
        return

    channel_id = peer_id.channel_id
    if channel_id == CHANNEL_ID:
        if text is None or len(text) == 0:
            print("Нет текста!")
            return
        print(text)


if __name__ == "__main__":
    app.start()
    app.run_until_disconnected()

Флаг incoming отвечает за фильтрацию входящих сообщений для исключения отправленных нами сообщений.

¶ Полезные ссылки

• Документация Telethon
• Документация типа события NewMessage

¶ Заключение

Оба примера выводят текст сообщения в консоль, однако вы можете свободно расширить логику. Для своего проекта я помещаю сообщения в очередь RabbitMQ для дальшейшей обработки. Надеюсь, это руководство поможет вам в реализации ваших проектов!