""" Drop-in replacement for telegram.Bot - this is the object stashed on context.bot (and returned by Update.get_bot() / Message.get_bot()) throughout the codebase. Every method here mirrors the exact keyword signature the original code calls it with (see MIGRATION_NOTES.md for the audit of every call site), backed entirely by a Telethon TelegramClient. Formatting: rather than hand-constructing raw telethon.tl.types.MessageEntity* objects and passing them via formatting_entities= (an earlier approach that turned out to be too fragile in practice - see MIGRATION_NOTES.md), text is converted from this codebase's MarkdownV2 into HTML (_markdown.py) and sent with parse_mode='html', letting Telethon's own parser/entity-builder do the actual work - the same code path used by every other Telethon-based bot. """ import logging import base64 from io import BytesIO from telethon.tl.types import ( DocumentAttributeAnimated, InputPhoto, InputDocument, ) from ._errors import translate_errors from ._markdown import markdown_v2_to_html from ._keyboard import convert_markup_to_buttons from ._mentions import resolve_mentions from ._types import ( is_media_ref, decode_media_ref, _get_chat_member, coerce_peer, ensure_resolvable_peer, ) from ._message import Message, answer_callback_query as _answer_callback_query import resources.Environment as Env logger = logging.getLogger(__name__) def _reconstruct_media_reference(decoded: dict): file_reference = base64.b64decode(decoded["fr"]) if decoded["k"] == "photo": return InputPhoto(id=decoded["id"], access_hash=decoded["ah"], file_reference=file_reference) return InputDocument(id=decoded["id"], access_hash=decoded["ah"], file_reference=file_reference) class Bot: """Mirrors telegram.Bot.""" def __init__(self, client, bot_user_id: int = None, bot_username: str = None): self._client = client self.id = bot_user_id self.username = bot_username # -- error reporting ---------------------------------------------------- async def report_error(self, text: str) -> None: """ Best-effort, last-resort error reporter: sends `text` (typically a traceback) to Env.ERROR_LOG_CHAT_ID as one or more MarkdownV2 code blocks (```python ... ```), so Telegram renders it with syntax highlighting and a one-tap copy button instead of as plain text. No truncation: text longer than fits in one message is split across as many messages as it takes, on line boundaries, so nothing gets cut off - getting the complete traceback through matters more here than fitting it in a single message. Deliberately self-contained rather than reusing message_service.py's log_error() (which needs a full context/update object and pulls in the whole markdown pipeline): this needs to work from low-level places that don't have either on hand - the generic background-task wrapper in runtime.py, the job-queue runner in _jobqueue.py - not just the top-level per-update handler in manage_message.py that already has both. And it must never itself raise: a failure while reporting an error could otherwise mask the original exception, or crash whatever's already in the middle of handling one. """ chat_id = Env.ERROR_LOG_CHAT_ID.get() if chat_id is None: return # Telegram's message length limit is 4096 characters. Leave headroom # for the ```python\n / \n``` fence (~14 characters) and for # backslash/backtick escaping potentially adding a character each # (see below) - and split on line boundaries so a line is never cut # mid-way through. max_chunk = 3800 chunks = [] current: list[str] = [] current_len = 0 for line in text.split("\n"): # +1 accounts for the newline that will join this line to the next if current and current_len + len(line) + 1 > max_chunk: chunks.append("\n".join(current)) current = [] current_len = 0 current.append(line) current_len += len(line) + 1 if current: chunks.append("\n".join(current)) for chunk in chunks: # Inside a MarkdownV2 pre/code block, only '\' and '`' need # escaping (per Bot API's spec) - escape backslashes first, or # the backslashes just added to escape backticks would # themselves get re-escaped. escaped = chunk.replace("\\", "\\\\").replace("`", "\\`") try: await self.send_message( chat_id=chat_id, text=f"```python\n{escaped}\n```", parse_mode="MarkdownV2", ) except Exception: logger.exception("Failed to report error to ERROR_LOG_CHAT_ID") # -- text preparation ------------------------------------------------- async def _prepare_text(self, text, parse_mode): """ Convert this codebase's MarkdownV2 text into HTML for Telethon's own parser to handle. Returns (text_to_send, telethon_parse_mode). """ if text is None: return None, None if parse_mode == "MarkdownV2": text = await resolve_mentions(self._client, text) return markdown_v2_to_html(text), "html" return text, None def _resolve_media_input(self, media_value, default_extension: str = None): """bytes -> upload fresh; cached "tgcref1:" reference -> reuse without re-uploading; anything else (path/URL) -> pass through as-is.""" if isinstance(media_value, (bytes, bytearray)): bio = BytesIO(media_value) # Telethon decides photo-vs-document (and picks a mime type) from # the file's *name extension*, not its content - see # telethon.utils.is_image(), which literally just checks the # extension. A bare BytesIO has no .name, so without this it # silently uploads as a generic "unnamed" document instead of a # photo/video/animation. bio.name = f"upload{default_extension or ''}" return bio if is_media_ref(media_value): return _reconstruct_media_reference(decode_media_ref(media_value)) return media_value @staticmethod def _reply_target(reply_to_message_id, message_thread_id): return reply_to_message_id if reply_to_message_id is not None else message_thread_id async def _warm_user_peer(self, chat_id): """Call before sending a brand-new message, right after coerce_peer(). Only user ids need this - group/channel ids are negative in this codebase's Bot-API-style numbering, and the bot already resolves those fine via its own group/channel membership, so this deliberately only fires for the "DMing a user by ID" case (e.g. notifications) where PeerIdInvalidError actually shows up.""" if isinstance(chat_id, int) and chat_id > 0: await ensure_resolvable_peer(self._client, chat_id) # -- messages ----------------------------------------------------------- async def send_message( self, chat_id, text, reply_markup=None, disable_web_page_preview=None, parse_mode="MarkdownV2", disable_notification=None, reply_to_message_id=None, allow_sending_without_reply=None, protect_content=None, message_thread_id=None, **_ignored, ) -> Message: chat_id = coerce_peer(chat_id) await self._warm_user_peer(chat_id) body, tl_parse_mode = await self._prepare_text(text, parse_mode) tl_message = await translate_errors( self._client.send_message( chat_id, body, parse_mode=tl_parse_mode, link_preview=(not disable_web_page_preview) if disable_web_page_preview is not None else True, silent=bool(disable_notification), reply_to=self._reply_target(reply_to_message_id, message_thread_id), buttons=convert_markup_to_buttons(reply_markup), ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def edit_message_text( self, text, chat_id=None, message_id=None, reply_markup=None, parse_mode="MarkdownV2", disable_web_page_preview=None, **_ignored, ) -> Message: chat_id = coerce_peer(chat_id) body, tl_parse_mode = await self._prepare_text(text, parse_mode) tl_message = await translate_errors( self._client.edit_message( chat_id, message_id, body, parse_mode=tl_parse_mode, link_preview=(not disable_web_page_preview) if disable_web_page_preview is not None else True, buttons=convert_markup_to_buttons(reply_markup), ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def edit_message_reply_markup(self, chat_id=None, message_id=None, reply_markup=None, **_ignored) -> Message: chat_id = coerce_peer(chat_id) tl_message = await translate_errors( self._client.edit_message(chat_id, message_id, buttons=convert_markup_to_buttons(reply_markup)) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def edit_message_caption( self, chat_id=None, message_id=None, caption=None, parse_mode="MarkdownV2", reply_markup=None, **_ignored ) -> Message: chat_id = coerce_peer(chat_id) body, tl_parse_mode = await self._prepare_text(caption, parse_mode) tl_message = await translate_errors( self._client.edit_message( chat_id, message_id, body, parse_mode=tl_parse_mode, buttons=convert_markup_to_buttons(reply_markup), ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def edit_message_media(self, chat_id=None, message_id=None, media=None, reply_markup=None, **_ignored) -> Message: chat_id = coerce_peer(chat_id) body, tl_parse_mode = await self._prepare_text(media.caption, media.parse_mode) if media else (None, None) ext_by_type = {"photo": ".jpg", "video": ".mp4", "animation": ".mp4"} default_extension = ext_by_type.get(getattr(media, "type", None), ".jpg") file_input = self._resolve_media_input(media.media, default_extension=default_extension) if media else None tl_message = await translate_errors( self._client.edit_message( chat_id, message_id, body, file=file_input, parse_mode=tl_parse_mode, buttons=convert_markup_to_buttons(reply_markup), ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def delete_message(self, chat_id, message_id) -> bool: chat_id = coerce_peer(chat_id) await translate_errors(self._client.delete_messages(chat_id, [message_id])) return True async def copy_message( self, chat_id, from_chat_id, message_id, message_thread_id=None, disable_notification=None, **_ignored ) -> Message: chat_id = coerce_peer(chat_id) await self._warm_user_peer(chat_id) from_chat_id = coerce_peer(from_chat_id) src = await translate_errors(self._client.get_messages(from_chat_id, ids=message_id)) if src is None: from .error import BadRequest raise BadRequest("Message to copy not found") reply_to = message_thread_id # Unlike every other method here, this one legitimately reuses raw # entities: `src.entities` came directly from Telegram itself (on the # message we're copying), not from hand-constructing anything - so # none of the fragility that motivated moving everything else to # HTML applies here. if getattr(src, "media", None): tl_message = await translate_errors( self._client.send_file( chat_id, file=src.media, caption=src.message or None, formatting_entities=list(src.entities) if src.entities else None, parse_mode=None, silent=bool(disable_notification), reply_to=reply_to, ) ) else: tl_message = await translate_errors( self._client.send_message( chat_id, src.message or "", formatting_entities=list(src.entities) if src.entities else None, parse_mode=None, silent=bool(disable_notification), reply_to=reply_to, ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def pin_chat_message(self, chat_id, message_id, disable_notification=None, **_ignored): chat_id = coerce_peer(chat_id) await translate_errors( self._client.pin_message(chat_id, message_id, notify=not disable_notification) ) return True async def unpin_chat_message(self, chat_id, message_id=None, **_ignored): chat_id = coerce_peer(chat_id) await translate_errors(self._client.unpin_message(chat_id, message_id)) return True # -- media -------------------------------------------------------------- async def send_photo( self, chat_id, photo, caption=None, reply_markup=None, parse_mode="MarkdownV2", disable_notification=None, reply_to_message_id=None, allow_sending_without_reply=None, protect_content=None, message_thread_id=None, **_ignored, ) -> Message: chat_id = coerce_peer(chat_id) await self._warm_user_peer(chat_id) body, tl_parse_mode = await self._prepare_text(caption, parse_mode) tl_message = await translate_errors( self._client.send_file( chat_id, file=self._resolve_media_input(photo, default_extension=".jpg"), caption=body, parse_mode=tl_parse_mode, buttons=convert_markup_to_buttons(reply_markup), silent=bool(disable_notification), reply_to=self._reply_target(reply_to_message_id, message_thread_id), force_document=False, ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def send_video( self, chat_id, video, caption=None, reply_markup=None, parse_mode="MarkdownV2", disable_notification=None, reply_to_message_id=None, allow_sending_without_reply=None, protect_content=None, message_thread_id=None, **_ignored, ) -> Message: chat_id = coerce_peer(chat_id) await self._warm_user_peer(chat_id) body, tl_parse_mode = await self._prepare_text(caption, parse_mode) tl_message = await translate_errors( self._client.send_file( chat_id, file=self._resolve_media_input(video, default_extension=".mp4"), caption=body, parse_mode=tl_parse_mode, buttons=convert_markup_to_buttons(reply_markup), silent=bool(disable_notification), reply_to=self._reply_target(reply_to_message_id, message_thread_id), force_document=False, supports_streaming=True, ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) async def send_animation( self, chat_id, animation, caption=None, reply_markup=None, parse_mode="MarkdownV2", disable_notification=None, reply_to_message_id=None, allow_sending_without_reply=None, protect_content=None, message_thread_id=None, **_ignored, ) -> Message: chat_id = coerce_peer(chat_id) await self._warm_user_peer(chat_id) body, tl_parse_mode = await self._prepare_text(caption, parse_mode) is_fresh_upload = isinstance(animation, (bytes, bytearray)) tl_message = await translate_errors( self._client.send_file( chat_id, file=self._resolve_media_input(animation, default_extension=".mp4"), caption=body, parse_mode=tl_parse_mode, buttons=convert_markup_to_buttons(reply_markup), silent=bool(disable_notification), reply_to=self._reply_target(reply_to_message_id, message_thread_id), force_document=False, attributes=[DocumentAttributeAnimated()] if is_fresh_upload else None, ) ) return await Message.from_telethon(tl_message, self._client, self, fetch_reply=False) # -- callback queries ----------------------------------------------------- async def answer_callback_query(self, callback_query_id, text=None, show_alert=None, **_ignored) -> bool: await _answer_callback_query(self._client, callback_query_id, text=text, show_alert=bool(show_alert)) return True # -- chat / members ------------------------------------------------------- async def get_chat_member(self, chat_id, user_id): return await _get_chat_member(self._client, chat_id, int(user_id))