Скрипт Отложенное сообщение Telegram в чаты

Скрипт для планирования сообщений в Telegram​

Автоматизируйте отправку сообщений в чаты Telegram с помощью этого удобного скрипта на Python!​

Что это за софт?

Этот скрипт на Python использует библиотеку Pyrogram для взаимодействия с Telegram API. Он позволяет вам запланировать отправку сообщений в любой чат Telegram (личный, группа или супергруппа) с гибкими настройками. Вы можете указать текст сообщения, количество сообщений, задержку перед первой отправкой и интервал между сообщениями. Скрипт также обрабатывает ограничения Telegram (например, flood wait) и поддерживает отправку в темы супергрупп.

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

• Гибкое планирование: Укажите, через сколько минут отправить первое сообщение и с каким интервалом отправлять остальные.
• Поддержка тем: Работает с темами в супергруппах — просто укажите ID темы.
• Обработка ошибок: Автоматически ждёт при flood wait и выводит понятные сообщения об ошибках.
• Цветной интерфейс: Использует библиотеку colorama для красивого вывода в консоли.
• Простота использования: Всё управление через консоль — просто вводите параметры и запускайте!

Как использовать:

1. Установите зависимости: pyrogram, tgcrypto, colorama (скрипт установит их автоматически, если их нет).
2. Запустите скрипт в Python.
3. Введите chat_id чата (например, -100123456789 или username).
4. Укажите текст сообщения или используйте стандартный (введите "-").
5. Введите количество сообщений, задержку перед первым сообщением и интервал между ними.
6. Скрипт начнёт планировать отправку и сообщит о результатах.

Пример использования:

Спойлер

Представьте, что вы хотите разослать рекламу в чат. Вы вводите chat_id, текст сообщения, указываете отправку 5 сообщений с интервалом в 10 минут, и скрипт автоматически запланирует всё за вас!

Полный исходный код:

import subprocess
import sys

def install_module(module_name):
try:
__import__(module_name)
except ImportError:
print(f"{module_name} не установлен. Устанавливаем...")
subprocess.check_call([sys.executable, "-m", "pip", "install", module_name])

install_module("pyrogram")
install_module("tgcrypto")
install_module("colorama")

from pyrogram import Client
import asyncio
from datetime import datetime, timedelta
from colorama import init, Fore, Style
import re

init()

api_id = ""
api_hash = ""

default_message = """
⛏ Плагины для FPC в продаже
Список плагинов

Автостарсы
15 000 ₽ (без демпинга) | 21 750 ₽ (с автодемпингом)
Включено: настройка, автодемпинг, обновления, чат поддержки

АвтоСММ
2 090 ₽
Включено: плагин с обновлениями, настройка, стартовые лоты, чат клиентов

Консультант по товарам ФП (GPT)
1 900 ₽

Оффлайн активация аккаунтов (GUARD)
2 375 ₽

Установка Python на VDS (автоустановщик)

Рассылка по публичным чатам
950 ₽

Автоотправка неподтвержденных заказов в ТП
4 750 ₽

Автоотправка гифтов за Старс
14 250 ₽

Автопополнение Steam-баланса (1:1, без %)
14 250 ₽

Имею большое кол-во отзывов

#fpc #Funpay #cardinal #funpaycardinal
"""

def print_box(text, color=Fore.CYAN):
lines = text.split('\n')
max_length = max(len(line) for line in lines)
print(f"{color}┌{'─' * (max_length + 2)}┐")
for line in lines:
print(f"│ {line.ljust(max_length)} │")
print(f"└{'─' * (max_length + 2)}┘{Style.RESET_ALL}")

async def send_message_with_retry(app, chat_id, message_text, schedule_time, message_thread_id=None):
while True:
try:
await app.send_message(
chat_id=chat_id,
text=message_text,
schedule_date=schedule_time,
message_thread_id=message_thread_id
)
print(f"{Fore.GREEN}✔ Сообщение запланировано на {schedule_time}{Style.RESET_ALL}")
return True
except Exception as e:
error_str = str(e)
if "FLOOD_WAIT_X" in error_str:
match = re.search(r"A wait of (\d+) seconds", error_str)
if match:
wait_time = int(match.group(1))
print(f"{Fore.YELLOW}Ожидание {wait_time} секунд из-за ограничений Telegram...{Style.RESET_ALL}")
await asyncio.sleep(wait_time)
else:
print(f"{Fore.RED}✘ Не удалось извлечь время ожидания: {error_str}{Style.RESET_ALL}")
return False
else:
error_text = ""
if "SCHEDULE_TOO_MUCH" in error_str:
error_text = "Превышен лимит запланированных сообщений в чате"
elif "TOPIC_CLOSED" in error_str:
error_text = "Тема закрыта"
elif "CHAT_WRITE_FORBIDDEN" in error_str:
error_text = "Нет прав на отправку"
else:
error_text = error_str
print(f"{Fore.RED}✘ Ошибка: {error_text}{Style.RESET_ALL}")
return False

async def main():
async with Client("my_account", api_id, api_hash) as app:
print_box("Клиент запущен!", Fore.GREEN)

chat_input = input(f"{Fore.CYAN}Введите chat_id (например, -100123456789 или @username): {Style.RESET_ALL}")
try:
if chat_input.startswith('@'):
chat = await app.get_chat(chat_input)
chat_id = chat.id
else:
chat_id = int(chat_input)
except Exception as e:
print_box(f"Ошибка в chat_id: {e}", Fore.RED)
return

message_thread_id = None
try:
chat = await app.get_chat(chat_id)
print(f"{Fore.YELLOW}Тип чата: {chat.type}{Style.RESET_ALL}")
if str(chat.type) == "ChatType.SUPERGROUP":
use_thread = input(f"{Fore.CYAN}В чате есть темы? (да/нет): {Style.RESET_ALL}").lower()
if use_thread == 'да':
thread_id = input(f"{Fore.CYAN}Введите ID темы (или оставьте пустым для основной ветки): {Style.RESET_ALL}")
if thread_id:
message_thread_id = int(thread_id)
except Exception as e:
print_box(f"Ошибка при проверке чата: {e}", Fore.RED)
return

message_text = input(f"{Fore.CYAN}Введите текст сообщения (или '-' для стандартного текста): {Style.RESET_ALL}")
if message_text.strip() == '-':
message_text = default_message

try:
message_count = int(input(f"{Fore.CYAN}Сколько сообщений отправить? (максимум 100): {Style.RESET_ALL}"))
if message_count < 1 or message_count > 100:
print_box("Ошибка: количество сообщений должно быть от 1 до 100", Fore.RED)
return
except ValueError:
print_box("Ошибка: введите корректное число сообщений", Fore.RED)
return

try:
delay = int(input(f"{Fore.CYAN}Через сколько минут отправить первое сообщение? {Style.RESET_ALL}"))
if delay < 0:
print_box("Ошибка: время не может быть отрицательным", Fore.RED)
return
start_time = datetime.now() + timedelta(minutes=delay)
except ValueError:
print_box("Ошибка: введите корректное число минут", Fore.RED)
return

try:
interval = int(input(f"{Fore.CYAN}Интервал между сообщениями (в минутах, ≥1): {Style.RESET_ALL}"))
if interval < 0:
print_box("Ошибка: интервал не может быть отрицательным", Fore.RED)
return
except ValueError:
print_box("Ошибка: введите корректное число минут для интервала", Fore.RED)
return

successful_count = 0
for i in range(message_count):
schedule_time = start_time + timedelta(minutes=interval * i)
if await send_message_with_retry(app, chat_id, message_text, schedule_time, message_thread_id):
successful_count += 1
else:
print(f"{Fore.YELLOW}Детали: chat_id={chat_id}, thread_id={message_thread_id}{Style.RESET_ALL}")
if "SCHEDULE_TOO_MUCH" in str(e):
break

print_box(f"Успешно запланировано {successful_count} из {message_count} сообщений", Fore.GREEN)

if __name__ == "__main__":
asyncio.run(main())

Код

import subprocess

import sys



def install_module(module_name):

    try:

        __import__(module_name)

    except ImportError:

        print(f"{module_name} не установлен. Устанавливаем...")

        subprocess.check_call([sys.executable, "-m", "pip", "install", module_name])



install_module("pyrogram")

install_module("tgcrypto")

install_module("colorama")



from pyrogram import Client

import asyncio

from datetime import datetime, timedelta

from colorama import init, Fore, Style

import re



init()



api_id = ""

api_hash = ""



default_message = """

⛏ Плагины для FPC в продаже

Список плагинов



Автостарсы

15 000 ₽ (без демпинга) | 21 750 ₽ (с автодемпингом)

Включено: настройка, автодемпинг, обновления, чат поддержки



АвтоСММ

2 090 ₽

Включено: плагин с обновлениями, настройка, стартовые лоты, чат клиентов



Консультант по товарам ФП (GPT)

1 900 ₽



Оффлайн активация аккаунтов (GUARD)

2 375 ₽



Установка Python на VDS (автоустановщик)



Рассылка по публичным чатам

950 ₽



Автоотправка неподтвержденных заказов в ТП

4 750 ₽



Автоотправка гифтов за Старс

14 250 ₽



Автопополнение Steam-баланса (1:1, без %)

14 250 ₽



Имею большое кол-во отзывов



#fpc #Funpay #cardinal #funpaycardinal

"""



def print_box(text, color=Fore.CYAN):

    lines = text.split('\n')

    max_length = max(len(line) for line in lines)

    print(f"{color}┌{'─' * (max_length + 2)}┐")

    for line in lines:

        print(f"│ {line.ljust(max_length)} │")

    print(f"└{'─' * (max_length + 2)}┘{Style.RESET_ALL}")



async def send_message_with_retry(app, chat_id, message_text, schedule_time, message_thread_id=None):

    while True:

        try:

            await app.send_message(

                chat_id=chat_id,

                text=message_text,

                schedule_date=schedule_time,

                message_thread_id=message_thread_id

            )

            print(f"{Fore.GREEN}✔ Сообщение запланировано на {schedule_time}{Style.RESET_ALL}")

            return True  

        except Exception as e:

            error_str = str(e)

            if "FLOOD_WAIT_X" in error_str:

                match = re.search(r"A wait of (\d+) seconds", error_str)

                if match:

                    wait_time = int(match.group(1))

                    print(f"{Fore.YELLOW}Ожидание {wait_time} секунд из-за ограничений Telegram...{Style.RESET_ALL}")

                    await asyncio.sleep(wait_time)

                else:

                    print(f"{Fore.RED}✘ Не удалось извлечь время ожидания: {error_str}{Style.RESET_ALL}")

                    return False

            else:

                error_text = ""

                if "SCHEDULE_TOO_MUCH" in error_str:

                    error_text = "Превышен лимит запланированных сообщений в чате"

                elif "TOPIC_CLOSED" in error_str:

                    error_text = "Тема закрыта"

                elif "CHAT_WRITE_FORBIDDEN" in error_str:

                    error_text = "Нет прав на отправку"

                else:

                    error_text = error_str

                print(f"{Fore.RED}✘ Ошибка: {error_text}{Style.RESET_ALL}")

                return False  



async def main():

    async with Client("my_account", api_id, api_hash) as app:

        print_box("Клиент запущен!", Fore.GREEN)

        

        chat_input = input(f"{Fore.CYAN}Введите chat_id (например, -100123456789 или @username): {Style.RESET_ALL}")

        try:

            if chat_input.startswith('@'):

                chat = await app.get_chat(chat_input)

                chat_id = chat.id

            else:

                chat_id = int(chat_input)

        except Exception as e:

            print_box(f"Ошибка в chat_id: {e}", Fore.RED)

            return



        message_thread_id = None

        try:

            chat = await app.get_chat(chat_id)

            print(f"{Fore.YELLOW}Тип чата: {chat.type}{Style.RESET_ALL}")

            if str(chat.type) == "ChatType.SUPERGROUP":

                use_thread = input(f"{Fore.CYAN}В чате есть темы? (да/нет): {Style.RESET_ALL}").lower()

                if use_thread == 'да':

                    thread_id = input(f"{Fore.CYAN}Введите ID темы (или оставьте пустым для основной ветки): {Style.RESET_ALL}")

                    if thread_id:

                        message_thread_id = int(thread_id)

        except Exception as e:

            print_box(f"Ошибка при проверке чата: {e}", Fore.RED)

            return



        message_text = input(f"{Fore.CYAN}Введите текст сообщения (или '-' для стандартного текста): {Style.RESET_ALL}")

        if message_text.strip() == '-':

            message_text = default_message



        try:

            message_count = int(input(f"{Fore.CYAN}Сколько сообщений отправить? (максимум 100): {Style.RESET_ALL}"))

            if message_count < 1 or message_count > 100:

                print_box("Ошибка: количество сообщений должно быть от 1 до 100", Fore.RED)

                return

        except ValueError:

            print_box("Ошибка: введите корректное число сообщений", Fore.RED)

            return



        try:

            delay = int(input(f"{Fore.CYAN}Через сколько минут отправить первое сообщение? {Style.RESET_ALL}"))

            if delay < 0:

                print_box("Ошибка: время не может быть отрицательным", Fore.RED)

                return

            start_time = datetime.now() + timedelta(minutes=delay)

        except ValueError:

            print_box("Ошибка: введите корректное число минут", Fore.RED)

            return



        try:

            interval = int(input(f"{Fore.CYAN}Интервал между сообщениями (в минутах, ≥1): {Style.RESET_ALL}"))

            if interval < 0:

                print_box("Ошибка: интервал не может быть отрицательным", Fore.RED)

                return

        except ValueError:

            print_box("Ошибка: введите корректное число минут для интервала", Fore.RED)

            return



        successful_count = 0

        for i in range(message_count):

            schedule_time = start_time + timedelta(minutes=interval * i)

            if await send_message_with_retry(app, chat_id, message_text, schedule_time, message_thread_id):

                successful_count += 1

            else:

                print(f"{Fore.YELLOW}Детали: chat_id={chat_id}, thread_id={message_thread_id}{Style.RESET_ALL}")

                if "SCHEDULE_TOO_MUCH" in str(e):

                    break

        

        print_box(f"Успешно запланировано {successful_count} из {message_count} сообщений", Fore.GREEN)



if __name__ == "__main__":

    asyncio.run(main())

Зависимости:

• Python 3.x
• pyrogram
• tgcrypto
• colorama