Building a Robust Telegram Alert Bot with PyFilesystem and FastAPI

In this post, we’ll explore a Python-based Telegram Alert Bot designed for reliability, subscriber management, and flexible logging. This bot doesn’t just send messages; it features a pending-approval system for unsubscribing, automated chat logging, and a virtualized filesystem.

Key Features

• Automated Subscriber Management: Anyone who messages the bot is automatically added to the database.
• Unsubscribe Approval Workflow: Users can’t just leave; they request to unsubscribe, and the owner approves or denies it.
• Smart Virtual Logging: Uses PyFilesystem2 to log data to memory (mem://) or disk, with an adaptive handler that throttles file operations based on traffic.
• FastAPI Integration: Wrapped in FastAPI to ensure compatibility with modern hosting platforms (like Hugging Face Spaces or Render) that require a web health check.

The Architecture

1. The Virtual Filesystem

Instead of hardcoding standard Python file I/O, this project uses fs (PyFilesystem2). This allows you to swap from an in-memory database to a real disk or even an S3 bucket by changing just one line:

# Use "mem://" for ephemeral data or "osfs://." for persistent local storage
filesystem = open_fs("mem://") 

2. Smart Logging

Logging can be a bottleneck. The SmartFSLogHandler included in this bot monitors the frequency of incoming logs. If logs are sparse, it opens/closes the file per entry. If traffic spikes, it keeps the file handle open to reduce I/O overhead.

3. Owner-Only Commands

The bot distinguishes between regular users and the administrator (OWNER_ID).

• Users can access: /subscribe, /unsubscribe, /status, and /myinfo.
• The Owner can access: /broadcast, /stats, /list_subscribers, and /process_unsubscribes.

Code Highlight: Broadcast System

The broadcast function allows the owner to send alerts to all active subscribers simultaneously. It handles errors gracefully, ensuring that one dead chat doesn’t stop the entire broadcast.

@bot.message_handler(commands=['broadcast'])
def broadcast(message):
    if message.from_user.id != OWNER_ID:
        bot.send_message(message.chat.id, "🚫 Unauthorized.")
        return

    broadcast_text = message.text[len('/broadcast'):].strip()
    subscribers = load_subscribers()
    
    for sub in subscribers:
        try:
            bot.send_message(sub.get("chat_id"), broadcast_text, parse_mode="HTML")
        except Exception as e:
            logging.error(f"Failed to send to {sub.get('chat_id')}: {e}")

Deployment with Docker

To make deployment seamless, we use a lightweight Python 3.9 image. Note that we expose port 7860, which is the standard for Hugging Face Spaces.

FROM python:3.9-slim
RUN apt-get update && apt-get install -y socat
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 7860
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

Running it Locally

1. Set Environment Variables:

   export BOT_TOKEN="your_telegram_bot_token"
   export CHAT_IDS="your_personal_telegram_id"
   

2. Install Dependencies:

   pip install -r requirements.txt
   

3. Run the App:

   uvicorn app:app --host 0.0.0.0 --port 7860
   

Download the Source Code

You can download the complete source code, including the Python script, requirements, and Dockerfile, as a ZIP archive below:

Note: Ensure you replace the environment variables in your deployment settings before running the bot.