¶ Предварительные требования

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

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

Как получить ID канала не выходя из Telegram

Самое простое - скопируйте ссылку на пост. Получите ссылку вида https://t.me/c/<ID канала>/<номер поста>. ID-канала получен!

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

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

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

В руководстве я буду использовать библиотеку telethon версии 1.38.0. Обратите внимание на совместимость версий, если используете отличную.

pip install telethon==1.38.0

¶ Скрипт

Данный скрипт обрабатывает только текстовые сообщения и посты с фото. Из-за особенностей telethon посты с фото “разбиваются” на несколько.

Полученные посты сохраняются в файл messages.json, в нем содержится:

• ID поста
• Дата публикации в UNIX-timestamp
• Текст поста
• Изображение поста (если есть)

Фото к сообщениям сохраняются в папку images. Предварительно создайте ее.

from telethon import TelegramClient
from telethon.helpers import TotalList
import asyncio
import json
from telethon.types import Message, Photo, MessageMediaPhoto
from telethon.tl.types import PeerChannel
import logging

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

logging.basicConfig(level=logging.INFO)

app = TelegramClient(
    session="userbot",
    api_id=0,  # INSERT API_ID
    api_hash="YOUR_HASH",
)

CHANNEL_ID = 0  # YOUR CHANNEL
payload: list[dict] = []


async def main(client: TelegramClient):
    min_id = 0
    max_id = 100

    while True:
        logging.info("Getting messages from {} to {}".format(min_id, max_id))
        res: TotalList | None | Message = await client.get_messages(
            PeerChannel(CHANNEL_ID),
            limit=(max_id - min_id),
            max_id=max_id,
            min_id=min_id,
        )
        assert isinstance(res, TotalList)
        logging.info("Got {} messages".format(len(res)))

        if len(res) == 0:
            break

        message: Message
        for message in reversed(res):
            media: str | None = None
            date: float | None = None
            message_text = ""
            if message.date is not None:
                date = message.date.timestamp()

            if isinstance(message.message, str) and len(message.message) > 0:
                message_text += message.message

            if (
                isinstance(message.media, MessageMediaPhoto)
                and message.media.photo is not None
                and isinstance(message.media.photo, Photo)
            ):
                path = "./images/{}.jpg".format(message.id)
                await client.download_media(message.media, file=path)  # type: ignore
                media = path
            elif message.media is not None:
                pass

            payload.append(
                {"id": message.id, "text": message_text, "media": media, "date": date}
            )
        with open("messages.json", "w", encoding="utf-8") as file:
            file.write(json.dumps(payload))
        max_id += 100
        min_id += 100


if __name__ == "__main__":
    with app as client:
        loop.run_until_complete(main(client))

¶ Объяснение работы скрипта

¶ Запуск функции

Создаем event loop для обработки нашей асинхронной функции и устанавливаем его.

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

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

logging.basicConfig(level=logging.INFO)

app = TelegramClient(
    session="userbot",
    api_id=0,  # INSERT API_ID
    api_hash="YOUR_HASH",
)

Создаем переменную, где хранится ID канала, который мы “читаем” и переменную, куда будут записываться все сообщения

CHANNEL_ID = 0  # YOUR CHANNEL
payload: list[dict] = []

В конце скрипта добавляем открытие подключения клиентом при помощи with и запускаем нашу функцию с переданным клиентом в ранее созданном event loop’e.

if __name__ == "__main__":
    with app as client:
        loop.run_until_complete(main(client))

¶ Функция обработки

Для соблюдения rate-лимитов и большей гибкости скрипт получает сообщения по 100 сообщений максимум (если посты не были удалены ранее, в таком случае меньше). Стандартная редакция скрипта начинает получения от постов с ID от 0 до 100 (т.е первые 100 сообщений в канале).

ID поста можно получить, скопировав ссылку на него.

24
async def main(client: TelegramClient):
25
    min_id = 0
26
    max_id = 100
27

28
    while True:
29
        logging.info("Getting messages from {} to {}".format(min_id, max_id))
30
        res: TotalList | None | Message = await client.get_messages(
31
            PeerChannel(CHANNEL_ID),
32
            limit=(max_id - min_id),
33
            max_id=max_id,
34
            min_id=min_id,
35
        )
36
        assert isinstance(res, TotalList)
37
        logging.info("Got {} messages".format(len(res)))

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

38
if len(res) == 0:
39
    break

Посты идут в обратном хронолическом порядке по умолчанию. Если у объекта сообщения есть текст и дата, они записываются в переменные для дальнейшей записи в payload.

42
message: Message
43
for message in reversed(res):
44
    media: str | None = None
45
    date: float | None = None
46
    message_text = ""
47
    if message.date is not None:
48
        date = message.date.timestamp()
49

50
    if isinstance(message.message, str) and len(message.message) > 0:
51
        message_text += message.message

Далее идет обработка медиавложений, с которой все сложнее. Поле media объекта Message может быть None, а также иметь следующий тип:

TypeMessageMedia = Union[
    MessageMediaEmpty,
	MessageMediaPhoto,
	MessageMediaGeo,
	MessageMediaContact,
	MessageMediaUnsupported,
	MessageMediaDocument,
	MessageMediaWebPage,
	MessageMediaVenue,
	MessageMediaGame,
	MessageMediaInvoice,
	MessageMediaGeoLive,
	MessageMediaPoll,
	MessageMediaDice,
	MessageMediaStory,
	MessageMediaGiveaway,
	MessageMediaGiveawayResults,
	MessageMediaPaidMedia
    ]

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

Обработка изображений также усложняется из-за системы типов, поскольку тип MessageMediaPhoto имеет опциональное поле photo, содержащее в себе тип TypePhoto выглядящий следующим образом:

TypePhoto = Union[PhotoEmpty, Photo]

Таким образом, проверка ожидает что у сообщения есть медиавложение в виде фото, в котором информация о фото присутствует. В таком случае клиент скачивает медиафайл в папку images с именем в виде ID сообщения. После, вся полученная информация добавляется в массив payload.

53
if (
54
    isinstance(message.media, MessageMediaPhoto)
55
    and message.media.photo is not None
56
    and isinstance(message.media.photo, Photo)
57
):
58
    path = "./images/{}.jpg".format(message.id)
59
    await client.download_media(message.media, file=path)  # type: ignore
60
    media = path
61
elif message.media is not None:
62
    pass
63

64
payload.append(
65
    {"id": message.id, "text": message_text, "media": media, "date": date}
66
)

В конце, мы сдвигаем порог ID постов на 100 и записываем имеющиеся данные в файл, чтобы они не “потерялись” в случае ошибки.

67
with open("messages.json", "w", encoding="utf-8") as file:
68
    file.write(json.dumps(payload))
69
max_id += 100
70
min_id += 100

¶ Заключение

Данный скрипт может развиваться под различные применения, поэтому я надеюсь что он полслужит отправной точкой для вашей задачи и поможет совладать с тяжелой структурой Telethon'а. Удачи!