Как автоматически удалить всю переписку в Telegram

Содержание

Особенности:
Требования для запуска:
Библиотека Telethon
API ID и API Hash
Как получить API:
Код:
Важные примечания:

Этот скрипт полностью удаляет все личные сообщения в Telegram, включая историю переписки у вас и у собеседника (при наличии прав).

Особенности:

Удаление пачками по 100 сообщений
Обработка FloodWait-ошибок с экспоненциальной задержкой
Подробное логирование в файл и консоль
7 попыток перезапуска при ошибках
Проверка результата после удаления

Требования для запуска:

Библиотека Telethon

pip install telethon

API ID и API Hash

Как получить API:

Перейдите на официальный портал разработчиков Telegram
Введите свой номер телефона в формате +79991234567

Во вкладке API development tools:

Нажмите «Create application»
Заполните поля (можно произвольно):

App title: MyApp

Short name: myapp

Platform: Web

Получите:

api_id → вставьте в API_ID

api_hash → вставьте в API_HASH

Код:

from telethon import TelegramClient, functions, types, errors
import asyncio
import logging
from datetime import datetime
import math

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('deletion.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Конфигурация (ВАШИ ДАННЫЕ)
API_ID = 11111111
API_HASH = '111111111111111111111111'
MAX_ATTEMPTS = 7
BASE_DELAY = 45
CHUNK_SIZE = 100
FLOOD_WAIT_SAFETY = 10

client = TelegramClient('session_name', API_ID, API_HASH)

async def get_message_count(user):
    """Возвращает приблизительное количество сообщений в чате."""
    try:
        history = await client(functions.messages.GetHistoryRequest(
            peer=user,
            offset_id=0,
            offset_date=None,
            add_offset=0,
            limit=1,
            max_id=0,
            min_id=0,
            hash=0
        ))
        return math.ceil(history.count / CHUNK_SIZE)
    except Exception as e:
        logger.error(f"Ошибка получения количества сообщений: {e}")
        return 1

async def delete_in_chunks(user, max_id):
    """Удаляет сообщения пачками с обработкой FloodWait."""
    try:
        await client(functions.messages.DeleteHistoryRequest(
            peer=user,
            max_id=max_id,
            just_clear=False,
            revoke=True
        ))
    except errors.FloodWaitError as e:
        logger.warning(f"FloodWait: Ожидание {e.seconds + FLOOD_WAIT_SAFETY} секунд")
        await asyncio.sleep(e.seconds + FLOOD_WAIT_SAFETY)
        return False
    except Exception as e:
        logger.error(f"Ошибка удаления: {type(e).__name__}: {str(e)}")
        return False
    return True

async def delete_entire_chat(user):
    """Удаляет переписку с расширенной логикой и пагинацией."""
    logger.info(f"Обработка чата с {user.first_name} (ID: {user.id})")
    
    total_chunks = await get_message_count(user)
    logger.info(f"Обнаружено ~{total_chunks * CHUNK_SIZE} сообщений")
    
    for attempt in range(1, MAX_ATTEMPTS + 1):
        success = True
        try:
            # Удаление пачками
            for chunk in range(total_chunks):
                result = await delete_in_chunks(user, 0)
                if not result:
                    success = False
                    break
                
                logger.info(f"Пачка {chunk+1}/{total_chunks} удалена")
                await asyncio.sleep(2)
            
            if success and await is_chat_empty(user):
                logger.info(f"Успешное удаление за {attempt} попытку")
                return True
            
            # Экспоненциальная задержка с базой BASE_DELAY
            delay = BASE_DELAY * (2 ** (attempt - 1))
            logger.info(f"Ожидание {delay} сек. перед повтором...")
            await asyncio.sleep(delay)
            
        except Exception as e:
            logger.error(f"Попытка {attempt} ошибка: {str(e)}")
            await asyncio.sleep(BASE_DELAY)
    
    logger.error(f"Не удалось удалить чат после {MAX_ATTEMPTS} попыток")
    return False

async def is_chat_empty(user):
    """Улучшенная проверка пустоты чата."""
    try:
        messages = await client.get_messages(user, limit=2)
        if len(messages) > 0:
            logger.warning(f"Обнаружено {len(messages)} остаточных сообщений")
            return False
        return True
    except Exception as e:
        logger.error(f"Ошибка проверки чата: {str(e)}")
        return False

async def delete_all_private_chats():
    """Основная функция обработки чатов."""
    dialogs = await client.get_dialogs()
    
    for dialog in dialogs:
        if isinstance(dialog.entity, types.User) and not dialog.entity.bot:
            start = datetime.now()
            logger.info(f"\n{'='*40}")
            
            success = await delete_entire_chat(dialog.entity)
            
            duration = (datetime.now() - start).total_seconds()
            logger.info(f"Результат: {'УСПЕХ' if success else 'НЕУДАЧА'} | Время: {duration:.1f} сек.")
            await asyncio.sleep(30)

async def main():
    await client.start()
    logger.info("Авторизация успешна")
    await delete_all_private_chats()
    await client.disconnect()

if __name__ == "__main__":
    try:
        with client:
            client.loop.run_until_complete(main())
    except Exception as e:
        logger.exception("Критическая ошибка:")
    finally:
        logger.info("Работа скрипта завершена")