1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
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()
|
