265 lines
8.1 KiB
Python
265 lines
8.1 KiB
Python
"""
|
|
Aetheel Telegram Adapter
|
|
=========================
|
|
Connects to Telegram via the Bot API using python-telegram-bot.
|
|
|
|
Features:
|
|
- Receives private messages and group @mentions
|
|
- Each chat = persistent conversation context
|
|
- Sends replies back to the same chat
|
|
- Extends BaseAdapter for multi-channel support
|
|
|
|
Setup:
|
|
1. Create a bot via @BotFather on Telegram
|
|
2. Set TELEGRAM_BOT_TOKEN in .env
|
|
3. Start with: python main.py --telegram
|
|
|
|
Usage:
|
|
from adapters.telegram_adapter import TelegramAdapter
|
|
|
|
adapter = TelegramAdapter()
|
|
adapter.on_message(my_handler)
|
|
adapter.start()
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
|
|
from adapters.base import BaseAdapter, IncomingMessage
|
|
|
|
logger = logging.getLogger("aetheel.telegram")
|
|
|
|
|
|
def resolve_telegram_token(explicit: str | None = None) -> str:
|
|
"""Resolve the Telegram bot token."""
|
|
token = (explicit or os.environ.get("TELEGRAM_BOT_TOKEN", "")).strip()
|
|
if not token:
|
|
raise ValueError(
|
|
"Telegram bot token is required. "
|
|
"Set TELEGRAM_BOT_TOKEN environment variable or pass it explicitly. "
|
|
"Get one from @BotFather on Telegram."
|
|
)
|
|
return token
|
|
|
|
|
|
class TelegramAdapter(BaseAdapter):
|
|
"""
|
|
Telegram channel adapter using python-telegram-bot.
|
|
|
|
Handles:
|
|
- Private messages (DMs)
|
|
- Group messages where the bot is @mentioned
|
|
- Inline replies in groups
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
bot_token: str | None = None,
|
|
):
|
|
super().__init__()
|
|
self._token = resolve_telegram_token(bot_token)
|
|
self._application = None
|
|
self._bot_username: str = ""
|
|
self._running = False
|
|
self._thread: threading.Thread | None = None
|
|
|
|
# -------------------------------------------------------------------
|
|
# BaseAdapter implementation
|
|
# -------------------------------------------------------------------
|
|
|
|
@property
|
|
def source_name(self) -> str:
|
|
return "telegram"
|
|
|
|
def start(self) -> None:
|
|
"""Start the Telegram adapter (blocking)."""
|
|
from telegram.ext import (
|
|
Application,
|
|
MessageHandler,
|
|
filters,
|
|
)
|
|
|
|
self._application = (
|
|
Application.builder().token(self._token).build()
|
|
)
|
|
|
|
# Register handler for all text messages
|
|
self._application.add_handler(
|
|
MessageHandler(
|
|
filters.TEXT & ~filters.COMMAND, self._handle_message
|
|
)
|
|
)
|
|
|
|
# Resolve bot identity
|
|
async def _resolve():
|
|
bot = self._application.bot
|
|
me = await bot.get_me()
|
|
self._bot_username = me.username or ""
|
|
logger.info(f"Telegram bot: @{self._bot_username} (id={me.id})")
|
|
|
|
asyncio.get_event_loop().run_until_complete(_resolve())
|
|
|
|
logger.info("=" * 60)
|
|
logger.info(" Aetheel Telegram Adapter")
|
|
logger.info("=" * 60)
|
|
logger.info(f" Bot: @{self._bot_username}")
|
|
logger.info(f" Mode: Polling")
|
|
logger.info(f" Handlers: {len(self._message_handlers)} registered")
|
|
logger.info("=" * 60)
|
|
|
|
self._running = True
|
|
self._application.run_polling(drop_pending_updates=True)
|
|
|
|
def start_async(self) -> None:
|
|
"""Start the adapter in a background thread (non-blocking)."""
|
|
self._thread = threading.Thread(
|
|
target=self.start, daemon=True, name="telegram-adapter"
|
|
)
|
|
self._thread.start()
|
|
logger.info("Telegram adapter started in background thread")
|
|
|
|
def stop(self) -> None:
|
|
"""Stop the Telegram adapter gracefully."""
|
|
self._running = False
|
|
if self._application:
|
|
try:
|
|
self._application.stop()
|
|
except Exception:
|
|
pass
|
|
logger.info("Telegram adapter stopped.")
|
|
|
|
def send_message(
|
|
self,
|
|
channel_id: str,
|
|
text: str,
|
|
thread_id: str | None = None,
|
|
) -> None:
|
|
"""Send a message to a Telegram chat."""
|
|
if not text.strip() or not self._application:
|
|
return
|
|
|
|
# Chunk long messages (Telegram limit = 4096 chars)
|
|
TELEGRAM_TEXT_LIMIT = 4096
|
|
chunks = _chunk_text(text, TELEGRAM_TEXT_LIMIT)
|
|
|
|
async def _send():
|
|
bot = self._application.bot
|
|
for chunk in chunks:
|
|
kwargs = {
|
|
"chat_id": int(channel_id),
|
|
"text": chunk,
|
|
"parse_mode": "Markdown",
|
|
}
|
|
if thread_id:
|
|
kwargs["reply_to_message_id"] = int(thread_id)
|
|
try:
|
|
await bot.send_message(**kwargs)
|
|
except Exception as e:
|
|
# Retry without parse_mode if Markdown fails
|
|
logger.debug(f"Markdown send failed, retrying plain: {e}")
|
|
kwargs.pop("parse_mode", None)
|
|
await bot.send_message(**kwargs)
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
loop.create_task(_send())
|
|
except RuntimeError:
|
|
asyncio.run(_send())
|
|
|
|
# -------------------------------------------------------------------
|
|
# Internal: Message handling
|
|
# -------------------------------------------------------------------
|
|
|
|
async def _handle_message(self, update, context) -> None:
|
|
"""Process an incoming Telegram message."""
|
|
message = update.effective_message
|
|
if not message or not message.text:
|
|
return
|
|
|
|
chat = update.effective_chat
|
|
user = update.effective_user
|
|
if not chat or not user:
|
|
return
|
|
|
|
is_private = chat.type == "private"
|
|
text = message.text
|
|
|
|
# In groups, only respond to @mentions
|
|
if not is_private:
|
|
if self._bot_username and f"@{self._bot_username}" in text:
|
|
text = text.replace(f"@{self._bot_username}", "").strip()
|
|
else:
|
|
return # Ignore non-mention messages in groups
|
|
|
|
if not text.strip():
|
|
return
|
|
|
|
# Build IncomingMessage
|
|
user_name = user.full_name or user.username or str(user.id)
|
|
channel_name = chat.title or (
|
|
f"DM with {user_name}" if is_private else str(chat.id)
|
|
)
|
|
|
|
msg = IncomingMessage(
|
|
text=text,
|
|
user_id=str(user.id),
|
|
user_name=user_name,
|
|
channel_id=str(chat.id),
|
|
channel_name=channel_name,
|
|
conversation_id=str(chat.id), # Telegram: one session per chat
|
|
source="telegram",
|
|
is_dm=is_private,
|
|
timestamp=datetime.fromtimestamp(
|
|
message.date.timestamp(), tz=timezone.utc
|
|
)
|
|
if message.date
|
|
else datetime.now(timezone.utc),
|
|
raw_event={
|
|
"thread_id": str(message.message_id) if not is_private else None,
|
|
"message_id": message.message_id,
|
|
"chat_type": chat.type,
|
|
},
|
|
)
|
|
|
|
logger.info(
|
|
f"📨 [TG] Message from {user_name} in {channel_name}: {text[:100]}"
|
|
)
|
|
|
|
# Dispatch to handlers (synchronous — handlers are sync functions)
|
|
self._dispatch(msg)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _chunk_text(text: str, limit: int = 4096) -> list[str]:
|
|
"""Split text into chunks respecting Telegram's character limit."""
|
|
if len(text) <= limit:
|
|
return [text]
|
|
|
|
chunks = []
|
|
remaining = text
|
|
while remaining:
|
|
if len(remaining) <= limit:
|
|
chunks.append(remaining)
|
|
break
|
|
|
|
cut = limit
|
|
newline_pos = remaining.rfind("\n", 0, limit)
|
|
if newline_pos > limit // 2:
|
|
cut = newline_pos + 1
|
|
else:
|
|
space_pos = remaining.rfind(" ", 0, limit)
|
|
if space_pos > limit // 2:
|
|
cut = space_pos + 1
|
|
|
|
chunks.append(remaining[:cut])
|
|
remaining = remaining[cut:]
|
|
|
|
return chunks
|