Бот для создания автоскринов каналов

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

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

Какие задачи решает бот для создания автоскринов каналов

Казалось бы, здесь все просто: бот решает проблему внезапного удаления контента, который вы еще даже не прочли. Да, но нет. Вернее — не только лишь да. На самом деле бот может:

• Обеспечить справедливый доступ к информации — одно дело, не иметь доступа к инфе, и совсем другое, даже не знать о ее существовании. Увы, даже в эру умирающих медиа контент могут попросить удалить через 2 минуты после публикации. И вы даже не узнаете, что он был. А благодаря боту — узнаете.

• Обеспечить защиту репутации — представьте, что про вас опубликовали статью, которая вас задевает. А потом удалили. И вы всем рассказываете, что на ваше честное имя посягают, а вам банально никто не верит — ведь вы пиздабол у вас нет пруфов. Благодаря этому боту пруфы будут.

• Обеспечить рост компетенций в сфере — предположим, что свои лучшие годы вы положили на алтарь ликбеза в арбитраже. Вы часами штудировали работу разных разгильдяев, чтобы рассказать всем, что они перепутали «что-то там» с «чем-то там» — и все ради того, чтоб сфера развивалась! И вот после очередной неравной битвы с разгильдяйством и восторженного рукоплескания вебов под крики «ну кабан!» первоисточник вашего праведного гнева удалили. И теперь уже вашему посту не верят. А так — вот вам пруф!

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

Принцип работы бота для создания автоскринов каналов

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

А вот что мы проигнорить никак не можем, так это странную логику Телеги, разрешающей скринить сразу десятки постов без авторизации, но не дающей заскринить конкретный пост. И при всем этом фактически подсвечивающий его же зеленым фоном. Зачем? Почему? В чем смысл — мы не знаем. Но из-за этого пришлось утяжелить код «вырезалкой» нужной части скрина.

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

Но вернемся к алгоритму. Алгоритмически бот делает следующее:

1. Слушает эфир.
2. Реагирует на введенные в «Избранное» команды.
3. При вводе команды /add канал — добавляет канал в отслеживаемые.
4. При вводе команды /del канал — удаляет канал из отслеживаемых.
5. При вводе команды /list — выводит список отслеживаемых каналов.
6. При вводе команды /time цифра — устанавливает срок «жизни» скриншота (в базе задано 720 минут — то есть половина суток).
7. При вводе команды /show канал цифра — выводит заданное количество последних скриншотов для указанного канала (в базе 9 скринов).
8. При вводе команды /show канал all — выводит все скриншоты для указанного канала.
9. При вводе команд с ошибкой или несоблюдении формата — оповещает об этом.
10. Реагирует на новые посты в указанных каналах, проверяя, являются ли они одним постом (технически каждое вложение в Телеге — отдельный пост). Если да — группирует их. Если нет — не группирует.
11. Делает скриншот через веб-версию, выбирая новый пост в качестве целевого.
12. Обрезает лишнее и сохраняет скриншот.
13. Проверяет раз в 60 минут, не истек ли срок жизни скриншотов. Если истек — удаляет их.

Ничего сложного. Но, увы, и ничего простого.

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

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

pip install pyrogram tgcrypto pillow playwright

playwright install

После завершения установки необходимых модулей — делаем следующее:

1. Переходим в веб-версию.
2. Ищем там «API development tools».
3. Заполняем поля анкеты.
4. Копипастим куда-то api_id и api_hash.

После чего :

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

2. Вносим в него вот это:

import os

import re

import asyncio

from datetime import datetime, timedelta

from pyrogram import Client, filters, idle

from pyrogram.types import Message

from pyrogram.enums import ChatType

from playwright.async_api import async_playwright

from PIL import Image

api_id = ваш API_ID

api_hash = "ваш API_HASH"

session_name = "userbot"

channels_file = "channels.txt"

screens_dir = "screens"

os.makedirs(screens_dir, exist_ok=True)

screen_lifetime = timedelta(minutes=720)

recent_posts = {}

app = Client(session_name, api_id=api_id, api_hash=api_hash)

async def take_post_screenshot(channel_url: str, post_id: str):

username = channel_url.rstrip("/").split("/")[-1]

save_dir = os.path.join(screens_dir, username)

os.makedirs(save_dir, exist_ok=True)

file_path = os.path.join(save_dir, f"{post_id}.png")

temp_path = os.path.join(save_dir, f"temp_{post_id}.png")

web_url = f"https://t.me/s/{username}/{post_id}"

try:

async with async_playwright() as p:

browser = await p.chromium.launch()

page = await browser.new_page()

await page.goto(web_url, timeout=30000)

await asyncio.sleep(5)

await page.screenshot(path=temp_path, full_page=True)

await browser.close()

except Exception as e:

print(f"[WARN] Скрин не удался: {e}")

return

try:

im = Image.open(temp_path).convert("RGB")

w, h = im.size

pixels = im.load()

def color_dist(c1, c2):

return sum((a - b) ** 2 for a, b in zip(c1, c2)) ** 0.5

def is_edge_bg(y, tol=12):

return color_dist(pixels[0, y], pixels[w - 1, y]) < tol

def is_full_bg(y, tol=12, ratio=0.9):

base = pixels[0, y]

if color_dist(base, pixels[w - 1, y]) >= tol:

return False

same = sum(1 for x in range(w) if color_dist(pixels[x, y], base) < tol)

return same / w >= ratio

top = 0

for y in range(h):

if not is_edge_bg(y):

top = y

break

gap = 0

bottom = h

for y in range(top + 1, h):

if is_full_bg(y):

gap += 1

if gap >= 8:

bottom = y - gap + 1

break

else:

gap = 0

im.crop((0, top, w, bottom)).save(file_path)

os.remove(temp_path)

except Exception as e:

print(f"[WARN] Ошибка обрезки: {e}")

def validate_channel_format(url: str) -> bool:

return re.fullmatch(r"https://t\.me/[a-zA-Z0-9_]{5,32}", url) is not None

async def check_channel_exists(url: str) -> bool:

try:

chat = await app.get_chat(url.split("/")[-1])

return chat.type == ChatType.CHANNEL

except:

return False

@app.on_message(filters.me & filters.command("add"))

async def add_channel(_, message: Message):

if len(message.command) < 2:

return await message.reply("Пример: /add https://t.me/channel")

url = message.command[1].rstrip("/")

if not validate_channel_format(url):

return await message.reply("Неверный формат.")

if not await check_channel_exists(url):

return await message.reply("Это не канал.")

channels = []

if os.path.exists(channels_file):

channels = [x.strip() for x in open(channels_file)]

if url not in channels:

channels.append(url)

open(channels_file, "w").write("\n".join(channels))

await message.reply("Канал добавлен.")

@app.on_message(filters.me & filters.command("del"))

async def del_channel(_, message: Message):

if len(message.command) < 2:

return

url = message.command[1].rstrip("/")

if not os.path.exists(channels_file):

return

channels = [x.strip() for x in open(channels_file)]

if url in channels:

channels.remove(url)

open(channels_file, "w").write("\n".join(channels))

await message.reply("Удалён.")

@app.on_message(filters.me & filters.command("list"))

async def list_channels(_, message: Message):

if not os.path.exists(channels_file):

return await message.reply("Пусто.")

channels = [x.strip() for x in open(channels_file)]

await message.reply("\n".join(channels) if channels else "Пусто.")

@app.on_message(filters.me & filters.command("time"))

async def set_time(_, message: Message):

global screen_lifetime

try:

minutes = int(message.command[1])

screen_lifetime = timedelta(minutes=minutes)

await message.reply(f"Время жизни: {minutes} мин.")

except:

await message.reply("Нужно число минут.")

@app.on_message(filters.me & filters.command("show"))

async def show_screens(_, message: Message):

if len(message.command) < 2:

return await message.reply("Пример: /show канал [all|число]")

channel = message.command[1].rstrip("/")

num_arg = message.command[2] if len(message.command) > 2 else None

if not os.path.exists(channels_file):

return await message.reply("Список каналов пуст.")

channels = [x.strip() for x in open(channels_file)]

if channel not in channels:

return await message.reply("Канал не добавлен через /add.")

if num_arg == "all":

count = None

else:

try:

count = int(num_arg) if num_arg else 9

except:

count = 9

username = channel.split("/")[-1]

dir_path = os.path.join(screens_dir, username)

if not os.path.exists(dir_path):

return await message.reply("Скриншоты ещё не созданы.")

files = [os.path.join(dir_path, f) for f in os.listdir(dir_path) if f.endswith(".png")]

files.sort(key=lambda x: os.path.getmtime(x), reverse=True)

if count:

files = files[:count]

if not files:

return await message.reply("Нет скринов для показа.")

CHUNK = 9

from pyrogram.types import InputMediaPhoto

for i in range(0, len(files), CHUNK):

chunk_files = files[i:i+CHUNK]

media_group = [InputMediaPhoto(media=f) for f in chunk_files]

await app.send_media_group(chat_id=message.chat.id, media=media_group)

@app.on_message(filters.channel)

async def channel_listener(_, message: Message):

chat = message.chat

if chat.type != ChatType.CHANNEL or not chat.username:

return

url = f"https://t.me/{chat.username}"

if not os.path.exists(channels_file):

return

channels = [x.strip() for x in open(channels_file)]

if url not in channels:

return

post_id = str(message.id)

key = f"{chat.id}_{message.media_group_id or post_id}"

if recent_posts.get(key):

return

recent_posts[key] = True

await asyncio.sleep(3)

await take_post_screenshot(url, post_id)

async def cleanup_screens():

while True:

now = datetime.now()

for root, _, files in os.walk(screens_dir):

for name in files:

if not name.endswith(".png"):

continue

path = os.path.join(root, name)

created = datetime.fromtimestamp(os.path.getmtime(path))

if now - created >= screen_lifetime:

os.remove(path)

await asyncio.sleep(600)

def handle_async_exception(loop, context):

exc = context.get("exception")

if exc:

if isinstance(exc, ValueError) and "Peer id invalid" in str(exc):

return

loop.default_exception_handler(context)

if __name__ == "__main__":

asyncio.get_event_loop().set_exception_handler(handle_async_exception)

asyncio.get_event_loop().create_task(cleanup_screens())

app.start()

print("Userbot запущен")

idle()

app.stop()

3. Заменяем строки api_id = ваш API_ID и api_hash = "ваш API_HASH" на свои значения.

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

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

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

Заходим в избранное, пишем /list — видим, что 1 канал уже есть (его автор внес до скрина). Добавляем еще один.

В один канал пишем текст. Во второй кидаем картинки. Смотрим в папку со скринами — видим, что скрины созданы. Проверяем, что будет, если сделать много постов сразу.

Видим, что бот корректно успевает — в папке канала 1 уже 10 скринов. В папке канала 2 — два. Пытаемся их вывести, но видим «Нет скринов для показа».

Проверяем — действительно нет. Почему? Потому что в целях теста автор поставил срок жизни скрина — 1 минута (и забыл :D). Задаем срок жизни 10 минут, заодно проверяя работоспособности команды time. Повторяем эксперимент, но в этот раз спамим сразу в два канала. В один — текстом, во второй — вложениями.

Видим, что бот все обрабатывает. В целях демонстрации автор специально спамил одним и тем же вложением, чтоб бот думал, что это все тот же пост, и визуально «наращивал» скриншот. В реальности же никто не спамил бы одной и той же картинкой, и на скрине было бы по 2 поста — целевой и следующий за ним. Либо же 1 — если «следующего» нет.

Наигравшись, ставим срок жизни 1 минута, чтоб очистить папки со скринами, и удаляем тестовые каналы из списка отслеживаемых. Ставим срок жизни полдня (720 минут).

Затем закидываем в бот реальные арбитражные каналы и смотрим, что он наскринит…

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

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