import asyncio from typing import Any, Awaitable, Callable import telebot from telebot.async_telebot import AsyncTeleBot from . import markups Message = telebot.types.Message Handler = Callable[[Message], Awaitable[Any]] Markups = ( telebot.types.ReplyKeyboardMarkup | telebot.types.InlineKeyboardMarkup | telebot.types.ReplyKeyboardRemove ) states: dict[tuple[int, int], Handler] = {} def register_state( message: Message, handler: Callable[[Message], Any], ) -> None: states[(message.chat.id, message.from_user.id)] = handler def reset_state(message: Message) -> None: try: del states[(message.chat.id, message.from_user.id)] except KeyError: pass def get_state(message: Message) -> Handler | None: return states.get((message.chat.id, message.from_user.id)) async def delete_message( bot: AsyncTeleBot, mes: Message, *, sleep_time: int = 0, ) -> bool: try: if sleep_time != 0: await asyncio.sleep(sleep_time) await bot.delete_message(mes.chat.id, mes.id) except telebot.apihelper.ApiException: return False else: return True async def send_tmp_message( bot: AsyncTeleBot, chat_id: telebot.types.Message, text: str, *, sleep_time: int = 5, markup: Markups | None = None, ) -> None: bot_mes = await bot.send_message( chat_id, text, parse_mode="MarkdownV2", reply_markup=markup ) await delete_message(bot, bot_mes, sleep_time=sleep_time) async def base_handler( bot: AsyncTeleBot, mes: Message, prev_mes: Message | None = None ) -> None: reset_state(mes) await delete_message(bot, mes) if prev_mes is not None: await delete_message(bot, prev_mes) async def send_deleteable_message( bot: AsyncTeleBot, chat_id: int, text: str, ) -> None: """Sends a message with a delete button""" markup = markups.deletion_markup() await bot.send_message( chat_id, text, parse_mode="MarkdownV2", reply_markup=markup, ) def escape(text: str) -> str: escaped_chars = "*_~|`[(" for char in escaped_chars: text = text.replace(char, rf"\\{char}") return text