Еще в 2020 году написал статью на vc.ru “Создание и развертывание ретранслятора Telegram каналов, используя Python и Heroku”. С тех пор, Heroku уже не предлагает бесплатный тариф. Но проект жив и обновляется.

</> Репозиторий проекта: https://github.com/khoben/telemirror

Собственно, нет ничего сверхъестественного в пересылке: получаем сообщение из канала-источника и отправляем его копию в канал-приёмник.

В простейшем случае это выглядит так:

from telethon import TelegramClient, events

client = TelegramClient("telemirror", api_id, api_hash)
SOURCE_CHANNELS = [-10001, -10002] # id каналов
TARGET_CHANNELS = [-10003, -10004]

@client.on(events.NewMessage(chats=SOURCE_CHANNELS))
async def handler(event):
    for target in TARGET_CHANNELS:
        await client.send_message(target, event.message)

client.start()
client.run_until_disconnected()

Приём сообщения → Отправка сообщения

Не хватает лишь хоть какой-то конфигурации и гибкости. Какие сообщения отбрасывать, изменять оригинальный текст, из какого канала в какой пересылать и т.д.

К простой схеме, описанной выше, добавляются модификаторы, которые можно назвать фильтрами.

В TeleMirror простейший фильтр для сообщений выглядит так:

class MessageFilter(Protocol):
    async def process(
        self, entity: EventEntity, event_type: Type[EventLike]
    ) -> FilterResult[EventEntity]:
        if isinstance(entity, EventMessage):
            # Обработка сообщения: продолжить, изменить, отбросить
            return await self._process_message(entity, event_type)

        if isinstance(entity, list):
            # Обработка группы сообщений - альбома
            return await self._process_album(entity, event_type)

        return FilterResult(FilterAction.CONTINUE, entity)

Фильтры могут видоизменять сообщения или отбрасывать их.

Такие фильтры можно объединить в группу, которая будет выполнять поочередно, обрабатывая сообщение:

class CompositeMessageFilter(MessageFilter):
    def __init__(self, filters: List[MessageFilter]) -> None:
        self._filters = filters

    async def process(
        self, entity: EventEntity, event_type: Type[EventLike]
    ) -> FilterResult[EventEntity]:
        for f in self._filters:
            filter_action, entity = await f.process(entity, event_type)
            match filter_action:
                case FilterAction.CONTINUE | True:
                    continue
                case FilterAction.DISCARD | False:
                    return FilterResult(FilterAction.DISCARD, entity)
                case FilterAction.FORCE_SEND:
                    return FilterResult(FilterAction.FORCE_SEND, entity)

        return FilterResult(FilterAction.CONTINUE, entity)

Приём сообщения → (фильтры) → Отправка (или нет) обработанного сообщения

Конфигурация для пересылки имеет вид:

CHAT_MAPPING: Dict[int, Dict[int, List["DirectionConfig"]]] = {}

@dataclass
class DirectionConfig:
    disable_delete: bool
    disable_edit: bool
    filters: MessageFilter
    from_topic_id: Optional[int] = None
    to_topic_id: Optional[int] = None
    mode: Literal["copy", "forward"] = "copy"

В общем, конфигурация для ретранслирования состоит из маппинга каналов (from->to) и конфига для этого направления.

Пример YAML конфигурации

# Направления пересылки
directions:
  - from: [-1001, -1002, -1003] # Каналы источники
    to: [-100203] # Каналы приёмники

  - from: [-1000#3] # Пересылка из топика в топик
    to: [-1001#4]

  - from: [-100226]
    to: [-1006, -1008]
    disable_edit: false
    disable_delete: false
    mode: forward  # Вид пересылки: копирование или пересылка
    filters: # Список фильтров, который будет применен поочередно
      - UrlMessageFilter:
          blacklist: !!set
            ? t.me
      - KeywordReplaceFilter:
          keywords:
            "google.com": "bing.com"
            "r'google\\.com.*'": "bing.com"
      - SkipWithKeywordsFilter:
          keywords: !!set
            ? "stopword"
            ? "r'badword.*'"

</> Подробнее смотрите в репозитории проекта: https://github.com/khoben/telemirror

Возможные проблемы

Бан аккаунта или деактивация сессии

Ретранслятор ведёт себя не как обычный пользователь: всегда онлайн, не смотрит рекламу и т.д. В качестве юзербота для ретранслятора используйте не свой основной аккаунт. Для вновь созданных аккаунтов большой риск бана, стоит некоторое время попользоваться им в режиме обычного пользователя.

Новые сообщения не приходят

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

Где запустить

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