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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
from math import ceil
from aiogram import Bot, Router
from aiogram.filters import Command
from aiogram.types import (
CallbackQuery,
InlineKeyboardButton,
InlineKeyboardMarkup,
Message,
)
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.functions import count
from libs.invoice import (
InvoiceStatus,
get_invoice_payments,
get_payment_status,
)
from libs.msg import eclipse_text
from libs.user import load_user_cache
from models import Invoice, PaymentStatus, User
from models.callback_data import InvoiceItemClb, InvoicePageClb, PayInvoiceClb
router = Router(name="invoices")
PAGE_SIZE = 5
PAYMENT_STATUS = {
PaymentStatus.PENDING: "🟡",
PaymentStatus.ACCEPTED: "🟢",
PaymentStatus.REJECTED: "🔴",
}
INVOICE_STATUS = {
InvoiceStatus.PAID: "🟢",
InvoiceStatus.UNPAID: "🔴",
}
def get_text(user: User) -> str:
if user.is_admin():
return "Выберете счёт для просмотра информации:"
else:
return "Выберете счёт для оплаты:"
async def get_reply_markup(
page: int,
user: User,
session: AsyncSession,
) -> InlineKeyboardMarkup | None:
total = await session.scalar(
select(count()).select_from(Invoice).where(Invoice.datetime >= user.datetime)
)
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(Invoice)
.where(Invoice.datetime >= user.datetime)
.offset(PAGE_SIZE * page)
.limit(PAGE_SIZE)
.order_by(Invoice.id.desc())
)
invoices = await session.scalars(query)
invoice_buttons = []
for i in invoices:
if user.is_admin():
invoice_payments = await get_invoice_payments(session, i)
status = INVOICE_STATUS[invoice_payments.status]
callback_data = InvoiceItemClb(page=page, invoice_id=i.id).pack()
else:
status = PAYMENT_STATUS[await get_payment_status(session, i.id, user.id)]
callback_data = PayInvoiceClb(invoice_id=i.id).pack()
button = InlineKeyboardButton(
text=(
f"{status} "
f"{eclipse_text(i.message.text, 10)} "
f"({i.datetime.strftime('%d %b %y г.')})"
),
callback_data=callback_data,
)
invoice_buttons.append([button])
page_buttons = []
if page > 0:
page_buttons.append(
InlineKeyboardButton(
text="◀️",
callback_data=InvoicePageClb(page=page - 1).pack(),
)
)
if page < total_pages - 1:
page_buttons.append(
InlineKeyboardButton(
text="▶️",
callback_data=InvoicePageClb(page=page + 1).pack(),
)
)
return InlineKeyboardMarkup(inline_keyboard=[*invoice_buttons, page_buttons])
@router.message(Command("invoices"))
async def command(msg: Message, session: AsyncSession, user: User) -> None:
reply_markup = await get_reply_markup(0, user, session)
if reply_markup is None:
await msg.answer("Нету счетов для оплаты.")
return
await msg.answer(get_text(user), reply_markup=reply_markup)
@router.callback_query(InvoicePageClb.filter())
async def page(
clb: CallbackQuery,
callback_data: InvoicePageClb,
session: AsyncSession,
user: User,
) -> None:
assert isinstance(clb.message, Message)
await clb.message.edit_text(
get_text(user),
reply_markup=await get_reply_markup(callback_data.page, user, session),
)
await clb.answer()
@router.callback_query(InvoiceItemClb.filter())
async def item(
clb: CallbackQuery,
bot: Bot,
callback_data: InvoiceItemClb,
session: AsyncSession,
user: User,
) -> None:
assert isinstance(clb.message, Message)
if not user.is_admin():
await clb.answer("У вас нет прав для данного действия.", show_alert=True)
return
invoice = await session.get(Invoice, callback_data.invoice_id)
assert invoice is not None
invoice_payments = await get_invoice_payments(session, invoice)
text_template = (
f"Статус оплаты счёта: {INVOICE_STATUS[invoice_payments.status]}\n"
"Пользователи оплатившие счёт:\n"
"{}"
)
reply_markup = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text="Назад к выбору",
callback_data=InvoicePageClb(page=callback_data.page).pack(),
)
]
]
)
user_status = []
for user_id, s in invoice_payments.user_status.items():
user_cache = await load_user_cache(bot, user_id)
user_status.append(f"{PAYMENT_STATUS[s]} - {user_cache.mention}")
await clb.message.edit_text(
text_template.format("\n".join(user_status)),
reply_markup=reply_markup,
disable_web_page_preview=True,
)
await clb.answer()
|
