#!/usr/bin/env python3 """Утилиты для форматирования и отправки сообщений.""" import asyncio import logging import re from typing import List, Tuple from telegram import Update logger = logging.getLogger(__name__) # Лимиты Telegram MAX_MESSAGE_LENGTH = 4096 # Максимальная длина сообщения RESERVED_FOR_HEADER = 20 # Резервируем место для "(N/N) " def escape_markdown(text: str) -> str: """ Экранирование специальных символов Markdown для Telegram API. Telegram Markdown v1 использует: * _ ` [ ] ( ) Эти символы нужно экранировать обратным слэшем. """ if not text: return text # Экранируем специальные символы Markdown # Порядок важен: сначала экранируем обратные слэши text = text.replace('\\', '\\\\') text = text.replace('`', '\\`') text = text.replace('*', '\\*') text = text.replace('_', '\\_') text = text.replace('[', '\\[') text = text.replace(']', '\\]') text = text.replace('(', '\\(') text = text.replace(')', '\\)') return text def find_code_blocks(text: str) -> List[Tuple[int, int]]: """ Найти все блоки кода (```) в тексте. Возвращает список кортежей (start, end) для каждого блока. """ blocks = [] pattern = re.compile(r'```') matches = list(pattern.finditer(text)) # Пары start-end для каждого блока i = 0 while i < len(matches) - 1: start = matches[i].start() end = matches[i + 1].end() blocks.append((start, end)) i += 2 return blocks def split_message(text: str, max_length: int = MAX_MESSAGE_LENGTH) -> List[Tuple[str, bool, bool, bool]]: """ Умно разбить длинный текст на сообщения <= max_length символов. Возвращает список кортежей (text, has_code, code_opened, code_closed): - text: текст сообщения - has_code: True если сообщение содержит часть блока кода - code_opened: True если блок кода ОТКРЫТ в этом сообщении (есть открывающий ```) - code_closed: True если блок кода ЗАКРЫТ в этом сообщении (есть закрывающий ```) Алгоритм: 1. Разбиваем текст на строки 2. Накапливаем строки до достижения лимита 3. Отслеживаем состояние блока кода (внутри/снаружи) """ def calc_code_flags(txt: str) -> Tuple[bool, bool, bool]: """Вычислить флаги code для данного текста.""" has_code = '```' in txt backtick_count = txt.count('```') code_opened = backtick_count >= 1 code_closed = backtick_count >= 2 or (backtick_count == 1 and txt.rstrip().endswith('```')) return has_code, code_opened, code_closed if len(text) <= max_length: has_code, code_opened, code_closed = calc_code_flags(text) return [(text, has_code, code_opened, code_closed)] parts = [] lines = text.split('\n') current = "" in_code_block = False # Состояние: внутри блока кода или нет code_block_started_in_current = False # Блок кода был открыт в текущей части for line in lines: # Проверяем, содержит ли строка ``` backticks_in_line = line.count('```') # Если строка содержит нечётное количество ```, она меняет состояние if backticks_in_line % 2 == 1: # Эта строка содержит ``` который меняет состояние if in_code_block: # Были внутри блока — эта строка закрывает его test_line = current + ('\n' if current else '') + line if len(test_line) > max_length - RESERVED_FOR_HEADER: # Сначала отправляем текущее (блок не закрыт в этой части!) if current: has_code, code_opened, _ = calc_code_flags(current) # code_closed=False потому что блок продолжится parts.append((current, has_code, code_opened, False)) current = line code_block_started_in_current = False else: current = test_line code_block_started_in_current = True in_code_block = False else: # Были снаружи — эта строка открывает блок test_line = current + ('\n' if current else '') + line if len(test_line) > max_length - RESERVED_FOR_HEADER: if current: has_code, code_opened, code_closed = calc_code_flags(current) parts.append((current, has_code, code_opened, code_closed)) current = line code_block_started_in_current = True else: current = test_line code_block_started_in_current = True in_code_block = True else: # Строка не меняет состояние test_line = current + ('\n' if current else '') + line if len(test_line) > max_length - RESERVED_FOR_HEADER: if current: has_code, code_opened, _ = calc_code_flags(current) # Если мы внутри блока кода — block не закрыт code_closed = not in_code_block parts.append((current, has_code, code_opened, code_closed)) current = line code_block_started_in_current = in_code_block else: current = test_line if current: has_code, code_opened, _ = calc_code_flags(current) code_closed = not in_code_block parts.append((current, has_code, code_opened, code_closed)) return parts async def send_long_message(update: Update, text: str, parse_mode: str = None, pause_every: int = 3): """ Отправить длинный текст, разбив на несколько сообщений. Умная разбивка: - Блоки кода не разрываются между сообщениями - Если блок кода не влезает — отправляется без Markdown - Нумерация (1/N), (2/N) только если сообщений > 1 - КАЖДЫЕ pause_every сообщений — пауза с кнопками "Продолжить" / "Отменить" - Ждём нажатия кнопки бесконечно, не продолжаем автоматически Args: update: Telegram update text: Текст для отправки parse_mode: Режим парсинга (Markdown) pause_every: Каждые сколько сообщений делать паузу (0 = без паузы) """ from telegram import InlineKeyboardButton, InlineKeyboardMarkup import asyncio parts = split_message(text) total = len(parts) messages_sent = 0 wait_msg = None for i, (part, has_code, code_opened, code_closed) in enumerate(parts): # Определяем parse_mode для этого сообщения # Используем parse_mode если сообщение имеет код ИЛИ если мы внутри блока кода # Мы внутри блока кода если: предыдущее не закрыло ИЛИ (текущее открыто и не закрыто) prev_closed = parts[i-1][3] if i > 0 else True in_code_block = not prev_closed or (code_opened and not code_closed) actual_parse_mode = parse_mode if parse_mode and (has_code or in_code_block) else None # Логика работы с блоками кода между сообщениями: # Если предыдущее сообщение не закрыло блок — нужно открыть в этом # Если текущее сообщение не закрыло блок — нужно закрыть в следующем if total > 1 and actual_parse_mode: # Проверяем нужно ли открыть блок в начале этого сообщения if i > 0 and not parts[i-1][3]: # предыдущее не закрыло блок part = "```\n" + part # Проверяем нужно ли закрыть блок в конце этого сообщения if i < total - 1 and not code_closed: # Следующее сообщение не начинается с ``` — закрываем блок if not parts[i+1][2]: # следующее не открывает блок part = part + "\n```" # Добавляем номер части ПОСЛЕ работы с блоками кода if total > 1: header = f"({i+1}/{total}) " if len(header) + len(part) <= MAX_MESSAGE_LENGTH: part = header + part try: await update.message.reply_text(part, parse_mode=actual_parse_mode) except Exception as e: # Фоллбэк: отправляем без разметки logger.debug(f"Ошибка Markdown, отправляем без разметки: {e}") await update.message.reply_text(part) messages_sent += 1 # Небольшая пауза между сообщениями await asyncio.sleep(0.1) # КАЖДЫЕ pause_every сообщений — спрашиваем продолжать ли if pause_every > 0 and messages_sent % pause_every == 0 and i < total - 1: remaining = total - (i + 1) keyboard = InlineKeyboardMarkup([ [ InlineKeyboardButton("▶️ Продолжить", callback_data=f"continue_output_{remaining}"), InlineKeyboardButton("❌ Отменить", callback_data="cancel_output") ] ]) wait_msg = await update.message.reply_text( f"📊 **Отправлено {messages_sent} из {total} сообщений**\n\n" f"Осталось ещё {remaining} сообщений.\n\n" f"Продолжить вывод?", parse_mode="Markdown", reply_markup=keyboard ) # Ждём ответа пользователя БЕСКОНЕЧНО from bot.config import state_manager user_id = update.effective_user.id # Сохраняем состояние ожидания state = state_manager.get(user_id) state.waiting_for_output_control = True state.output_remaining = remaining state.output_wait_message = wait_msg # Ждём пока пользователь не нажмёт кнопку while state.waiting_for_output_control: await asyncio.sleep(1) state = state_manager.get(user_id) # Пользователь ответил if state.continue_output: # Продолжаем - удаляем кнопки try: await wait_msg.delete() except: pass # Очищаем состояние state.output_remaining = None state.output_wait_message = None else: # Отменил - редактируем сообщение и убираем кнопки try: await wait_msg.edit_text("❌ **Вывод отменен пользователем**", parse_mode="Markdown") except: pass return # Прерываем вывод def format_long_output(text: str, max_lines: int = 100, head_lines: int = 50, tail_lines: int = 50) -> str: """ Форматировать длинный вывод: показать первые и последние строки. По умолчанию: первые 50 + последние 50 строк = 100 строк максимум. """ lines = text.split('\n') total_lines = len(lines) if total_lines <= max_lines: return text # Показываем первые head_lines и последние tail_lines head = lines[:head_lines] tail = lines[-tail_lines:] skipped = total_lines - head_lines - tail_lines result = '\n'.join(head) result += f'\n\n... ({skipped} строк пропущено) ...\n' result += '\n'.join(tail) return result