feat(global): init structure

This commit is contained in:
h
2025-07-01 12:55:01 +03:00
commit 1b21f23294
37 changed files with 1208 additions and 0 deletions

View File

View File

@@ -0,0 +1,3 @@
from .handler import on_error
__all__ = ["on_error"]

View File

@@ -0,0 +1,41 @@
from aiogram import Bot
from aiogram.dispatcher import router as s_router
from aiogram.types.error_event import ErrorEvent
from rich.traceback import Traceback
from utils.logging import console
async def on_error(event: ErrorEvent, bot: Bot):
import base64
import os
error_id = base64.urlsafe_b64encode(os.urandom(6)).decode()
traceback = Traceback.from_exception(
type(event.exception),
event.exception,
event.exception.__traceback__,
show_locals=True,
suppress=[s_router],
)
if event.update.chosen_inline_result:
await bot.edit_message_caption(
inline_message_id=event.update.chosen_inline_result.inline_message_id,
caption=f"💔 <b>ERROR</b> occurred. Use this code to search in logs: "
f"<code>{error_id}</code>",
parse_mode="HTML",
)
if event.update.message:
await event.update.message.answer(
text=f"💔 <b>ERROR</b> occurred. Use this code to search in logs: "
f"<code>{error_id}</code>",
parse_mode="HTML",
)
console.print(f"[red]{error_id} occurred[/]")
console.print(event)
console.print(traceback)
console.print(f"-{error_id} occurred-")

View File

@@ -0,0 +1 @@
from .paginator import Paginator

View File

@@ -0,0 +1,178 @@
from itertools import islice
from typing import Any, Callable, Coroutine, Iterable, Iterator
from aiogram import Dispatcher, F, Router, types
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery
from aiogram.utils.keyboard import InlineKeyboardBuilder
class Paginator:
def __init__(
self,
data: (
types.InlineKeyboardMarkup
| Iterable[types.InlineKeyboardButton]
| Iterable[Iterable[types.InlineKeyboardButton]]
| InlineKeyboardBuilder
),
before_data: list[list[types.InlineKeyboardButton]] = None,
after_data: list[list[types.InlineKeyboardButton]] = None,
state: FSMContext = None,
callback_startswith: str = "page_",
size: int = 8,
page_separator: str = "/",
dp: Dispatcher | Router | None = None,
):
"""
Example: paginator = Paginator(data=kb, size=5)
:param data: An iterable object that stores an InlineKeyboardButton.
:param callback_startswith: What should callback_data begin with in handler pagination. Default = 'page_'.
:param size: Number of lines per page. Default = 8.
:param state: Current state.
:param page_separator: Separator for page numbers. Default = '/'.
"""
self.dp = dp
self.page_separator = page_separator
self._state = state
self._size = size
self._startswith = callback_startswith
self._before_data = before_data or []
self._after_data = after_data or []
if isinstance(data, types.InlineKeyboardMarkup):
self._list_kb = list(self._chunk(it=data.inline_keyboard, size=self._size))
elif isinstance(data, Iterable):
self._list_kb = list(self._chunk(it=list(data), size=self._size))
elif isinstance(data, InlineKeyboardBuilder):
self._list_kb = list(self._chunk(it=data.export(), size=self._size))
else:
raise ValueError(f"{data} is not valid data")
"""
Class for pagination's in aiogram inline keyboards
"""
def __call__(self, current_page=0, *args, **kwargs) -> types.InlineKeyboardMarkup:
"""
Example:
await message.answer(
text='Some menu',
reply_markup=paginator()
)
:return: InlineKeyboardMarkup
"""
if current_page >= len(self._list_kb):
current_page = 0
_list_current_page = self._list_kb[current_page]
paginations = self._get_paginator(
counts=len(self._list_kb),
page=current_page,
page_separator=self.page_separator,
startswith=self._startswith,
)
keyboard = types.InlineKeyboardMarkup(
inline_keyboard=self._before_data
+ [
*_list_current_page,
paginations,
]
+ self._after_data
)
if self.dp:
self.paginator_handler()
return keyboard
@staticmethod
def _get_page(call: types.CallbackQuery) -> int:
"""
:param call: CallbackQuery in paginator handler.
:return: Current page.
"""
return int(call.data[-1])
@staticmethod
def _chunk(it, size) -> Iterator[tuple[Any, ...]]:
"""
:param it: Source iterable object.
:param size: Chunk size.
:return: Iterator chunks pages.
"""
it = iter(it)
return iter(lambda: tuple(islice(it, size)), ())
@staticmethod
def _get_paginator(
counts: int, page: int, page_separator: str = "/", startswith: str = "page_"
) -> list[types.InlineKeyboardButton]:
"""
:param counts: Counts total buttons.
:param page: Current page.
:param page_separator: Separator for page numbers. Default = '/'.
:return: Page control line buttons.
"""
counts -= 1
paginations = []
if page > 0:
paginations.append(
types.InlineKeyboardButton(text="⏮️️", callback_data=f"{startswith}0")
)
paginations.append(
types.InlineKeyboardButton(
text="⬅️", callback_data=f"{startswith}{page - 1}"
),
)
paginations.append(
types.InlineKeyboardButton(
text=f"{page + 1}{page_separator}{counts + 1}", callback_data="pass"
),
)
if counts > page:
paginations.append(
types.InlineKeyboardButton(
text="➡️", callback_data=f"{startswith}{page + 1}"
)
)
paginations.append(
types.InlineKeyboardButton(
text="⏭️", callback_data=f"{startswith}{counts}"
)
)
return paginations
def paginator_handler(
self,
) -> tuple[Callable[[CallbackQuery, FSMContext], Coroutine[Any, Any, None]], Any]:
"""
Example:
args, kwargs = paginator.paginator_handler()
dp.register_callback_query_handler(*args, **kwargs)
:return: Data for register handler pagination.
"""
async def _page(call: types.CallbackQuery, state: FSMContext):
page = self._get_page(call)
await call.message.edit_reply_markup(
reply_markup=self.__call__(current_page=page)
)
await state.update_data({f"last_page_{self._startswith}": page})
if not self.dp:
return _page, F.data.startswith(self._startswith)
else:
self.dp.callback_query.register(
_page,
F.data.startswith(self._startswith),
)