| 1 | import os |
| 2 | import re |
| 3 | |
| 4 | import aiohttp |
| 5 | from aiogram import Bot |
| 6 | from aiogram.exceptions import TelegramBadRequest |
| 7 | from aiogram.types import ( |
| 8 | FSInputFile, |
| 9 | InlineKeyboardButton, |
| 10 | InlineKeyboardMarkup, |
| 11 | ) |
| 12 | |
| 13 | from helpers.errors import format_error |
| 14 | from helpers.print_style import PrintStyle |
| 15 | |
| 16 | _UNSET = object() # sentinel: "not provided" (lets Bot default apply) |
| 17 | |
| 18 | # Text messages |
| 19 | |
| 20 | MAX_MESSAGE_LENGTH: int = 4096 # Telegram message length limit |
| 21 | TELEGRAM_API_BASE: str = "https://api.telegram.org" |
| 22 | |
| 23 | |
| 24 | async def send_text( |
| 25 | bot: Bot, |
| 26 | chat_id: int, |
| 27 | text: str, |
| 28 | reply_to_message_id: int | None = None, |
| 29 | parse_mode: object = _UNSET, |
| 30 | ) -> int | None: |
| 31 | """Send text message, splitting if too long. Returns last message_id or None on error. |
| 32 | |
| 33 | parse_mode behaviour: |
| 34 | - _UNSET (default): omitted from send_message → Bot's DefaultBotProperties applies. |
| 35 | - None: explicitly no formatting. |
| 36 | - "HTML"/"Markdown"/etc.: that specific mode. |
| 37 | """ |
| 38 | try: |
| 39 | chunks = _split_text(text, MAX_MESSAGE_LENGTH) |
| 40 | last_msg_id = None |
| 41 | pm_kwargs: dict = {} if parse_mode is _UNSET else {"parse_mode": parse_mode} |
| 42 | for chunk in chunks: |
| 43 | try: |
| 44 | msg = await bot.send_message( |
| 45 | chat_id=chat_id, |
| 46 | text=chunk, |
| 47 | reply_to_message_id=reply_to_message_id, |
| 48 | **pm_kwargs, |
| 49 | ) |
| 50 | last_msg_id = msg.message_id |
| 51 | except TelegramBadRequest: |
| 52 | # Retry as plain text, stripping HTML tags |
| 53 | plain = re.sub(r"<[^>]+>", "", chunk) |
| 54 | msg = await bot.send_message( |
| 55 | chat_id=chat_id, |
| 56 | text=plain, |
| 57 | reply_to_message_id=reply_to_message_id, |
| 58 | parse_mode=None, |
| 59 | ) |
| 60 | last_msg_id = msg.message_id |
| 61 | return last_msg_id |
| 62 | except Exception as e: |
| 63 | PrintStyle.error(f"Telegram send_text failed: {format_error(e)}") |
| 64 | return None |
| 65 | |
| 66 | # Files and images |
| 67 | |
| 68 | async def send_file( |
| 69 | bot: Bot, |
| 70 | chat_id: int, |
| 71 | file_path: str, |
| 72 | caption: str = "", |
| 73 | reply_to_message_id: int | None = None, |
| 74 | ) -> int | None: |
| 75 | """Send a file from local path. Returns message_id or None on error.""" |
| 76 | try: |
| 77 | if not os.path.isfile(file_path): |
| 78 | PrintStyle.error(f"Telegram: file not found: {file_path}") |
| 79 | return None |
| 80 | input_file = FSInputFile(file_path) |
| 81 | msg = await bot.send_document( |
| 82 | chat_id=chat_id, |
| 83 | document=input_file, |
| 84 | caption=caption[:1024] if caption else None, |
| 85 | reply_to_message_id=reply_to_message_id, |
| 86 | ) |
| 87 | return msg.message_id |
| 88 | except Exception as e: |
| 89 | PrintStyle.error(f"Telegram send_file failed: {format_error(e)}") |
| 90 | return None |
| 91 | |
| 92 | |
| 93 | async def send_photo( |
| 94 | bot: Bot, |
| 95 | chat_id: int, |
| 96 | photo_path: str, |
| 97 | caption: str = "", |
| 98 | reply_to_message_id: int | None = None, |
| 99 | ) -> int | None: |
| 100 | """Send a photo from local path. Returns message_id or None on error.""" |
| 101 | try: |
| 102 | if not os.path.isfile(photo_path): |
| 103 | PrintStyle.error(f"Telegram: photo not found: {photo_path}") |
| 104 | return None |
| 105 | input_file = FSInputFile(photo_path) |
| 106 | msg = await bot.send_photo( |
| 107 | chat_id=chat_id, |
| 108 | photo=input_file, |
| 109 | caption=caption[:1024] if caption else None, |
| 110 | reply_to_message_id=reply_to_message_id, |
| 111 | ) |
| 112 | return msg.message_id |
| 113 | except Exception as e: |
| 114 | PrintStyle.error(f"Telegram send_photo failed: {format_error(e)}") |
| 115 | return None |
| 116 | |
| 117 | |
| 118 | async def send_voice( |
| 119 | bot: Bot, |
| 120 | chat_id: int, |
| 121 | voice_path: str, |
| 122 | caption: str = "", |
| 123 | reply_to_message_id: int | None = None, |
| 124 | ) -> int | None: |
| 125 | """Send a voice message from local path. Returns message_id or None on error.""" |
| 126 | try: |
| 127 | if not os.path.isfile(voice_path): |
| 128 | PrintStyle.error(f"Telegram: voice file not found: {voice_path}") |
| 129 | return None |
| 130 | input_file = FSInputFile(voice_path) |
| 131 | msg = await bot.send_voice( |
| 132 | chat_id=chat_id, |
| 133 | voice=input_file, |
| 134 | caption=caption[:1024] if caption else None, |
| 135 | reply_to_message_id=reply_to_message_id, |
| 136 | ) |
| 137 | return msg.message_id |
| 138 | except Exception as e: |
| 139 | PrintStyle.error(f"Telegram send_voice failed: {format_error(e)}") |
| 140 | return None |
| 141 | |
| 142 | |
| 143 | async def send_audio( |
| 144 | bot: Bot, |
| 145 | chat_id: int, |
| 146 | audio_path: str, |
| 147 | caption: str = "", |
| 148 | reply_to_message_id: int | None = None, |
| 149 | ) -> int | None: |
| 150 | """Send an audio message from local path. Returns message_id or None on error.""" |
| 151 | try: |
| 152 | if not os.path.isfile(audio_path): |
| 153 | PrintStyle.error(f"Telegram: audio file not found: {audio_path}") |
| 154 | return None |
| 155 | input_file = FSInputFile(audio_path) |
| 156 | msg = await bot.send_audio( |
| 157 | chat_id=chat_id, |
| 158 | audio=input_file, |
| 159 | caption=caption[:1024] if caption else None, |
| 160 | reply_to_message_id=reply_to_message_id, |
| 161 | ) |
| 162 | return msg.message_id |
| 163 | except Exception as e: |
| 164 | PrintStyle.error(f"Telegram send_audio failed: {format_error(e)}") |
| 165 | return None |
| 166 | |
| 167 | |
| 168 | async def send_video( |
| 169 | bot: Bot, |
| 170 | chat_id: int, |
| 171 | video_path: str, |
| 172 | caption: str = "", |
| 173 | reply_to_message_id: int | None = None, |
| 174 | ) -> int | None: |
| 175 | """Send a video message from local path. Returns message_id or None on error.""" |
| 176 | try: |
| 177 | if not os.path.isfile(video_path): |
| 178 | PrintStyle.error(f"Telegram: video file not found: {video_path}") |
| 179 | return None |
| 180 | input_file = FSInputFile(video_path) |
| 181 | msg = await bot.send_video( |
| 182 | chat_id=chat_id, |
| 183 | video=input_file, |
| 184 | caption=caption[:1024] if caption else None, |
| 185 | reply_to_message_id=reply_to_message_id, |
| 186 | ) |
| 187 | return msg.message_id |
| 188 | except Exception as e: |
| 189 | PrintStyle.error(f"Telegram send_video failed: {format_error(e)}") |
| 190 | return None |
| 191 | |
| 192 | |
| 193 | # Inline keyboards |
| 194 | |
| 195 | def build_inline_keyboard( |
| 196 | buttons: list[list[dict]], |
| 197 | ) -> InlineKeyboardMarkup: |
| 198 | """Build inline keyboard from a list of rows. |
| 199 | Each row is a list of dicts with keys: text, callback_data or url. |
| 200 | """ |
| 201 | rows = [] |
| 202 | for row in buttons: |
| 203 | row_buttons = [] |
| 204 | for btn in row: |
| 205 | if "url" in btn: |
| 206 | row_buttons.append(InlineKeyboardButton( |
| 207 | text=btn["text"], url=btn["url"], |
| 208 | )) |
| 209 | else: |
| 210 | row_buttons.append(InlineKeyboardButton( |
| 211 | text=btn["text"], |
| 212 | callback_data=btn.get("callback_data", btn["text"]), |
| 213 | )) |
| 214 | rows.append(row_buttons) |
| 215 | return InlineKeyboardMarkup(inline_keyboard=rows) |
| 216 | |
| 217 | |
| 218 | async def send_text_with_keyboard( |
| 219 | bot: Bot, |
| 220 | chat_id: int, |
| 221 | text: str, |
| 222 | buttons: list[list[dict]], |
| 223 | reply_to_message_id: int | None = None, |
| 224 | parse_mode: object = _UNSET, |
| 225 | ) -> int | None: |
| 226 | """Send text with inline keyboard buttons.""" |
| 227 | try: |
| 228 | keyboard = build_inline_keyboard(buttons) |
| 229 | pm_kwargs: dict = {} if parse_mode is _UNSET else {"parse_mode": parse_mode} |
| 230 | msg = await bot.send_message( |
| 231 | chat_id=chat_id, |
| 232 | text=text, |
| 233 | reply_markup=keyboard, |
| 234 | reply_to_message_id=reply_to_message_id, |
| 235 | **pm_kwargs, |
| 236 | ) |
| 237 | return msg.message_id |
| 238 | except Exception as e: |
| 239 | PrintStyle.error(f"Telegram send_text_with_keyboard failed: {format_error(e)}") |
| 240 | return None |
| 241 | |
| 242 | # Typing indicator |
| 243 | |
| 244 | async def send_typing(bot: Bot, chat_id: int): |
| 245 | """Send 'typing...' action to chat.""" |
| 246 | try: |
| 247 | await bot.send_chat_action(chat_id=chat_id, action="typing") |
| 248 | except Exception: |
| 249 | pass |
| 250 | |
| 251 | |
| 252 | async def raw_send_text( |
| 253 | token: str, |
| 254 | chat_id: int, |
| 255 | text: str, |
| 256 | reply_to_message_id: int | None = None, |
| 257 | parse_mode: str | None = "HTML", |
| 258 | reply_markup: dict | None = None, |
| 259 | ) -> int | None: |
| 260 | payload: dict[str, object] = { |
| 261 | "chat_id": chat_id, |
| 262 | "text": text[:MAX_MESSAGE_LENGTH], |
| 263 | } |
| 264 | if parse_mode: |
| 265 | payload["parse_mode"] = parse_mode |
| 266 | if reply_to_message_id: |
| 267 | payload["reply_parameters"] = {"message_id": int(reply_to_message_id)} |
| 268 | if reply_markup: |
| 269 | payload["reply_markup"] = reply_markup |
| 270 | data = await _raw_post(token, "sendMessage", payload) |
| 271 | result = data.get("result") if isinstance(data, dict) else None |
| 272 | if isinstance(result, dict): |
| 273 | return result.get("message_id") |
| 274 | return None |
| 275 | |
| 276 | |
| 277 | async def raw_edit_text( |
| 278 | token: str, |
| 279 | chat_id: int, |
| 280 | message_id: int, |
| 281 | text: str, |
| 282 | parse_mode: str | None = "HTML", |
| 283 | reply_markup: dict | None = None, |
| 284 | ) -> bool: |
| 285 | payload: dict[str, object] = { |
| 286 | "chat_id": chat_id, |
| 287 | "message_id": message_id, |
| 288 | "text": text[:MAX_MESSAGE_LENGTH], |
| 289 | } |
| 290 | if parse_mode: |
| 291 | payload["parse_mode"] = parse_mode |
| 292 | if reply_markup: |
| 293 | payload["reply_markup"] = reply_markup |
| 294 | data = await _raw_post(token, "editMessageText", payload) |
| 295 | if not isinstance(data, dict): |
| 296 | return False |
| 297 | if data.get("ok"): |
| 298 | return True |
| 299 | description = str(data.get("description") or "").lower() |
| 300 | return "message is not modified" in description |
| 301 | |
| 302 | |
| 303 | async def raw_edit_reply_markup( |
| 304 | token: str, |
| 305 | chat_id: int, |
| 306 | message_id: int, |
| 307 | reply_markup: dict | None = None, |
| 308 | ) -> bool: |
| 309 | payload: dict[str, object] = { |
| 310 | "chat_id": chat_id, |
| 311 | "message_id": message_id, |
| 312 | } |
| 313 | if reply_markup: |
| 314 | payload["reply_markup"] = reply_markup |
| 315 | data = await _raw_post(token, "editMessageReplyMarkup", payload) |
| 316 | return bool(isinstance(data, dict) and data.get("ok")) |
| 317 | |
| 318 | |
| 319 | async def _raw_post(token: str, method: str, payload: dict[str, object]) -> dict: |
| 320 | url = f"{TELEGRAM_API_BASE}/bot{token}/{method}" |
| 321 | try: |
| 322 | timeout = aiohttp.ClientTimeout(total=10) |
| 323 | async with aiohttp.ClientSession(timeout=timeout) as session: |
| 324 | async with session.post(url, json=payload) as response: |
| 325 | data = await response.json(content_type=None) |
| 326 | if response.status != 200 or not data.get("ok"): |
| 327 | PrintStyle.debug(f"Telegram {method} failed: {data}") |
| 328 | return data if isinstance(data, dict) else {} |
| 329 | except Exception as e: |
| 330 | PrintStyle.debug(f"Telegram {method} failed: {format_error(e)}") |
| 331 | return {} |
| 332 | |
| 333 | # File download |
| 334 | |
| 335 | async def download_file( |
| 336 | bot: Bot, |
| 337 | file_id: str, |
| 338 | destination: str, |
| 339 | ) -> str | None: |
| 340 | """Download a file by file_id to destination path. Returns path or None on error.""" |
| 341 | try: |
| 342 | file = await bot.get_file(file_id) |
| 343 | if not file.file_path: |
| 344 | return None |
| 345 | os.makedirs(os.path.dirname(destination), exist_ok=True) |
| 346 | await bot.download_file(file.file_path, destination) |
| 347 | return destination |
| 348 | except Exception as e: |
| 349 | PrintStyle.error(f"Telegram download failed: {format_error(e)}") |
| 350 | return None |
| 351 | |
| 352 | # Helpers |
| 353 | |
| 354 | def _split_text(text: str, max_len: int) -> list[str]: |
| 355 | if len(text) <= max_len: |
| 356 | return [text] |
| 357 | chunks = [] |
| 358 | while text: |
| 359 | if len(text) <= max_len: |
| 360 | chunks.append(text) |
| 361 | break |
| 362 | # Try to split at newline |
| 363 | split_at = text.rfind("\n", 0, max_len) |
| 364 | if split_at == -1 or split_at < max_len // 2: |
| 365 | split_at = max_len |
| 366 | chunks.append(text[:split_at]) |
| 367 | text = text[split_at:].lstrip("\n") |
| 368 | return chunks |
| 369 | |
| 370 | |
| 371 | _IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"} |
| 372 | _VOICE_EXTENSIONS = {".ogg", ".oga", ".opus"} |
| 373 | _AUDIO_EXTENSIONS = {".mp3", ".m4a", ".aac", ".wav", ".flac"} |
| 374 | _VIDEO_EXTENSIONS = {".mp4", ".m4v", ".mov", ".webm"} |
| 375 | |
| 376 | |
| 377 | def is_image_file(path: str) -> bool: |
| 378 | _, ext = os.path.splitext(path.lower()) |
| 379 | return ext in _IMAGE_EXTENSIONS |
| 380 | |
| 381 | |
| 382 | def is_voice_file(path: str) -> bool: |
| 383 | _, ext = os.path.splitext(path.lower()) |
| 384 | return ext in _VOICE_EXTENSIONS |
| 385 | |
| 386 | |
| 387 | def is_audio_file(path: str) -> bool: |
| 388 | _, ext = os.path.splitext(path.lower()) |
| 389 | return ext in _AUDIO_EXTENSIONS |
| 390 | |
| 391 | |
| 392 | def is_video_file(path: str) -> bool: |
| 393 | _, ext = os.path.splitext(path.lower()) |
| 394 | return ext in _VIDEO_EXTENSIONS |
| 395 | |
| 396 | |
| 397 | def md_to_telegram_html(text: str) -> str: |
| 398 | """Convert Markdown to Telegram-compatible HTML.""" |
| 399 | stash: list[str] = [] |
| 400 | |
| 401 | def _put(html: str) -> str: |
| 402 | stash.append(html) |
| 403 | return f"\x00B{len(stash) - 1}\x00" |
| 404 | |
| 405 | def _esc(t: str) -> str: |
| 406 | return t.replace("&", "&").replace("<", "<").replace(">", ">") |
| 407 | |
| 408 | # Stash code blocks & inline code |
| 409 | def _code_block(m): |
| 410 | lang, body = m.group(1), m.group(2).rstrip("\n") |
| 411 | if lang: |
| 412 | return _put(f'<pre><code class="language-{lang}">{_esc(body)}</code></pre>') |
| 413 | return _put(f"<pre>{_esc(body)}</pre>") |
| 414 | |
| 415 | text = re.sub(r"(?:```|~~~)(\w*)\n?(.*?)(?:```|~~~)", _code_block, text, flags=re.DOTALL) |
| 416 | text = re.sub(r"`([^`]+)`", lambda m: _put(f"<code>{_esc(m.group(1))}</code>"), text) |
| 417 | |
| 418 | # Strip unsupported syntax |
| 419 | text = _strip_tables(text) |
| 420 | text = re.sub(r"^[ \t]*[-*_=]{3,}[ \t]*$", "", text, flags=re.MULTILINE) |
| 421 | text = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"[\1](\2)", text) |
| 422 | |
| 423 | # Stash links |
| 424 | text = re.sub( |
| 425 | r"\[([^\]]+)\]\(([^)]+)\)", |
| 426 | lambda m: _put( |
| 427 | f'<a href="{m.group(2).replace("&", "&").replace(chr(34), """)}">{_esc(m.group(1))}</a>' |
| 428 | ), |
| 429 | text, |
| 430 | ) |
| 431 | |
| 432 | # Escape HTML & apply inline formatting |
| 433 | text = _esc(text) |
| 434 | text = re.sub(r"\*\*\*(.+?)\*\*\*", r"<b><i>\1</i></b>", text) |
| 435 | text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text) |
| 436 | text = re.sub(r"__(.+?)__", r"<b>\1</b>", text) |
| 437 | text = re.sub(r"(?<!\w)\*([^*]+?)\*(?!\w)", r"<i>\1</i>", text) |
| 438 | text = re.sub(r"(?<!\w)_([^_]+?)_(?!\w)", r"<i>\1</i>", text) |
| 439 | text = re.sub(r"~~(.+?)~~", r"<s>\1</s>", text) |
| 440 | text = re.sub(r"^#{1,6}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE) |
| 441 | |
| 442 | # Block-level formatting |
| 443 | text = _convert_blockquotes(text) |
| 444 | text = _convert_lists(text) |
| 445 | |
| 446 | # Restore stash |
| 447 | for i, block in enumerate(stash): |
| 448 | text = text.replace(f"\x00B{i}\x00", block) |
| 449 | return text |
| 450 | |
| 451 | |
| 452 | |
| 453 | _TABLE_RE = re.compile(r"^\|(.+)\|$") |
| 454 | _TABLE_SEP = re.compile(r"^[\s|:-]+$") |
| 455 | |
| 456 | |
| 457 | def _strip_tables(text: str) -> str: |
| 458 | """Strip Markdown table pipe syntax, keeping cell content as plain text.""" |
| 459 | out: list[str] = [] |
| 460 | for line in text.split("\n"): |
| 461 | stripped = line.strip() |
| 462 | if _TABLE_SEP.match(stripped): |
| 463 | continue |
| 464 | m = _TABLE_RE.match(stripped) |
| 465 | if m: |
| 466 | out.append(" ".join(c.strip() for c in m.group(1).split("|"))) |
| 467 | else: |
| 468 | out.append(line) |
| 469 | return "\n".join(out) |
| 470 | |
| 471 | |
| 472 | _LIST_RE = re.compile(r"^( *)([-*+]|\d+\.)\s+(.*)$") |
| 473 | _BULLETS = ("\u2022", "\u25e6", "\u25aa") |
| 474 | |
| 475 | |
| 476 | def _convert_lists(text: str) -> str: |
| 477 | """Convert Markdown list markers to Unicode bullets.""" |
| 478 | out: list[str] = [] |
| 479 | for line in text.split("\n"): |
| 480 | m = _LIST_RE.match(line) |
| 481 | if m: |
| 482 | depth = len(m.group(1)) // 2 |
| 483 | marker, content = m.group(2), m.group(3) |
| 484 | px = " " * depth |
| 485 | if marker.rstrip(".").isdigit(): |
| 486 | out.append(f"{px}{marker} {content}") |
| 487 | else: |
| 488 | out.append(f"{px}{_BULLETS[min(depth, len(_BULLETS) - 1)]} {content}") |
| 489 | else: |
| 490 | out.append(line) |
| 491 | return "\n".join(out) |
| 492 | |
| 493 | |
| 494 | def _convert_blockquotes(text: str) -> str: |
| 495 | """Convert Markdown blockquotes to Telegram <blockquote> tags.""" |
| 496 | out: list[str] = [] |
| 497 | buf: list[str] = [] |
| 498 | |
| 499 | def _flush(): |
| 500 | if buf: |
| 501 | out.append("<blockquote>" + "\n".join(buf) + "</blockquote>") |
| 502 | buf.clear() |
| 503 | |
| 504 | for line in text.split("\n"): |
| 505 | m = re.match(r"^>\s?(.*)", line) |
| 506 | if m: |
| 507 | buf.append(m.group(1)) |
| 508 | else: |
| 509 | _flush() |
| 510 | out.append(line) |
| 511 | _flush() |
| 512 | return "\n".join(out) |