app/bot/handler_utils.py - master - artyuhovs/AIFeedBackTrainingBot

app/bot/handler_utils.py
85 строк3 KB
Codex
Завершить архитектурное оздоровление и исправить итоги
15 июл 2026, 11:05
15 июл 2026, 11:05c65f9a6
from __future__ import annotations

from collections.abc import Callable
from typing import Any, cast

from aiogram.types import CallbackQuery, FSInputFile, InaccessibleMessage, Message

from app.bot.callbacks import parse_entity_callback, parse_framework_callback, parse_pair_callback
from app.utils.text_utils import split_telegram_text, text_stats


def _new_progress_reporter(callback: Callable[..., Any]) -> Any:
    # Resolve through the compatibility facade so existing monkeypatch targets keep working.
    from app.bot import handlers

    return handlers.TelegramProgressReporter(callback)


def _is_video_note_only(response_format: object | None) -> bool:
    return str(response_format or "").strip().casefold() == "video_note"


async def _send_text_section(
    message: Message,
    title: str,
    text: str,
    include_stats: bool = False,
    parse_mode: str | None = None,
    reply_to_message_id: int | None = None,
) -> None:
    header = f"{title}:\n\n" if title else ""
    if include_stats:
        stats = text_stats(text)
        header = f"Слов: {stats.words}, символов: {stats.characters}\n{header}"
    for chunk in split_telegram_text(f"{header}{text}".rstrip()):
        await _answer_reply_to_source(message, chunk, reply_to_message_id=reply_to_message_id, parse_mode=parse_mode)


async def _answer_reply_to_source(
    message: Message, text: str, reply_to_message_id: int | None = None, **kwargs: object
) -> None:
    answer = cast(Any, message.answer)
    await answer(text, **_reply_kwargs(message, reply_to_message_id), **kwargs)


async def _answer_document_reply_to_source(
    message: Message,
    document: FSInputFile,
    reply_to_message_id: int | None = None,
    **kwargs: object,
) -> None:
    answer_document = cast(Any, message.answer_document)
    await answer_document(document, **_reply_kwargs(message, reply_to_message_id), **kwargs)


def _reply_kwargs(message: Message, reply_to_message_id: int | None = None) -> dict[str, object]:
    return {
        "reply_to_message_id": message.message_id if reply_to_message_id is None else reply_to_message_id,
        "allow_sending_without_reply": True,
    }


def accessible_callback_message(callback: CallbackQuery) -> Message | None:
    """Narrow aiogram's callback message union without trusting inaccessible messages."""
    value = callback.message
    if value is None or isinstance(value, InaccessibleMessage):
        return None
    return value


async def answer_long(message: Message, text: str) -> None:
    for chunk in split_telegram_text(text):
        await message.answer(chunk)


def callback_int(data: str | None, prefix: str) -> int | None:
    return parse_entity_callback(data, prefix)


def callback_pair(data: str | None, prefix: str) -> tuple[int, int] | None:
    return parse_pair_callback(data, prefix)


def callback_framework(data: str | None) -> tuple[int, int] | None:
    return parse_framework_callback(data)