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
|
from datetime import UTC, datetime
from aiogram import Bot, F, Router
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 (
ContentType,
KeyboardButton,
Message,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
)
from pydantic.main import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from libs.fsm import get_data, set_data
from libs.msg import publish
from models import Announcement, RichText, User
router = Router(name="new_announcement")
class NewAnnouncementStates(StatesGroup):
message = State()
class NewAnnouncementData(BaseModel):
rich_text: RichText | None = None
SEND_BUTTON = "Опубликовать"
CANCEL_BUTTON = "Отменить создание"
@router.message(Command("new_announcement"))
async def new_announcement_command(msg: Message, state: FSMContext) -> None:
await msg.answer(
"Укажите сообщение для анонса.",
reply_markup=ReplyKeyboardMarkup(
keyboard=[
[
KeyboardButton(text=SEND_BUTTON),
KeyboardButton(text=CANCEL_BUTTON),
]
],
resize_keyboard=True,
),
)
await state.set_state(NewAnnouncementStates.message)
@router.message(NewAnnouncementStates.message, F.text == SEND_BUTTON)
async def announcement_send(
msg: Message,
bot: Bot,
state: FSMContext,
session: AsyncSession,
) -> None:
users = await session.scalars(select(User.id).where(User.id != msg.chat.id))
data = await get_data(state, NewAnnouncementData)
if data.rich_text is None:
await msg.answer("Для публикации анонса укажите текст сообщения.")
return
status_template = "Публикация анонса...\nОпубликовано: {}"
status_msg = await msg.answer(status_template.format(0))
async for n in publish(bot, users, data.rich_text):
try:
await status_msg.edit_text(status_template.format(n))
except TelegramAPIError:
pass
announcement = Announcement(message=data, datetime=datetime.now(UTC))
session.add(announcement)
await status_msg.delete()
await msg.answer(
"Анонс отправлен всем пользователям",
reply_markup=ReplyKeyboardRemove(),
)
await state.clear()
@router.message(NewAnnouncementStates.message, F.text == CANCEL_BUTTON)
async def announcement_cancel(msg: Message, state: FSMContext) -> None:
await msg.answer("Создание анонса отменено", reply_markup=ReplyKeyboardRemove())
await state.clear()
@router.message(NewAnnouncementStates.message)
async def announcement_message(msg: Message, bot: Bot, state: FSMContext) -> None:
if msg.content_type != ContentType.TEXT or msg.text is None:
await msg.answer(
"Неверный тип сообщения.\n"
"Бот поддерживает отправку только текстовых анонсов."
)
return
rich_text = RichText.from_message(msg)
await set_data(state, NewAnnouncementData(rich_text=rich_text))
msg_rich_text = RichText.from_text("Сообщение вашего анонса:\n", rich_text)
await msg_rich_text.send(bot, msg.chat.id)
|
