Исходник Telegram-бота рассылки и оповещений

# -*- coding: utf-8 -*-
# ------------------------------------------------------------
#  Telegram-бот рассылки и оповещений
#  Создано Filtor (filtori.com)
# ------------------------------------------------------------
#  С этим ботом вы можете отправить всем, кто запустил бота,
#  одно сообщение (оповещение, скидку, новость) сразу:
#   - каждый пользователь, нажавший /start, добавляется в список подписчиков
#   - только «админ» может командой  /send текст_сообщения  написать всем
#  Список подписчиков сохраняется в файл subscribers.txt.
#
#  Замечание: Telegram-бот должен запускаться на сервере,
#     имеющем доступ к api.telegram.org.
#
#  Полное руководство:
#     https://filtori.com/tj/telegram-bot-source/broadcast-bot/
#  Индивидуальная версия с панелью управления и планировщиком?  +98 912 287 4862
#  Наш Telegram: https://t.me/filtori
# ------------------------------------------------------------

import time
import requests

TOKEN = "ВСТАВЬТЕ_ТОКЕН_ВАШЕГО_БОТА"
BASE = f"https://api.telegram.org/bot{TOKEN}"

# ------------------------------------------------------------
#  Вставьте сюда свой числовой chat_id, чтобы только вы могли делать рассылку.
#  Как найти свой chat_id? Нажмите /start у бота и отправьте сообщение;
#  в терминале будет напечатан chat_id (строка print ниже).
# ------------------------------------------------------------
ADMIN_CHAT_ID = 123456789  # <-- замените это число на свой chat_id

SUBSCRIBERS_FILE = "subscribers.txt"

# Telegram позволяет отправлять примерно до 30 сообщений в секунду разным
# пользователям; чтобы не получить ошибку 429, между отправками делаем паузу.
SEND_DELAY = 0.05


def send_message(chat_id, text):
    url = f"{BASE}/sendMessage"
    try:
        requests.post(url, json={"chat_id": chat_id, "text": text}, timeout=15)
    except requests.RequestException as error:
        print("Ошибка отправки сообщения:", error)


def get_updates(offset):
    url = f"{BASE}/getUpdates"
    params = {"offset": offset, "timeout": 30}
    try:
        return requests.get(url, params=params, timeout=40).json()
    except requests.RequestException as error:
        print("Ошибка получения сообщений:", error)
        return {"ok": False, "result": []}


def load_subscribers():
    """Чтение списка chat_id подписчиков из файла."""
    try:
        with open(SUBSCRIBERS_FILE, "r", encoding="utf-8") as file:
            return set(line.strip() for line in file if line.strip())
    except FileNotFoundError:
        return set()


def add_subscriber(chat_id):
    """Добавление нового подписчика (если его ещё не было)."""
    subscribers = load_subscribers()
    if str(chat_id) not in subscribers:
        with open(SUBSCRIBERS_FILE, "a", encoding="utf-8") as file:
            file.write(f"{chat_id}\n")


def broadcast(text):
    """Отправка одного сообщения всем подписчикам."""
    subscribers = load_subscribers()
    sent = 0
    for chat_id in subscribers:
        send_message(chat_id, text)
        sent += 1
        time.sleep(SEND_DELAY)  # соблюдение ограничения скорости отправки Telegram
    return sent


def handle(chat_id, text):
    if text == "/start":
        add_subscriber(chat_id)
        send_message(
            chat_id,
            "Привет 👋 Вы подписались на наш канал оповещений!\n"
            "Отныне мы будем присылать вам новости и скидки прямо сюда.\n"
            "— Filtor (filtori.com)",
        )
        return

    if text.startswith("/send"):
        if chat_id != ADMIN_CHAT_ID:
            send_message(chat_id, "Только админ бота может делать рассылку.")
            return
        payload = text[len("/send"):].strip()
        if not payload:
            send_message(chat_id, "Напишите текст сообщения после команды:\n/send Привет всем!")
            return
        count = broadcast(payload)
        send_message(chat_id, f"Сообщение отправлено {count} подписчикам ✅")
        return


def main():
    print("Бот оповещений Filtor запущен... (Ctrl+C для остановки)")
    offset = 0
    while True:
        updates = get_updates(offset)
        if not updates.get("ok"):
            time.sleep(2)
            continue

        for update in updates.get("result", []):
            offset = update["update_id"] + 1
            message = update.get("message")
            if not message:
                continue
            chat_id = message["chat"]["id"]
            text = message.get("text", "")
            print("Сообщение от chat_id:", chat_id, "->", text)
            if text:
                handle(chat_id, text)

        time.sleep(0.5)


if __name__ == "__main__":
    main()