270 lines
12 KiB
Python
270 lines
12 KiB
Python
#!/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 # Состояние: внутри блока кода или нет
|
||
|
||
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
|
||
else:
|
||
current = test_line
|
||
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
|
||
else:
|
||
current = test_line
|
||
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
|
||
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):
|
||
"""
|
||
Отправить длинный текст, разбив на несколько сообщений.
|
||
Использует polling для ожидания нажатия кнопки (не блокирует event loop).
|
||
"""
|
||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
|
||
|
||
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 для этого сообщения
|
||
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 and 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
|
||
)
|
||
|
||
# Ждём через polling (короткие паузы дают event loop обработать callback)
|
||
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
|
||
state.continue_output = None # None = ещё не решил
|
||
|
||
logger.info(f"send_long_message: ждём нажатия кнопки (user_id={user_id}, remaining={remaining})")
|
||
|
||
# Polling с короткими паузами (даём event loop обработать callback)
|
||
for _ in range(600): # Максимум 600 * 0.5 = 300 секунд = 5 минут
|
||
await asyncio.sleep(0.5)
|
||
state = state_manager.get(user_id)
|
||
if state.continue_output is not None:
|
||
# Пользователь нажал кнопку
|
||
break
|
||
|
||
logger.info(f"send_long_message: кнопка нажата, continue_output={state.continue_output}")
|
||
|
||
# Проверяем решение пользователя
|
||
if not state.continue_output:
|
||
# Отменил
|
||
try:
|
||
await wait_msg.delete()
|
||
except:
|
||
pass
|
||
state.waiting_for_output_control = False
|
||
state.output_remaining = None
|
||
state.output_wait_message = None
|
||
return
|
||
|
||
# Продолжаем
|
||
try:
|
||
await wait_msg.delete()
|
||
except:
|
||
pass
|
||
|
||
state.waiting_for_output_control = False
|
||
state.output_remaining = None
|
||
state.output_wait_message = None
|
||
|
||
|
||
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
|