Бот для отслеживания удаленных постов

Ботов, которые «следят» за каналами или даже пересылают их посты куда-то в укромное место, полным полно. Это очень удобно, ведь позволяет сохранить инфу, которая может быть удалена. Да что уж там — в свое время мы даже сделали бот для сохранения удаленных сообщений в Telegram. Вот только у всех этих ботов обычно есть существенный недостаток — мелочь, которая сильно снижает их потенциал.

Они сохраняют посты, но не следят за тем, был ли пост удален. Как итог — нужно вручную все просматривать. И это ок, если вы следите за парочкой каналов. А если за парой десятков или сотен? Было бы супер, если бы за этим следил бот, да? Ну вот именно такой бот мы и решили создать. А заодно рассказать вам, как сделать такой же бот для себя любимых.

Погнали!

Какие задачи решает бот для отслеживания удаленных постов

Собственно, этих задач всего две, а именно:

• Сохранение информации от возможного удаления.

• Оповещение в случае этого самого удаления.

Хотя чисто технически бот также решает задачу отделения постов с вложениями от постов без вложений, но юзер этого никак не узнает, не залезая боту под капот. Ну или не пробуя сделать что-то подобное самостоятельно — без гайдов.

Так что если бы юзер попробовал, он бы узнал, что каждое вложение в Телеге — это отдельный пост, и, чтобы пост с кучей картинок при сохранении не превратился в 9 разных постов, нужен фильтр. Но внешне это никак не проявляется, кроме, собственно, результата. А потому и в список решаемых ботом задач не включено.

Принцип работы бота для отслеживания удаленных постов

Что касается алгоритма, здесь все достаточно просто, хоть и объемно:

• Бот считывает из своего же кода инфу о чате для управления и канале для пересылки — решение зашить в код призвано исключить риск того, что пользователь перепутает чаты и пошлет что-то не туда.

• Слушает эфир и реагирует на команды в чате для управления.

• При вводе команды /add — добавляет канал в список отслеживаемых.

• При вводе команды /del — удаляет канал из списка отслеживаемых.

• При вводе команды /list — выводит список отслеживаемых.

• При появлении новых постов в каналах из списка отслеживаемых — пересылает их в канал для пересылки, параллельно создавая пару «ссылка на оригинал» + «ссылка на копию» в JSON-файле.

• Раз в час бот перебирает «ссылки на оригинал», проверяя, не был ли удален тот или иной пост.

• Если пост был удален — оповещает об этом в чате для управления, выводя соответствующую «ссылке на оригинал» (ныне удаленный) «ссылку на копию» из канала для пересылки.

Ну и еще удаляет инфу о постах старше недели, ибо зачем за ними следить?

Пошаговая инструкция, как сделать бот для отслеживания удаленных постов

Чтобы сделать такой бот, естественно, нужен Python-сервер. Если у вас такого еще нет — вот инфа, как сделать. Ну или можно просто арендовать нормальный сервер. Затем сервер нужно настроить. Для этого вводим:

pip install telethon

pip install asyncio

В нашем случае мы ввели еще и:

python.exe -m pip install --upgrade pip

Нужно, чтобы обновить установленную библиотеку. Но если вы делаете это впервые — будет достаточно первой команды.

Затем, после того как все установится, делаем вот что:

1. Открываем веб-версию Телеги.
2. Жмем «API development tools».
3. Вводим инфу в поля анкеты.
4. Сохраняем свои api_id и api_hash.

После этого:

1. Создаем bot.py в корневой папке сервера.

2. Копируем туда:

import asyncio

import re

from datetime import datetime, timedelta

from telethon.sync import TelegramClient, events

from telethon.tl.functions.channels import JoinChannelRequest

api_id = 'ваш API ID'

api_hash = 'ваш API Hash'

destination_chat_id = -ID чата для управления

destination_chan_id = -100ID канала для сохранения

client = TelegramClient('session_name', api_id, api_hash)

timers = {}

message_ids = {}

tracked_messages = []

CHECK_INTERVAL = 60 * 100

MAX_AGE_DAYS = 7

async def add_or_delete_or_list_channels(event):

file_path = 'source_channels.txt'

message = event.message.message

channel_link = re.search(r'https?://\S+', message)

if event.pattern_match.group(0) == '/list':

try:

with open(file_path, 'r') as file:

source_channel_links = [line.strip() for line in file.readlines()]

await event.respond('\n'.join(source_channel_links) or 'Список каналов пуст.')

except FileNotFoundError:

await event.respond('Файл со списком каналов не найден.')

elif channel_link:

try:

with open(file_path, 'r') as file:

source_channel_links = [line.strip() for line in file.readlines()]

if channel_link.group(0) not in source_channel_links:

if event.pattern_match.group(0) == '/add':

try:

entity = await client.get_entity(channel_link.group(0))

await client(JoinChannelRequest(entity))

if entity:

with open(file_path, 'a') as file:

file.write(channel_link.group(0) + '\n')

await event.respond('Ссылка на канал успешно добавлена!')

else:

await event.respond('Канал по указанной ссылке не существует.')

except Exception as e:

await event.respond(f'Ошибка при добавлении канала: {e}')

else:

if event.pattern_match.group(0) == '/add':

await event.respond('Ссылка на канал уже существует в списке.')

elif event.pattern_match.group(0) == '/del':

if channel_link.group(0) in source_channel_links:

source_channel_links.remove(channel_link.group(0))

source_channel_links = list(filter(None, source_channel_links))

with open(file_path, 'w') as file:

file.write('\n'.join(source_channel_links) + '\n')

await event.respond('Ссылка на канал успешно удалена из списка.')

else:

await event.respond('Указанный канал не найден в списке.')

except FileNotFoundError:

await event.respond('Файл со списком каналов не найден.')

else:

await event.respond('Не удалось извлечь ссылку на канал из сообщения или неверная команда.')

async def forward_message(event):

message = event.message

sender_id = message.sender_id

if message.forward or message.fwd_from:

print("Это репост!")

return

if sender_id not in message_ids:

message_ids[sender_id] = []

message_ids[sender_id].append(message.id)

async def forward_message_after_delay():

forwarded = await client.forward_messages(destination_chan_id, message_ids[sender_id], from_peer=sender_id)

for msg_id in message_ids[sender_id]:

tracked_messages.append({

"channel_id": message.chat_id,

"message_id": msg_id,

"forwarded_id": forwarded.id if hasattr(forwarded, "id") else None,

"timestamp": datetime.now()

})

del message_ids[sender_id]

print("Таймер удален и сообщение добавлено в tracked_messages")

if sender_id in timers and not timers[sender_id].done():

timers[sender_id].cancel()

timers[sender_id] = asyncio.create_task(asyncio.sleep(5))

print("IDs сообщений:", message_ids[sender_id])

print("Ожидание завершения таймера...")

await timers[sender_id]

print("Таймер завершен.")

await forward_message_after_delay()

async def check_deleted_messages():

while True:

now = datetime.now()

to_remove = []

for msg in tracked_messages:

age = now - msg["timestamp"]

if age > timedelta(days=MAX_AGE_DAYS):

to_remove.append(msg)

continue

m = await client.get_messages(msg["channel_id"], ids=msg["message_id"])

if m is None:

text = f"Сообщение {msg['message_id']} удалено в канале {msg['channel_id']}"

await client.send_message('me', text)

to_remove.append(msg)

print(f"Сообщение {msg['message_id']} удалено — алерт отправлен в Избранное")

for r in to_remove:

tracked_messages.remove(r)

await asyncio.sleep(CHECK_INTERVAL)

async def main():

await client.start()

await update_sources()

client.add_event_handler(add_or_delete_or_list_channels,

events.NewMessage(chats=destination_chat_id, pattern=r'(/add|/del|/list)'))

client.add_event_handler(update_sources,

events.NewMessage(chats=destination_chat_id, pattern=r'(/upd|/add|/del)'))

asyncio.create_task(check_deleted_messages())

await client.run_until_disconnected()

async def update_sources(event=None):

file_path = 'source_channels.txt'

client.remove_event_handler(forward_message)

try:

with open(file_path, 'r') as file:

source_channel_links = [line.strip() for line in file.readlines()]

except FileNotFoundError:

await client.send_message(destination_chat_id, 'Файл со списком каналов не найден.')

return

source_channel_ids = await get_channel_ids(source_channel_links)

client.add_event_handler(forward_message, events.NewMessage(chats=source_channel_ids))

async def get_channel_ids(links):

channel_ids = []

for link in links:

entity = await client.get_entity(link)

channel_ids.append(entity.id)

return channel_ids

asyncio.get_event_loop().run_until_complete(main())

3. Заменяем api_id = 'ваш API ID', api_hash = 'ваш API Hash', destination_chat_id = -ID чата для управления и destination_chan_id = -100ID канала для сохранения на свои данные. Обратите внимание, что перед ID чата знак минус, а перед ID канала -100 — это важно.

4. Запускаем бот командой python bot.py

5. Проверяем, все ли работает.

Демонстрация работы

Все работает!

Подводя итоги

Как видите, создать бот для отслеживания удаленных постов не так уж и сложно — достаточно следить за нашей Телегой, вовремя узнавать про новые боты и просто следовать инструкции.