Untitled
4ae4d
plain_text
2 months ago
58 kB
6
Indexable
--- services/bot/core/handlers/events_ui.py ---
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from dialog_bot_sdk.interactive_media import Button, InteractiveMediaGroup, MediaGroupBuilder
from core.bot_kit.fsm import FSMContext
from core.config import bot
from core.handlers.events_pages import EVENTS_PAGE_SIZE, fetch_events_page
from core.markups import (
all_events_pagination_keyboard,
event_actions_keyboard,
events_filters_button_keyboard,
events_select_keyboard,
format_event_details,
my_events_pagination_keyboard,
part_events_pagination_keyboard,
points_events_pagination_keyboard,
points_menu_keyboard,
user_events_pagination_keyboard,
)
from core.schemas import EventCardsPageSchema, EventSchema, UserSchema
from core.services import EventService, ReportService
from core.utils import format_events_filter_summary, get_events_filter, logger
if TYPE_CHECKING:
from core.schemas.event import EventCardSchema
ADMIN_CAN_MANAGE_OTHERS = True # имеет ли администратор доступ ко всем мероприятиям?
@dataclass(frozen=True)
class EventUIContext: # legacy
name: str
# list
list_open_media_id: str
pagination_builder: Callable[..., list[InteractiveMediaGroup]]
extra_groups_builder: Callable[[], list[InteractiveMediaGroup]] | None
# card
back_value: str | None
back_label: str
enter_code_media_id: str
sign_up_media_id: str
sign_out_media_id: str
_UI: dict[str, EventUIContext] = {
# дом волонтёра -> все мероприятия
"events": EventUIContext(
name="events",
list_open_media_id="user_events_open",
pagination_builder=user_events_pagination_keyboard,
extra_groups_builder=None,
back_value="events",
back_label="⬅️ К списку мероприятий",
enter_code_media_id="event_user_enter_code",
sign_up_media_id="event_user_sign_up",
sign_out_media_id="event_user_sign_out",
),
# дом волонтёра -> мои меропрития
"part_events": EventUIContext(
name="part_events",
list_open_media_id="user_events_open",
pagination_builder=part_events_pagination_keyboard,
extra_groups_builder=None,
back_value="part_events",
back_label="⬅️ К списку мероприятий",
enter_code_media_id="event_user_enter_code",
sign_up_media_id="event_user_sign_up",
sign_out_media_id="event_user_sign_out",
),
# дом волонтёра -> посмотреть по uuid
"volunteer_home": EventUIContext(
name="volunteer_home",
list_open_media_id="user_events_open", # не используется в этом контексте
pagination_builder=user_events_pagination_keyboard,
extra_groups_builder=None,
back_value="volunteer_home",
back_label="⬅️ В дом волонтёра",
enter_code_media_id="event_user_enter_code",
sign_up_media_id="event_user_sign_up",
sign_out_media_id="event_user_sign_out",
),
# дом модерации -> все мероприятия
"moderation": EventUIContext(
name="moderation",
list_open_media_id="all_events_open",
pagination_builder=all_events_pagination_keyboard,
extra_groups_builder=None,
back_value="all_events",
back_label="⬅️ К списку мероприятий",
enter_code_media_id="event_menu_enter_code",
sign_up_media_id="event_menu_sign_up",
sign_out_media_id="event_menu_sign_out",
),
# меню баллов -> мероприятия с начисленными баллами
"points": EventUIContext(
name="points",
list_open_media_id="points_event_open",
pagination_builder=points_events_pagination_keyboard,
extra_groups_builder=points_menu_keyboard, # лидерборд + назад
back_value="points",
back_label="⬅️ К баллам",
enter_code_media_id="event_user_enter_code",
sign_up_media_id="event_user_sign_up",
sign_out_media_id="event_user_sign_out",
),
# дом модерации -> посмотреть по uuid
"moderation_menu": EventUIContext(
name="moderation_menu",
list_open_media_id="all_events_open", # не используется в этом контексте
pagination_builder=all_events_pagination_keyboard,
extra_groups_builder=None,
back_value="moderation",
back_label="⬅️ В меню модерации",
enter_code_media_id="event_menu_enter_code",
sign_up_media_id="event_menu_sign_up",
sign_out_media_id="event_menu_sign_out",
),
# дом модерации -> мои мероприятия
"my_events": EventUIContext(
name="my_events",
list_open_media_id="my_events_open",
pagination_builder=my_events_pagination_keyboard,
extra_groups_builder=None,
back_value="my_events",
back_label="⬅️ К моим мероприятиям",
enter_code_media_id="event_menu_enter_code",
sign_up_media_id="event_menu_sign_up",
sign_out_media_id="event_menu_sign_out",
),
# ответ нейронки -> карточка мероприятия
"ai_dialog_event": EventUIContext(
name="ai_dialog_event",
list_open_media_id="user_events_open",
pagination_builder=user_events_pagination_keyboard,
extra_groups_builder=None,
back_value=None,
back_label="",
enter_code_media_id="event_user_enter_code",
sign_up_media_id="event_user_sign_up",
sign_out_media_id="event_user_sign_out",
),
}
def get_ui(ctx_name: str) -> EventUIContext: # legacy
return _UI.get(ctx_name, _UI["events"])
def build_back_keyboard(ctx_name: str) -> list[InteractiveMediaGroup]:
ui = get_ui(ctx_name)
if not ui.back_value:
return []
return MediaGroupBuilder(
[Button(media_id="leave", value=ui.back_value, label=ui.back_label)]
).build()
def split_event_value(raw: Any) -> tuple[str, str | None]: # legacy
"""
ожидаем:
- "<event_id>"
- "<event_id>|<ctx>"
"""
s = str(raw or "").strip()
if "|" in s:
ev_id, ctx = s.split("|", 1)
return ev_id.strip(), (ctx.strip() or None)
return s, None
def can_manage_event(event: EventSchema, user: UserSchema | None) -> bool:
if user is not None and user.admin and ADMIN_CAN_MANAGE_OTHERS:
logger.debug("[events_ui] can_manage_event admin -> True")
return True
uid = user.messenger_id if user is not None else None
if not uid:
logger.debug("[events_ui] can_manage_event no uid -> False")
return False
result = bool(event.creator_id) and (str(event.creator_id) == str(uid))
logger.debug(
"[events_ui] can_manage_event user=%s event=%s result=%s",
user.__repr__(),
event.__repr__(),
result,
)
return result
def can_view_code(event: EventSchema, user: UserSchema | None) -> bool:
# кто может управлять - тот видит код
return can_manage_event(event, user)
def _send_events_list_screen( # legacy
peer,
*,
ui: EventUIContext,
ctx_name: str,
offset: int,
limit: int,
page_data: EventCardsPageSchema,
note: str | None = None,
header: str | None = None,
empty_text: str | None = None,
empty_groups: list[InteractiveMediaGroup] | None = None,
filter_summary: str | None = None,
show_active: bool = False,
):
# ИСПРАВЛЕНО: используем event_cards вместо events
event_cards = page_data.event_cards # list[EventCardSchema]
has_prev = page_data.has_prev
has_next = page_data.has_next
prefix = f"{note}\n\n" if note else ""
header_part = (header if header is not None else "📅 Мероприятия") + "\n\n"
filter_part = f"\n\n{filter_summary}" if filter_summary else ""
if not event_cards: # проверяем карточки
text = (
prefix
+ header_part
+ (
empty_text
or "⚠️ Мероприятий не найдено.\n\nПопробуй изменить критерии или вернись назад."
)
+ filter_part
)
im: list[InteractiveMediaGroup] = []
footer: list[InteractiveMediaGroup] = []
if empty_groups is not None:
footer = empty_groups
elif ui.extra_groups_builder:
footer = ui.extra_groups_builder()
im += events_filters_button_keyboard(ctx_name=ctx_name, offset=offset)
im += ui.pagination_builder(
offset=offset,
limit=limit,
has_prev=has_prev,
has_next=has_next,
)
im += footer
bot.messaging.send_message(peer=peer, text=text, interactive_media_groups=im)
return
text = (
prefix
+ header_part
+ f"Страница: {page_data.current_page} / {page_data.total_pages}\n\n"
+ "Выбери мероприятие:"
+ filter_part
)
im: list[InteractiveMediaGroup] = []
im += events_select_keyboard(
event_cards,
ctx=ctx_name,
open_media_id=ui.list_open_media_id,
start_index=offset + 1,
per_row=1,
)
im += events_filters_button_keyboard(ctx_name=ctx_name, offset=offset)
im += ui.pagination_builder(offset=offset, limit=limit, has_prev=has_prev, has_next=has_next)
if ui.extra_groups_builder:
im += ui.extra_groups_builder()
bot.messaging.send_message(peer=peer, text=text, interactive_media_groups=im)
def send_event_cards_page(
peer,
*,
offset: int,
event_service: EventService,
ctx_name: str,
note: str | None = None,
context: FSMContext | None = None,
):
ui = get_ui(ctx_name)
limit = EVENTS_PAGE_SIZE
flt = get_events_filter(context, ctx_name=ctx_name) if context is not None else None
summary = format_events_filter_summary(flt) if context is not None else None
page_data = fetch_events_page(
event_service=event_service,
offset=offset,
limit=limit,
requester_messenger_id=peer.id,
filters=flt,
)
_send_events_list_screen(
peer,
ui=ui,
ctx_name=ctx_name,
offset=offset,
limit=limit,
page_data=page_data,
note=note,
filter_summary=summary,
)
def send_part_events_page( # legacy
peer,
*,
offset: int,
event_service: EventService,
context: FSMContext | None = None,
note: str | None = None,
):
bot.messaging.send_message(
peer=peer,
text=("f{note}\n\n" if note else "Мои мероприятия\n\nПока не реализовано"),
interactive_media_groups=build_back_keyboard("part_events"),
)
def send_points_page( # legacy
peer,
*,
offset: int,
event_service: EventService,
context: FSMContext | None = None,
):
bot.messaging.send_message(
peer=peer,
text=("Меорприятия за которые я получил баллы\n\nПока не реализовано"),
interactive_media_groups=build_back_keyboard("part_events"),
)
def send_my_events_page( # legacy
peer,
*,
offset: int,
event_service: EventService,
requester_messenger_id: int,
ctx_name: str,
note: str | None = None,
context: FSMContext | None = None,
):
ui = get_ui(ctx_name)
limit = EVENTS_PAGE_SIZE
flt = get_events_filter(context, ctx_name=ctx_name) if context is not None else None
summary = format_events_filter_summary(flt) if context is not None else None
page_data = fetch_events_page(
event_service=event_service,
offset=offset,
limit=limit,
requester_messenger_id=peer.id,
filters=flt,
mine=True,
)
_send_events_list_screen(
peer,
ui=ui,
ctx_name=ctx_name,
offset=offset,
limit=limit,
page_data=page_data,
note=note,
header="👤 Мои мероприятия",
filter_summary=summary,
show_active=True,
)
def send_event_card(
peer,
*,
event_id: str,
event_service: EventService,
report_service: ReportService | None = None,
user: UserSchema | None,
ctx_name: str,
note: str | None = None,
show_organizer: bool = True,
):
ui = get_ui(ctx_name)
card: EventCardSchema | None = event_service.get_event_by_id(event_id, peer.id)
if card is None:
text = f"{note}\n\n⚠️ Мероприятие не найдено." if note else "⚠️ Мероприятие не найдено."
bot.messaging.send_message(
peer=peer,
text=text,
interactive_media_groups=build_back_keyboard(ctx_name),
)
return
# TODO: 1. Убрать проверку права редактирования в сервис
# TODO: 2. Бекенд будет возвращать список организаторов события
# TODO: 3. Редактировать может не creator_id, а волонтер, входящий в список организаторов
can_manage = can_manage_event(card.event, user)
if can_manage:
managed_ev = event_service.get_event_by_id(event_id, peer.id, mine=True)
if managed_ev is not None:
# TODO убрать заглушку: backend не возвращает participation для ручки GET /events/card/list/mine
participation = card.participation # TODO REMOVE MOCK
card = managed_ev
card.participation = participation # TODO REMOVE MOCK
is_participant = card.participation is not None
# TODO: Код не приходит, если его нет, эта логика на бекенде, подумать, как упростить это здесь (Optional?)
show_code = can_view_code(card.event, user)
can_access_reports = can_manage
report_exists: bool | None = None
if can_access_reports and report_service is not None:
report_exists = report_service.report_exists(
event_id=str(card.event.id),
requester_messenger_id=peer.id,
)
else:
report_exists = None
can_create_report = bool(can_manage and (report_exists is False))
can_view_report = bool(can_access_reports and (report_exists is True))
show_active = ctx_name == "my_events"
base = format_event_details(
card,
show_active=show_active,
show_code=show_code,
show_organizer=show_organizer,
show_report_status=bool(can_access_reports and report_exists is not None),
report_exists=bool(report_exists) if report_exists is not None else None,
)
text = f"{note}\n\n{base}" if note else base
bot.messaging.send_message(
peer=peer,
text=text,
interactive_media_groups=event_actions_keyboard(
card.event.id,
is_participant=is_participant,
can_manage=can_manage,
back_value=ui.back_value,
back_label=ui.back_label,
enter_code_media_id=ui.enter_code_media_id,
sign_up_media_id=ui.sign_up_media_id,
sign_out_media_id=ui.sign_out_media_id,
ctx=ui.name, # важно: чтобы value был "<event_id>|<ctx>"
can_create_report=can_create_report,
can_view_report=can_view_report,
),
)
def build_back_to_event_keyboard( # TODO перенести клавиатуру туда где ей место
event_id: str, ctx_name: str
) -> list[InteractiveMediaGroup]: # legacy
ui = get_ui(ctx_name)
return MediaGroupBuilder(
[
Button(
media_id=ui.list_open_media_id,
value=f"{event_id}|{ui.name}",
label="⬅️ К мероприятию",
)
]
).build()
def build_back_to_event_editing_keyboard( # TODO перенести клавиатуру туда где ей место
event_id: str, ctx_name: str
) -> list[InteractiveMediaGroup]: # legacy
ui = get_ui(ctx_name)
return MediaGroupBuilder(
[
Button(
media_id=ui.list_open_media_id,
value=f"{event_id}|{ui.name}",
label="⬅️ К мероприятию",
)
]
).build()
--- services/bot/core/handlers/moderation.py ---
import re
from datetime import datetime
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent, UpdateMessage
from dialog_bot_sdk.interactive_media import Button
from core.bot_kit.fsm import FSMContext, State, StatesGroup
from core.bot_kit.router import Router
from core.config import bot
from core.handlers.events_ui import (
build_back_to_event_keyboard,
send_event_card,
send_event_cards_page,
send_my_events_page,
split_event_value,
)
from core.markups import (
back_to_moderation_keyboard,
choices_from_backend_tags,
event_edit_fields_keyboard,
format_event_details,
gosb_select_keyboard,
moderation_menu_keyboard,
tags_toggle_keyboard,
tb_select_keyboard,
)
from core.schemas import EventCardSchema, TagSchema, UserSchema
from core.services import EventService, ReportService, TagService, TerbankService
from core.utils import (
clear_context_keep_events_filters,
delete_prev_message,
delete_prev_message_by_peer,
logger,
)
TAGS_MEDIA_ID = "event_tags_toggle"
TAGS_TOGGLE_PREFIX = "toggle:"
TAGS_DONE_VALUE = "done"
# TODO убрать хардкод
events_rt = Router()
class EventEditTagsState(StatesGroup): # legacy
tags = State()
class EventEditState(StatesGroup): # legacy
wait_value = State()
class EventEnterCodeState(StatesGroup): # legacy
wait_code = State()
class EventViewState(StatesGroup): # legacy
wait_event_id = State()
class EventCreateState(StatesGroup): # legacy
name = State()
date = State()
time = State()
# location удалён
gosb = State()
project = State()
description = State()
hours = State()
tags = State()
code = State()
@bot.di
def moderation_menu_handler(event: UpdateInteractiveMediaEvent, context: FSMContext): # legacy
clear_context_keep_events_filters(context)
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text=("Ты в доме модерации!\n\nПожалуйста, выбери, что хочешь сделать 👇"),
interactive_media_groups=moderation_menu_keyboard(),
)
@bot.di
def all_events_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_event_cards_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def my_events_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_my_events_page(
event.peer,
offset=0,
event_service=event_service,
requester_messenger_id=event.peer.id,
ctx_name="my_events",
context=context,
)
@bot.di
def my_events_page_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_my_events_page(
event.peer,
offset=offset,
event_service=event_service,
requester_messenger_id=event.peer.id,
ctx_name="my_events",
context=context,
)
@bot.di
def my_events_open_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id = str(event.data.value or "").strip()
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name="my_events",
report_service=report_service,
)
@bot.di
def all_events_page_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_event_cards_page(
event.peer,
offset=offset,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def all_events_open_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name,
report_service=report_service,
)
@bot.di
def create_event_start_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
):
"""старт мастера создания мероприятия"""
delete_prev_message_by_peer(bot, event.peer)
context.clear()
bot.messaging.send_message(
peer=event.peer,
text=("Создание мероприятия.\n\nШаг 1/10: введи, пожалуйста, название мероприятия:"),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.name)
@events_rt.message(state=EventCreateState.name)
@bot.di
def event_create_name_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
name = message.message.text_message.text.strip()
context.update_data({"event_name": name})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 2/10: введи, пожалуйста, дату мероприятия например, в формате `20.02.2002`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.date)
@events_rt.message(state=EventCreateState.date)
@bot.di
def event_create_date_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
date = message.message.text_message.text.strip()
formats = []
for first_gap in ". -_":
for second_gap in ". -_":
formats.append(f"%d{first_gap}%m{second_gap}%Y")
date_obj = None
for format in formats:
try:
date_obj = datetime.strptime(date, format).date()
except ValueError:
continue
if date_obj is None:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Я не смог распознать дату.\n"
"Пожалуйста, введи дату, например, в формате `20.02.2002`:"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
if date_obj <= datetime.now().date():
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\n"
"Пожалуйста, введи дату, например, в формате `20.02.2002`:"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"event_date": date_obj})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 3/10: введи, пожалуйста, время мероприятия например, в формате `10:30`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.time)
@events_rt.message(state=EventCreateState.time)
@bot.di
def event_create_time_step(
message: UpdateMessage, context: FSMContext, terbank_service: TerbankService
):
delete_prev_message(bot, message)
time = message.message.text_message.text.strip()
time_obj = None
for format in ["%H:%M", "%H %M", "%H-%M", "%H_%M"]:
try:
time_obj = datetime.strptime(time, format).time()
except ValueError:
continue
if time_obj is None:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Я не смог распознать время.\n"
"Пожалуйста, введи в формате `HH:MM`, например: `10:30`:"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"event_time": time_obj})
tbs = terbank_service.get_all_tbs(message.peer.id)
bot.messaging.send_message(
peer=message.peer,
text="Шаг 4/10: Выбери тербанк:",
interactive_media_groups=tb_select_keyboard(tbs),
)
@bot.di
def tb_update_selected(event: UpdateInteractiveMediaEvent, context: FSMContext):
# keeping what user have selected in "tb_id"
context.update_data({"tb_id": event.data.value})
@bot.di
def tb_approved(
event: UpdateInteractiveMediaEvent, terbank_service: TerbankService, context: FSMContext
):
# tb was selected and approved. now selecting gosb:
delete_prev_message_by_peer(bot, event.peer)
selected_tb_id = context.get_data().get("tb_id")
if not isinstance(selected_tb_id, str):
logger.error(
"error while creating event: error while selecting tb: selected tb must be a string, got {selected_tb_id!r}"
)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Произошла ошибка. Попробуй позже.",
)
context.clear()
return
gosbs = terbank_service.get_gosbs_for_tb(tb_id=selected_tb_id, messenger_id=event.peer.id)
bot.messaging.send_message(
peer=event.peer,
text="Шаг 5/10: Выбери ГОСБ",
interactive_media_groups=gosb_select_keyboard(gosbs),
)
@bot.di
def gosb_update_selected(event: UpdateInteractiveMediaEvent, context: FSMContext):
# keeping what user have selected in "gosb_id"
context.update_data({"gosb_id": event.data.value})
@bot.di
def gosb_approved(event: UpdateInteractiveMediaEvent, context: FSMContext):
# gosb was selected and approved.
delete_prev_message_by_peer(bot, event.peer)
selected_gosb_id = context.get_data().get("gosb_id")
if not isinstance(selected_gosb_id, str):
logger.error(
"error while creating event: error while selecting gosb: selected gosb must be a string, got {selected_gosb_id!r}"
)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Произошла ошибка. Попробуй позже.",
)
context.clear()
return
bot.messaging.send_message(
peer=event.peer,
text=(
"Шаг 6/10: Введи id проекта, если хочешь создать мероприятие в проекте\n"
"Введи `-`, если хочешь создать событийное мероприятие - мероприятие вне проекта."
),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.project)
@events_rt.message(state=EventCreateState.gosb)
@bot.di
def event_create_gosb_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
gosb = message.message.text_message.text.strip()
context.update_data({"gosb_id": gosb})
bot.messaging.send_message(
peer=message.peer,
text=(
"Шаг 6/10: Введи id проекта, если хочешь создать мероприятие в проекте\n"
"Введи `-`, если хочешь создать событийное мероприятие - мероприятие вне проекта."
),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.project)
@events_rt.message(state=EventCreateState.project)
@bot.di
def event_create_project_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
raw = message.message.text_message.text.strip()
project_id = "" if raw == "-" else raw
context.update_data({"project_id": project_id})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 7/10: Введи описание:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.description)
@events_rt.message(state=EventCreateState.description)
@bot.di
def event_create_description_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
description = message.message.text_message.text.strip()
context.update_data({"event_description": description})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 8/10: Введи количество волонтёрских часов, начисляемых за мероприятие:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.hours)
@events_rt.message(state=EventCreateState.hours)
@bot.di
def event_create_hours_step(
message: UpdateMessage,
context: FSMContext,
tag_service: TagService,
):
delete_prev_message(bot, message)
text = message.message.text_message.text.strip()
try:
if not re.fullmatch(r"^[0-9]{1,2}$", text):
raise ValueError
hours = int(text)
if hours <= 0 or hours >= 19:
raise ValueError
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Количество часов должно быть натуральным числом, меньшим 19. Попробуй ещё раз.",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"hours": hours})
all_tags = tag_service.get_all_tags()
choices = choices_from_backend_tags(all_tags)
context.update_data(
{
"event_tag_keys": [],
"all_tags": [{"id": t.id, "title": t.title} for t in all_tags],
},
)
_send_tags_toggle_ui(
peer=message.peer,
title="Шаг 9/10: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:",
choices=choices,
selected_keys=set(),
footer_buttons=[Button(media_id="leave", value="moderation", label="⬅️ В меню модерации")],
)
context.set_state(EventCreateState.tags)
@bot.di
def event_tags_toggle_handler(
event: UpdateInteractiveMediaEvent,
event_service: EventService,
context: FSMContext,
):
state = context.get_state()
is_create = state == EventCreateState.tags or str(state) == str(EventCreateState.tags)
is_edit = state == EventEditTagsState.tags or str(state) == str(EventEditTagsState.tags)
if not (is_create or is_edit):
return
raw = str(event.data.value or "")
data = context.get_data()
key_field = "event_tag_keys" if is_create else "edit_tag_keys"
selected = set(data.get(key_field) or [])
all_tags = data.get("all_tags", [])
if not all_tags:
logger.info("No tags were taken while editing event tags. Please check seeding file.")
if raw == TAGS_DONE_VALUE:
title_to_id = {t["title"]: t["id"] for t in all_tags if t.get("title") and t.get("id")}
app_tag_ids = [title_to_id[n] for n in selected if n in title_to_id]
delete_prev_message_by_peer(bot, event.peer)
if is_create:
context.update_data({"event_app_tag_ids": app_tag_ids})
bot.messaging.send_message(
peer=event.peer,
text="Шаг 10/10: введи код мероприятия:\nБез пробелов, не больше 16 символов. Например: WELCOME2025",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.code)
return
if is_edit:
current = set(data.get("edit_current_tag_keys") or [])
event_id = str(data.get("edit_event_id"))
if not event_id or event_id == "None":
logger.error(
"Error while editing event tags: "
f"no event was found by saved id {event_id!r}, "
f"saved in data by the key 'edit_event_id'. "
"exiting editing session"
)
context.set_state(None)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново.",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
to_add = selected - current
to_remove = current - selected
failed = False
# add
for title in to_add:
tag_id = title_to_id.get(title)
if not tag_id:
logger.error(
"Error while editing event tags: "
f"selected tag with title={title!r} "
f"have no entrance in all_tags={all_tags}. "
"skipping this tag."
)
elif not event_service.add_tag(
event_id=event_id, tag_id=tag_id, messenger_id=event.peer.id
):
failed = True
# remove
for title in to_remove:
tag_id = title_to_id.get(title)
if not tag_id:
logger.error(
"Error while editing event tags: "
f"selected tag with title={title!r} "
f"have no entrance in all_tags={all_tags}. "
"skipping this tag."
)
elif not event_service.remove_tag(
event_id=event_id, tag_id=tag_id, messenger_id=event.peer.id
):
failed = True
context.set_state(None)
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Ошибка изменения тегов! Некоторые теги могли не измениться."
if failed
else "✅ Теги успешно изменены.",
event_obj=event_service.get_event_by_id(event_id, event.peer.id),
show_organizer=True,
)
return
if raw.startswith(TAGS_TOGGLE_PREFIX):
key = raw[len(TAGS_TOGGLE_PREFIX) :].strip()
if key in selected:
selected.remove(key)
else:
selected.add(key)
context.update_data({key_field: list(selected)})
delete_prev_message_by_peer(bot, event.peer)
if is_create:
footer = [Button(media_id="leave", value="moderation", label="⬅️ В меню модерации")]
title = "Шаг 9/10: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:"
else:
event_id = data.get("edit_event_id") or ""
footer = [
Button(
media_id="event_menu_edit",
value=event_id,
label="⬅️ К редактированию мероприятия",
)
]
title = "Редактирование направления: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:"
normalized_tags: list[TagSchema] = []
for tag in all_tags:
if isinstance(tag, TagSchema):
normalized_tags.append(tag)
elif isinstance(tag, dict):
normalized_tags.append(TagSchema.from_dict(tag))
choices = choices_from_backend_tags(normalized_tags)
_send_tags_toggle_ui(
peer=event.peer,
title=title,
choices=choices,
selected_keys=set(selected),
footer_buttons=footer,
)
return
@events_rt.message(state=EventCreateState.code)
@bot.di
def event_create_code_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
):
delete_prev_message(bot, message)
code = message.message.text_message.text.strip()
if not re.fullmatch(r"^[A-Za-zА-Яа-я0-9]{1,16}$", code):
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Код должен быть комбинацией русских и латинских букв и цифр, длиной не больше 16 символов. Попробуй еще раз:",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"event_verification_code": code})
data = context.get_data()
try:
created = event_service.create_from_wizard(data, message.peer)
except Exception as e:
logger.error("Error while creating an event via wizard: %s", e)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Произошла ошибка.",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.clear()
return
context.clear()
note = {
"REQUEST": "✅ Заявка на мероприятие успешно создана.",
"COMPLETELY": "✅ Мероприятие успешно создано.",
"FAILED": "⚠️ Произошла ошибка: не удалось создать заявку/мероприятие.",
}.get(
created.get("TYPE", ""),
"⚠️ Произошла ошибка: не получить статус заявки/мероприятия. Проверь результат в 'Своих заявках' или в 'Своих мероприятиях'.",
)
formatted_data = (
str(data).strip("{}").replace("), ", "),\n").replace("], ", "],\n").replace("', ", "',\n")
)
bot.messaging.send_message(
peer=message.peer,
text=f"{note}\n{formatted_data}",
interactive_media_groups=moderation_menu_keyboard(),
)
@bot.di
def event_view_by_id_start(event: UpdateInteractiveMediaEvent, context: FSMContext): # legacy
"""старт диалога - просим у модератора UUID мероприятия"""
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Введи UUID мероприятия, которое хочешь посмотреть.\n\n",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventViewState.wait_event_id)
@events_rt.message(state=EventViewState.wait_event_id)
@bot.di
def event_view_by_id_step( # legacy
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
user: UserSchema,
):
"""читаем UUID, тянем мероприятие из бэкенда и показываем информацию"""
delete_prev_message(bot, message)
raw_id = message.message.text_message.text.strip()
event = event_service.get_event_by_id(raw_id, message.peer.id)
if event is None:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Мероприятие с таким UUID не найдено.\nПопробуй ещё раз. Пример UUID: `932cf1cb-1ee7-4a45-bc96-41a232f3f538`",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"current_event_id": event.id})
context.set_state(None)
send_event_card(
message.peer,
event_id=event.id,
event_service=event_service,
user=user,
ctx_name="moderation_menu",
)
@bot.di
def event_menu_sign_up_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
ok = event_service.sign_up(event_id, event.peer.id)
if not ok:
note = "⚠️ Не удалось записаться на мероприятие."
else:
note = "✅ Запись на мероприятие прошла успешно!"
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_sign_out_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
# TODO may be a source of bugs. do we really need to deafult context?
status = event_service.get_participation_status(event_id, event.peer.id)
if status is None:
note = "⚠️ Не удалось отписаться: ты не записан на это мероприятие."
elif status.state == "confirmed":
note = "⚠️ Нельзя отписаться: часы уже получен за это участие."
else:
ok = event_service.sign_out_by_event(event_id, event.peer.id)
if ok:
note = "✅ Ты успешно отписался от мероприятия."
else:
note = "⚠️ Не удалось отписаться. Попробуй ещё раз."
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_enter_code_start_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext
): # legacy
delete_prev_message_by_peer(bot, event.peer)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
context.update_data({"current_event_id": event_id, "current_event_ctx": ctx_name})
context.set_state(EventEnterCodeState.wait_code)
bot.messaging.send_message(
peer=event.peer,
text="🎟️ Введи бонус-код мероприятия (без пробелов):",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
@events_rt.message(state=EventEnterCodeState.wait_code)
@bot.di
def event_menu_enter_code_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema,
):
delete_prev_message(bot, message)
event_id = str(context.get_data().get("current_event_id"))
# если event_id is None, то клиент не найдёт event с id == "None", и всё ок
code = message.message.text_message.text.strip()
note: str
status = event_service.get_participation_status(event_id, message.peer.id)
if status is None:
note = "⚠️ Код не принят, сначала нужно записаться на мероприятие"
elif status.state == "confirmed":
note = "⚠️ Код не принят, баллы уже получены"
else:
ok = event_service.enter_code_by_event(event_id, message.peer.id, code)
if not ok:
note = "⚠️ Код не принят, проверь правильность кода"
else:
note = "✅ Код принят. Часы начислены."
data = context.get_data()
ctx_name = data.get("current_event_ctx") or "moderation"
context.set_state(None)
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_delete_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
if ctx_name not in ("moderation", "moderation_menu"):
ctx_name = "moderation"
try:
ok = event_service.delete_event(
event_id,
requester_messenger_id=event.peer.id,
)
except Exception:
ok = False
if ok:
context.clear()
if ctx_name == "moderation":
send_event_cards_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="moderation",
note="Мероприятие удалено.",
)
else:
bot.messaging.send_message(
peer=event.peer,
text="Мероприятие удалено.\n\nТы в доме модерации",
interactive_media_groups=moderation_menu_keyboard(),
)
return
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note="⚠️ Не удалось удалить мероприятие",
)
def _send_event_edit_menu( # legacy
peer,
*,
event_id: str,
event_service: EventService,
note: str | None = None,
event_obj=None, # temporary remove type since it was fixed in another branch
show_organizer: bool,
):
refreshed = event_obj or event_service.get_event_by_id(event_id, peer.id)
base = (
format_event_details(refreshed, show_code=True, show_organizer=show_organizer)
if refreshed
else "⚠️ Мероприятие не найдено"
)
prefix = f"{note}\n\n" if note else ""
bot.messaging.send_message(
peer=peer,
text=(f"{prefix}Редактирование мероприятия\n\n{base}\n\nВыбери поле для Редактирования:"),
interactive_media_groups=event_edit_fields_keyboard(event_id),
)
@bot.di
def event_menu_edit_handler( # legacy
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, _ctx = split_event_value(event.data.value)
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
show_organizer=True,
)
def _get_field_label(field: str) -> str: # legacy
return {
"title": "Название",
"date": "Дата",
"time": "Время",
"tags": "Теги",
"event_organizers": "Организаторы",
"hours": "Баллы",
"description": "Описание",
"code": "Код",
}.get(field, field)
def _get_current_value(event: EventCardSchema, field: str) -> str: # legacy
val = getattr(event, field, None)
if val is None or val == "":
return "-"
if field == "date":
try:
return str(event.event.date_time) # TODO надел чтобы работало пока
except Exception:
return str(val)
if field == "time":
try:
return str(event.event.date_time) # TODO надел чтобы работало пока
except Exception:
return str(val)
return str(val)
def _send_tags_toggle_ui(
*,
peer,
title: str,
choices,
selected_keys: set[str],
footer_buttons: list[Button],
):
bot.messaging.send_message(
peer=peer,
text=title,
interactive_media_groups=tags_toggle_keyboard(
choices=choices,
selected_keys=selected_keys,
media_id=TAGS_MEDIA_ID,
toggle_value_prefix=TAGS_TOGGLE_PREFIX,
done_value=TAGS_DONE_VALUE,
done_label="✅ Готово",
extra_footer=footer_buttons,
),
)
@bot.di
def event_menu_edit_field_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
tag_service: TagService,
):
delete_prev_message_by_peer(bot, event.peer)
raw = str(event.data.value or "")
if "|" not in raw:
logger.error("Got unexpected event.data.value={raw!r} while trying to edit event")
bot.messaging.send_message(peer=event.peer, text="⚠️ Некорректная команда редактирования.")
return
event_id, field = raw.split("|", 1)
field = field.strip()
refreshed_event_schema = event_service.get_event_by_id(event_id, event.peer.id)
if not refreshed_event_schema:
bot.messaging.send_message(peer=event.peer, text="⚠️ Мероприятие не найдено.")
return
if field == "tags":
# редактируем теги через toggle UI
all_tags = tag_service.get_all_tags()
choices = choices_from_backend_tags(all_tags)
selected_titles: set[str] = set()
for t in refreshed_event_schema.tags or []:
if isinstance(t, dict) and t.get("title"):
selected_titles.add(str(t["title"]))
else:
title = getattr(t, "title", None)
if title:
selected_titles.add(str(title))
context.update_data(
{
"edit_event_id": event_id,
"all_tags": [{"id": t.id, "title": t.title} for t in all_tags],
"edit_tag_keys": list(selected_titles),
"edit_current_tag_keys": list(selected_titles),
}
)
context.set_state(EventEditTagsState.tags)
_send_tags_toggle_ui(
peer=event.peer,
title="Редактирование направления: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:",
choices=choices,
selected_keys=set(selected_titles),
footer_buttons=[
Button(
media_id="event_menu_edit",
value=event_id,
label="⬅️ К редактированию мероприятия",
)
],
)
return
context.update_data({"edit_event_id": event_id, "edit_field": field})
context.set_state(EventEditState.wait_value)
label = _get_field_label(field)
cur = _get_current_value(refreshed_event_schema, field)
if field == "event_organizers":
bot.messaging.send_message(
peer=event.peer,
text="Пока не реализовано",
interactive_media_groups=build_back_to_event_keyboard(
event_id=event_id,
ctx_name="moderation", # TODO remove hardcode
),
)
return
if field == "date":
hint = "Введи дату в формате `dd.mm.yyyy`, например: `20.02.2002`"
elif field == "time":
hint = "Введи время в формате `HH:MM`, например: `10:30`"
else:
hint = 'Введи новое значение. Чтобы очистить поле - отправь: "-"'
bot.messaging.send_message(
peer=event.peer,
text=f"Редактирование: {label}\nТекущее значение: {cur}\n{hint}",
)
@events_rt.message(state=EventEditState.wait_value)
@bot.di
def event_edit_value_step( # legacy
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
delete_prev_message(bot, message)
data = context.get_data()
event_id: str = data.get("edit_event_id", "")
field: str = data.get("edit_field", "")
if not event_id or not field:
context.set_state(None)
bot.messaging.send_message(
peer=message.peer, text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново."
)
return
raw = message.message.text_message.text.strip()
value: str | int = "" if raw == "-" else raw
if field == "event_organizers":
bot.messaging.send_message(
peer=message.peer,
text="Пока не реализовано",
interactive_media_groups=build_back_to_event_keyboard(
event_id=event_id,
ctx_name="moderation", # TODO remove hardcode
),
)
context.set_state(None)
return
if field == "hours":
if raw == "-":
value = None
else:
try:
value = int(raw)
if value <= 0 or value >= 2**31:
raise ValueError
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Баллы должны быть натуральным числом,"
f" не большим {2**31}, попробуй еще раз."
),
)
return
if (
field == "code"
and value is not None
and not re.fullmatch(r"^[A-Za-zА-Яа-я0-9]{1,16}$", str(value))
):
bot.messaging.send_message(
peer=message.peer,
text=(
"Код должен быть до 16 символов: латинские и кириллические"
" буквы и цифры без пробелов до 16 символов."
),
)
return
if field == "date" and value is not None:
raise NotImplementedError
if field == "time" and value is not None:
raise NotImplementedError
if field in ("date", "time"):
raise NotImplementedError
payload = {field: value}
updated = event_service.update_event(
event_id=event_id,
messenger_id=message.peer.id,
payload=payload, # type(value) in (int, str)
)
context.set_state(None)
_send_event_edit_menu(
message.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Не удалось обновить поле." if not updated else "✅ Поле обновлено.",
event_obj=event_service.get_event_by_id(event_id, message.peer.id),
show_organizer=True,
)
Editor is loading...
Leave a Comment