Бот для регулярной отправки сообщений

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

Какие задачи решает бот для регулярной отправки сообщений

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

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

Принцип работы бота для регулярной отправки сообщений

Алгоритмически бот несложный, и делает он следующее:

1. Слушает эфир.
2. При вводе /newtask в ответ на сообщение — создает задачу.
3. При вводе /starttask ID в ответ на сообщение — запускает задачу с соответствующим ID.
4. При вводе /stoptask ID в ответ на сообщение — останавливает задачу с соответствующим ID.
5. При вводе /setperiod ID 10m в ответ на сообщение — меняет частоту повторения задачи с соответствующим ID.
6. При вводе /tasklist в ответ на сообщение — выводит список задач.
7. При вводе /deltask ID в ответ на сообщение — удаляет задачу с соответствующим ID (и связанные с ней вложения).

При этом сами ID бот создает автоматически. Равно как автоматически он сохраняет и все прикрепленные к сообщению вложения. Чат/ЛС — всегда те же, где была создана задача. Это сделано для того, чтобы нельзя было насоздавать кучу задач без ведома собеседника. Но вы можете сами доработать этот момент, если оно вам надо. Правда, информируем, что это незаконно.

Раз уж мы разобрались с алгоритмом — перейдем к созданию бота!

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

Сперва настраиваем сервер. Если у вас его нет и где его взять вы не в курсе — вам сюда. Если же вы в курсе, откуда берутся серверы, то нужен тот, что может в Python. В этом случает вводим в консоль:

pip install telethon

Затем нужно получить api_id и api_hash, для этого:

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

После этого:

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

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

import os

import json

import asyncio

import re

from telethon import TelegramClient, events

API_ID = ваш API_ID

API_HASH = "ваш API_HASH"

DATA_FILE = "data.json"

ATTACH_DIR = "attachments"

os.makedirs(ATTACH_DIR, exist_ok=True)

client = TelegramClient("userbot", API_ID, API_HASH)

def load_data():

if not os.path.exists(DATA_FILE):

return {"tasks": {}}

try:

with open(DATA_FILE, "r", encoding="utf-8") as f:

data = json.load(f)

except Exception:

return {"tasks": {}}

if "tasks" not in data or not isinstance(data["tasks"], dict):

data["tasks"] = {}

return data

def save_data(data):

with open(DATA_FILE, "w", encoding="utf-8") as f:

json.dump(data, f, indent=2, ensure_ascii=False)

data = load_data()

running_tasks = {}

def parse_interval(t: str):

m = re.fullmatch(r"(\d+)([smhd])", t)

if not m:

return None

value = int(m.group(1))

unit = m.group(2)

if unit == "s":

return value

elif unit == "m":

return value * 60

elif unit == "h":

return value * 3600

elif unit == "d":

if value > 3:

return None

return value * 86400

return None

async def loop_task(task_id):

while True:

info = data["tasks"][task_id]

await asyncio.sleep(info["interval"])

chat = info["chat_id"]

msg = info["text"]

attach = info.get("attachment")

await client.send_message(chat, msg, file=attach if attach else None)

@client.on(events.NewMessage(pattern=r"^/newtask$"))

async def new_task(event):

if not event.is_reply:

return await event.reply("Ответь этой командой на сообщение, которое нужно повторять.")

reply = await event.get_reply_message()

if data["tasks"]:

next_id = max(map(int, data["tasks"].keys())) + 1

else:

next_id = 1

task_id = str(next_id)

text = reply.message or ""

attachment = None

if reply.media:

name = reply.file.name

ext = reply.file.ext

if not ext and name and "." in name:

ext = "." + name.split(".")[-1]

if not ext:

ext = ".bin"

filename = f"task_{task_id}{ext}"

path = os.path.join(ATTACH_DIR, filename)

await reply.download_media(file=path)

attachment = path

data["tasks"][task_id] = {

"chat_id": event.chat_id,

"text": text,

"attachment": attachment,

"interval": 3600,

"active": False

}

save_data(data)

await event.reply(

f"Создана задача **{task_id}**.\n"

f"Интервал: 1h.\n"

f"Запуск: `/starttask {task_id}`\n"

f"Изменить интервал: `/setperiod {task_id} 10m`"

)

@client.on(events.NewMessage(pattern=r"^/starttask (\d+)$"))

async def start_task(event):

task_id = event.pattern_match.group(1)

if task_id not in data["tasks"]:

return await event.reply("Нет такой задачи.")

if data["tasks"][task_id]["active"]:

return await event.reply("Задача уже запущена.")

data["tasks"][task_id]["active"] = True

save_data(data)

running_tasks[task_id] = asyncio.create_task(loop_task(task_id))

await event.reply(f"Задача **{task_id}** запущена.")

@client.on(events.NewMessage(pattern=r"^/stoptask (\d+)$"))

async def stop_task(event):

task_id = event.pattern_match.group(1)

if task_id not in data["tasks"]:

return await event.reply("Нет такой задачи.")

if not data["tasks"][task_id]["active"]:

return await event.reply("Задача уже остановлена.")

data["tasks"][task_id]["active"] = False

save_data(data)

if task_id in running_tasks:

running_tasks[task_id].cancel()

del running_tasks[task_id]

await event.reply(f"Задача **{task_id}** остановлена.")

@client.on(events.NewMessage(pattern=r"^/setperiod (\d+) (.+)$"))

async def set_period(event):

task_id = event.pattern_match.group(1)

interval_str = event.pattern_match.group(2)

if task_id not in data["tasks"]:

return await event.reply("Нет такой задачи.")

seconds = parse_interval(interval_str)

if seconds is None:

return await event.reply("Используй формат: 5s, 10m, 2h, 3d (до 3d).")

data["tasks"][task_id]["interval"] = seconds

save_data(data)

await event.reply(f"Интервал задачи **{task_id}** установлен: {interval_str}")

@client.on(events.NewMessage(pattern=r"^/tasklist$"))

async def task_list(event):

if not data["tasks"]:

return await event.reply("Нет задач.")

txt = "📌 **Задачи:**\n\n"

for tid, info in data["tasks"].items():

active = "🟢" if info["active"] else "⚪"

txt += (

f"ID {tid}: {active}\n"

f" интервал: {info['interval']} сек\n"

f" чат: {info['chat_id']}\n"

)

await event.reply(txt)

@client.on(events.NewMessage(pattern=r"^/deltask (\d+)$"))

async def delete_task(event):

task_id = event.pattern_match.group(1)

if task_id not in data["tasks"]:

return await event.reply("Нет такой задачи.")

if data["tasks"][task_id]["active"] and task_id in running_tasks:

running_tasks[task_id].cancel()

del running_tasks[task_id]

attach = data["tasks"][task_id].get("attachment")

if attach and os.path.exists(attach):

os.remove(attach)

del data["tasks"][task_id]

save_data(data)

await event.reply(f"Задача **{task_id}** удалена.")

async def restore_tasks():

await client.connect()

for tid, info in data["tasks"].items():

if info.get("active"):

running_tasks[tid] = asyncio.create_task(loop_task(tid))

async def main():

print("Userbot starting…")

await restore_tasks()

await client.run_until_disconnected()

client.start()

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

3. Меняем API_ID = ваш API_ID и API_HASH = "ваш API_HASH" на свои.

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

5. Проверяем работоспособность.

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

P. S. В разделе «Демонстрация работы» использованы секундные таймеры — просто для демонстрации работы. В реальности же подразумевается, что интервалы будут хотя бы в несколько часов, это позволит избежать бана. При слишком частой отправке одних и тех же сообщений шанс бана повышается.

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

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