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
|
from datetime import UTC, datetime
from aiogram import Bot, F, Router
from aiogram.enums import ButtonStyle
from aiogram.exceptions import TelegramAPIError
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import (
KeyboardButton,
Message,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
)
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from libs.fsm import get_data, set_data
from libs.msg import send_invoice
from models import Invoice, RichText, User, UserRole
router = Router(name="new_invoice")
class NewInvoiceStates(StatesGroup):
message = State()
class NewInvoiceData(BaseModel):
rich_text: RichText | None = None
CREATE_BUTTON = "Создать"
CANCEL_BUTTON = "Отменить создание"
@router.message(Command("new_invoice"))
async def command(msg: Message, state: FSMContext) -> None:
await msg.answer(
"Укажите сообщение для создания счёта",
reply_markup=ReplyKeyboardMarkup(
keyboard=[
[
KeyboardButton(text=CREATE_BUTTON, style=ButtonStyle.SUCCESS),
KeyboardButton(text=CANCEL_BUTTON, style=ButtonStyle.DANGER),
]
],
resize_keyboard=True,
),
)
await state.set_state(NewInvoiceStates.message)
@router.message(NewInvoiceStates.message, F.text == CREATE_BUTTON)
async def create(
msg: Message,
bot: Bot,
state: FSMContext,
session: AsyncSession,
) -> None:
users = await session.scalars(select(User.id).where(User.role != UserRole.ADMIN))
data = await get_data(state, NewInvoiceData)
if data.rich_text is None:
await msg.answer("Для создания счёта укажите сообщение.")
return
status_template = "Рассылка счёта...\nОтправлено: {}"
status_msg = await msg.answer(status_template.format(0))
invoice = Invoice(message=data.rich_text, datetime=datetime.now(UTC))
session.add(invoice)
await session.flush()
async for n in send_invoice(bot, users, data.rich_text, invoice.id):
try:
await status_msg.edit_text(status_template.format(n))
except TelegramAPIError:
pass
await status_msg.delete()
await msg.answer(
"Счёт отправлен всем пользователям",
reply_markup=ReplyKeyboardRemove(),
)
await state.clear()
@router.message(NewInvoiceStates.message, F.text == CANCEL_BUTTON)
async def cancel(msg: Message, state: FSMContext) -> None:
await msg.answer("Создание счёта отменено", reply_markup=ReplyKeyboardRemove())
await state.clear()
@router.message(NewInvoiceStates.message)
async def set_message(msg: Message, bot: Bot, state: FSMContext) -> None:
if msg.text is None:
await msg.answer(
"Неверный тип сообщения.\n"
"Бот поддерживает отправку только текстовых счетов."
)
return
rich_text = RichText.from_message(msg)
await set_data(state, NewInvoiceData(rich_text=rich_text))
msg_rich_text = RichText.from_text("Сообщение вашего счёта:\n", rich_text)
await msg_rich_text.send(bot, msg.chat.id)
|
