104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Утилиты для форматирования и отправки сообщений."""
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import List
|
||
from telegram import Update
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Лимиты Telegram
|
||
MAX_MESSAGE_LENGTH = 4096 # Максимальная длина сообщения
|
||
|
||
|
||
def escape_markdown(text: str) -> str:
|
||
"""
|
||
Экранирование специальных символов Markdown для Telegram API.
|
||
"""
|
||
text = text.replace('```', '\\`\\`\\`')
|
||
return text
|
||
|
||
|
||
def split_message(text: str, max_length: int = MAX_MESSAGE_LENGTH) -> List[str]:
|
||
"""
|
||
Разбить длинный текст на сообщения <= max_length символов.
|
||
Старается разбивать по границам строк или блоков кода.
|
||
"""
|
||
if len(text) <= max_length:
|
||
return [text]
|
||
|
||
parts = []
|
||
current = ""
|
||
|
||
for line in text.split('\n'):
|
||
# Если добавление строки превысит лимит
|
||
if len(current) + len(line) + 1 > max_length:
|
||
if current:
|
||
parts.append(current)
|
||
# Если строка сама по себе длиннее лимита — режем её
|
||
while len(line) > max_length:
|
||
parts.append(line[:max_length])
|
||
line = line[max_length:]
|
||
current = line
|
||
else:
|
||
current += ('\n' if current else '') + line
|
||
|
||
if current:
|
||
parts.append(current)
|
||
|
||
return parts
|
||
|
||
|
||
async def send_long_message(update: Update, text: str, parse_mode: str = None):
|
||
"""
|
||
Отправить длинный текст, разбив на несколько сообщений.
|
||
Если parse_mode="Markdown" и текст содержит блоки кода — отправляет без разметки.
|
||
"""
|
||
parts = split_message(text)
|
||
|
||
for i, part in enumerate(parts):
|
||
# Добавляем номер части если их несколько
|
||
if len(parts) > 1:
|
||
header = f"({i+1}/{len(parts)}) "
|
||
if len(header) + len(part) <= MAX_MESSAGE_LENGTH:
|
||
part = header + part
|
||
|
||
# Если это не первая часть и был Markdown — убираем parse_mode
|
||
# чтобы не было проблем с разорванной разметкой
|
||
actual_parse_mode = parse_mode if i == 0 else None
|
||
|
||
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)
|
||
|
||
# Небольшая пауза между сообщениями
|
||
await asyncio.sleep(0.1)
|
||
|
||
|
||
def format_long_output(text: str, max_lines: int = 20, head_lines: int = 10, tail_lines: int = 10) -> str:
|
||
"""
|
||
Форматировать длинный вывод: показать первые и последние строки.
|
||
По умолчанию: первые 10 + последние 10 строк = 20 строк максимум.
|
||
"""
|
||
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
|