from math import ceil from aiogram import Bot, Router from aiogram.filters import Command from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message from aiogram.types.callback_query import CallbackQuery from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.functions import count from libs.msg import eclipse_text from models import Announcement from models.callback_data import AnnounceItemClb, AnnouncePageClb router = Router(name="announcements") PAGE_SIZE = 5 async def get_reply_markup( page: int, session: AsyncSession, ) -> InlineKeyboardMarkup | None: total = await session.scalar(select(count()).select_from(Announcement)) assert total is not None total_pages = ceil(total / PAGE_SIZE) if total == 0: return None page = max(0, min(page, total_pages - 1)) query = ( select(Announcement) .offset(PAGE_SIZE * page) .limit(PAGE_SIZE) .order_by(Announcement.id.desc()) ) announcements = await session.scalars(query) announcement_buttons = [ [ InlineKeyboardButton( text=( f"{eclipse_text(a.message.text, 10)} " f"({a.datetime.strftime('%d %b %y г.')})" ), callback_data=AnnounceItemClb(page=page, announce_id=a.id).pack(), ) ] for a in announcements ] page_buttons = [] if page > 0: page_buttons.append( InlineKeyboardButton( text="◀️", callback_data=AnnouncePageClb(page=page - 1).pack(), ) ) if page < total_pages - 1: page_buttons.append( InlineKeyboardButton( text="▶️", callback_data=AnnouncePageClb(page=page + 1).pack(), ) ) return InlineKeyboardMarkup(inline_keyboard=[*announcement_buttons, page_buttons]) @router.message(Command("announcements")) async def command(msg: Message, bot: Bot, session: AsyncSession) -> None: reply_markup = await get_reply_markup(0, session) if reply_markup is None: await msg.answer("Нету анонсов для просмотра.") return await msg.answer("Выберете анонс для просмотра.", reply_markup=reply_markup) @router.callback_query(AnnouncePageClb.filter()) async def page( clb: CallbackQuery, callback_data: AnnouncePageClb, session: AsyncSession, ) -> None: assert isinstance(clb.message, Message) reply_markup = await get_reply_markup(callback_data.page, session) await clb.message.edit_text( "Выберете анонс для просмотра:", reply_markup=reply_markup, ) await clb.answer() @router.callback_query(AnnounceItemClb.filter()) async def item( clb: CallbackQuery, callback_data: AnnounceItemClb, session: AsyncSession, ) -> None: assert isinstance(clb.message, Message) announcement = await session.get(Announcement, callback_data.announce_id) assert announcement is not None rich_text = announcement.message reply_markup = InlineKeyboardMarkup( inline_keyboard=[ [ InlineKeyboardButton( text="Назад к выбору", callback_data=AnnouncePageClb(page=callback_data.page).pack(), ) ] ] ) await clb.message.edit_text( text=rich_text.text, entities=rich_text.entities, parse_mode=None, reply_markup=reply_markup, ) await clb.answer()