Untitled
4ae4d
plain_text
02/10/2026 10:00 AM
104.9 KB
14
Indexable
--- src/bot/core/handlers/events_pages.py ---
from core.schemas.event import EventSchema
from core.services.event import EventService
from core.utils.events_filter import EventsFilter, EventsSearchBase, build_events_search_params
EVENTS_PAGE_SIZE = 10
def fetch_events_page(
*,
event_service: EventService,
city_id: str,
offset: int,
limit: int,
requester_messenger_id: int | None = None,
only_claimed: bool = False,
only_organizer: bool = False,
only_participant: bool = False,
has_report: bool | None = None,
tag_ids: list[str] | None = None,
filters: EventsFilter | None = None,
lock_mode: bool = False,
) -> tuple[list[EventSchema], bool, bool, int | None]:
base = EventsSearchBase(
city_id=city_id,
requester_messenger_id=requester_messenger_id,
only_claimed=only_claimed,
only_organizer=only_organizer,
only_participant=only_participant,
has_report=has_report,
tag_ids=list(tag_ids or []),
lock_mode=bool(lock_mode),
)
params = build_events_search_params(
base=base,
flt=filters,
offset=offset,
limit=limit + 1,
)
events_raw, total = event_service.search_events_with_total(**params)
has_next = len(events_raw) > limit
has_prev = offset > 0
page_events = events_raw[:limit]
return page_events, has_prev, has_next, total
def render_events_list_text(
*, events: list[EventSchema], offset: int, total: int | None, footer: str | None = None
) -> str:
if not events:
return "⚠️ Мероприятий не найдено."
header_total = str(total) if total is not None else "?"
lines = [f"Мероприятия (показаны {offset + 1} - {offset + len(events)} из {header_total}):\n"]
for event in events:
date_str = event.date.strftime("%d.%m.%Y %H:%M")
lines.append(f" - {event.name} --- {date_str} --- ({event.location}) --- `{event.id}`")
if footer:
lines.append("")
lines.append(footer)
return "\n".join(lines)
--- src/bot/core/handlers/event.py ---
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent, UpdateMessage
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_city_events_page,
send_event_card,
split_event_value,
)
from core.schemas.user import UserSchema
from core.services.event import EventService
from core.services.report import ReportService
from core.utils import delete_prev_message, delete_prev_message_by_peer
from core.utils.events_filter import clear_context_keep_events_filters
DEFAULT_CITY_UUID = "be9da3ba-a0e4-42f9-9ae2-9875c8625083" # SPB
user_events_rt = Router()
class UserEventEnterCodeState(StatesGroup):
wait_code = State()
@bot.di
def events_menu_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_city_events_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="events",
context=context,
)
@bot.di
def user_events_page_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_city_events_page(
event.peer,
offset=offset,
event_service=event_service,
ctx_name="events",
context=context,
)
@bot.di
def user_events_open_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 "events"
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 event_user_sign_up_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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 "events"
ok = event_service.sign_up(event_id, event.peer.id)
note = (
"✅ Запись на мероприятие прошла успешно!"
if ok
else "⚠️ Не удалось записаться на мероприятие."
)
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_user_sign_out_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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 "events"
status = event_service.get_participation_status(event_id, event.peer.id)
if not status:
note = "⚠️ Не удалось отписаться: ты не записан на это мероприятие."
elif status.get("is_claimed"):
note = "⚠️ Нельзя отписаться: баллы уже получены за это мероприятие."
else:
ok = event_service.sign_out_by_event(event_id, event.peer.id)
note = "✅ Ты успешно отписался от мероприятия." if ok else "⚠️ Не удалось отписаться."
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_user_enter_code_start_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
delete_prev_message_by_peer(bot, event.peer)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "events"
context.update_data({"current_event_id": event_id, "current_event_ctx": ctx_name})
context.set_state(UserEventEnterCodeState.wait_code)
bot.messaging.send_message(
peer=event.peer,
text="Введи бонус-код мероприятия (без пробелов):",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
@user_events_rt.message(state=UserEventEnterCodeState.wait_code)
@bot.di
def event_user_enter_code_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
delete_prev_message(bot, message)
data = context.get_data()
event_id = data.get("current_event_id")
ctx_name = data.get("current_event_ctx") or "events"
code = message.message.text_message.text.strip()
note: str
status = event_service.get_participation_status(event_id, message.peer.id)
if not status:
note = "⚠️ Код не принят: сначала нужно записаться на мероприятие."
elif status.get("is_claimed"):
note = "⚠️ Код не принят: баллы уже получены."
else:
result = event_service.enter_code_by_event(event_id, message.peer.id, code)
if not result:
note = "⚠️ Код не принят: проверь правильность кода."
else:
note = f"✅ Код принят. Текущие баллы: {result.get('points', '-')}"
context.set_state(None)
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_participants_open_handler(
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)
ctx_name = ctx or "events"
items = event_service.get_event_participants(event_id, event.peer.id, offset=0, limit=200)
if not items:
text = "👥 *Участники*\n\nНе удалось найти волонтёров, записанных на это мероприятие."
else:
lines = ["👥 *Участники*", ""]
for i, it in enumerate(items, start=1):
name = (it.get("firstname") or "—").strip()
claimed = bool(it.get("is_claimed"))
status = "☑️ Получил баллы" if claimed else "📝 Записан"
points = it.get("points", "—")
lines.append(f"{i}. {name} — {status} — баллы: {points}")
text = "\n".join(lines)
bot.messaging.send_message(
peer=event.peer,
text=text,
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
--- src/bot/core/handlers/points.py ---
from datetime import datetime
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent
from core.bot_kit.fsm import FSMContext
from core.bot_kit.router import Router
from core.config import bot
from core.handlers.events_ui import send_event_card, send_points_page
from core.schemas.event import EventSchema
from core.schemas.user import UserSchema
from core.services.event import EventService
from core.utils import delete_prev_message_by_peer
from core.utils.events_filter import clear_context_keep_events_filters
DEFAULT_CITY_UUID = "be9da3ba-a0e4-42f9-9ae2-9875c8625083" # SPB
# TODO уйти от хардкода, брать город пользователя
points_rt = Router()
def _format_event_line(event: EventSchema) -> str:
date_val = getattr(event, "date", None)
if isinstance(date_val, datetime):
date_str = date_val.strftime("%d.%m.%Y %H:%M")
else:
# на случай, если backend отдаст строку/иное
date_str = str(date_val)
name = getattr(event, "name", "-")
location = getattr(event, "location", "-")
event_id = getattr(event, "id", "-")
return f" - {name} --- {date_str} --- ({location}) --- `{event_id}`"
def _safe_int(value, default: int = 0) -> int:
try:
if value is None:
return default
return int(value)
except Exception:
return default
@bot.di
def points_menu_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_points_page(
event.peer,
offset=0,
event_service=event_service,
context=context,
)
@bot.di
def points_page_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_points_page(
event.peer,
offset=offset,
event_service=event_service,
context=context,
)
@bot.di
def points_event_open_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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="points"
)
--- src/bot/core/handlers/moderation.py ---
import re
from datetime import datetime
from uuid import UUID
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_city_events_page,
send_event_card,
send_my_events_page,
split_event_value,
)
from core.markups.event import event_actions_keyboard, format_event_details
from core.markups.tags import choices_from_backend_tags, tags_toggle_keyboard
from core.schemas.event import EventSchema
from core.schemas.user import UserSchema
from core.services.event import EventService
from core.services.report import ReportService
from core.services.tag import TagService
from core.utils import delete_prev_message, delete_prev_message_by_peer
from core.utils.events_filter import clear_context_keep_events_filters
from ..markups import (
back_to_moderation_keyboard,
event_edit_fields_keyboard,
moderation_menu_keyboard,
)
TAGS_MEDIA_ID = "event_tags_toggle"
TAGS_TOGGLE_PREFIX = "toggle:"
TAGS_DONE_VALUE = "done"
DEFAULT_CITY_UUID = "be9da3ba-a0e4-42f9-9ae2-9875c8625083" # SPB
# TODO убрать хардкод
events_rt = Router()
class EventEditTagsState(StatesGroup):
tags = State()
class EventEditState(StatesGroup):
wait_value = State()
class EventEnterCodeState(StatesGroup):
wait_code = State()
class EventViewState(StatesGroup):
wait_event_id = State()
class EventCreateState(StatesGroup):
name = State()
date = State()
time = State()
location = State()
organizer = State()
description = State()
points = State()
tags = State()
code = State()
@bot.di
def moderation_menu_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
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(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
"""
показывает модератору список всех мероприятий в городе (пока один город)
привязан к кнопке value="all_events" в moderation_menu_keyboard()
реализована пагинация, в планах фильтры и интерактивная кнопка на каждое мероприятие
дефолтный фильтр: ваш город, планируется возможность изменения дефолтного фильтра
"""
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_city_events_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def my_events_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
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,
context=context,
)
@bot.di
def my_events_page_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
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,
context=context,
)
@bot.di
def my_events_open_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 = 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(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
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_city_events_page(
event.peer,
offset=offset,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def all_events_open_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"
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/9: введи, пожалуйста, название мероприятия:"),
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/9: введи, пожалуйста, дату мероприятия (dd.mm.yyyy)\nНапример: `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()
try:
date_obj = datetime.strptime(date, "%d.%m.%Y").date()
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Я не смог распознать дату.\n"
"Пожалуйста, введи в формате `dd.mm.yyyy`, например: `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/9: введи время мероприятия в формате `HH:MM`\nНапример: `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):
delete_prev_message(bot, message)
time = message.message.text_message.text.strip()
try:
time_obj = datetime.strptime(time, "%H:%M").time()
except ValueError:
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})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 4/9: Введи локацию:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.location)
@events_rt.message(state=EventCreateState.location)
@bot.di
def event_create_location_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
location = message.message.text_message.text.strip()
context.update_data({"event_location": location})
bot.messaging.send_message(
peer=message.peer,
text='Шаг 5/9: Введи UUID организатора (введи минус "-", чтобы поставить организатором себя):',
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.organizer)
@events_rt.message(state=EventCreateState.organizer)
@bot.di
def event_create_organizer_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
raw = message.message.text_message.text.strip()
if raw != "-":
try:
UUID(raw)
except Exception:
bot.messaging.send_message(
peer=message.peer,
text=(
"Организатор должен быть UUID пользователя или `-`.\n"
"Отправь `-`, чтобы поставить себя организатором, или напиши UUID.\n"
"пример UUID: `d3ac9bc9-19ae-4ded-9687-cbde5ea14eb0`"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"event_organizer": raw})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 6/9: Введи описание:",
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="Шаг 7/9: Введи количество очков начисляемых за мероприятие:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.points)
@events_rt.message(state=EventCreateState.points)
@bot.di
def event_create_points_step(message: UpdateMessage, context: FSMContext, tag_service: TagService):
delete_prev_message(bot, message)
text = message.message.text_message.text.strip()
try:
points = int(text)
if 0 >= points or points >= 2**31:
raise ValueError
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text=f"Количество баллов должно быть натуральным числом, меньшим {2**31}. Попробуй ещё раз.",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
all_tags = tag_service.get_all_tags()
choices = choices_from_backend_tags(all_tags)
context.update_data(
{
"event_points": points,
"event_tag_keys": [],
"event_all_tags": [{"id": t.id, "name": t.name} for t in all_tags],
}
)
_send_tags_toggle_ui(
peer=message.peer,
title="Шаг 8/9: Выбери теги мероприятий, затем нажми «Готово»:",
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,
context: FSMContext,
tag_service: TagService,
event_service: EventService,
):
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()
# выбранные имена (keys)
key_field = "event_tag_keys" if is_create else "edit_tag_keys"
selected = set(data.get(key_field) or [])
# теги backend (id+name)
all_tags = data.get("event_all_tags") or data.get("edit_all_tags")
if not all_tags:
# если по какой-то причине нет в контексте, догрузим
tags = tag_service.get_all_tags()
all_tags = [{"id": t.id, "name": t.name} for t in tags]
else:
norm = []
for t in all_tags:
if isinstance(t, dict):
norm.append(t)
else:
norm.append({"id": getattr(t, "id", None), "name": getattr(t, "name", None)})
all_tags = norm
# общий список choices
choices = choices_from_backend_tags(all_tags)
# DONE
if raw == TAGS_DONE_VALUE:
name_to_id = {t["name"]: t["id"] for t in all_tags if t.get("name") and t.get("id")}
app_tag_ids = [name_to_id[n] for n in selected if n in name_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=(
"Шаг 9/9: введи код мероприятия:\n"
"Без пробелов, не больше 16 символов. Например: WELCOME2025"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.code)
return
# EDIT TAGS
event_id = data.get("edit_event_id")
if not event_id:
context.set_state(None)
bot.messaging.send_message(
peer=event.peer, text="Сессия редактирования сброшена. Открой мероприятие заново."
)
return
updated = event_service.update_event(event_id, {"app_tag_ids": app_tag_ids})
context.set_state(None)
if not updated:
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Не удалось обновить теги.",
event_obj=event_service.get_event_by_id(event_id),
show_organizer=True,
)
return
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
note="✅ Теги обновлены.",
event_obj=event_service.get_event_by_id(event_id),
show_organizer=True,
)
return
# TOGGLE
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 = "Шаг 8/9: Выбери теги мероприятия, затем нажми «Готово»:"
else:
# для редактирования возвращаемся в edit menu мероприятия
event_id = data.get("edit_event_id") or ""
footer = [
Button(
media_id="event_menu_edit",
value=event_id,
label="⬅️ К редактированию мероприятия",
)
]
title = "Редактирование тегов: выбери теги, затем нажми «Готово»:"
_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,
user: UserSchema | None,
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_bonus_code": code})
data = context.get_data()
try:
created = event_service.create_from_wizard(user, data)
except Exception as e:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Не удалось создать мероприятие.",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.clear()
raise e
return
context.clear()
send_event_card(
message.peer,
event_id=created.id,
event_service=event_service,
user=user,
ctx_name="moderation_menu",
note="Мероприятие создано успешно",
)
@bot.di
def event_view_by_id_start(event: UpdateInteractiveMediaEvent, context: FSMContext):
"""старт диалога - просим у модератора 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(
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)
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",
)
def _update_event_card(
peer,
event_service: EventService,
event_id: str,
can_manage: bool,
show_organizer: bool,
): # deprecated?
refreshed = event_service.get_event_by_id(event_id)
part = event_service.get_participation_status(event_id, peer.id)
is_participant = part is not None
prev = bot.messaging.load_message_history_sync(peer=peer, limit=1)[0]
bot.messaging.update_message_sync(
prev,
text=format_event_details(refreshed, show_code=True, show_organizer=show_organizer)
if refreshed
else "⚠️ Мероприятие не найдено",
interactive_media_groups=event_actions_keyboard(
event_id,
is_participant=is_participant,
can_manage=can_manage,
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",
ctx="moderation_menu",
),
)
@bot.di
def event_menu_sign_up_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_sign_out_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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"
status = event_service.get_participation_status(event_id, event.peer.id)
if not status:
note = "⚠️ Не удалось отписаться: ты не записан на это мероприятие."
elif status.get("is_claimed"):
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,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_enter_code_start_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
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, user: UserSchema
):
delete_prev_message(bot, message)
event_id = context.get_data().get("current_event_id")
code = message.message.text_message.text.strip()
note = None
status = event_service.get_participation_status(event_id, message.peer.id)
if not status:
note = "⚠️ Код не принят, сначала нужно записаться на мероприятие"
elif status.get("is_claimed"):
note = "⚠️ Код не принят, баллы уже получены"
else:
result = event_service.enter_code_by_event(event_id, message.peer.id, code)
if not result:
note = "⚠️ Код не принят, проверь правильность кода"
else:
note = f"✅ Код принят. Текущие баллы: {result.get('points', '-')}"
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,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_delete_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
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)
except Exception:
ok = False
if ok:
context.clear()
if ctx_name == "moderation":
send_city_events_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,
user=user,
ctx_name=ctx_name,
note="⚠️ Не удалось удалить мероприятие",
)
def _send_event_edit_menu(
peer,
*,
event_id: str,
event_service: EventService,
note: str | None = None,
event_obj: EventSchema | None = None,
show_organizer: bool,
):
refreshed = event_obj or event_service.get_event_by_id(event_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(
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:
return {
"name": "Название",
"date": "Дата",
"time": "Время",
"location": "Локация",
"tags": "Теги",
"creator_id": "Организатор",
"points": "Баллы",
"description": "Описание",
"bonus_code": "Код",
}.get(field, field)
def _get_current_value(event: EventSchema, field: str) -> str:
val = getattr(event, field, None)
if val is None or val == "":
return "-"
if field == "date":
try:
return event.date.strftime("%d.%m.%Y")
except Exception:
return str(val)
if field == "time":
try:
return event.date.strftime("%H:%M")
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(
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:
bot.messaging.send_message(peer=event.peer, text="⚠️ Некорректная команда редактирования.")
return
event_id, field = raw.split("|", 1)
field = field.strip()
refreshed = event_service.get_event_by_id(event_id)
if not refreshed:
bot.messaging.send_message(peer=event.peer, text="⚠️ Мероприятие не найдено.")
return
if field == "tags":
# редактируем теги через toggle UI
all_tags = tag_service.get_all_tags()
all_tags_dump = [{"id": t.id, "name": t.name} for t in all_tags]
choices = choices_from_backend_tags(all_tags)
selected_names: set[str] = set()
for t in refreshed.tags or []:
if isinstance(t, dict) and t.get("name"):
selected_names.add(str(t["name"]))
else:
name = getattr(t, "name", None)
if name:
selected_names.add(str(name))
context.update_data(
{
"edit_event_id": event_id,
"edit_all_tags": all_tags_dump,
"edit_tag_keys": list(selected_names),
}
)
context.set_state(EventEditTagsState.tags)
_send_tags_toggle_ui(
peer=event.peer,
title="Редактирование тегов: выбери теги, затем нажми «Готово»:",
choices=choices,
selected_keys=set(selected_names),
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, field)
if field == "creator_id":
hint = 'Введи UUID организатора или отправь минус "-", чтобы поставить организатором себя.'
elif 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(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
user: UserSchema | None,
):
delete_prev_message(bot, message)
data = context.get_data()
event_id = data.get("edit_event_id")
field = 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: object = None if raw == "-" else raw
if field == "creator_id":
if raw == "-":
if not user.id:
bot.messaging.send_message(
peer=message.peer, text="⚠️ Не удалось определить ваш user.id."
)
return
value = user.id
else:
try:
value = str(UUID(raw))
except Exception:
bot.messaging.send_message(
peer=message.peer,
text=(
'Организатор должен быть UUID пользователя или "-". Попробуй ещё раз.\n'
"Пример UUID: `d0a794d0-a138-47b7-b5f3-da53d19bfb07`"
),
)
return
if field == "points":
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 == "bonus_code" and value is not None:
if 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:
try:
value = datetime.strptime(value, "%d.%m.%Y").date()
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Я не смог распознать дату.\n"
"Пожалуйста, введи в формате `dd.mm.yyyy`, например: `20.02.2002`:"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
if field == "time" and value is not None:
try:
value = datetime.strptime(value, "%H:%M").time()
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text=(
"⚠️ Я не смог распознать время.\n"
"Пожалуйста, введи в формате `HH:MM`, например: `10:30`:"
),
interactive_media_groups=back_to_moderation_keyboard(),
)
return
if field in ("date", "time"):
current = event_service.get_event_by_id(event_id)
if not current:
context.set_state(None)
bot.messaging.send_message(peer=message.peer, text="⚠️ Мероприятие не найдено.")
return
current_dt = current.date
tz = getattr(current_dt, "tzinfo", None)
if value is None:
bot.messaging.send_message(
peer=message.peer, text="⚠️ Дату/время нельзя очистить. Выбери другое значение."
)
return
if field == "date":
new_dt = datetime.combine(value, current_dt.time())
else:
new_dt = datetime.combine(current_dt.date(), value)
if tz is not None:
new_dt = new_dt.replace(tzinfo=tz)
iso = new_dt.isoformat()
if iso.endswith("+00:00"):
iso = iso.replace("+00:00", "Z")
payload = {"date": iso}
else:
payload = {field: value}
updated = event_service.update_event(event_id, payload)
context.set_state(None)
if not updated:
_send_event_edit_menu(
message.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Не удалось обновить поле.",
event_obj=event_service.get_event_by_id(event_id),
show_organizer=True,
)
return
_send_event_edit_menu(
message.peer,
event_id=event_id,
event_service=event_service,
note="✅ Поле обновлено.",
event_obj=event_service.get_event_by_id(event_id),
show_organizer=True,
)
--- src/bot/core/handlers/events_ui.py ---
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import 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.event import (
event_actions_keyboard,
events_select_keyboard,
format_event_details,
my_events_pagination_keyboard,
user_events_pagination_keyboard,
)
from core.markups.events_filters import events_filters_button_keyboard
from core.markups.inline import all_events_pagination_keyboard
from core.markups.points import points_events_pagination_keyboard, points_menu_keyboard
from core.schemas.event import EventSchema
from core.schemas.user import UserSchema
from core.services.event import EventService
from core.services.report import ReportService
from core.utils.events_filter import format_events_filter_summary, get_events_filter
DEFAULT_CITY_UUID = "be9da3ba-a0e4-42f9-9ae2-9875c8625083" # SPB
ADMIN_CAN_MANAGE_OTHERS = True # имеет ли администратор доступ ко всем мероприятиям?
@dataclass(frozen=True)
class EventUIContext:
name: str
# list
list_open_media_id: str
pagination_builder: Callable[..., list[InteractiveMediaGroup]]
extra_groups_builder: Callable[[], list[InteractiveMediaGroup]] | None
# card
back_value: str
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",
),
# дом волонтёра -> посмотреть по 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",
),
}
def get_ui(ctx_name: str) -> EventUIContext:
return _UI.get(ctx_name, _UI["events"])
def build_back_keyboard(ctx_name: str) -> list[InteractiveMediaGroup]:
ui = get_ui(ctx_name)
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]:
"""
ожидаем:
- "<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(ev: EventSchema, user: UserSchema | None) -> bool:
if getattr(user, "role", None) == "admin" and ADMIN_CAN_MANAGE_OTHERS:
return True
uid = getattr(user, "id", None)
if not uid:
return False
return bool(getattr(ev, "creator_id", None)) and (str(ev.creator_id) == str(uid))
def can_view_code(ev: EventSchema, user: UserSchema | None) -> bool:
# кто может управлять - тот видит код
return can_manage_event(ev, user)
def _send_events_list_screen(
peer,
*,
ui: EventUIContext,
ctx_name: str,
offset: int,
limit: int,
page_events: list[EventSchema],
total: int | None,
has_prev: bool,
has_next: bool,
note: str | None = None,
header: str | None = None,
empty_text: str | None = None,
empty_groups: list[InteractiveMediaGroup] | None = None,
filter_summary: str | None = None,
):
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 page_events:
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
total_str = str(total) if total is not None else "?"
text = (
prefix
+ header_part
+ f"Показаны: {offset + 1} — {offset + len(page_events)} из {total_str}\n\n"
+ "Выбери мероприятие:"
+ filter_part
)
im: list[InteractiveMediaGroup] = []
im += events_select_keyboard(
page_events, 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_city_events_page(
peer,
*,
offset: int,
event_service: EventService,
ctx_name: str,
city_id: str = DEFAULT_CITY_UUID,
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_events, has_prev, has_next, total = fetch_events_page(
event_service=event_service,
city_id=city_id,
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_events=page_events,
total=total,
has_prev=has_prev,
has_next=has_next,
note=note,
filter_summary=summary,
)
def send_points_page(
peer,
*,
offset: int,
event_service: EventService,
context: FSMContext | None = None,
):
limit = EVENTS_PAGE_SIZE
points_total = "?"
flt = get_events_filter(context, ctx_name="points") if context is not None else None
summary = format_events_filter_summary(flt) if context is not None else None
page_events, has_prev, has_next, total = fetch_events_page(
event_service=event_service,
city_id=DEFAULT_CITY_UUID,
offset=offset,
limit=limit,
requester_messenger_id=peer.id,
only_claimed=True,
filters=flt,
)
_send_events_list_screen(
peer,
ui=get_ui("points"),
ctx_name="points",
offset=offset,
limit=limit,
page_events=page_events,
total=total,
has_prev=has_prev,
has_next=has_next,
header=f"🏆 Твои баллы: {points_total}\n\n",
empty_text=(
"Пока нет мероприятий, за которые уже начислены баллы.\n\n"
"Как получить баллы:\n"
"1) Запишитесь на мероприятие\n"
"2) После участия введи код\n"
),
empty_groups=points_menu_keyboard(),
filter_summary=summary,
)
def send_my_events_page(
peer,
*,
offset: int,
event_service: EventService,
requester_messenger_id: int,
city_id: str = DEFAULT_CITY_UUID,
note: str | None = None,
context: FSMContext | None = None,
):
ui = get_ui("my_events")
limit = EVENTS_PAGE_SIZE
flt = get_events_filter(context, ctx_name="my_events") if context is not None else None
summary = format_events_filter_summary(flt) if context is not None else None
page_events, has_prev, has_next, total = fetch_events_page(
event_service=event_service,
city_id=city_id,
offset=offset,
limit=limit,
requester_messenger_id=requester_messenger_id,
only_organizer=True,
filters=flt,
lock_mode=True,
)
_send_events_list_screen(
peer,
ui=ui,
ctx_name="my_events",
offset=offset,
limit=limit,
page_events=page_events,
total=total,
has_prev=has_prev,
has_next=has_next,
note=note,
header="📋 Мои мероприятия\n\n",
empty_text="Пока у тебя нет созданных мероприятий.",
filter_summary=summary,
)
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)
ev = event_service.get_event_by_id(event_id)
if ev 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
part = event_service.get_participation_status(ev.id, peer.id)
is_participant = part is not None
can_manage = can_manage_event(ev, user)
show_code = can_view_code(ev, user)
role = getattr(user, "role", None) if user else None
can_access_reports = (role in ("admin", "moderator")) or 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(ev.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))
base = format_event_details(
ev,
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(
ev.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(event_id: str, ctx_name: str) -> list[InteractiveMediaGroup]:
ui = get_ui(ctx_name)
return MediaGroupBuilder(
[
Button(
media_id=ui.list_open_media_id,
value=f"{event_id}|{ui.name}",
label="⬅️ К мероприятию",
)
]
).build()
--- src/bot/core/services/event.py ---
from datetime import datetime, timezone
from typing import Any
from uuid import UUID
from core.clients.event import EventClient
from core.schemas.event import EventSchema
from core.schemas.user import UserSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
from core.utils.logger import logger
DEFAULT_CITY_UUID = "be9da3ba-a0e4-42f9-9ae2-9875c8625083" # SPB
# TODO убрать хардкод, брать city_id из профиля пользователя
class EventService(BaseService):
"""
сервис мероприятий на стороне бота.
на данный момент:
- ходит в backend через EventClient
- умеет получать список мероприятий по городу
- умеет получать одно мероприятие по id
- создает мероприятие через клиент используя мастера создания
"""
def __init__(self, registry: BaseServiceRegistry, client: EventClient):
# DI может вызвать клиент без аргументов.
super().__init__(registry)
self.client = client
def get_event_participants(
self,
event_id: str,
requester_messenger_id: int,
offset: int = 0,
limit: int = 50,
) -> list[dict[str, Any]]:
"""
возвращает список EventSchema для города
пока что используем дефолтный город - спб
"""
resp = self.client.get_event_participants(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
offset=offset,
limit=limit,
)
if resp.status_code != 200:
raise Exception(
f"Error while getting event participants: {resp.status_code}\n{resp.text}"
)
body = resp.json()
return body if isinstance(body, list) else []
def get_events(self, city_id: str, offset: int = 0, limit: int = 50) -> list[EventSchema]:
"""
возвращает список EventSchema для города
пока что используем дефолтный город - спб
"""
resp = self.client.get_city_events(city_id=city_id, offset=offset, limit=limit)
if resp.status_code != 200:
raise Exception(f"Error while getting events: {resp.status_code}\n{resp.text}")
body = resp.json()
return [self._map_event(item) for item in body]
def _parse_total_count(self, resp) -> int | None:
try:
raw = None
headers = getattr(resp, "headers", None)
if headers is not None:
raw = headers.get("X-Total-Count")
if raw is None:
return None
return int(raw)
except Exception:
logger.warning("EventService failed to get total events count from header.")
return None
def search_events(
self,
*,
city_id: str | None = None,
offset: int = 0,
limit: int = 50,
requester_messenger_id: int | None = None,
only_claimed: bool = False,
only_organizer: bool = False,
only_participant: bool = False,
has_report: bool | None = None,
tag_ids: list[str] | None = None,
cursor_date: str | None = None,
cursor_id: str | None = None,
) -> list[EventSchema]:
resp = self.client.search_events(
city_id=city_id,
offset=offset,
limit=limit,
requester_messenger_id=requester_messenger_id,
only_claimed=only_claimed,
only_organizer=only_organizer,
only_participant=only_participant,
has_report=has_report,
tag_ids=tag_ids,
cursor_date=cursor_date,
cursor_id=cursor_id,
)
if resp.status_code != 200:
raise Exception(f"Error while searching events: {resp.status_code}\n{resp.text}")
body = resp.json()
return [self._map_event(item) for item in (body or [])]
def search_events_with_total(
self,
*,
city_id: str | None = None,
offset: int = 0,
limit: int = 0,
requester_messenger_id: int | None = None,
only_claimed: bool = False,
only_organizer: bool = False,
only_participant: bool = False,
has_report: bool | None = None,
tag_ids: list[str] | None = None,
cursor_date: str | None = None,
cursor_id: str | None = None,
) -> tuple[list[EventSchema], int | None]:
resp = self.client.search_events(
city_id=city_id,
offset=offset,
limit=limit,
requester_messenger_id=requester_messenger_id,
only_claimed=only_claimed,
only_organizer=only_organizer,
only_participant=only_participant,
has_report=has_report,
tag_ids=tag_ids,
cursor_date=cursor_date,
cursor_id=cursor_id,
)
if resp.status_code != 200:
raise Exception(f"Error while searching events: {resp.status_code}\n{resp.text}")
body = resp.json()
items = [self._map_event(item) for item in (body or [])]
total = self._parse_total_count(resp)
return items, total
def get_claimed_events_page(
self,
*,
messenger_id: int,
city_id: str | None,
offset: int,
limit: int,
) -> tuple[list[EventSchema], int | None]:
# requester_messenger_id обязателен для only_[claimed|organizer|participant] на backend
return self.search_events_with_total(
city_id=city_id,
offset=offset,
limit=limit,
requester_messenger_id=messenger_id,
only_claimed=True,
)
def get_event_by_id(self, id: str) -> EventSchema | None:
resp = self.client.get_event_info(id)
logger.info(
"[EventService::get_event_by_id] event_id=%r status=%s url=%r method=%r body=%r",
id,
resp.status_code,
getattr(getattr(resp, "request", None), "url", None),
getattr(getattr(resp, "request", None), "method", None),
resp.text,
)
if resp.status_code == 404:
return None
if resp.status_code != 200:
# TODO сделать нормальную обработку
logger.error(f"[EventService] get_events failed: {resp.status_code} {resp.text}")
return None
return self._map_event(resp.json())
def sign_up(self, event_id: str, messenger_id: int) -> bool:
"""messenger_id= = peer.id (как на бэкенде: user.get_by_messenger_id)"""
resp = self.client.sign_up(event_id=event_id, messenger_id=messenger_id)
return resp.status_code == 200
def delete_event(self, event_id: str) -> bool:
resp = self.client.delete_event(event_id)
return resp.status_code == 204 # in [200, 204]
def update_event(self, event_id: str, payload: dict) -> EventSchema | None:
resp = self.client.update_event(event_id, payload)
if resp.status_code != 200:
logger.error(f"[EventService] update_event failed: {resp.status_code}\n{resp.text}")
return None
return self._map_event(resp.json())
def _get_user_participations_raw(self, messenger_id: int) -> list[dict]:
"""
GET /events/my?user_id=<messenger_id>
на бекэнд возвращается список participation (без response_model),
поэтому на стороне бота парсим максимально терпимо
"""
resp = self.client.get_my_events(messenger_id=messenger_id)
if resp.status_code != 200:
raise Exception(f"Error while getting user events: {resp.status_code}\n{resp.text}")
body = resp.json()
# ожидаем list[dict]
if isinstance(body, list):
return body
return []
def _find_participation_for_event(
self, messenger_id: int, event_id: str
) -> tuple[str | None, bool | None]:
"""
ищем participation_id для конкретного event_id
возвращаем (participation_id, is claimed)
"""
items = self._get_user_participations_raw(messenger_id)
target_event_id = str(event_id)
for item in items:
if not isinstance(item, dict):
continue
pid = item.get("id")
ev_id = item.get("event_id")
is_claimed = item.get("is_claimed")
# если event вложен
if ev_id is None and isinstance(item.get("event"), dict):
ev_id = item["event"].get("id")
if ev_id is None:
continue
if str(ev_id) == target_event_id:
return (
str(pid) if pid else None,
bool(is_claimed) if is_claimed is not None else None,
)
return (None, None)
def sign_out_by_event(self, event_id: str, messenger_id: int) -> bool:
"""
backend sign-out requires participation_id, so:
1. GET /event/my
2. находим participation_id для event_id
3. DELETE /events/sign-out
"""
participation_id, _ = self._find_participation_for_event(messenger_id, event_id)
if not participation_id:
return False
resp = self.client.sign_out(participation_id=participation_id)
return resp.status_code == 200
def enter_code_by_event(
self, event_id: str, messenger_id: int, bonus_code: str
) -> dict[str, Any] | None:
"""
аналогично sign-out: нужен participation_id
возвращает обновленного пользователя (ResponseUserSchema) на успехе
"""
participation_id, _ = self._find_participation_for_event(messenger_id, event_id)
if not participation_id:
return None
resp = self.client.enter_code(participation_id=participation_id, bonus_code=bonus_code)
if resp.status_code != 200:
return None
return resp.json()
def get_participation_status(self, event_id: str, messenger_id: int) -> dict[str, Any] | None:
participation_id, is_claimed = self._find_participation_for_event(messenger_id, event_id)
if not participation_id:
return None
return {
"participation_id": participation_id,
"is_claimed": bool(is_claimed) if is_claimed is not None else False,
}
def enter_code(self, user_id: int, code: str):
# TODO: replace with client usage
return code in ["code", "код"]
def create_event(
self,
*,
name: str,
description: str,
bonus_code: str,
points: int,
date_obj: datetime,
city_id: str,
creator_id: str | None = None,
):
"""
создает мероприятие через backend и возвращает dict из backend'a
- ожидает, что все данные уже провалидированы и разобраны
- метод не мапит ответ backend'a в EventSchema и не делает никакой обработки, а возвращает ровно тот json, который вернул backend (dect)
- может бросить Exception, если status_code не 201/200
"""
date_iso = date_obj.isoformat()
resp = self.client.create_event(
name=name,
description=description,
bonus_code=bonus_code,
points=points,
date=date_iso,
city_id=city_id,
creator_id=creator_id,
)
if resp.status_code not in [200, 201]:
raise Exception(f"Error while creating event: {resp.status_code}\n{resp.text}")
return resp.json()
def create_from_wizard(self, user: UserSchema | None, data: dict[str, Any]) -> dict[str, Any]:
"""
обрабатываем данные полученный из мастера создания,
посылаем их в бекенд.
"""
name = data.get("event_name")
date_obj = data.get("event_date")
time_obj = data.get("event_time")
location = data.get("event_location")
organizer_raw = data.get("event_organizer") or ""
description = data.get("event_description") or ""
points = data.get("event_points")
app_tag_ids = data.get("event_app_tag_ids") or []
bonus_code = data.get("event_bonus_code")
event_dt = datetime.combine(date_obj, time_obj).replace(tzinfo=timezone.utc)
date = event_dt.isoformat().replace("+00:00", "Z")
creator_id: str | None = None
if organizer_raw == "-":
if not user.id:
raise Exception("Cannot use '-' for organizer, current user has no id")
creator_id = user.id
else:
try:
creator_id = str(UUID(organizer_raw))
except Exception as e:
raise ValueError(f"Invalid organizer UUID: {organizer_raw!r}") from e
resp = self.client.create_event(
name=name,
description=description,
bonus_code=bonus_code,
points=points,
date=date,
city_id=DEFAULT_CITY_UUID,
location=location,
app_tag_ids=app_tag_ids,
creator_id=creator_id,
)
if resp.status_code not in [200, 201]:
raise Exception(f"Error while creating event: {resp.status_code}\n{resp.text}")
return self._map_event(resp.json())
def _map_event(self, body: dict) -> EventSchema:
"""преобразует json из backend'a в EventSchema бота"""
date_raw = body.get("date")
date = self._parse_iso_datetime(date_raw)
return EventSchema(
id=body.get("id"),
name=body.get("name", ""),
date=date,
description=body.get("description", "") or "",
location=body.get("location"),
tags=body.get("tags"),
bonus_code=body.get("bonus_code"),
points=body.get("points"),
city_id=body.get("city_id"),
creator_id=body.get("creator_id"),
creator_name=body.get("creator_name"),
)
def _parse_iso_datetime(self, value: str | None) -> datetime:
if not value:
return datetime.now()
if value.endswith("Z"):
value = value[:-1] + "+00:00"
try:
return datetime.fromisoformat(value)
except Exception as e:
logger.error(f"[EventService] failed to parse date {value!r}, returing now(): {e}")
return datetime.now()
Editor is loading...
Leave a Comment