TelegramManager 0.1.16

Simple Interface for Telegram read and listen

TelegramManager

A Python CLI tool and module for fetching and monitoring Telegram messages from public channels and groups. Built with Telethon for reliable Telegram API integration. Supports both synchronous and asynchronous operations.

Installation

Install TelegramManager using pip:

pip install .

For development installation:

pip install -e .[dev]

Configuration

Environment Configuration

Create a .env file in your project root directory:

TELEGRAM_API_ID=your_api_id
TELEGRAM_API_HASH=your_api_hash
TELEGRAM_PHONE_NUMBER=[phone]

The default TelegramManager() and AsyncTelegramManager() constructors automatically load these environment variables.

Manual Configuration

For programmatic usage without environment files:

from telegram_manager.controller import TelegramManager, AsyncTelegramManager

# Synchronous version
tg = TelegramManager(
    api_id=123456,
    api_hash="your_api_hash_here",
    phone_number="[phone]"
)

# Asynchronous version
async_tg = AsyncTelegramManager(
    api_id=123456,
    api_hash="your_api_hash_here",
    phone_number="[phone]"
)

Command Line Interface

The tm command provides two primary operations:

Fetch Messages

Retrieve historical messages from a channel or group:

tm fetch <channel> [--min-id <id>] [--limit <n>] [--since <relative-time>] [--search <text>] [--json] [--verbose]

Options:

• --min-id: Minimum message ID to fetch from.
• --limit: Maximum number of messages to retrieve.
• --since: Filter messages newer than a relative time expression.
• --search: Filter messages containing the given search string.
• --json: Output each message in JSON format.
• --verbose: Print detailed metadata per message.

Supported --since formats:

• mo: months (e.g. 1mo)
• w: weeks (e.g. 2w)
• d: days (e.g. 3d)
• h: hours (e.g. 4h)
• m: minutes (e.g. 30m)

You can combine units:

tm fetch @openai --since "1mo 2w 3d 4h 30m" --search GPT --verbose

Listen for Messages

Monitor channels for new messages in real-time:

tm listen <channel>

Example:

tm listen "Some Group Chat"

Verbose Mode

When --verbose is enabled in fetch, each message will include:

• Message ID
• Date in local time and UTC
• Sender username and ID
• Message type (text, photo, document, video)
• Reply-to message ID (if any)
• Raw text content

A final summary is also printed, including:

• Total messages
• Unique user count
• Breakdown by media type
• Minimum message ID fetched

JSON Output

Use --json to emit each message as a structured JSON object. This is useful for piping into other programs or saving to file.

Python API

TelegramManager provides both synchronous and asynchronous APIs. Choose the one that best fits your application's architecture.

Synchronous API

The synchronous API is simpler and suitable for most use cases, especially scripts and applications that don't require concurrent operations.

Basic Usage

from telegram_manager import TelegramManager

# Using environment variables
tg = TelegramManager()

# Or with explicit credentials
tg = TelegramManager(
    api_id=12345,
    api_hash="your_api_hash",
    phone_number="[phone]"
)

Context Manager Usage

with TelegramManager() as tg:
    # Fetch messages
    messages = tg.fetch_messages("@channel_name", limit=10)
    
    # Send message
    tg.send_message("@username", "Hello!")
    
    # Get chat info
    info = tg.get_chat_info("@channel_name")

Fetching Messages

# Basic message fetching
messages = tg.fetch_messages("@somechannel", limit=5)

# With custom message processor
def process_message(message):
    print(f"ID: {message.id}, Text: {message.raw_text}")

tg.fetch_messages(
    chat_identifier="@somechannel",
    message_processor=process_message,
    limit=10
)

# With filtering options
messages = tg.fetch_messages(
    "@channel",
    limit=100,
    min_id=12345,
    search_text="important"
)

Real-time Message Monitoring

def handle_new_message(message):
    print(f"New message: {message.message}")

tg.listen("@somechannel", message_handler=handle_new_message)

Asynchronous API

The asynchronous API is ideal for applications that need to handle multiple operations concurrently or integrate with async frameworks like FastAPI, aiohttp, or Discord.py.

Basic Usage

import asyncio
from telegram_manager import AsyncTelegramManager

async def main():
    # Using environment variables
    async_tg = AsyncTelegramManager()
    
    # Or with explicit credentials
    async_tg = AsyncTelegramManager(
        api_id=12345,
        api_hash="your_api_hash",
        phone_number="[phone]"
    )

# Run the async function
asyncio.run(main())

Context Manager Usage

async def main():
    async with AsyncTelegramManager() as tg:
        # Fetch messages
        messages = await tg.fetch_messages("@channel_name", limit=10)
        
        # Send message
        await tg.send_message("@username", "Hello from async!")
        
        # Get chat info
        info = await tg.get_chat_info("@channel_name")
        print(f"Chat info: {info}")

asyncio.run(main())

Fetching Messages Asynchronously

async def fetch_example():
    async with AsyncTelegramManager() as tg:
        # Basic message fetching
        messages = await tg.fetch_messages("@somechannel", limit=5)
        
        # With custom message processor
        async def process_message(message):
            print(f"ID: {message.id}, Text: {message.raw_text}")
            # Can perform async operations here
            await some_async_operation(message)
        
        await tg.fetch_messages(
            chat_identifier="@somechannel",
            message_processor=process_message,
            limit=10
        )

asyncio.run(fetch_example())

Concurrent Operations

async def concurrent_example():
    async with AsyncTelegramManager() as tg:
        # Fetch from multiple channels concurrently
        tasks = [
            tg.fetch_messages("@channel1", limit=10),
            tg.fetch_messages("@channel2", limit=10),
            tg.fetch_messages("@channel3", limit=10)
        ]
        
        results = await asyncio.gather(*tasks)
        
        for i, messages in enumerate(results, 1):
            print(f"Channel {i} has {len(messages)} messages")

asyncio.run(concurrent_example())

Real-time Message Monitoring (Async)

async def handle_new_message(message):
    print(f"New async message: {message.message}")
    # Can perform async operations
    await process_message_async(message)

async def listen_example():
    async with AsyncTelegramManager() as tg:
        await tg.listen("@somechannel", message_handler=handle_new_message)

asyncio.run(listen_example())

Integration with Web Frameworks

# FastAPI example
from fastapi import FastAPI
from telegram_manager import AsyncTelegramManager

app = FastAPI()
tg = AsyncTelegramManager()

@app.on_event("startup")
async def startup():
    await tg.start()

@app.on_event("shutdown")
async def shutdown():
    await tg.disconnect()

@app.get("/messages/{channel}")
async def get_messages(channel: str, limit: int = 10):
    messages = await tg.fetch_messages(f"@{channel}", limit=limit)
    return {"messages": [msg.raw_text for msg in messages]}

@app.post("/send/{channel}")
async def send_message(channel: str, message: str):
    await tg.send_message(f"@{channel}", message)
    return {"status": "sent"}

Complete Usage Examples

Synchronous Example

def sync_example():
    """Example of synchronous usage."""
    manager = TelegramManager(
        api_id=12345,
        api_hash="your_api_hash",
        phone_number="[phone]"
    )

    with manager:
        # Fetch messages
        messages = manager.fetch_messages("@channel_name", limit=10)
        print(f"Fetched {len(messages)} messages")

        # Send message
        manager.send_message("@username", "Hello from sync!")

        # Get chat info
        info = manager.get_chat_info("@channel_name")
        print(f"Chat info: {info}")

        # Listen for new messages (blocks)
        def message_handler(msg):
            print(f"New: {msg.raw_text}")
        
        manager.listen("@channel_name", message_handler=message_handler)

Asynchronous Example

async def async_example():
    """Example of asynchronous usage."""
    manager = AsyncTelegramManager(
        api_id=12345,
        api_hash="your_api_hash",
        phone_number="[phone]"
    )

    async with manager:
        # Fetch messages
        messages = await manager.fetch_messages("@channel_name", limit=10)
        print(f"Fetched {len(messages)} messages")

        # Send message
        await manager.send_message("@username", "Hello from async!")

        # Get chat info
        info = await manager.get_chat_info("@channel_name")
        print(f"Chat info: {info}")

        # Listen for new messages (non-blocking with other async operations)
        async def message_handler(msg):
            print(f"New: {msg.raw_text}")
            await some_async_processing(msg)
        
        # Can run concurrently with other async operations
        listen_task = asyncio.create_task(
            manager.listen("@channel_name", message_handler=message_handler)
        )
        
        # Do other async work while listening
        await other_async_operations()
        
        # Cancel listening when done
        listen_task.cancel()

# Run the async example
asyncio.run(async_example())

API Comparison

Feature	Synchronous API	Asynchronous API
Context Manager	with TelegramManager():	async with AsyncTelegramManager():
Fetch Messages	messages = tg.fetch_messages()	messages = await tg.fetch_messages()
Send Message	tg.send_message()	await tg.send_message()
Listen	tg.listen() (blocks)	await tg.listen() (non-blocking)
Concurrency	Sequential operations	Concurrent operations with asyncio.gather()
Integration	Simple scripts, CLI tools	Web frameworks, concurrent applications

When to Use Which API

Use Synchronous API when:

• Building simple scripts or CLI tools
• Operations are naturally sequential
• You don't need concurrent Telegram operations
• Working with sync-only libraries

Use Asynchronous API when:

• Building web applications (FastAPI, aiohttp)
• Need to handle multiple channels concurrently
• Integrating with other async libraries
• Building real-time applications
• Performance and concurrency are important

Supported Input Formats

TelegramManager accepts multiple channel identifier formats:

• Telegram URLs: https://t.me/channelname
• Username format: @channelname
• Dialog names: "Channel Display Name"

Authentication

• Session files are created locally to maintain authentication across sessions
• First-time usage requires verification code entry
• Authentication state persists between program runs
• Both sync and async APIs share the same session management

Requirements

• Python 3.7 or higher
• Valid Telegram API credentials
• Network connectivity for Telegram API access
• For async usage: Basic understanding of Python asyncio