nanobot/channels/telegram.py

@@ -4,9 +4,11 @@ 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, MessageHandler, filters, ContextTypes
+from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
 from telegram.request import HTTPXRequest
 
 from nanobot.bus.events import OutboundMessage
@@ -21,72 +23,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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
-    
+
     # 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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
         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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
         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"),
@@ -94,7 +96,7 @@ class TelegramChannel(BaseChannel):
         BotCommand("help", "Show available commands"),
         BotCommand("quota", "Show current quota status"),
     ]
-    
+
     def __init__(
         self,
         config: TelegramConfig,
@@ -107,15 +109,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)
@@ -123,63 +125,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:
@@ -193,17 +195,21 @@ 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)
-            await self._app.bot.send_message(
-                chat_id=chat_id,
-                text=html_content,
-                parse_mode="HTML"
-            )
+
+            # 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"
+                )
         except ValueError:
             logger.error(f"Invalid chat_id: {msg.chat_id}")
         except Exception as e:
@@ -211,24 +217,178 @@ class TelegramChannel(BaseChannel):
             logger.warning(f"HTML parse failed, falling back to plain text: {e}")
             try:
                 await self._app.bot.send_message(
-                    chat_id=int(msg.chat_id),
+                    chat_id=chat_id,
                     text=msg.content
                 )
-            except Exception as e2:
-                logger.error(f"Error sending Telegram message: {e2}")
-    
+            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=100_000_000)
+                    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}, sending original")
+
+                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
+        for i, (path, kind, content) in enumerate(processed_media):
+            if path in grouping["album"]:
+                continue  # Already sent in album
+
+            # Only first separate item gets caption
+            item_caption = media_caption if i == 0 else None
+
+            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"
+            )
+
     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:
@@ -238,38 +398,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"
@@ -282,23 +442,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
@@ -311,21 +471,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,
@@ -340,19 +500,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:
@@ -363,7 +523,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}")
@@ -377,6 +537,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, "")

tests/test_telegram_channel_media.py

@@ -0,0 +1,101 @@
+"""Integration tests for Telegram media sending."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.telegram import TelegramChannel
+
+
+@pytest.fixture
+def mock_telegram_app():
+    """Mock python-telegram-bot Application."""
+    app = MagicMock()
+    app.bot = MagicMock()
+    app.bot.send_photo = AsyncMock()
+    app.bot.send_video = AsyncMock()
+    app.bot.send_audio = AsyncMock()
+    app.bot.send_document = AsyncMock()
+    app.bot.send_media_group = AsyncMock()
+    app.bot.send_message = AsyncMock()
+    return app
+
+
+@pytest.mark.asyncio
+async def test_send_single_image(mock_telegram_app, tmp_path):
+    """Test sending single image."""
+    # Create test image
+    from PIL import Image
+    img_path = tmp_path / "test.jpg"
+    img = Image.new("RGB", (100, 100), color="red")
+    img.save(img_path, format="JPEG")
+
+    # Setup channel
+    bus = MessageBus()
+    config = MagicMock()
+    config.token = "fake_token"
+    config.proxy = None
+
+    channel = TelegramChannel(config, bus)
+    channel._app = mock_telegram_app
+    channel._running = True
+
+    # Send message with media
+    msg = OutboundMessage(
+        channel="telegram",
+        chat_id="12345",
+        content="Test image",
+        media=[str(img_path)]
+    )
+
+    with patch("nanobot.channels.telegram._markdown_to_telegram_html", return_value="Test image"):
+        await channel.send(msg)
+
+    # Verify send_photo was called
+    mock_telegram_app.bot.send_photo.assert_called_once()
+    call_args = mock_telegram_app.bot.send_photo.call_args
+    assert call_args.kwargs["chat_id"] == 12345
+    assert call_args.kwargs["caption"] == "Test image"
+
+
+@pytest.mark.asyncio
+async def test_send_album(mock_telegram_app, tmp_path):
+    """Test sending multiple images as album."""
+    from PIL import Image
+
+    # Create test images
+    img_paths = []
+    for i in range(3):
+        img_path = tmp_path / f"test{i}.jpg"
+        img = Image.new("RGB", (100, 100), color="red")
+        img.save(img_path, format="JPEG")
+        img_paths.append(str(img_path))
+
+    # Setup channel
+    bus = MessageBus()
+    config = MagicMock()
+    config.token = "fake_token"
+    config.proxy = None
+
+    channel = TelegramChannel(config, bus)
+    channel._app = mock_telegram_app
+    channel._running = True
+
+    # Send message with multiple images
+    msg = OutboundMessage(
+        channel="telegram",
+        chat_id="12345",
+        content="Album test",
+        media=img_paths
+    )
+
+    with patch("nanobot.channels.telegram._markdown_to_telegram_html", return_value="Album test"):
+        await channel.send(msg)
+
+    # Verify send_media_group was called
+    mock_telegram_app.bot.send_media_group.assert_called_once()
+    call_args = mock_telegram_app.bot.send_media_group.call_args
+    assert call_args.kwargs["chat_id"] == 12345
+    assert len(call_args.kwargs["media"]) == 3