Автоматизированная очистка собственных сообщений в телеграме

[Oleg Polikhronidi]

Обычный юзер тул для тг. Автоматом чистит только ваши сообщения в указанных чатах без ручного тыканья. Работает по принципу - логинится под аккаунтом, тянет историю, отсекает сообщения по автору и таймстемпу, при необходимости сначала редактирует текст (чтобы не оставалось старого контента у клиентов с кэшем), после чего по очереди удаляет сообщения с заданными задержками, чтобы не упираться в лимиты, нормально переваривает большие объёмы истории, не падает на длинных чатах и закрывает типичную боль, когда в группе сотни ваших сообщений и удалять их руками. Указываете ид и хэш которые берёте на my.telegram.org, номер телефона нужен для авторизации сессии, дальше идут конфиги по типу кд между проходами по чатам, чтобы не спамить в ботапи, кдмежду удалениями сообщений, чтобы не словить флудвейт, лимит по времени в часах, потому что старые сообщения тг уже не даёт редачить. Кд на количество сообщений за один проход, чтобы не тянуть всю историю разом, список чатов, где скрипт будет работать (поддерживаются иди @юзер), и строка для замены текста, если нужно сначала перезаписать сообщение перед удалением.

Требования:
asyncio==3.4.3
pyrofork==2.3.61
configparser==7.2.0
tgcrypto-pyrofork==1.2.7

config.ini:

[Telegram]
# данные тг (берутся на my.telegram.org)
API_ID = <api_id>
API_HASH = <api_hash>
PHONE = +1231231232

[cleaner]
# кд между проходами по чатам (сек)
CHAT_DELAY = 1.0

# кд между операциями над сообщениями (сек)
MESSAGE_DELAY = 0.5

# кд на окно для обработки сообщений (часы)
# но старше указанного не редактируем, ток удаляем
TIME_LIMIT_HOURS = 48

# макс к-во сообщений за проход
# 0 = без ограна
FETCH_LIMIT = 1000

# ид чатов для обработки (id или @юзер если паблик чат)
TARGET_CHATS = @username,-100XXXXXXXXXX,XXXXXXXXXX

# заглушка для предварительного редактирования
REWRITE_TEXT = "."

app/utils/detector.py:

import asyncio
from pyrogram import Client
from datetime import datetime, timedelta
from data.config import config
from app.utils.logger import logger


async def clean(app: Client):
    logger.info('start')

    try:
        await app.get_users(config.CHATS_ID)
    except:
        async for _ in app.get_dialogs():
            pass

    for chat in config.CHATS_ID:
        try:
            await app.get_chat(chat)
        except:
            logger.error(f'chat {chat} not found')
            continue

        async for msg in app.get_chat_history(chat, limit=config.COUNT_MESSAGES):
            if config.LIMIT_HOURS and msg.date:
                if datetime.utcnow() - msg.date > timedelta(hours=config.LIMIT_HOURS):
                    break

            if not msg.from_user or not msg.from_user.is_self:
                continue

            try:
                await msg.edit_text(config.TEXT_REPLACE)
            except:
                pass

            try:
                await app.delete_messages(chat, msg.id)
            except:
                pass
            await asyncio.sleep(config.INTERVAL_MESSAGE)
        await asyncio.sleep(config.INTERVAL_CHAT)
    logger.info('done')


run = clean

app/utils/logger.py:

import sys
import logging
from datetime import datetime


class SimpleFormatter(logging.Formatter):
    def format(self, record):
        ts = datetime.now().strftime('%d.%m.%y %H:%M:%S')
        return f'[{ts}] - [{record.levelname}]: {record.getMessage()}'


logger = logging.getLogger('bot')
logger.setLevel(logging.DEBUG)

stream = logging.StreamHandler(sys.stdout)
stream.setLevel(logging.DEBUG)
stream.setFormatter(SimpleFormatter())

if not logger.handlers:
    logger.addHandler(stream)


def info(msg):
    print('\r', end='')
    logger.info(msg)


def warn(msg):
    print('\r', end='')
    logger.warning(msg)


def error(msg):
    print('\r', end='')
    logger.error(msg)


def log_same_line(msg, level='INFO'):
    ts = datetime.now().strftime('%d.%m.%y %H:%M:%S')
    print(f'\r[{ts}] - [{level.upper()}]: {msg}', end='', flush=True)

main.py:

import asyncio
from pyrogram import Client
from data.config import config
from app.utils.detector import cleaner
from app.utils.logger import error


async def run():
    async with Client(
        config.SESSION,
        api_id=config.API_ID,
        api_hash=config.API_HASH,
        phone_number=config.PHONE_NUMBER
    ) as app:
        await cleaner(app)


if __name__ == "__main__":
    try:
        asyncio.run(run())
    except Exception as e:
        error(f'ошибка: {e}')