| 1 | """Telegram channel implementation using python-telegram-bot.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import asyncio |
| 6 | import re |
| 7 | import time |
| 8 | import unicodedata |
| 9 | from dataclasses import dataclass |
| 10 | from typing import Any, Literal |
| 11 | |
| 12 | from loguru import logger |
| 13 | from pydantic import Field |
| 14 | from telegram import BotCommand, InlineKeyboardButton, InlineKeyboardMarkup, ReactionTypeEmoji, ReplyParameters, Update |
| 15 | from telegram.error import BadRequest, NetworkError, TimedOut |
| 16 | from telegram.ext import Application, CallbackQueryHandler, ContextTypes, MessageHandler, filters |
| 17 | from telegram.request import HTTPXRequest |
| 18 | |
| 19 | from nanobot.bus.events import OutboundMessage |
| 20 | from nanobot.bus.queue import MessageBus |
| 21 | from nanobot.channels.base import BaseChannel |
| 22 | from nanobot.command.builtin import build_help_text |
| 23 | from nanobot.config.paths import get_media_dir |
| 24 | from nanobot.config.schema import Base |
| 25 | from nanobot.security.network import validate_url_target |
| 26 | from nanobot.utils.helpers import split_message |
| 27 | |
| 28 | TELEGRAM_MAX_MESSAGE_LEN = 4000 # Telegram message character limit |
| 29 | # Telegram's actual API limit is 4096; we split raw markdown at 4000 as a |
| 30 | # safety margin for mid-stream edits (plain text). For _stream_end, we |
| 31 | # convert to HTML first and then split at the true 4096-char boundary so |
| 32 | # the final rendered message never overflows. |
| 33 | TELEGRAM_HTML_MAX_LEN = 4096 |
| 34 | TELEGRAM_REPLY_CONTEXT_MAX_LEN = TELEGRAM_MAX_MESSAGE_LEN # Max length for reply context in user message |
| 35 | |
| 36 | |
| 37 | def _escape_telegram_html(text: str) -> str: |
| 38 | """Escape text for Telegram HTML parse mode.""" |
| 39 | return text.replace("&", "&").replace("<", "<").replace(">", ">") |
| 40 | |
| 41 | |
| 42 | def _tool_hint_to_telegram_blockquote(text: str) -> str: |
| 43 | """Render tool hints as an expandable blockquote (collapsed by default).""" |
| 44 | return f"<blockquote expandable>{_escape_telegram_html(text)}</blockquote>" if text else "" |
| 45 | |
| 46 | |
| 47 | def _strip_md(s: str) -> str: |
| 48 | """Strip markdown inline formatting from text.""" |
| 49 | s = re.sub(r'\*\*(.+?)\*\*', r'\1', s) |
| 50 | s = re.sub(r'__(.+?)__', r'\1', s) |
| 51 | s = re.sub(r'~~(.+?)~~', r'\1', s) |
| 52 | s = re.sub(r'`([^`]+)`', r'\1', s) |
| 53 | return s.strip() |
| 54 | |
| 55 | |
| 56 | def _strip_md_block(text: str) -> str: |
| 57 | """Strip block-level and inline markdown for readable plain-text preview. |
| 58 | |
| 59 | Used during streaming mid-edits so users see clean text instead of raw |
| 60 | markdown syntax while the response is still being generated. |
| 61 | """ |
| 62 | # Code blocks -> just the code |
| 63 | text = re.sub(r'```[\w]*\n?([\s\S]*?)```', r'\1', text) |
| 64 | # Headers -> plain text |
| 65 | text = re.sub(r'^#{1,6}\s+(.+)$', r'\1', text, flags=re.MULTILINE) |
| 66 | # Blockquotes |
| 67 | text = re.sub(r'^>\s*(.*)$', r'\1', text, flags=re.MULTILINE) |
| 68 | # Bold / italic / strikethrough |
| 69 | text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) |
| 70 | text = re.sub(r'__(.+?)__', r'\1', text) |
| 71 | text = re.sub(r'(?<![a-zA-Z0-9])_([^_]+)_(?![a-zA-Z0-9])', r'\1', text) |
| 72 | text = re.sub(r'~~(.+?)~~', r'\1', text) |
| 73 | # Inline code |
| 74 | text = re.sub(r'`([^`]+)`', r'\1', text) |
| 75 | # Links [text](url) -> text |
| 76 | text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', text) |
| 77 | # Bullet lists |
| 78 | text = re.sub(r'^[-*]\s+', '• ', text, flags=re.MULTILINE) |
| 79 | # Numbered lists (normalize spacing) |
| 80 | text = re.sub(r'^(\d+)\.\s+', r'\1. ', text, flags=re.MULTILINE) |
| 81 | return text |
| 82 | |
| 83 | |
| 84 | def _render_table_box(table_lines: list[str]) -> str: |
| 85 | """Convert markdown pipe-table to compact aligned text for <pre> display.""" |
| 86 | |
| 87 | def dw(s: str) -> int: |
| 88 | return sum(2 if unicodedata.east_asian_width(c) in ('W', 'F') else 1 for c in s) |
| 89 | |
| 90 | rows: list[list[str]] = [] |
| 91 | has_sep = False |
| 92 | for line in table_lines: |
| 93 | cells = [_strip_md(c) for c in line.strip().strip('|').split('|')] |
| 94 | if all(re.match(r'^:?-+:?$', c) for c in cells if c): |
| 95 | has_sep = True |
| 96 | continue |
| 97 | rows.append(cells) |
| 98 | if not rows or not has_sep: |
| 99 | return '\n'.join(table_lines) |
| 100 | |
| 101 | ncols = max(len(r) for r in rows) |
| 102 | for r in rows: |
| 103 | r.extend([''] * (ncols - len(r))) |
| 104 | widths = [max(dw(r[c]) for r in rows) for c in range(ncols)] |
| 105 | |
| 106 | def dr(cells: list[str]) -> str: |
| 107 | return ' '.join(f'{c}{" " * (w - dw(c))}' for c, w in zip(cells, widths)) |
| 108 | |
| 109 | out = [dr(rows[0])] |
| 110 | out.append(' '.join('─' * w for w in widths)) |
| 111 | for row in rows[1:]: |
| 112 | out.append(dr(row)) |
| 113 | return '\n'.join(out) |
| 114 | |
| 115 | |
| 116 | def _markdown_to_telegram_html(text: str) -> str: |
| 117 | """ |
| 118 | Convert markdown to Telegram-safe HTML. |
| 119 | """ |
| 120 | if not text: |
| 121 | return "" |
| 122 | |
| 123 | # 1. Extract and protect code blocks (preserve content from other processing) |
| 124 | code_blocks: list[str] = [] |
| 125 | def save_code_block(m: re.Match) -> str: |
| 126 | code_blocks.append(m.group(1)) |
| 127 | return f"\x00CB{len(code_blocks) - 1}\x00" |
| 128 | |
| 129 | text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text) |
| 130 | |
| 131 | # 1.5. Convert markdown tables to box-drawing (reuse code_block placeholders) |
| 132 | lines = text.split('\n') |
| 133 | rebuilt: list[str] = [] |
| 134 | li = 0 |
| 135 | while li < len(lines): |
| 136 | if re.match(r'^\s*\|.+\|', lines[li]): |
| 137 | tbl: list[str] = [] |
| 138 | while li < len(lines) and re.match(r'^\s*\|.+\|', lines[li]): |
| 139 | tbl.append(lines[li]) |
| 140 | li += 1 |
| 141 | box = _render_table_box(tbl) |
| 142 | if box != '\n'.join(tbl): |
| 143 | code_blocks.append(box) |
| 144 | rebuilt.append(f"\x00CB{len(code_blocks) - 1}\x00") |
| 145 | else: |
| 146 | rebuilt.extend(tbl) |
| 147 | else: |
| 148 | rebuilt.append(lines[li]) |
| 149 | li += 1 |
| 150 | text = '\n'.join(rebuilt) |
| 151 | |
| 152 | # 2. Extract and protect inline code |
| 153 | inline_codes: list[str] = [] |
| 154 | def save_inline_code(m: re.Match) -> str: |
| 155 | inline_codes.append(m.group(1)) |
| 156 | return f"\x00IC{len(inline_codes) - 1}\x00" |
| 157 | |
| 158 | text = re.sub(r'`([^`]+)`', save_inline_code, text) |
| 159 | |
| 160 | # 3. Headers # Title -> <b>Title</b> (preserve visual hierarchy) |
| 161 | text = re.sub(r'^#{1,6}\s+(.+)$', r'⟪B⟫\1⟪/B⟫', text, flags=re.MULTILINE) |
| 162 | |
| 163 | # 4. Blockquotes > text -> just the text (before HTML escaping) |
| 164 | text = re.sub(r'^>\s*(.*)$', r'\1', text, flags=re.MULTILINE) |
| 165 | |
| 166 | # 5. Escape HTML special characters |
| 167 | text = _escape_telegram_html(text) |
| 168 | |
| 169 | # 6. Links [text](url) - must be before bold/italic to handle nested cases |
| 170 | text = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', text) |
| 171 | |
| 172 | # 7. Bold **text** or __text__ |
| 173 | text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text) |
| 174 | text = re.sub(r'__(.+?)__', r'<b>\1</b>', text) |
| 175 | |
| 176 | # 8. Italic _text_ (avoid matching inside words like some_var_name) |
| 177 | text = re.sub(r'(?<![a-zA-Z0-9])_([^_]+)_(?![a-zA-Z0-9])', r'<i>\1</i>', text) |
| 178 | |
| 179 | # 9. Strikethrough ~~text~~ |
| 180 | text = re.sub(r'~~(.+?)~~', r'<s>\1</s>', text) |
| 181 | |
| 182 | # 10. Bullet lists - item -> • item |
| 183 | text = re.sub(r'^[-*]\s+', '• ', text, flags=re.MULTILINE) |
| 184 | |
| 185 | # 10.5. Numbered lists 1. item -> 1. item (keep number, normalize indent) |
| 186 | text = re.sub(r'^(\d+)\.\s+', r'\1. ', text, flags=re.MULTILINE) |
| 187 | |
| 188 | # 11. Restore inline code with HTML tags |
| 189 | for i, code in enumerate(inline_codes): |
| 190 | # Escape HTML in code content |
| 191 | escaped = _escape_telegram_html(code) |
| 192 | text = text.replace(f"\x00IC{i}\x00", f"<code>{escaped}</code>") |
| 193 | |
| 194 | # 12. Restore code blocks with HTML tags |
| 195 | for i, code in enumerate(code_blocks): |
| 196 | # Escape HTML in code content |
| 197 | escaped = _escape_telegram_html(code) |
| 198 | text = text.replace(f"\x00CB{i}\x00", f"<pre><code>{escaped}</code></pre>") |
| 199 | |
| 200 | # 13. Restore header bold markers (inserted in step 3, after HTML escaping) |
| 201 | text = text.replace('⟪B⟫', '<b>').replace('⟪/B⟫', '</b>') |
| 202 | |
| 203 | return text |
| 204 | |
| 205 | |
| 206 | _SEND_MAX_RETRIES = 3 |
| 207 | _SEND_RETRY_BASE_DELAY = 0.5 # seconds, doubled each retry |
| 208 | _STREAM_EDIT_INTERVAL_DEFAULT = 0.6 # min seconds between edit_message_text calls |
| 209 | |
| 210 | |
| 211 | @dataclass |
| 212 | class _StreamBuf: |
| 213 | """Per-chat streaming accumulator for progressive message editing.""" |
| 214 | text: str = "" |
| 215 | message_id: int | None = None |
| 216 | last_edit: float = 0.0 |
| 217 | stream_id: str | None = None |
| 218 | |
| 219 | |
| 220 | class TelegramConfig(Base): |
| 221 | """Telegram channel configuration.""" |
| 222 | |
| 223 | enabled: bool = False |
| 224 | token: str = "" |
| 225 | allow_from: list[str] = Field(default_factory=list) |
| 226 | proxy: str | None = None |
| 227 | reply_to_message: bool = False |
| 228 | react_emoji: str = "👀" |
| 229 | group_policy: Literal["open", "mention"] = "mention" |
| 230 | connection_pool_size: int = 32 |
| 231 | pool_timeout: float = 5.0 |
| 232 | streaming: bool = True |
| 233 | # Enable inline keyboard buttons in Telegram messages. |
| 234 | inline_keyboards: bool = False |
| 235 | stream_edit_interval: float = Field(default=_STREAM_EDIT_INTERVAL_DEFAULT, ge=0.1) |
| 236 | |
| 237 | |
| 238 | class TelegramChannel(BaseChannel): |
| 239 | """ |
| 240 | Telegram channel using long polling. |
| 241 | |
| 242 | Simple and reliable - no webhook/public IP needed. |
| 243 | """ |
| 244 | |
| 245 | name = "telegram" |
| 246 | display_name = "Telegram" |
| 247 | |
| 248 | # Commands registered with Telegram's command menu |
| 249 | BOT_COMMANDS = [ |
| 250 | BotCommand("start", "Start the bot"), |
| 251 | BotCommand("new", "Start a new conversation"), |
| 252 | BotCommand("stop", "Stop the current task"), |
| 253 | BotCommand("restart", "Restart the bot"), |
| 254 | BotCommand("status", "Show bot status"), |
| 255 | BotCommand("dream", "Run Dream memory consolidation now"), |
| 256 | BotCommand("dream_log", "Show the latest Dream memory change"), |
| 257 | BotCommand("dream_restore", "Restore Dream memory to an earlier version"), |
| 258 | BotCommand("help", "Show available commands"), |
| 259 | ] |
| 260 | |
| 261 | @classmethod |
| 262 | def default_config(cls) -> dict[str, Any]: |
| 263 | return TelegramConfig().model_dump(by_alias=True) |
| 264 | |
| 265 | def __init__(self, config: Any, bus: MessageBus): |
| 266 | if isinstance(config, dict): |
| 267 | config = TelegramConfig.model_validate(config) |
| 268 | super().__init__(config, bus) |
| 269 | self.config: TelegramConfig = config |
| 270 | self._app: Application | None = None |
| 271 | self._chat_ids: dict[str, int] = {} # Map sender_id to chat_id for replies |
| 272 | self._typing_tasks: dict[str, asyncio.Task] = {} # chat_id -> typing loop task |
| 273 | self._media_group_buffers: dict[str, dict] = {} |
| 274 | self._media_group_tasks: dict[str, asyncio.Task] = {} |
| 275 | self._message_threads: dict[tuple[str, int], int] = {} |
| 276 | self._bot_user_id: int | None = None |
| 277 | self._bot_username: str | None = None |
| 278 | self._stream_bufs: dict[str, _StreamBuf] = {} # chat_id -> streaming state |
| 279 | |
| 280 | def is_allowed(self, sender_id: str) -> bool: |
| 281 | """Preserve Telegram's legacy id|username allowlist matching.""" |
| 282 | if super().is_allowed(sender_id): |
| 283 | return True |
| 284 | |
| 285 | allow_list = getattr(self.config, "allow_from", []) |
| 286 | if not allow_list or "*" in allow_list: |
| 287 | return False |
| 288 | |
| 289 | sender_str = str(sender_id) |
| 290 | if sender_str.count("|") != 1: |
| 291 | return False |
| 292 | |
| 293 | sid, username = sender_str.split("|", 1) |
| 294 | if not sid.isdigit() or not username: |
| 295 | return False |
| 296 | |
| 297 | return sid in allow_list or username in allow_list |
| 298 | |
| 299 | @staticmethod |
| 300 | def _normalize_telegram_command(content: str) -> str: |
| 301 | """Map Telegram-safe command aliases back to canonical nanobot commands.""" |
| 302 | if not content.startswith("/"): |
| 303 | return content |
| 304 | if content == "/dream_log" or content.startswith("/dream_log "): |
| 305 | return content.replace("/dream_log", "/dream-log", 1) |
| 306 | if content == "/dream_restore" or content.startswith("/dream_restore "): |
| 307 | return content.replace("/dream_restore", "/dream-restore", 1) |
| 308 | return content |
| 309 | |
| 310 | async def start(self) -> None: |
| 311 | """Start the Telegram bot with long polling.""" |
| 312 | if not self.config.token: |
| 313 | logger.error("Telegram bot token not configured") |
| 314 | return |
| 315 | |
| 316 | self._running = True |
| 317 | |
| 318 | proxy = self.config.proxy or None |
| 319 | |
| 320 | # Separate pools so long-polling (getUpdates) never starves outbound sends. |
| 321 | api_request = HTTPXRequest( |
| 322 | connection_pool_size=self.config.connection_pool_size, |
| 323 | pool_timeout=self.config.pool_timeout, |
| 324 | connect_timeout=30.0, |
| 325 | read_timeout=30.0, |
| 326 | proxy=proxy, |
| 327 | ) |
| 328 | poll_request = HTTPXRequest( |
| 329 | connection_pool_size=4, |
| 330 | pool_timeout=self.config.pool_timeout, |
| 331 | connect_timeout=30.0, |
| 332 | read_timeout=30.0, |
| 333 | proxy=proxy, |
| 334 | ) |
| 335 | builder = ( |
| 336 | Application.builder() |
| 337 | .token(self.config.token) |
| 338 | .request(api_request) |
| 339 | .get_updates_request(poll_request) |
| 340 | ) |
| 341 | self._app = builder.build() |
| 342 | self._app.add_error_handler(self._on_error) |
| 343 | |
| 344 | # Add command handlers (using Regex to support @username suffixes before bot initialization) |
| 345 | self._app.add_handler(MessageHandler(filters.Regex(r"^/start(?:@\w+)?$"), self._on_start)) |
| 346 | self._app.add_handler( |
| 347 | MessageHandler( |
| 348 | filters.Regex(r"^/(new|stop|restart|status|dream)(?:@\w+)?(?:\s+.*)?$"), |
| 349 | self._forward_command, |
| 350 | ) |
| 351 | ) |
| 352 | self._app.add_handler( |
| 353 | MessageHandler( |
| 354 | filters.Regex(r"^/(dream-log|dream_log|dream-restore|dream_restore)(?:@\w+)?(?:\s+.*)?$"), |
| 355 | self._forward_command, |
| 356 | ) |
| 357 | ) |
| 358 | self._app.add_handler(MessageHandler(filters.Regex(r"^/help(?:@\w+)?$"), self._on_help)) |
| 359 | |
| 360 | # Add message handler for text, photos, voice, documents, and locations |
| 361 | self._app.add_handler( |
| 362 | MessageHandler( |
| 363 | (filters.TEXT | filters.PHOTO | filters.VOICE | filters.AUDIO | filters.Document.ALL | filters.LOCATION) |
| 364 | & ~filters.COMMAND, |
| 365 | self._on_message |
| 366 | ) |
| 367 | ) |
| 368 | |
| 369 | # Conditionally register inline keyboard callback handler |
| 370 | if self.config.inline_keyboards: |
| 371 | self._app.add_handler(CallbackQueryHandler(self._on_callback_query)) |
| 372 | allowed_updates = ["message", "callback_query"] |
| 373 | logger.debug("Telegram inline keyboards enabled") |
| 374 | else: |
| 375 | allowed_updates = ["message"] |
| 376 | |
| 377 | logger.info("Starting Telegram bot (polling mode)...") |
| 378 | |
| 379 | # Initialize and start polling |
| 380 | await self._app.initialize() |
| 381 | await self._app.start() |
| 382 | |
| 383 | # Get bot info and register command menu |
| 384 | bot_info = await self._app.bot.get_me() |
| 385 | self._bot_user_id = getattr(bot_info, "id", None) |
| 386 | self._bot_username = getattr(bot_info, "username", None) |
| 387 | logger.info("Telegram bot @{} connected", bot_info.username) |
| 388 | |
| 389 | try: |
| 390 | await self._app.bot.set_my_commands(self.BOT_COMMANDS) |
| 391 | logger.debug("Telegram bot commands registered") |
| 392 | except Exception as e: |
| 393 | logger.warning("Failed to register bot commands: {}", e) |
| 394 | |
| 395 | # Start polling (this runs until stopped) |
| 396 | await self._app.updater.start_polling( |
| 397 | allowed_updates=allowed_updates, |
| 398 | drop_pending_updates=False, # Process pending messages on startup |
| 399 | error_callback=self._on_polling_error, |
| 400 | ) |
| 401 | |
| 402 | # Keep running until stopped |
| 403 | while self._running: |
| 404 | await asyncio.sleep(1) |
| 405 | |
| 406 | async def stop(self) -> None: |
| 407 | """Stop the Telegram bot.""" |
| 408 | self._running = False |
| 409 | |
| 410 | # Cancel all typing indicators |
| 411 | for chat_id in list(self._typing_tasks): |
| 412 | self._stop_typing(chat_id) |
| 413 | |
| 414 | for task in self._media_group_tasks.values(): |
| 415 | task.cancel() |
| 416 | self._media_group_tasks.clear() |
| 417 | self._media_group_buffers.clear() |
| 418 | |
| 419 | if self._app: |
| 420 | logger.info("Stopping Telegram bot...") |
| 421 | await self._app.updater.stop() |
| 422 | await self._app.stop() |
| 423 | await self._app.shutdown() |
| 424 | self._app = None |
| 425 | |
| 426 | @staticmethod |
| 427 | def _get_media_type(path: str) -> str: |
| 428 | """Guess media type from file extension.""" |
| 429 | ext = path.rsplit(".", 1)[-1].lower() if "." in path else "" |
| 430 | if ext in ("jpg", "jpeg", "png", "gif", "webp"): |
| 431 | return "photo" |
| 432 | if ext == "ogg": |
| 433 | return "voice" |
| 434 | if ext in ("mp3", "m4a", "wav", "aac"): |
| 435 | return "audio" |
| 436 | return "document" |
| 437 | |
| 438 | @staticmethod |
| 439 | def _is_remote_media_url(path: str) -> bool: |
| 440 | return path.startswith(("http://", "https://")) |
| 441 | |
| 442 | async def send(self, msg: OutboundMessage) -> None: |
| 443 | """Send a message through Telegram.""" |
| 444 | if not self._app: |
| 445 | logger.warning("Telegram bot not running") |
| 446 | return |
| 447 | |
| 448 | # Only stop typing indicator and remove reaction for final responses |
| 449 | if not msg.metadata.get("_progress", False): |
| 450 | self._stop_typing(msg.chat_id) |
| 451 | if reply_to_message_id := msg.metadata.get("message_id"): |
| 452 | try: |
| 453 | await self._remove_reaction(msg.chat_id, int(reply_to_message_id)) |
| 454 | except ValueError: |
| 455 | pass |
| 456 | |
| 457 | try: |
| 458 | chat_id = int(msg.chat_id) |
| 459 | except ValueError: |
| 460 | logger.error("Invalid chat_id: {}", msg.chat_id) |
| 461 | return |
| 462 | reply_to_message_id = msg.metadata.get("message_id") |
| 463 | message_thread_id = msg.metadata.get("message_thread_id") |
| 464 | if message_thread_id is None and reply_to_message_id is not None: |
| 465 | message_thread_id = self._message_threads.get((msg.chat_id, reply_to_message_id)) |
| 466 | thread_kwargs = {} |
| 467 | if message_thread_id is not None: |
| 468 | thread_kwargs["message_thread_id"] = message_thread_id |
| 469 | |
| 470 | reply_params = None |
| 471 | if self.config.reply_to_message: |
| 472 | if reply_to_message_id: |
| 473 | reply_params = ReplyParameters( |
| 474 | message_id=reply_to_message_id, |
| 475 | allow_sending_without_reply=True |
| 476 | ) |
| 477 | |
| 478 | # Send media files |
| 479 | for media_path in (msg.media or []): |
| 480 | try: |
| 481 | media_type = self._get_media_type(media_path) |
| 482 | sender = { |
| 483 | "photo": self._app.bot.send_photo, |
| 484 | "voice": self._app.bot.send_voice, |
| 485 | "audio": self._app.bot.send_audio, |
| 486 | }.get(media_type, self._app.bot.send_document) |
| 487 | param = "photo" if media_type == "photo" else media_type if media_type in ("voice", "audio") else "document" |
| 488 | |
| 489 | # Telegram Bot API accepts HTTP(S) URLs directly for media params. |
| 490 | if self._is_remote_media_url(media_path): |
| 491 | ok, error = validate_url_target(media_path) |
| 492 | if not ok: |
| 493 | raise ValueError(f"unsafe media URL: {error}") |
| 494 | await self._call_with_retry( |
| 495 | sender, |
| 496 | chat_id=chat_id, |
| 497 | **{param: media_path}, |
| 498 | reply_parameters=reply_params, |
| 499 | **thread_kwargs, |
| 500 | ) |
| 501 | continue |
| 502 | |
| 503 | with open(media_path, "rb") as f: |
| 504 | await sender( |
| 505 | chat_id=chat_id, |
| 506 | **{param: f}, |
| 507 | reply_parameters=reply_params, |
| 508 | **thread_kwargs, |
| 509 | ) |
| 510 | except Exception as e: |
| 511 | filename = media_path.rsplit("/", 1)[-1] |
| 512 | logger.error("Failed to send media {}: {}", media_path, e) |
| 513 | await self._app.bot.send_message( |
| 514 | chat_id=chat_id, |
| 515 | text=f"[Failed to send: {filename}]", |
| 516 | reply_parameters=reply_params, |
| 517 | **thread_kwargs, |
| 518 | ) |
| 519 | |
| 520 | # Send text content |
| 521 | if msg.content and msg.content != "[empty message]": |
| 522 | render_as_blockquote = bool(msg.metadata.get("_tool_hint")) |
| 523 | buttons = getattr(msg, "buttons", None) or [] |
| 524 | reply_markup = self._build_keyboard(buttons) if buttons else None |
| 525 | text = msg.content |
| 526 | # Fallback: no native keyboard → splice labels into the message so the choices survive. |
| 527 | if buttons and reply_markup is None: |
| 528 | text = f"{text}\n\n{self._buttons_as_text(buttons)}" |
| 529 | chunks = split_message(text, TELEGRAM_MAX_MESSAGE_LEN) |
| 530 | for i, chunk in enumerate(chunks): |
| 531 | is_last = (i == len(chunks) - 1) |
| 532 | await self._send_text( |
| 533 | chat_id, chunk, reply_params, thread_kwargs, |
| 534 | render_as_blockquote=render_as_blockquote, |
| 535 | reply_markup=reply_markup if is_last else None, |
| 536 | ) |
| 537 | |
| 538 | async def _call_with_retry(self, fn, *args, **kwargs): |
| 539 | """Call an async Telegram API function with retry on pool/network timeout and RetryAfter.""" |
| 540 | from telegram.error import RetryAfter |
| 541 | |
| 542 | for attempt in range(1, _SEND_MAX_RETRIES + 1): |
| 543 | try: |
| 544 | return await fn(*args, **kwargs) |
| 545 | except TimedOut: |
| 546 | if attempt == _SEND_MAX_RETRIES: |
| 547 | raise |
| 548 | delay = _SEND_RETRY_BASE_DELAY * (2 ** (attempt - 1)) |
| 549 | logger.warning( |
| 550 | "Telegram timeout (attempt {}/{}), retrying in {:.1f}s", |
| 551 | attempt, _SEND_MAX_RETRIES, delay, |
| 552 | ) |
| 553 | await asyncio.sleep(delay) |
| 554 | except RetryAfter as e: |
| 555 | if attempt == _SEND_MAX_RETRIES: |
| 556 | raise |
| 557 | delay = float(e.retry_after) |
| 558 | logger.warning( |
| 559 | "Telegram Flood Control (attempt {}/{}), retrying in {:.1f}s", |
| 560 | attempt, _SEND_MAX_RETRIES, delay, |
| 561 | ) |
| 562 | await asyncio.sleep(delay) |
| 563 | |
| 564 | async def _send_text( |
| 565 | self, |
| 566 | chat_id: int, |
| 567 | text: str, |
| 568 | reply_params=None, |
| 569 | thread_kwargs: dict | None = None, |
| 570 | render_as_blockquote: bool = False, |
| 571 | reply_markup=None, |
| 572 | ) -> None: |
| 573 | """Send a plain text message with HTML fallback.""" |
| 574 | try: |
| 575 | html = _tool_hint_to_telegram_blockquote(text) if render_as_blockquote else _markdown_to_telegram_html(text) |
| 576 | await self._call_with_retry( |
| 577 | self._app.bot.send_message, |
| 578 | chat_id=chat_id, text=html, parse_mode="HTML", |
| 579 | reply_parameters=reply_params, |
| 580 | reply_markup=reply_markup, |
| 581 | **(thread_kwargs or {}), |
| 582 | ) |
| 583 | except BadRequest as e: |
| 584 | logger.warning("HTML parse failed, falling back to plain text: {}", e) |
| 585 | try: |
| 586 | await self._call_with_retry( |
| 587 | self._app.bot.send_message, |
| 588 | chat_id=chat_id, |
| 589 | text=text, |
| 590 | reply_parameters=reply_params, |
| 591 | reply_markup=reply_markup, |
| 592 | **(thread_kwargs or {}), |
| 593 | ) |
| 594 | except Exception as e2: |
| 595 | logger.error("Error sending Telegram message: {}", e2) |
| 596 | raise |
| 597 | |
| 598 | @staticmethod |
| 599 | def _is_not_modified_error(exc: Exception) -> bool: |
| 600 | return isinstance(exc, BadRequest) and "message is not modified" in str(exc).lower() |
| 601 | |
| 602 | async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: |
| 603 | """Progressive message editing: send on first delta, edit on subsequent ones.""" |
| 604 | if not self._app: |
| 605 | return |
| 606 | meta = metadata or {} |
| 607 | int_chat_id = int(chat_id) |
| 608 | stream_id = meta.get("_stream_id") |
| 609 | |
| 610 | if meta.get("_stream_end"): |
| 611 | buf = self._stream_bufs.get(chat_id) |
| 612 | if not buf or not buf.message_id or not buf.text: |
| 613 | return |
| 614 | if stream_id is not None and buf.stream_id is not None and buf.stream_id != stream_id: |
| 615 | return |
| 616 | self._stop_typing(chat_id) |
| 617 | if reply_to_message_id := meta.get("message_id"): |
| 618 | try: |
| 619 | await self._remove_reaction(chat_id, int(reply_to_message_id)) |
| 620 | except ValueError: |
| 621 | pass |
| 622 | thread_kwargs = {} |
| 623 | if message_thread_id := meta.get("message_thread_id"): |
| 624 | thread_kwargs["message_thread_id"] = message_thread_id |
| 625 | raw_text = buf.text |
| 626 | html = _markdown_to_telegram_html(raw_text) |
| 627 | if len(html) <= TELEGRAM_HTML_MAX_LEN: |
| 628 | primary_html = html |
| 629 | extra_html_chunks = [] |
| 630 | else: |
| 631 | html_chunks = split_message(html, TELEGRAM_HTML_MAX_LEN) |
| 632 | primary_html = html_chunks[0] |
| 633 | extra_html_chunks = html_chunks[1:] |
| 634 | try: |
| 635 | await self._call_with_retry( |
| 636 | self._app.bot.edit_message_text, |
| 637 | chat_id=int_chat_id, message_id=buf.message_id, |
| 638 | text=primary_html, parse_mode="HTML", |
| 639 | ) |
| 640 | except BadRequest as e: |
| 641 | # Only fall back to plain text on actual HTML parse/format errors. |
| 642 | # Network errors (TimedOut, NetworkError) should propagate immediately |
| 643 | # to avoid doubling connection demand during pool exhaustion. |
| 644 | if self._is_not_modified_error(e): |
| 645 | logger.debug("Final stream edit already applied for {}", chat_id) |
| 646 | self._stream_bufs.pop(chat_id, None) |
| 647 | return |
| 648 | logger.debug("Final stream edit failed (HTML), trying plain: {}", e) |
| 649 | # Fall back to raw markdown (not HTML) so users don't see raw tags. |
| 650 | primary_plain = split_message(raw_text, TELEGRAM_MAX_MESSAGE_LEN)[0] if len(raw_text) > TELEGRAM_MAX_MESSAGE_LEN else raw_text |
| 651 | try: |
| 652 | await self._call_with_retry( |
| 653 | self._app.bot.edit_message_text, |
| 654 | chat_id=int_chat_id, message_id=buf.message_id, |
| 655 | text=primary_plain, |
| 656 | ) |
| 657 | except Exception as e2: |
| 658 | if self._is_not_modified_error(e2): |
| 659 | logger.debug("Final stream plain edit already applied for {}", chat_id) |
| 660 | else: |
| 661 | logger.warning("Final stream edit failed: {}", e2) |
| 662 | raise # Let ChannelManager handle retry |
| 663 | for extra_html_chunk in extra_html_chunks: |
| 664 | try: |
| 665 | await self._call_with_retry( |
| 666 | self._app.bot.send_message, |
| 667 | chat_id=int_chat_id, text=extra_html_chunk, |
| 668 | parse_mode="HTML", |
| 669 | **thread_kwargs, |
| 670 | ) |
| 671 | except Exception: |
| 672 | # Fall back to _send_text which handles HTML→plain gracefully. |
| 673 | await self._send_text(int_chat_id, extra_html_chunk) |
| 674 | self._stream_bufs.pop(chat_id, None) |
| 675 | return |
| 676 | |
| 677 | buf = self._stream_bufs.get(chat_id) |
| 678 | if buf is None or (stream_id is not None and buf.stream_id is not None and buf.stream_id != stream_id): |
| 679 | buf = _StreamBuf(stream_id=stream_id) |
| 680 | self._stream_bufs[chat_id] = buf |
| 681 | elif buf.stream_id is None: |
| 682 | buf.stream_id = stream_id |
| 683 | buf.text += delta |
| 684 | |
| 685 | if not buf.text.strip(): |
| 686 | return |
| 687 | |
| 688 | now = time.monotonic() |
| 689 | thread_kwargs = {} |
| 690 | if message_thread_id := meta.get("message_thread_id"): |
| 691 | thread_kwargs["message_thread_id"] = message_thread_id |
| 692 | if buf.message_id is None: |
| 693 | preview = _strip_md_block(buf.text) |
| 694 | try: |
| 695 | sent = await self._call_with_retry( |
| 696 | self._app.bot.send_message, |
| 697 | chat_id=int_chat_id, text=preview, |
| 698 | **thread_kwargs, |
| 699 | ) |
| 700 | buf.message_id = sent.message_id |
| 701 | buf.last_edit = now |
| 702 | except Exception as e: |
| 703 | logger.warning("Stream initial send failed: {}", e) |
| 704 | raise # Let ChannelManager handle retry |
| 705 | elif (now - buf.last_edit) >= self.config.stream_edit_interval: |
| 706 | if len(buf.text) > TELEGRAM_MAX_MESSAGE_LEN: |
| 707 | await self._flush_stream_overflow(int_chat_id, buf, thread_kwargs) |
| 708 | buf.last_edit = now |
| 709 | return |
| 710 | preview = _strip_md_block(buf.text) |
| 711 | try: |
| 712 | await self._call_with_retry( |
| 713 | self._app.bot.edit_message_text, |
| 714 | chat_id=int_chat_id, message_id=buf.message_id, |
| 715 | text=preview, |
| 716 | ) |
| 717 | buf.last_edit = now |
| 718 | except Exception as e: |
| 719 | if self._is_not_modified_error(e): |
| 720 | buf.last_edit = now |
| 721 | return |
| 722 | logger.warning("Stream edit failed: {}", e) |
| 723 | raise # Let ChannelManager handle retry |
| 724 | |
| 725 | async def _flush_stream_overflow( |
| 726 | self, |
| 727 | chat_id: int, |
| 728 | buf: "_StreamBuf", |
| 729 | thread_kwargs: dict, |
| 730 | ) -> None: |
| 731 | """Split an oversized stream buffer mid-flight. |
| 732 | |
| 733 | Edits the current stream message with the first chunk, sends any |
| 734 | intermediate chunks as standalone messages, then opens a new message |
| 735 | for the tail so subsequent deltas continue streaming into it. |
| 736 | """ |
| 737 | chunks = split_message(buf.text, TELEGRAM_MAX_MESSAGE_LEN) |
| 738 | if len(chunks) <= 1: |
| 739 | return |
| 740 | try: |
| 741 | await self._call_with_retry( |
| 742 | self._app.bot.edit_message_text, |
| 743 | chat_id=chat_id, message_id=buf.message_id, |
| 744 | text=chunks[0], |
| 745 | ) |
| 746 | except Exception as e: |
| 747 | if not self._is_not_modified_error(e): |
| 748 | logger.warning("Stream overflow edit failed: {}", e) |
| 749 | raise |
| 750 | for chunk in chunks[1:-1]: |
| 751 | await self._call_with_retry( |
| 752 | self._app.bot.send_message, |
| 753 | chat_id=chat_id, text=chunk, **thread_kwargs, |
| 754 | ) |
| 755 | tail = chunks[-1] |
| 756 | sent = await self._call_with_retry( |
| 757 | self._app.bot.send_message, |
| 758 | chat_id=chat_id, text=tail, **thread_kwargs, |
| 759 | ) |
| 760 | buf.message_id = sent.message_id |
| 761 | buf.text = tail |
| 762 | |
| 763 | async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 764 | """Handle /start command.""" |
| 765 | if not update.message or not update.effective_user: |
| 766 | return |
| 767 | |
| 768 | user = update.effective_user |
| 769 | await update.message.reply_text( |
| 770 | f"👋 Hi {user.first_name}! I'm nanobot.\n\n" |
| 771 | "Send me a message and I'll respond!\n" |
| 772 | "Type /help to see available commands." |
| 773 | ) |
| 774 | |
| 775 | async def _on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 776 | """Handle /help command, bypassing ACL so all users can access it.""" |
| 777 | if not update.message: |
| 778 | return |
| 779 | await update.message.reply_text(build_help_text()) |
| 780 | |
| 781 | @staticmethod |
| 782 | def _sender_id(user) -> str: |
| 783 | """Build sender_id with username for allowlist matching.""" |
| 784 | sid = str(user.id) |
| 785 | return f"{sid}|{user.username}" if user.username else sid |
| 786 | |
| 787 | @staticmethod |
| 788 | def _derive_topic_session_key(message) -> str | None: |
| 789 | """Derive topic-scoped session key for Telegram chats with threads.""" |
| 790 | message_thread_id = getattr(message, "message_thread_id", None) |
| 791 | if message_thread_id is None: |
| 792 | return None |
| 793 | return f"telegram:{message.chat_id}:topic:{message_thread_id}" |
| 794 | |
| 795 | @staticmethod |
| 796 | def _build_message_metadata(message, user) -> dict: |
| 797 | """Build common Telegram inbound metadata payload.""" |
| 798 | reply_to = getattr(message, "reply_to_message", None) |
| 799 | return { |
| 800 | "message_id": message.message_id, |
| 801 | "user_id": user.id, |
| 802 | "username": user.username, |
| 803 | "first_name": user.first_name, |
| 804 | "is_group": message.chat.type != "private", |
| 805 | "message_thread_id": getattr(message, "message_thread_id", None), |
| 806 | "is_forum": bool(getattr(message.chat, "is_forum", False)), |
| 807 | "reply_to_message_id": getattr(reply_to, "message_id", None) if reply_to else None, |
| 808 | } |
| 809 | |
| 810 | async def _extract_reply_context(self, message) -> str | None: |
| 811 | """Extract text from the message being replied to, if any.""" |
| 812 | reply = getattr(message, "reply_to_message", None) |
| 813 | if not reply: |
| 814 | return None |
| 815 | text = getattr(reply, "text", None) or getattr(reply, "caption", None) or "" |
| 816 | if len(text) > TELEGRAM_REPLY_CONTEXT_MAX_LEN: |
| 817 | text = text[:TELEGRAM_REPLY_CONTEXT_MAX_LEN] + "..." |
| 818 | |
| 819 | if not text: |
| 820 | return None |
| 821 | |
| 822 | bot_id, _ = await self._ensure_bot_identity() |
| 823 | reply_user = getattr(reply, "from_user", None) |
| 824 | |
| 825 | if bot_id and reply_user and getattr(reply_user, "id", None) == bot_id: |
| 826 | return f"[Reply to bot: {text}]" |
| 827 | elif reply_user and getattr(reply_user, "username", None): |
| 828 | return f"[Reply to @{reply_user.username}: {text}]" |
| 829 | elif reply_user and getattr(reply_user, "first_name", None): |
| 830 | return f"[Reply to {reply_user.first_name}: {text}]" |
| 831 | else: |
| 832 | return f"[Reply to: {text}]" |
| 833 | |
| 834 | async def _download_message_media( |
| 835 | self, msg, *, add_failure_content: bool = False |
| 836 | ) -> tuple[list[str], list[str]]: |
| 837 | """Download media from a message (current or reply). Returns (media_paths, content_parts).""" |
| 838 | media_file = None |
| 839 | media_type = None |
| 840 | if getattr(msg, "photo", None): |
| 841 | media_file = msg.photo[-1] |
| 842 | media_type = "image" |
| 843 | elif getattr(msg, "voice", None): |
| 844 | media_file = msg.voice |
| 845 | media_type = "voice" |
| 846 | elif getattr(msg, "audio", None): |
| 847 | media_file = msg.audio |
| 848 | media_type = "audio" |
| 849 | elif getattr(msg, "document", None): |
| 850 | media_file = msg.document |
| 851 | media_type = "file" |
| 852 | elif getattr(msg, "video", None): |
| 853 | media_file = msg.video |
| 854 | media_type = "video" |
| 855 | elif getattr(msg, "video_note", None): |
| 856 | media_file = msg.video_note |
| 857 | media_type = "video" |
| 858 | elif getattr(msg, "animation", None): |
| 859 | media_file = msg.animation |
| 860 | media_type = "animation" |
| 861 | if not media_file or not self._app: |
| 862 | return [], [] |
| 863 | try: |
| 864 | file = await self._app.bot.get_file(media_file.file_id) |
| 865 | ext = self._get_extension( |
| 866 | media_type, |
| 867 | getattr(media_file, "mime_type", None), |
| 868 | getattr(media_file, "file_name", None), |
| 869 | ) |
| 870 | media_dir = get_media_dir("telegram") |
| 871 | unique_id = getattr(media_file, "file_unique_id", media_file.file_id) |
| 872 | file_path = media_dir / f"{unique_id}{ext}" |
| 873 | await file.download_to_drive(str(file_path)) |
| 874 | path_str = str(file_path) |
| 875 | if media_type in ("voice", "audio"): |
| 876 | transcription = await self.transcribe_audio(file_path) |
| 877 | if transcription: |
| 878 | logger.info("Transcribed {}: {}...", media_type, transcription[:50]) |
| 879 | return [path_str], [f"[transcription: {transcription}]"] |
| 880 | return [path_str], [f"[{media_type}: {path_str}]"] |
| 881 | return [path_str], [f"[{media_type}: {path_str}]"] |
| 882 | except Exception as e: |
| 883 | logger.warning("Failed to download message media: {}", e) |
| 884 | if add_failure_content: |
| 885 | return [], [f"[{media_type}: download failed]"] |
| 886 | return [], [] |
| 887 | |
| 888 | async def _ensure_bot_identity(self) -> tuple[int | None, str | None]: |
| 889 | """Load bot identity once and reuse it for mention/reply checks.""" |
| 890 | if self._bot_user_id is not None or self._bot_username is not None: |
| 891 | return self._bot_user_id, self._bot_username |
| 892 | if not self._app: |
| 893 | return None, None |
| 894 | bot_info = await self._app.bot.get_me() |
| 895 | self._bot_user_id = getattr(bot_info, "id", None) |
| 896 | self._bot_username = getattr(bot_info, "username", None) |
| 897 | return self._bot_user_id, self._bot_username |
| 898 | |
| 899 | @staticmethod |
| 900 | def _has_mention_entity( |
| 901 | text: str, |
| 902 | entities, |
| 903 | bot_username: str, |
| 904 | bot_id: int | None, |
| 905 | ) -> bool: |
| 906 | """Check Telegram mention entities against the bot username.""" |
| 907 | handle = f"@{bot_username}".lower() |
| 908 | for entity in entities or []: |
| 909 | entity_type = getattr(entity, "type", None) |
| 910 | if entity_type == "text_mention": |
| 911 | user = getattr(entity, "user", None) |
| 912 | if user is not None and bot_id is not None and getattr(user, "id", None) == bot_id: |
| 913 | return True |
| 914 | continue |
| 915 | if entity_type != "mention": |
| 916 | continue |
| 917 | offset = getattr(entity, "offset", None) |
| 918 | length = getattr(entity, "length", None) |
| 919 | if offset is None or length is None: |
| 920 | continue |
| 921 | if text[offset : offset + length].lower() == handle: |
| 922 | return True |
| 923 | return handle in text.lower() |
| 924 | |
| 925 | async def _is_group_message_for_bot(self, message) -> bool: |
| 926 | """Allow group messages when policy is open, @mentioned, or replying to the bot.""" |
| 927 | if message.chat.type == "private" or self.config.group_policy == "open": |
| 928 | return True |
| 929 | |
| 930 | bot_id, bot_username = await self._ensure_bot_identity() |
| 931 | if bot_username: |
| 932 | text = message.text or "" |
| 933 | caption = message.caption or "" |
| 934 | if self._has_mention_entity( |
| 935 | text, |
| 936 | getattr(message, "entities", None), |
| 937 | bot_username, |
| 938 | bot_id, |
| 939 | ): |
| 940 | return True |
| 941 | if self._has_mention_entity( |
| 942 | caption, |
| 943 | getattr(message, "caption_entities", None), |
| 944 | bot_username, |
| 945 | bot_id, |
| 946 | ): |
| 947 | return True |
| 948 | |
| 949 | reply_user = getattr(getattr(message, "reply_to_message", None), "from_user", None) |
| 950 | return bool(bot_id and reply_user and reply_user.id == bot_id) |
| 951 | |
| 952 | def _remember_thread_context(self, message) -> None: |
| 953 | """Cache Telegram thread context by chat/message id for follow-up replies.""" |
| 954 | message_thread_id = getattr(message, "message_thread_id", None) |
| 955 | if message_thread_id is None: |
| 956 | return |
| 957 | key = (str(message.chat_id), message.message_id) |
| 958 | self._message_threads[key] = message_thread_id |
| 959 | if len(self._message_threads) > 1000: |
| 960 | self._message_threads.pop(next(iter(self._message_threads))) |
| 961 | |
| 962 | async def _forward_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 963 | """Forward slash commands to the bus for unified handling in AgentLoop.""" |
| 964 | if not update.message or not update.effective_user: |
| 965 | return |
| 966 | message = update.message |
| 967 | user = update.effective_user |
| 968 | self._remember_thread_context(message) |
| 969 | |
| 970 | # Strip @bot_username suffix if present |
| 971 | content = message.text or "" |
| 972 | if content.startswith("/") and "@" in content: |
| 973 | cmd_part, *rest = content.split(" ", 1) |
| 974 | cmd_part = cmd_part.split("@")[0] |
| 975 | content = f"{cmd_part} {rest[0]}" if rest else cmd_part |
| 976 | content = self._normalize_telegram_command(content) |
| 977 | |
| 978 | await self._handle_message( |
| 979 | sender_id=self._sender_id(user), |
| 980 | chat_id=str(message.chat_id), |
| 981 | content=content, |
| 982 | metadata=self._build_message_metadata(message, user), |
| 983 | session_key=self._derive_topic_session_key(message), |
| 984 | ) |
| 985 | |
| 986 | async def _on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 987 | """Handle incoming messages (text, photos, voice, documents).""" |
| 988 | if not update.message or not update.effective_user: |
| 989 | return |
| 990 | |
| 991 | message = update.message |
| 992 | user = update.effective_user |
| 993 | chat_id = message.chat_id |
| 994 | sender_id = self._sender_id(user) |
| 995 | self._remember_thread_context(message) |
| 996 | |
| 997 | # Store chat_id for replies |
| 998 | self._chat_ids[sender_id] = chat_id |
| 999 | |
| 1000 | if not await self._is_group_message_for_bot(message): |
| 1001 | return |
| 1002 | |
| 1003 | # Build content from text and/or media |
| 1004 | content_parts = [] |
| 1005 | media_paths = [] |
| 1006 | |
| 1007 | # Text content |
| 1008 | if message.text: |
| 1009 | content_parts.append(message.text) |
| 1010 | if message.caption: |
| 1011 | content_parts.append(message.caption) |
| 1012 | |
| 1013 | # Location content |
| 1014 | if message.location: |
| 1015 | lat = message.location.latitude |
| 1016 | lon = message.location.longitude |
| 1017 | content_parts.append(f"[location: {lat}, {lon}]") |
| 1018 | |
| 1019 | # Download current message media |
| 1020 | current_media_paths, current_media_parts = await self._download_message_media( |
| 1021 | message, add_failure_content=True |
| 1022 | ) |
| 1023 | media_paths.extend(current_media_paths) |
| 1024 | content_parts.extend(current_media_parts) |
| 1025 | if current_media_paths: |
| 1026 | logger.debug("Downloaded message media to {}", current_media_paths[0]) |
| 1027 | |
| 1028 | # Reply context: text and/or media from the replied-to message |
| 1029 | reply = getattr(message, "reply_to_message", None) |
| 1030 | if reply is not None: |
| 1031 | reply_ctx = await self._extract_reply_context(message) |
| 1032 | reply_media, reply_media_parts = await self._download_message_media(reply) |
| 1033 | if reply_media: |
| 1034 | media_paths = reply_media + media_paths |
| 1035 | logger.debug("Attached replied-to media: {}", reply_media[0]) |
| 1036 | tag = reply_ctx or (f"[Reply to: {reply_media_parts[0]}]" if reply_media_parts else None) |
| 1037 | if tag: |
| 1038 | content_parts.insert(0, tag) |
| 1039 | content = "\n".join(content_parts) if content_parts else "[empty message]" |
| 1040 | |
| 1041 | logger.debug("Telegram message from {}: {}...", sender_id, content[:50]) |
| 1042 | |
| 1043 | str_chat_id = str(chat_id) |
| 1044 | metadata = self._build_message_metadata(message, user) |
| 1045 | session_key = self._derive_topic_session_key(message) |
| 1046 | |
| 1047 | # Telegram media groups: buffer briefly, forward as one aggregated turn. |
| 1048 | if media_group_id := getattr(message, "media_group_id", None): |
| 1049 | key = f"{str_chat_id}:{media_group_id}" |
| 1050 | if key not in self._media_group_buffers: |
| 1051 | self._media_group_buffers[key] = { |
| 1052 | "sender_id": sender_id, "chat_id": str_chat_id, |
| 1053 | "contents": [], "media": [], |
| 1054 | "metadata": metadata, |
| 1055 | "session_key": session_key, |
| 1056 | } |
| 1057 | self._start_typing(str_chat_id) |
| 1058 | await self._add_reaction(str_chat_id, message.message_id, self.config.react_emoji) |
| 1059 | buf = self._media_group_buffers[key] |
| 1060 | if content and content != "[empty message]": |
| 1061 | buf["contents"].append(content) |
| 1062 | buf["media"].extend(media_paths) |
| 1063 | if key not in self._media_group_tasks: |
| 1064 | self._media_group_tasks[key] = asyncio.create_task(self._flush_media_group(key)) |
| 1065 | return |
| 1066 | |
| 1067 | # Start typing indicator before processing |
| 1068 | self._start_typing(str_chat_id) |
| 1069 | await self._add_reaction(str_chat_id, message.message_id, self.config.react_emoji) |
| 1070 | |
| 1071 | # Forward to the message bus |
| 1072 | await self._handle_message( |
| 1073 | sender_id=sender_id, |
| 1074 | chat_id=str_chat_id, |
| 1075 | content=content, |
| 1076 | media=media_paths, |
| 1077 | metadata=metadata, |
| 1078 | session_key=session_key, |
| 1079 | ) |
| 1080 | |
| 1081 | async def _flush_media_group(self, key: str) -> None: |
| 1082 | """Wait briefly, then forward buffered media-group as one turn.""" |
| 1083 | try: |
| 1084 | await asyncio.sleep(0.6) |
| 1085 | if not (buf := self._media_group_buffers.pop(key, None)): |
| 1086 | return |
| 1087 | content = "\n".join(buf["contents"]) or "[empty message]" |
| 1088 | await self._handle_message( |
| 1089 | sender_id=buf["sender_id"], chat_id=buf["chat_id"], |
| 1090 | content=content, media=list(dict.fromkeys(buf["media"])), |
| 1091 | metadata=buf["metadata"], |
| 1092 | session_key=buf.get("session_key"), |
| 1093 | ) |
| 1094 | finally: |
| 1095 | self._media_group_tasks.pop(key, None) |
| 1096 | |
| 1097 | def _start_typing(self, chat_id: str) -> None: |
| 1098 | """Start sending 'typing...' indicator for a chat.""" |
| 1099 | # Cancel any existing typing task for this chat |
| 1100 | self._stop_typing(chat_id) |
| 1101 | self._typing_tasks[chat_id] = asyncio.create_task(self._typing_loop(chat_id)) |
| 1102 | |
| 1103 | def _stop_typing(self, chat_id: str) -> None: |
| 1104 | """Stop the typing indicator for a chat.""" |
| 1105 | task = self._typing_tasks.pop(chat_id, None) |
| 1106 | if task and not task.done(): |
| 1107 | task.cancel() |
| 1108 | |
| 1109 | async def _add_reaction(self, chat_id: str, message_id: int, emoji: str) -> None: |
| 1110 | """Add emoji reaction to a message (best-effort, non-blocking).""" |
| 1111 | if not self._app or not emoji: |
| 1112 | return |
| 1113 | try: |
| 1114 | await self._app.bot.set_message_reaction( |
| 1115 | chat_id=int(chat_id), |
| 1116 | message_id=message_id, |
| 1117 | reaction=[ReactionTypeEmoji(emoji=emoji)], |
| 1118 | ) |
| 1119 | except Exception as e: |
| 1120 | logger.debug("Telegram reaction failed: {}", e) |
| 1121 | |
| 1122 | async def _remove_reaction(self, chat_id: str, message_id: int) -> None: |
| 1123 | """Remove emoji reaction from a message (best-effort, non-blocking).""" |
| 1124 | if not self._app: |
| 1125 | return |
| 1126 | try: |
| 1127 | await self._app.bot.set_message_reaction( |
| 1128 | chat_id=int(chat_id), |
| 1129 | message_id=message_id, |
| 1130 | reaction=[], |
| 1131 | ) |
| 1132 | except Exception as e: |
| 1133 | logger.debug("Telegram reaction removal failed: {}", e) |
| 1134 | |
| 1135 | async def _typing_loop(self, chat_id: str) -> None: |
| 1136 | """Repeatedly send 'typing' action until cancelled.""" |
| 1137 | try: |
| 1138 | while self._app: |
| 1139 | await self._app.bot.send_chat_action(chat_id=int(chat_id), action="typing") |
| 1140 | await asyncio.sleep(4) |
| 1141 | except asyncio.CancelledError: |
| 1142 | pass |
| 1143 | except Exception as e: |
| 1144 | logger.debug("Typing indicator stopped for {}: {}", chat_id, e) |
| 1145 | |
| 1146 | @staticmethod |
| 1147 | def _format_telegram_error(exc: Exception) -> str: |
| 1148 | """Return a short, readable error summary for logs.""" |
| 1149 | text = str(exc).strip() |
| 1150 | if text: |
| 1151 | return text |
| 1152 | if exc.__cause__ is not None: |
| 1153 | cause = exc.__cause__ |
| 1154 | cause_text = str(cause).strip() |
| 1155 | if cause_text: |
| 1156 | return f"{exc.__class__.__name__} ({cause_text})" |
| 1157 | return f"{exc.__class__.__name__} ({cause.__class__.__name__})" |
| 1158 | return exc.__class__.__name__ |
| 1159 | |
| 1160 | def _on_polling_error(self, exc: Exception) -> None: |
| 1161 | """Keep long-polling network failures to a single readable line.""" |
| 1162 | summary = self._format_telegram_error(exc) |
| 1163 | if isinstance(exc, (NetworkError, TimedOut)): |
| 1164 | logger.warning("Telegram polling network issue: {}", summary) |
| 1165 | else: |
| 1166 | logger.error("Telegram polling error: {}", summary) |
| 1167 | |
| 1168 | async def _on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 1169 | """Log polling / handler errors instead of silently swallowing them.""" |
| 1170 | summary = self._format_telegram_error(context.error) |
| 1171 | |
| 1172 | if isinstance(context.error, (NetworkError, TimedOut)): |
| 1173 | logger.warning("Telegram network issue: {}", summary) |
| 1174 | else: |
| 1175 | logger.error("Telegram error: {}", summary) |
| 1176 | |
| 1177 | def _get_extension( |
| 1178 | self, |
| 1179 | media_type: str, |
| 1180 | mime_type: str | None, |
| 1181 | filename: str | None = None, |
| 1182 | ) -> str: |
| 1183 | """Get file extension based on media type or original filename.""" |
| 1184 | if mime_type: |
| 1185 | ext_map = { |
| 1186 | "image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif", |
| 1187 | "audio/ogg": ".ogg", "audio/mpeg": ".mp3", "audio/mp4": ".m4a", |
| 1188 | } |
| 1189 | if mime_type in ext_map: |
| 1190 | return ext_map[mime_type] |
| 1191 | |
| 1192 | type_map = {"image": ".jpg", "voice": ".ogg", "audio": ".mp3", "file": ""} |
| 1193 | if ext := type_map.get(media_type, ""): |
| 1194 | return ext |
| 1195 | |
| 1196 | if filename: |
| 1197 | from pathlib import Path |
| 1198 | |
| 1199 | return "".join(Path(filename).suffixes) |
| 1200 | |
| 1201 | return "" |
| 1202 | |
| 1203 | def _build_keyboard(self, buttons: list) -> InlineKeyboardMarkup | None: |
| 1204 | """Build inline keyboard markup if inline_keyboards is enabled.""" |
| 1205 | if not buttons or not self.config.inline_keyboards: |
| 1206 | return None |
| 1207 | keyboard = [ |
| 1208 | [InlineKeyboardButton(label, callback_data=self._safe_callback_data(label)) for label in row] |
| 1209 | for row in buttons |
| 1210 | ] |
| 1211 | return InlineKeyboardMarkup(keyboard) |
| 1212 | |
| 1213 | @staticmethod |
| 1214 | def _safe_callback_data(label: str) -> str: |
| 1215 | # Telegram caps callback_data at 64 bytes UTF-8; truncate at a char boundary so the keyboard still sends. |
| 1216 | encoded = label.encode("utf-8") |
| 1217 | if len(encoded) <= 64: |
| 1218 | return label |
| 1219 | return encoded[:64].decode("utf-8", errors="ignore") |
| 1220 | |
| 1221 | @staticmethod |
| 1222 | def _buttons_as_text(buttons: list[list[str]]) -> str: |
| 1223 | # Buttons are semantic options; when we can't render a keyboard, the user still needs to see them. |
| 1224 | return "\n".join(" ".join(f"[{label}]" for label in row) for row in buttons if row) |
| 1225 | |
| 1226 | async def _on_callback_query(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: |
| 1227 | """Handle inline keyboard button clicks (callback queries).""" |
| 1228 | if not update.callback_query or not update.effective_user: |
| 1229 | return |
| 1230 | query = update.callback_query |
| 1231 | user = update.effective_user |
| 1232 | chat_id = query.message.chat_id if query.message else None |
| 1233 | sender_id = self._sender_id(user) |
| 1234 | if not chat_id: |
| 1235 | logger.warning("Callback query without chat_id") |
| 1236 | return |
| 1237 | button_label = query.data or "" |
| 1238 | await query.answer() |
| 1239 | if query.message: |
| 1240 | try: |
| 1241 | await query.message.edit_reply_markup(reply_markup=None) |
| 1242 | except Exception: |
| 1243 | pass |
| 1244 | logger.debug("Inline button tap from {}: {}", sender_id, button_label) |
| 1245 | self._start_typing(str(chat_id)) |
| 1246 | await self._handle_message( |
| 1247 | sender_id=sender_id, |
| 1248 | chat_id=str(chat_id), |
| 1249 | content=button_label, |
| 1250 | metadata={ |
| 1251 | "callback_query_id": query.id, |
| 1252 | "button_label": button_label, |
| 1253 | "user_id": user.id, |
| 1254 | "username": user.username, |
| 1255 | "first_name": user.first_name, |
| 1256 | "is_callback": True, |
| 1257 | }, |
| 1258 | ) |
| 1259 |