Broadcast & notification Telegram bot source

# -*- coding: utf-8 -*-
# ------------------------------------------------------------
#  Broadcast & notification Telegram bot
#  Built by Filtor (filtori.com)
# ------------------------------------------------------------
#  With this bot you can send a message (announcement, discount,
#  news) to everyone who started the bot, all at once:
#   - whenever a user presses /start, they are added to the subscriber list
#   - only the "admin" can broadcast to everyone with  /send message_text
#  The subscriber list is saved in the subscribers.txt file.
#
#  Note: a Telegram bot must run on a server that has access
#     to api.telegram.org.
#
#  Full guide:
#     https://filtori.com/en/telegram-bot-source/broadcast-bot/
#  Custom version with an admin panel and scheduling?  +98 912 287 4862
#  Our Telegram: https://t.me/filtori
# ------------------------------------------------------------

import time
import requests

TOKEN = "PASTE_YOUR_BOT_TOKEN_HERE"
BASE = f"https://api.telegram.org/bot{TOKEN}"

# ------------------------------------------------------------
#  Put your own numeric chat_id here so only you can broadcast.
#  How to find your chat_id? /start the bot and send a message;
#  in the terminal, the chat_id is printed (the print line below).
# ------------------------------------------------------------
ADMIN_CHAT_ID = 123456789  # <-- replace this number with your own chat_id

SUBSCRIBERS_FILE = "subscribers.txt"

# Telegram allows about 30 messages per second to different users;
# to avoid a 429 error, we add a small pause between each send.
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 sending message:", 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 fetching messages:", error)
        return {"ok": False, "result": []}


def load_subscribers():
    """Read the list of subscriber chat_ids from the file."""
    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):
    """Add a new subscriber (if not already present)."""
    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):
    """Send a message to all subscribers."""
    subscribers = load_subscribers()
    sent = 0
    for chat_id in subscribers:
        send_message(chat_id, text)
        sent += 1
        time.sleep(SEND_DELAY)  # respect Telegram send rate limit
    return sent


def handle(chat_id, text):
    if text == "/start":
        add_subscriber(chat_id)
        send_message(
            chat_id,
            "Hi 👋 You joined our notification channel!\n"
            "From now on we will send news and discounts right here.\n"
            "— Filtor (filtori.com)",
        )
        return

    if text.startswith("/send"):
        if chat_id != ADMIN_CHAT_ID:
            send_message(chat_id, "Only the bot admin can send a broadcast.")
            return
        payload = text[len("/send"):].strip()
        if not payload:
            send_message(chat_id, "Write the message text after the command:\n/send Hello everyone!")
            return
        count = broadcast(payload)
        send_message(chat_id, f"Message sent to {count} subscribers ✅")
        return


def main():
    print("Filtor broadcast bot is running... (Ctrl+C to stop)")
    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("Message from chat_id:", chat_id, "->", text)
            if text:
                handle(chat_id, text)

        time.sleep(0.5)


if __name__ == "__main__":
    main()