tgcast (seinarukiro2/telecast-python)

tgcast

https://github.com/seinarukiro2/telecast-python
Admin
tgcast is a lightweight Telegram Bot API broadcast engine for Python that enables sending messages...

Tokens:5,734
Snippets:48
Trust Score:7.5
Update:1 week ago
Show doc for...
Context Summary (auto-generated)
Raw
# tgcast

tgcast is a lightweight Telegram Bot API broadcast engine for Python designed for sending transactional notifications and opt-in messages at scale. It provides a self-contained solution requiring no external servers—just `pip install` and go—with built-in priority queues using weighted round-robin scheduling, automatic rate limiting that adapts to Telegram's 429 responses, crash-safe campaigns with idempotent message enqueue, and SQLite WAL storage for zero-configuration persistence.

The library handles the complexity of reliable Telegram message delivery including global and per-chat rate limiting, exponential backoff with jitter for failed messages, dead-letter queue (DLQ) management for permanently failed tasks, and stale lease recovery to prevent stuck tasks. It supports sending text messages, photos, and documents via file_id, URL, or local file path, with YAML-based templates featuring locale fallback chains or custom renderer implementations.

## Installation

Install tgcast from PyPI. The base package handles text messages; add the templates extra for YAML template support.

```bash
# Basic installation
pip install tgcast

# With YAML template support
pip install tgcast[templates]
```

## Initialize Telecast Engine

The Telecast class is the main entry point for the broadcast engine. It manages the message queue, background workers, rate limiting, and database storage. Only `bot_token` is required; all other parameters have sensible defaults.

```python
import os
from tgcast import Telecast

# Minimal initialization
eng = Telecast(bot_token=os.environ["BOT_TOKEN"])

# Full configuration
eng = Telecast(
    bot_token=os.environ["BOT_TOKEN"],
    store_dsn="tgcast.db",           # SQLite database path
    templates_path="templates.yaml",  # YAML templates file
    global_rps=25.0,                  # Global messages/sec limit
    per_chat_rps=1.0,                 # Per-chat messages/sec limit
    max_concurrency=8,                # Worker pool size
    lease_ttl=30.0,                   # Task lease duration (seconds)
    max_retries=5,                    # Max retries before DLQ
    base_backoff=1.0,                 # Initial retry delay (seconds)
    max_backoff=300.0,                # Max retry delay (seconds)
)
```

## Start and Shutdown Engine

The engine must be started before enqueuing tasks and should be gracefully shut down when done. The start() method is non-blocking and spawns background threads for task scheduling, campaign processing, and lease recovery.

```python
import os
import signal
from tgcast import Telecast, Task

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Enqueue tasks here...
task_id = eng.enqueue(Task(chat_id=123456789, text="Hello!"))

# Wait for interrupt signal
signal.pause()

# Graceful shutdown with timeout
eng.shutdown(timeout=10.0)
```

## Context Manager Usage

Telecast supports the context manager protocol for automatic startup and cleanup. This is ideal for scripts that need to send messages and exit cleanly.

```python
import os
import time
from tgcast import Telecast, Task

with Telecast(bot_token=os.environ["BOT_TOKEN"], store_dsn=":memory:") as eng:
    task_id = eng.enqueue(Task(chat_id=123456789, text="Hello!"))
    time.sleep(2)
    status = eng.task_status(task_id)
    print(f"{task_id}: {status.state}")
# Engine automatically shuts down when exiting the context
```

## Enqueue Text Message Task

The enqueue() method adds a message task to the queue and returns a task ID. Tasks are processed asynchronously by background workers with automatic rate limiting and retry logic.

```python
import os
from tgcast import Telecast, Task, Priority

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Simple text message
task_id = eng.enqueue(Task(
    chat_id=123456789,
    text="Hello from tgcast!",
))

# Message with priority and formatting
task_id = eng.enqueue(Task(
    chat_id=123456789,
    text="*Bold* and _italic_ text",
    parse_mode="MarkdownV2",
    priority=Priority.HIGH,
    disable_notification=True,
))

print(f"Enqueued task: {task_id}")
# Output: Enqueued task: a1b2c3d4e5f6...
```

## Check Task Status

The task_status() method retrieves the current state of a task including delivery status, attempt count, and timing information.

```python
import os
from tgcast import Telecast, Task

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

task_id = eng.enqueue(Task(chat_id=123456789, text="Test message"))

status = eng.task_status(task_id)
if status:
    print(f"Task ID: {status.id}")
    print(f"State: {status.state}")      # queued, leased, sent, failed, dead
    print(f"Attempt: {status.attempt}")
    print(f"Created: {status.created_at}")
else:
    print("Task not found")

# Output:
# Task ID: a1b2c3d4...
# State: queued
# Attempt: 0
# Created: 2024-01-15T10:30:00+00:00
```

## Idempotent Message Delivery

Set an idempotency_key to prevent duplicate sends. If the same key is used twice, a DuplicateKeyError is raised, ensuring each message is delivered exactly once.

```python
import os
from tgcast import Telecast, Task, DuplicateKeyError

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# First attempt succeeds
eng.enqueue(Task(
    chat_id=123456789,
    text="Order #456 confirmed",
    idempotency_key="order:456:confirmation",
))

# Second attempt with same key raises error
try:
    eng.enqueue(Task(
        chat_id=123456789,
        text="Order #456 confirmed",
        idempotency_key="order:456:confirmation",
    ))
except DuplicateKeyError as e:
    print(f"Duplicate key rejected: {e.key}")
    # Output: Duplicate key rejected: order:456:confirmation
```

## Send Photo

Send photos via URL, local file path, or Telegram file_id. The photo parameter accepts any of these formats and handles them automatically.

```python
import os
from tgcast import Telecast, Task, TaskKind

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Send photo from URL
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.SEND_PHOTO,
    photo="https://example.com/image.jpg",
    caption="Check out this image!",
    parse_mode="HTML",
))

# Send local file
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.SEND_PHOTO,
    photo="/path/to/local/image.png",
    caption="<b>Local photo</b>",
    parse_mode="HTML",
))

# Send via file_id (previously uploaded photo)
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.SEND_PHOTO,
    photo="AgACAgIAAxkBAAI...",  # Telegram file_id
    caption="Reusing uploaded photo",
))
```

## Send Document

Send documents (files) with the same flexibility as photos—supporting URLs, local paths, and file_ids.

```python
import os
from tgcast import Telecast, Task, TaskKind

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Send PDF from local path
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.SEND_DOCUMENT,
    document="/path/to/report.pdf",
    caption="Monthly report attached",
))

# Send document from URL
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.SEND_DOCUMENT,
    document="https://example.com/data.xlsx",
    caption="Spreadsheet download",
))
```

## Edit Existing Message

Edit a previously sent message by specifying the message_id. Useful for updating status messages or interactive content.

```python
import os
from tgcast import Telecast, Task, TaskKind

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Edit an existing message
eng.enqueue(Task(
    chat_id=123456789,
    kind=TaskKind.EDIT_MESSAGE,
    message_id=98765,  # ID of the message to edit
    text="Updated content here",
    parse_mode="HTML",
))
```

## YAML Templates with Locale Fallback

Load templates from a YAML file or raw bytes. The template engine supports locale fallback: exact match → base locale (ru-RU → ru) → "en" → any available.

```python
import os
from tgcast import Telecast, Task

# Templates as YAML bytes
templates_yaml = b"""
welcome:
  en: "Hello, {name}! Welcome aboard."
  ru: "Привет, {name}! Добро пожаловать."
  es: "¡Hola, {name}! Bienvenido."

order_shipped:
  en: "Your order #{order_id} has been shipped!"
  ru: "Ваш заказ #{order_id} отправлен!"
"""

eng = Telecast(
    bot_token=os.environ["BOT_TOKEN"],
    templates_data=templates_yaml,
)
eng.start()

# Send with Russian locale
eng.enqueue(Task(
    chat_id=123456789,
    template_key="welcome",
    locale="ru",
    vars={"name": "Иван"},
))
# Sends: "Привет, Иван! Добро пожаловать."

# Fallback: fr-FR → fr → en (no French, falls back to English)
eng.enqueue(Task(
    chat_id=123456789,
    template_key="welcome",
    locale="fr-FR",
    vars={"name": "Pierre"},
))
# Sends: "Hello, Pierre! Welcome aboard."
```

## Custom Template Renderer

Implement the Renderer protocol to use your own template system (Jinja2, database-backed, etc.) instead of YAML.

```python
import os
import signal
from tgcast import Telecast, Task

class MarkdownRenderer:
    """Custom renderer that generates Markdown output."""

    def render(self, key: str, locale: str, vars: dict) -> str:
        if key == "order_confirmation":
            return f"*Order #{vars['order_id']}*\nStatus: {vars['status']}"
        elif key == "welcome":
            greeting = "Привет" if locale.startswith("ru") else "Hello"
            return f"{greeting}, *{vars['name']}*!"
        return f"Unknown template: {key}"

eng = Telecast(
    bot_token=os.environ["BOT_TOKEN"],
    template_renderer=MarkdownRenderer(),
)
eng.start()

eng.enqueue(Task(
    chat_id=123456789,
    template_key="order_confirmation",
    vars={"order_id": "42", "status": "shipped"},
    parse_mode="MarkdownV2",
))
# Sends: "*Order #42*\nStatus: shipped"
```

## Create and Run Campaign

Campaigns enable broadcasting messages to thousands or millions of recipients. Create a campaign with a template, add recipients, then start it. The engine processes recipients in batches with automatic idempotency keys.

```python
import os
import time
from tgcast import Telecast, CampaignConfig, Recipient, Priority

eng = Telecast(
    bot_token=os.environ["BOT_TOKEN"],
    templates_data=b"""
welcome:
  en: "Hello, {name}! Welcome aboard."
  ru: "Привет, {name}! Добро пожаловать."
""",
)
eng.start()

# Create campaign
campaign_id = eng.create_campaign(CampaignConfig(
    name="Welcome wave",
    template_key="welcome",
    priority=Priority.NORMAL,
))
print(f"Campaign created: {campaign_id}")

# Add recipients with per-user locale and variables
eng.add_recipients(campaign_id, [
    Recipient(chat_id=100, locale="en", vars={"name": "Alice"}),
    Recipient(chat_id=200, locale="ru", vars={"name": "Боб"}),
    Recipient(chat_id=300, locale="en", vars={"name": "Charlie"}),
])

# Start campaign
eng.start_campaign(campaign_id)

# Poll progress until complete
for _ in range(60):
    stats = eng.campaign_stats(campaign_id)
    print(f"sent={stats.sent} failed={stats.failed} pending={stats.pending}")
    if stats.pending == 0:
        print(f"Campaign complete! Status: {stats.status}")
        break
    time.sleep(1)

eng.shutdown()
```

## Campaign with Fixed Locale

Use LocaleStrategy.FIXED to send all recipients the same locale instead of per-user locales.

```python
import os
from tgcast import Telecast, CampaignConfig, Recipient, LocaleStrategy

eng = Telecast(
    bot_token=os.environ["BOT_TOKEN"],
    templates_data=b"""
announcement:
  en: "Important: {message}"
  ru: "Важно: {message}"
""",
)
eng.start()

# All recipients get English regardless of their locale setting
campaign_id = eng.create_campaign(CampaignConfig(
    name="System announcement",
    template_key="announcement",
    locale_strategy=LocaleStrategy.FIXED,
    fixed_locale="en",
    vars={"message": "Scheduled maintenance tonight at 10 PM UTC"},
))

eng.add_recipients(campaign_id, [
    Recipient(chat_id=100),
    Recipient(chat_id=200),
    Recipient(chat_id=300),
])

eng.start_campaign(campaign_id)
```

## Pause and Resume Campaign

Campaigns can be paused and resumed. Pausing stops enqueueing new recipients; already queued tasks continue processing.

```python
import os
import time
from tgcast import Telecast, CampaignConfig, Recipient

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

campaign_id = eng.create_campaign(CampaignConfig(
    name="Large broadcast",
    template_key="update",
))
eng.add_recipients(campaign_id, [Recipient(chat_id=i) for i in range(1000)])

eng.start_campaign(campaign_id)
time.sleep(5)

# Pause campaign
eng.pause_campaign(campaign_id)
stats = eng.campaign_stats(campaign_id)
print(f"Paused at sent={stats.sent}, pending={stats.pending}")

# Resume later
time.sleep(10)
eng.start_campaign(campaign_id)
print("Campaign resumed")
```

## Dead Letter Queue Management

Tasks that fail permanently (blocked users, invalid chat_ids) are moved to the DLQ after max_retries. You can list DLQ tasks and requeue them after fixing the issue.

```python
import os
from tgcast import Telecast

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# List dead-letter tasks with pagination
dead_tasks, total = eng.dlq_list(limit=20, offset=0)
print(f"DLQ contains {total} tasks")

for task in dead_tasks:
    print(f"  Task {task.id}: chat_id={task.chat_id}, "
          f"attempts={task.attempt}, error at {task.updated_at}")

# Requeue a task for retry (e.g., after user unblocked the bot)
if dead_tasks:
    task_id = dead_tasks[0].id
    eng.dlq_requeue(task_id)
    print(f"Requeued task {task_id}")
```

## Priority Queues

Tasks support three priority levels with weighted round-robin scheduling (high:normal:low = 5:3:1). Higher priority tasks are processed more frequently but lower priority tasks are never starved.

```python
import os
from tgcast import Telecast, Task, Priority

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# High priority: OTPs, security alerts
eng.enqueue(Task(
    chat_id=123456789,
    text="Your verification code: 123456",
    priority=Priority.HIGH,
))

# Normal priority: order updates, notifications
eng.enqueue(Task(
    chat_id=123456789,
    text="Your order has shipped!",
    priority=Priority.NORMAL,
))

# Low priority: marketing, newsletters
eng.enqueue(Task(
    chat_id=123456789,
    text="Check out our weekly deals!",
    priority=Priority.LOW,
))
```

## Scheduled Messages

Use the not_before parameter to schedule messages for future delivery. Tasks won't be processed until the specified time.

```python
import os
from datetime import datetime, timedelta, timezone
from tgcast import Telecast, Task

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

# Schedule message for 1 hour from now
future_time = datetime.now(timezone.utc) + timedelta(hours=1)

eng.enqueue(Task(
    chat_id=123456789,
    text="Reminder: Your appointment is in 30 minutes!",
    not_before=future_time,
))

print(f"Message scheduled for {future_time.isoformat()}")
```

## Reply Markup (Inline Keyboards)

Add interactive inline keyboards to messages using the reply_markup parameter with JSON-serialized keyboard data.

```python
import os
import json
from tgcast import Telecast, Task

eng = Telecast(bot_token=os.environ["BOT_TOKEN"])
eng.start()

keyboard = {
    "inline_keyboard": [
        [
            {"text": "Yes ✓", "callback_data": "confirm_yes"},
            {"text": "No ✗", "callback_data": "confirm_no"},
        ],
        [
            {"text": "Visit Website", "url": "https://example.com"},
        ],
    ]
}

eng.enqueue(Task(
    chat_id=123456789,
    text="Do you want to proceed?",
    reply_markup=keyboard,
))
```

---

tgcast is ideal for applications requiring reliable Telegram message delivery including transactional notifications (order confirmations, OTPs, alerts), opt-in broadcast campaigns, and any scenario needing rate-limited delivery with retry logic. The self-contained SQLite storage means no external dependencies like Redis or RabbitMQ, making deployment simple for both small scripts and production services.

Integration follows a straightforward pattern: initialize Telecast with your bot token, start the engine, enqueue tasks or create campaigns, and let the background workers handle delivery with automatic rate limiting and retries. For high-volume scenarios, tune the global_rps, per_chat_rps, and max_concurrency parameters; for crash resilience, rely on the idempotency keys and SQLite WAL storage to resume without duplicate sends.