Rebase onto upstream (a4d95fd)
#12
+75
-247
@@ -4,11 +4,9 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from loguru import logger
|
||||
from telegram import BotCommand, Update
|
||||
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
|
||||
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
|
||||
from telegram.request import HTTPXRequest
|
||||
|
||||
from nanobot.bus.events import OutboundMessage
|
||||
@@ -16,10 +14,6 @@ from nanobot.bus.queue import MessageBus
|
||||
from nanobot.channels.base import BaseChannel
|
||||
from nanobot.config.schema import TelegramConfig
|
||||
|
||||
# Telegram API limits
|
||||
REMOTE_MEDIA_SIZE_LIMIT = 100_000_000 # 100MB for remote URLs
|
||||
TELEGRAM_PHOTO_SIZE_LIMIT = 10_000_000 # 10MB for photos
|
||||
|
||||
|
||||
def _markdown_to_telegram_html(text: str) -> str:
|
||||
"""
|
||||
@@ -27,72 +21,72 @@ def _markdown_to_telegram_html(text: str) -> str:
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
|
||||
|
||||
# 1. Extract and protect code blocks (preserve content from other processing)
|
||||
code_blocks: list[str] = []
|
||||
def save_code_block(m: re.Match) -> str:
|
||||
code_blocks.append(m.group(1))
|
||||
return f"\x00CB{len(code_blocks) - 1}\x00"
|
||||
|
||||
|
||||
text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text)
|
||||
|
||||
|
||||
# 2. Extract and protect inline code
|
||||
inline_codes: list[str] = []
|
||||
def save_inline_code(m: re.Match) -> str:
|
||||
inline_codes.append(m.group(1))
|
||||
return f"\x00IC{len(inline_codes) - 1}\x00"
|
||||
|
||||
|
||||
text = re.sub(r'`([^`]+)`', save_inline_code, text)
|
||||
|
||||
|
||||
# 3. Headers # Title -> just the title text
|
||||
text = re.sub(r'^#{1,6}\s+(.+)$', r'\1', text, flags=re.MULTILINE)
|
||||
|
||||
|
||||
# 4. Blockquotes > text -> just the text (before HTML escaping)
|
||||
text = re.sub(r'^>\s*(.*)$', r'\1', text, flags=re.MULTILINE)
|
||||
|
||||
|
||||
# 5. Escape HTML special characters
|
||||
text = text.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
|
||||
|
||||
# 6. Links [text](url) - must be before bold/italic to handle nested cases
|
||||
text = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', text)
|
||||
|
||||
|
||||
# 7. Bold **text** or __text__
|
||||
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
|
||||
text = re.sub(r'__(.+?)__', r'<b>\1</b>', text)
|
||||
|
||||
|
||||
# 8. Italic _text_ (avoid matching inside words like some_var_name)
|
||||
text = re.sub(r'(?<![a-zA-Z0-9])_([^_]+)_(?![a-zA-Z0-9])', r'<i>\1</i>', text)
|
||||
|
||||
|
||||
# 9. Strikethrough ~~text~~
|
||||
text = re.sub(r'~~(.+?)~~', r'<s>\1</s>', text)
|
||||
|
||||
|
||||
# 10. Bullet lists - item -> • item
|
||||
text = re.sub(r'^[-*]\s+', '• ', text, flags=re.MULTILINE)
|
||||
|
||||
|
||||
# 11. Restore inline code with HTML tags
|
||||
for i, code in enumerate(inline_codes):
|
||||
# Escape HTML in code content
|
||||
escaped = code.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
text = text.replace(f"\x00IC{i}\x00", f"<code>{escaped}</code>")
|
||||
|
||||
|
||||
# 12. Restore code blocks with HTML tags
|
||||
for i, code in enumerate(code_blocks):
|
||||
# Escape HTML in code content
|
||||
escaped = code.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
text = text.replace(f"\x00CB{i}\x00", f"<pre><code>{escaped}</code></pre>")
|
||||
|
||||
|
||||
return text
|
||||
|
||||
|
||||
class TelegramChannel(BaseChannel):
|
||||
"""
|
||||
Telegram channel using long polling.
|
||||
|
||||
|
||||
Simple and reliable - no webhook/public IP needed.
|
||||
"""
|
||||
|
||||
|
||||
name = "telegram"
|
||||
|
||||
|
||||
# Commands registered with Telegram's command menu
|
||||
BOT_COMMANDS = [
|
||||
BotCommand("start", "Start the bot"),
|
||||
@@ -100,7 +94,7 @@ class TelegramChannel(BaseChannel):
|
||||
BotCommand("help", "Show available commands"),
|
||||
BotCommand("quota", "Show current quota status"),
|
||||
]
|
||||
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: TelegramConfig,
|
||||
@@ -113,15 +107,15 @@ class TelegramChannel(BaseChannel):
|
||||
self._app: Application | None = None
|
||||
self._chat_ids: dict[str, int] = {} # Map sender_id to chat_id for replies
|
||||
self._typing_tasks: dict[str, asyncio.Task] = {} # chat_id -> typing loop task
|
||||
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Start the Telegram bot with long polling."""
|
||||
if not self.config.token:
|
||||
logger.error("Telegram bot token not configured")
|
||||
return
|
||||
|
||||
|
||||
self._running = True
|
||||
|
||||
|
||||
# Build the application with larger connection pool to avoid pool-timeout on long runs
|
||||
req = HTTPXRequest(connection_pool_size=16, pool_timeout=5.0, connect_timeout=30.0, read_timeout=30.0)
|
||||
builder = Application.builder().token(self.config.token).request(req).get_updates_request(req)
|
||||
@@ -129,63 +123,63 @@ class TelegramChannel(BaseChannel):
|
||||
builder = builder.proxy(self.config.proxy).get_updates_proxy(self.config.proxy)
|
||||
self._app = builder.build()
|
||||
self._app.add_error_handler(self._on_error)
|
||||
|
||||
|
||||
# Add command handlers
|
||||
self._app.add_handler(CommandHandler("start", self._on_start))
|
||||
self._app.add_handler(CommandHandler("new", self._forward_command))
|
||||
self._app.add_handler(CommandHandler("help", self._forward_command))
|
||||
self._app.add_handler(CommandHandler("quota", self._forward_command))
|
||||
|
||||
|
||||
# Add message handler for text, photos, voice, documents
|
||||
self._app.add_handler(
|
||||
MessageHandler(
|
||||
(filters.TEXT | filters.PHOTO | filters.VOICE | filters.AUDIO | filters.Document.ALL)
|
||||
& ~filters.COMMAND,
|
||||
(filters.TEXT | filters.PHOTO | filters.VOICE | filters.AUDIO | filters.Document.ALL)
|
||||
& ~filters.COMMAND,
|
||||
self._on_message
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
logger.info("Starting Telegram bot (polling mode)...")
|
||||
|
||||
|
||||
# Initialize and start polling
|
||||
await self._app.initialize()
|
||||
await self._app.start()
|
||||
|
||||
|
||||
# Get bot info and register command menu
|
||||
bot_info = await self._app.bot.get_me()
|
||||
logger.info(f"Telegram bot @{bot_info.username} connected")
|
||||
|
||||
|
||||
try:
|
||||
await self._app.bot.set_my_commands(self.BOT_COMMANDS)
|
||||
logger.debug("Telegram bot commands registered")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to register bot commands: {e}")
|
||||
|
||||
|
||||
# Start polling (this runs until stopped)
|
||||
await self._app.updater.start_polling(
|
||||
allowed_updates=["message"],
|
||||
drop_pending_updates=True # Ignore old messages on startup
|
||||
)
|
||||
|
||||
|
||||
# Keep running until stopped
|
||||
while self._running:
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Stop the Telegram bot."""
|
||||
self._running = False
|
||||
|
||||
|
||||
# Cancel all typing indicators
|
||||
for chat_id in list(self._typing_tasks):
|
||||
self._stop_typing(chat_id)
|
||||
|
||||
|
||||
if self._app:
|
||||
logger.info("Stopping Telegram bot...")
|
||||
await self._app.updater.stop()
|
||||
await self._app.stop()
|
||||
await self._app.shutdown()
|
||||
self._app = None
|
||||
|
||||
|
||||
async def send(self, msg: OutboundMessage) -> None:
|
||||
"""Send a message through Telegram."""
|
||||
if not self._app:
|
||||
@@ -199,21 +193,17 @@ class TelegramChannel(BaseChannel):
|
||||
if msg.metadata.get("suppressed", False):
|
||||
logger.debug(f"Suppressed output (not sent to Telegram): {msg.content[:100]}...")
|
||||
return # Don't send to Telegram API
|
||||
|
||||
|
||||
try:
|
||||
# chat_id should be the Telegram chat ID (integer)
|
||||
chat_id = int(msg.chat_id)
|
||||
# Convert markdown to Telegram HTML
|
||||
html_content = _markdown_to_telegram_html(msg.content)
|
||||
|
||||
# Handle media if present
|
||||
if msg.media:
|
||||
await self._send_with_media(chat_id, html_content, msg.media)
|
||||
else:
|
||||
# Text-only message
|
||||
await self._app.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=html_content,
|
||||
parse_mode="HTML"
|
||||
)
|
||||
await self._app.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=html_content,
|
||||
parse_mode="HTML"
|
||||
)
|
||||
except ValueError:
|
||||
logger.error(f"Invalid chat_id: {msg.chat_id}")
|
||||
except Exception as e:
|
||||
@@ -221,186 +211,24 @@ class TelegramChannel(BaseChannel):
|
||||
logger.warning(f"HTML parse failed, falling back to plain text: {e}")
|
||||
try:
|
||||
await self._app.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
chat_id=int(msg.chat_id),
|
||||
text=msg.content
|
||||
)
|
||||
except Exception as inner_e:
|
||||
logger.error(f"Error sending Telegram message: {inner_e}")
|
||||
|
||||
async def _send_with_media(self, chat_id: int, caption: str, media_paths: list[str]) -> None:
|
||||
"""
|
||||
Send message with media attachments.
|
||||
|
||||
Args:
|
||||
chat_id: Telegram chat ID
|
||||
caption: Message caption
|
||||
media_paths: List of file paths or URLs
|
||||
"""
|
||||
from telegram import InputMediaPhoto, InputMediaVideo
|
||||
|
||||
from nanobot.channels.telegram_media import (
|
||||
MediaKind,
|
||||
classify_media,
|
||||
detect_mime,
|
||||
fetch_media,
|
||||
group_media_for_album,
|
||||
optimize_image,
|
||||
)
|
||||
|
||||
# Process each media item
|
||||
processed_media: list[tuple[str, MediaKind, bytes]] = []
|
||||
|
||||
for path in media_paths:
|
||||
try:
|
||||
# Fetch remote URLs
|
||||
if path.startswith(("http://", "https://")):
|
||||
content, mime = await fetch_media(path, max_bytes=REMOTE_MEDIA_SIZE_LIMIT)
|
||||
kind = classify_media(mime)
|
||||
else:
|
||||
# Local file
|
||||
file_path = Path(path)
|
||||
if not file_path.exists():
|
||||
logger.warning(f"Media file not found: {path}")
|
||||
continue
|
||||
|
||||
with open(file_path, "rb") as f:
|
||||
content = f.read()
|
||||
|
||||
mime = detect_mime(path, content)
|
||||
kind = classify_media(mime)
|
||||
|
||||
# Optimize images
|
||||
if kind == MediaKind.IMAGE:
|
||||
try:
|
||||
content = optimize_image(path, max_bytes=6_000_000)
|
||||
except Exception as e:
|
||||
logger.warning(f"Image optimization failed: {e}")
|
||||
# Check if original is under Telegram's limit
|
||||
if len(content) <= TELEGRAM_PHOTO_SIZE_LIMIT:
|
||||
logger.info("Using original image (under 10MB limit)")
|
||||
else:
|
||||
logger.error(f"Original image too large ({len(content)} bytes), skipping")
|
||||
continue
|
||||
|
||||
processed_media.append((path, kind, content))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to process media {path}: {e}")
|
||||
continue
|
||||
|
||||
if not processed_media:
|
||||
# No media could be processed, send text only
|
||||
await self._app.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=caption,
|
||||
parse_mode="HTML"
|
||||
)
|
||||
return
|
||||
|
||||
# Group media for album sending
|
||||
media_items = [(path, kind) for path, kind, _ in processed_media]
|
||||
grouping = group_media_for_album(media_items)
|
||||
|
||||
# Handle caption length (Telegram limit: 1024 chars)
|
||||
if len(caption) > 1024:
|
||||
# Send media without caption, then follow-up text
|
||||
media_caption = None
|
||||
followup_text = caption
|
||||
else:
|
||||
media_caption = caption
|
||||
followup_text = None
|
||||
|
||||
# Send album if grouped
|
||||
if grouping["album"]:
|
||||
album_paths = grouping["album"]
|
||||
album_media = []
|
||||
|
||||
for path, kind, content in processed_media:
|
||||
if path not in album_paths:
|
||||
continue
|
||||
|
||||
if kind == MediaKind.IMAGE:
|
||||
media_obj = InputMediaPhoto(
|
||||
media=content,
|
||||
caption=media_caption if len(album_media) == 0 else None,
|
||||
parse_mode="HTML" if media_caption else None
|
||||
)
|
||||
elif kind == MediaKind.VIDEO:
|
||||
media_obj = InputMediaVideo(
|
||||
media=content,
|
||||
caption=media_caption if len(album_media) == 0 else None,
|
||||
parse_mode="HTML" if media_caption else None
|
||||
)
|
||||
else:
|
||||
continue # Skip non-album types
|
||||
|
||||
album_media.append(media_obj)
|
||||
|
||||
if album_media:
|
||||
await self._app.bot.send_media_group(
|
||||
chat_id=chat_id,
|
||||
media=album_media
|
||||
)
|
||||
|
||||
# Send separate media
|
||||
first_separate = True
|
||||
for path, kind, content in processed_media:
|
||||
if path in grouping["album"]:
|
||||
continue # Already sent in album
|
||||
|
||||
# Only first separate item gets caption
|
||||
item_caption = media_caption if first_separate else None
|
||||
first_separate = False
|
||||
|
||||
if kind == MediaKind.IMAGE:
|
||||
await self._app.bot.send_photo(
|
||||
chat_id=chat_id,
|
||||
photo=content,
|
||||
caption=item_caption,
|
||||
parse_mode="HTML" if item_caption else None
|
||||
)
|
||||
elif kind == MediaKind.VIDEO:
|
||||
await self._app.bot.send_video(
|
||||
chat_id=chat_id,
|
||||
video=content,
|
||||
caption=item_caption,
|
||||
parse_mode="HTML" if item_caption else None
|
||||
)
|
||||
elif kind == MediaKind.AUDIO:
|
||||
await self._app.bot.send_audio(
|
||||
chat_id=chat_id,
|
||||
audio=content,
|
||||
caption=item_caption,
|
||||
parse_mode="HTML" if item_caption else None
|
||||
)
|
||||
elif kind == MediaKind.DOCUMENT:
|
||||
await self._app.bot.send_document(
|
||||
chat_id=chat_id,
|
||||
document=content,
|
||||
caption=item_caption,
|
||||
parse_mode="HTML" if item_caption else None
|
||||
)
|
||||
|
||||
# Send follow-up text if caption was too long
|
||||
if followup_text:
|
||||
await self._app.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=followup_text,
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
except Exception as e2:
|
||||
logger.error(f"Error sending Telegram message: {e2}")
|
||||
|
||||
async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Handle /start command."""
|
||||
if not update.message or not update.effective_user:
|
||||
return
|
||||
|
||||
|
||||
user = update.effective_user
|
||||
await update.message.reply_text(
|
||||
f"👋 Hi {user.first_name}! I'm nanobot.\n\n"
|
||||
"Send me a message and I'll respond!\n"
|
||||
"Type /help to see available commands."
|
||||
)
|
||||
|
||||
|
||||
async def _forward_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Forward slash commands to the bus for unified handling in AgentLoop."""
|
||||
if not update.message or not update.effective_user:
|
||||
@@ -410,38 +238,38 @@ class TelegramChannel(BaseChannel):
|
||||
chat_id=str(update.message.chat_id),
|
||||
content=update.message.text,
|
||||
)
|
||||
|
||||
|
||||
async def _on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Handle incoming messages (text, photos, voice, documents)."""
|
||||
if not update.message or not update.effective_user:
|
||||
return
|
||||
|
||||
|
||||
message = update.message
|
||||
user = update.effective_user
|
||||
chat_id = message.chat_id
|
||||
|
||||
|
||||
# Use stable numeric ID, but keep username for allowlist compatibility
|
||||
sender_id = str(user.id)
|
||||
if user.username:
|
||||
sender_id = f"{sender_id}|{user.username}"
|
||||
|
||||
|
||||
# Store chat_id for replies
|
||||
self._chat_ids[sender_id] = chat_id
|
||||
|
||||
|
||||
# Build content from text and/or media
|
||||
content_parts = []
|
||||
media_paths = []
|
||||
|
||||
|
||||
# Text content
|
||||
if message.text:
|
||||
content_parts.append(message.text)
|
||||
if message.caption:
|
||||
content_parts.append(message.caption)
|
||||
|
||||
|
||||
# Handle media files
|
||||
media_file = None
|
||||
media_type = None
|
||||
|
||||
|
||||
if message.photo:
|
||||
media_file = message.photo[-1] # Largest photo
|
||||
media_type = "image"
|
||||
@@ -454,23 +282,23 @@ class TelegramChannel(BaseChannel):
|
||||
elif message.document:
|
||||
media_file = message.document
|
||||
media_type = "file"
|
||||
|
||||
|
||||
# Download media if present
|
||||
if media_file and self._app:
|
||||
try:
|
||||
file = await self._app.bot.get_file(media_file.file_id)
|
||||
ext = self._get_extension(media_type, getattr(media_file, 'mime_type', None))
|
||||
|
||||
|
||||
# Save to workspace/media/
|
||||
from pathlib import Path
|
||||
media_dir = Path.home() / ".nanobot" / "media"
|
||||
media_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
|
||||
await file.download_to_drive(str(file_path))
|
||||
|
||||
|
||||
media_paths.append(str(file_path))
|
||||
|
||||
|
||||
# Handle voice transcription
|
||||
if media_type == "voice" or media_type == "audio":
|
||||
from nanobot.providers.transcription import GroqTranscriptionProvider
|
||||
@@ -483,21 +311,21 @@ class TelegramChannel(BaseChannel):
|
||||
content_parts.append(f"[{media_type}: {file_path}]")
|
||||
else:
|
||||
content_parts.append(f"[{media_type}: {file_path}]")
|
||||
|
||||
|
||||
logger.debug(f"Downloaded {media_type} to {file_path}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to download media: {e}")
|
||||
content_parts.append(f"[{media_type}: download failed]")
|
||||
|
||||
|
||||
content = "\n".join(content_parts) if content_parts else "[empty message]"
|
||||
|
||||
|
||||
logger.debug(f"Telegram message from {sender_id}: {content[:50]}...")
|
||||
|
||||
|
||||
str_chat_id = str(chat_id)
|
||||
|
||||
|
||||
# Start typing indicator before processing
|
||||
self._start_typing(str_chat_id)
|
||||
|
||||
|
||||
# Forward to the message bus
|
||||
await self._handle_message(
|
||||
sender_id=sender_id,
|
||||
@@ -512,19 +340,19 @@ class TelegramChannel(BaseChannel):
|
||||
"is_group": message.chat.type != "private"
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _start_typing(self, chat_id: str) -> None:
|
||||
"""Start sending 'typing...' indicator for a chat."""
|
||||
# Cancel any existing typing task for this chat
|
||||
self._stop_typing(chat_id)
|
||||
self._typing_tasks[chat_id] = asyncio.create_task(self._typing_loop(chat_id))
|
||||
|
||||
|
||||
def _stop_typing(self, chat_id: str) -> None:
|
||||
"""Stop the typing indicator for a chat."""
|
||||
task = self._typing_tasks.pop(chat_id, None)
|
||||
if task and not task.done():
|
||||
task.cancel()
|
||||
|
||||
|
||||
async def _typing_loop(self, chat_id: str) -> None:
|
||||
"""Repeatedly send 'typing' action until cancelled."""
|
||||
try:
|
||||
@@ -535,7 +363,7 @@ class TelegramChannel(BaseChannel):
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.debug(f"Typing indicator stopped for {chat_id}: {e}")
|
||||
|
||||
|
||||
async def _on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Log polling / handler errors instead of silently swallowing them."""
|
||||
logger.error(f"Telegram error: {context.error}")
|
||||
@@ -549,6 +377,6 @@ class TelegramChannel(BaseChannel):
|
||||
}
|
||||
if mime_type in ext_map:
|
||||
return ext_map[mime_type]
|
||||
|
||||
|
||||
type_map = {"image": ".jpg", "voice": ".ogg", "audio": ".mp3", "file": ""}
|
||||
return type_map.get(media_type, "")
|
||||
|
||||
@@ -38,8 +38,6 @@ dependencies = [
|
||||
"qq-botpy>=1.0.0",
|
||||
"python-socks[asyncio]>=2.4.0",
|
||||
"prompt-toolkit>=3.0.0",
|
||||
"pillow-heif>=0.10.0",
|
||||
"python-magic>=0.4.27",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
Reference in New Issue
Block a user