Untitled
4ae4d
plain_text
02/10/2026 1:47 PM
58.4 KB
10
Indexable
f=src/bot/core/handlers/events_filters.py
f=src/bot/core/handlers/events_ui.py
--- src/bot/core/handlers/events_filters.py ---
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent
from core.utils.events_filter import EventsMode
from core.bot_kit.fsm import FSMContext, State, StatesGroup
from core.config import bot
from core.handlers.events_ui import send_city_events_page, send_my_events_page, send_points_page
from core.markups.events_filters import (
events_filters_city_keyboard,
events_filters_menu_keyboard,
events_filters_mode_keyboard,
)
from core.markups.tags import TagChoice
from core.services.city import CityService
from core.services.event import EventService
from core.services.tag import TagService
from core.utils import delete_prev_message_by_peer
from core.utils.events_filter import (
format_events_filter_summary,
get_events_filter,
reset_events_filter,
set_events_filter,
)
from core.utils.tags_toggle_ui import TagsToggleCfg, is_done, parse_toggle_key, send_tags_toggle_ui
class EventsFiltersTagsState(StatesGroup):
tags = State()
class EventsFiltersCityState(StatesGroup):
city = State()
class EventsFiltersModeState(StatesGroup):
mode = State()
FILTERS_TAGS_CFG = TagsToggleCfg(
media_id="events_filters_tags_toggle",
toggle_prefix="toggle:",
done_value="done",
)
_CTX_KEY = "events_filters_edit_ctx"
_OFFSET_KEY = "events_filters_edit_offset"
_TAG_IDS_KEY = "events_filters_edit_tags_selected"
_CITIES_OFFSET_KEY = "events_filters_city_offset"
_CITY_SELECTED_ID_KEY = "events_filters_city_selected_id"
_CITY_SELECTED_NAME_KEY = "events_filters_city_selected_name"
_CITIES_PAGE_LIMIT = 10
_MODE_KEY = "events_filters_mode_selected"
_MODE_LOCKED_KEY = "events_filters_mode_locked"
def _locked_mode_for_ctx(ctx: str) -> str | None:
if ctx == "my_events":
return "organizer"
if ctx in ("part_events", "points"):
return "participant"
return None
def _split_ctx_offset(raw: object) -> tuple[str, int]:
s = str(raw or "").strip()
if "|" not in s:
return (s or "events", 0)
a, b = s.split("|", 1)
ctx = (a or "events").strip()
try:
off = int(str(b).strip())
if off < 0:
off = 0
except Exception:
off = 0
return ctx, off
def _send_list(
*, peer, ctx_name: str, offset: int, context: FSMContext, event_service: EventService
):
# apply/back используют эту маршрутизацию
if ctx_name == "points":
return send_points_page(peer, offset=offset, event_service=event_service, context=context)
if ctx_name == "my_events":
return send_my_events_page(
peer,
offset=offset,
event_service=event_service,
requester_messenger_id=peer.id,
context=context,
)
# events / moderation
return send_city_events_page(
peer,
offset=offset,
event_service=event_service,
ctx_name=ctx_name,
context=context,
)
@bot.di
def events_filters_open_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, offset = _split_ctx_offset(event.data.value)
flt = get_events_filter(context, ctx_name=ctx_name)
summary = format_events_filter_summary(flt)
locked_mode = _locked_mode_for_ctx(ctx_name)
locked_note = ""
if locked_mode == "participant":
locked_note = "\n\n🔒 Режим зафиксирован: *участник*"
elif locked_mode == "organizer":
locked_note = "\n\n🔒 Режим зафиксирован: *организатор*"
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}{locked_note}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=ctx_name, offset=offset),
)
@bot.di
def events_filters_reset_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, offset = _split_ctx_offset(event.data.value)
reset_events_filter(context, ctx_name=ctx_name)
# остаёмся в меню, чтобы пользователь видел что “сбросилось”
flt = get_events_filter(context, ctx_name=ctx_name)
summary = format_events_filter_summary(flt)
locked_mode = _locked_mode_for_ctx(ctx_name)
locked_note = ""
if locked_mode == "participant":
locked_note = "\n\n🔒 Режим зафиксирован: *участник*"
elif locked_mode == "organizer":
locked_note = "\n\n🔒 Режим зафиксирован: *организатор*"
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}{locked_note}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=ctx_name, offset=offset),
)
@bot.di
def events_filters_apply_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, _offset = _split_ctx_offset(event.data.value)
# безопаснее возвращать на первую страницу (фильтр мог сузить выдачу)
_send_list(
peer=event.peer, ctx_name=ctx_name, offset=0, context=context, event_service=event_service
)
@bot.di
def events_filters_back_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, offset = _split_ctx_offset(event.data.value)
_send_list(
peer=event.peer,
ctx_name=ctx_name,
offset=offset,
context=context,
event_service=event_service,
)
@bot.di
def events_filters_not_implemented_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
# handlers для tags/city/date/mode до следующих коммитов
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, offset = _split_ctx_offset(event.data.value)
flt = get_events_filter(context, ctx_name=ctx_name)
summary = format_events_filter_summary(flt)
bot.messaging.send_message(
peer=event.peer,
text=f"⚠️ Пока не реализовано в этом коммите.\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=ctx_name, offset=offset),
)
def _tag_choices_from_backend(all_tags) -> list[TagChoice]:
out: list[TagChoice] = []
for t in all_tags or []:
tid = getattr(t, "id", None)
name = getattr(t, "name", None)
if tid and name:
out.append(TagChoice(key=str(tid), label=str(name)))
return out
@bot.di
def events_filters_tags_open_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
tag_service: TagService,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, offset = _split_ctx_offset(event.data.value)
flt = get_events_filter(context, ctx_name=ctx_name)
selected_ids = set(flt.tag_ids or [])
all_tags = tag_service.get_all_tags()
choices = _tag_choices_from_backend(all_tags)
# сохраняем "куда вернуться"
context.update_data(
{
_CTX_KEY: ctx_name,
_OFFSET_KEY: int(offset),
_TAG_IDS_KEY: list(selected_ids),
}
)
context.set_state(EventsFiltersTagsState.tags)
send_tags_toggle_ui(
peer=event.peer,
title="🏷️ *Фильтры: теги*\n\nВыбери теги, затем нажми «Готово»:",
choices=choices,
selected_keys=set(selected_ids),
cfg=FILTERS_TAGS_CFG,
done_label="✅ Готово",
)
@bot.di
def events_filters_tags_toggle_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
tag_service: TagService,
):
state = context.get_state()
if not (state == EventsFiltersTagsState.tags or str(state) == str(EventsFiltersTagsState.tags)):
return
delete_prev_message_by_peer(bot, event.peer)
raw = str(event.data.value or "")
data = context.get_data() or {}
ctx_name = str(data.get(_CTX_KEY) or "events")
offset = int(data.get(_OFFSET_KEY) or 0)
selected = set(data.get(_TAG_IDS_KEY) or [])
# DONE -> сохранить filter.tag_ids и вернуться в меню фильтров
if is_done(raw, FILTERS_TAGS_CFG):
flt = get_events_filter(context, ctx_name=ctx_name)
flt.tag_ids = sorted([str(x) for x in selected if str(x).strip()])
all_tags = tag_service.get_all_tags()
id_to_name = {
str(getattr(t, "id", "")).strip(): str(getattr(t, "name", "")).strip()
for t in (all_tags or [])
if str(getattr(t, "id", "")).strip() and str(getattr(t, "name", "")).strip()
}
flt.tag_names = [id_to_name[i] for i in flt.tag_ids if i in id_to_name]
set_events_filter(context, ctx_name=ctx_name, flt=flt)
context.set_state(None)
# чистим только "служебные" ключи экрана выбора тегов
context.update_data({_CTX_KEY: None, _OFFSET_KEY: None, _TAG_IDS_KEY: None})
summary = format_events_filter_summary(get_events_filter(context, ctx_name=ctx_name))
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=ctx_name, offset=offset),
)
return
# TOGGLE
key = parse_toggle_key(raw, FILTERS_TAGS_CFG)
if not key:
return
if key in selected:
selected.remove(key)
else:
selected.add(key)
context.update_data({_TAG_IDS_KEY: list(selected)})
all_tags = tag_service.get_all_tags()
choices = _tag_choices_from_backend(all_tags)
send_tags_toggle_ui(
peer=event.peer,
title="🏷️ *Фильтры: теги*\n\nВыбери теги, затем нажми «Готово»:",
choices=choices,
selected_keys=set(selected),
cfg=FILTERS_TAGS_CFG,
done_label="✅ Готово",
)
def _safe_int(value: object, default: int = 0) -> int:
try:
if value is None:
return default
n = int(value)
return n if n >= 0 else default
except Exception:
return default
def _send_city_picker(
*,
peer,
ctx_name: str,
back_offset: int,
cities_offset: int,
selected_city_id: str | None,
city_service: CityService,
):
cities = city_service.get_all(offset=cities_offset, limit=_CITIES_PAGE_LIMIT)
has_prev = cities_offset > 0
has_next = len(cities) >= _CITIES_PAGE_LIMIT
bot.messaging.send_message(
peer=peer,
text="🏙️ *Фильтры: город*\n\nВыбери один город и нажми «Готово»:",
interactive_media_groups=events_filters_city_keyboard(
ctx_name=ctx_name,
back_offset=back_offset,
cities=cities,
cities_offset=cities_offset,
limit=_CITIES_PAGE_LIMIT,
selected_city_id=selected_city_id,
has_prev=has_prev,
has_next=has_next,
),
)
@bot.di
def events_filters_city_open_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
city_service: CityService,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, back_offset = _split_ctx_offset(event.data.value)
flt = get_events_filter(context, ctx_name=ctx_name)
# сохраняем "куда вернуться" + текущий выбор
context.update_data(
{
_CTX_KEY: ctx_name,
_OFFSET_KEY: int(back_offset),
_CITIES_OFFSET_KEY: 0,
_CITY_SELECTED_ID_KEY: flt.city_id,
_CITY_SELECTED_NAME_KEY: flt.city_name,
}
)
context.set_state(EventsFiltersCityState.city)
_send_city_picker(
peer=event.peer,
ctx_name=ctx_name,
back_offset=back_offset,
cities_offset=0,
selected_city_id=flt.city_id,
city_service=city_service,
)
@bot.di
def events_filters_city_select_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
city_service: CityService,
):
state = context.get_state()
if not (state == EventsFiltersCityState.city or str(state) == str(EventsFiltersCityState.city)):
return
delete_prev_message_by_peer(bot, event.peer)
raw = str(event.data.value or "").strip()
data = context.get_data() or {}
ctx_name = str(data.get(_CTX_KEY) or "events")
back_offset = _safe_int(data.get(_OFFSET_KEY), 0)
cities_offset = _safe_int(data.get(_CITIES_OFFSET_KEY), 0)
# selected_city_id = data.get(_CITY_SELECTED_ID_KEY)
# BACK / DONE несут ctx|offset
if raw.startswith("back:"):
_ctx, _off = _split_ctx_offset(raw[len("back:") :])
summary = format_events_filter_summary(get_events_filter(context, ctx_name=_ctx))
context.set_state(None)
context.update_data(
{
_CTX_KEY: None,
_OFFSET_KEY: None,
_CITIES_OFFSET_KEY: None,
_CITY_SELECTED_ID_KEY: None,
_CITY_SELECTED_NAME_KEY: None,
}
)
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=_ctx, offset=_off),
)
return
if raw.startswith("done:"):
_ctx, _off = _split_ctx_offset(raw[len("done:") :])
sel_id = str(data.get(_CITY_SELECTED_ID_KEY) or "").strip() or None
sel_name = str(data.get(_CITY_SELECTED_NAME_KEY) or "").strip() or None
flt = get_events_filter(context, ctx_name=_ctx)
flt.city_id = sel_id
flt.city_name = sel_name
set_events_filter(context, ctx_name=_ctx, flt=flt)
summary = format_events_filter_summary(get_events_filter(context, ctx_name=_ctx))
context.set_state(None)
context.update_data(
{
_CTX_KEY: None,
_OFFSET_KEY: None,
_CITIES_OFFSET_KEY: None,
_CITY_SELECTED_ID_KEY: None,
_CITY_SELECTED_NAME_KEY: None,
}
)
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=_ctx, offset=_off),
)
return
# CLEAR
if raw == "clear":
context.update_data({_CITY_SELECTED_ID_KEY: None, _CITY_SELECTED_NAME_KEY: None})
_send_city_picker(
peer=event.peer,
ctx_name=ctx_name,
back_offset=back_offset,
cities_offset=cities_offset,
selected_city_id=None,
city_service=city_service,
)
return
# PAGE
if raw.startswith("page:"):
new_off = _safe_int(raw[len("page:") :], 0)
context.update_data({_CITIES_OFFSET_KEY: new_off})
_send_city_picker(
peer=event.peer,
ctx_name=ctx_name,
back_offset=back_offset,
cities_offset=new_off,
selected_city_id=str(data.get(_CITY_SELECTED_ID_KEY) or "").strip() or None,
city_service=city_service,
)
return
# SELECT
if raw.startswith("select:"):
cid = raw[len("select:") :].strip()
if not cid:
return
city = city_service.get_by_id(cid)
context.update_data(
{
_CITY_SELECTED_ID_KEY: cid,
_CITY_SELECTED_NAME_KEY: (city.name if city else None),
}
)
_send_city_picker(
peer=event.peer,
ctx_name=ctx_name,
back_offset=back_offset,
cities_offset=cities_offset,
selected_city_id=cid,
city_service=city_service,
)
return
def _is_mode_locked_for_ctx(ctx: str) -> str | None:
if ctx == "my_events":
return "organizer"
if ctx in ("part_events", "points"):
return "participant"
return None
def _send_mode_picker(*, peer, ctx_name: str, back_offset: int, context: FSMContext) -> None:
data = context.get_data() or {}
locked_mode = str(data.get(_MODE_LOCKED_KEY) or "").strip() or None
if locked_mode not in ("all", "participant", "organizer"):
locked_mode = None
sel = str(data.get(_MODE_KEY) or "all").strip()
if sel not in ("all", "participant", "organizer"):
sel = "all"
note = ""
if locked_mode == "organizer":
note = "\n\nℹ️ Для этой вкладки режим зафиксирован: *организатор*."
if locked_mode == "participant":
note = "\n\nℹ️ Для этой вкладки режим зафиксирован: *участник*."
bot.messaging.send_message(
peer=peer,
text="🎛️ *Фильтры: режим*\n\nВыбери режим и нажми «Готово»:" + note,
interactive_media_groups=events_filters_mode_keyboard(
ctx_name=ctx_name,
back_offset=back_offset,
selected_mode=sel, # type: ignore[arg-type]
locked_mode=locked_mode,
),
)
@bot.di
def events_filters_mode_open_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
ctx_name, back_offset = _split_ctx_offset(event.data.value)
flt = get_events_filter(context, ctx_name=ctx_name)
locked_mode = _locked_mode_for_ctx(ctx_name)
context.update_data(
{
_CTX_KEY: ctx_name,
_OFFSET_KEY: int(back_offset),
_MODE_KEY: flt.mode,
_MODE_LOCKED_KEY: locked_mode,
}
)
context.set_state(EventsFiltersModeState.mode)
_send_mode_picker(peer=event.peer, ctx_name=ctx_name, back_offset=back_offset, context=context)
@bot.di
def events_filters_mode_select_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
state = context.get_state()
if not (state == EventsFiltersModeState.mode or str(state) == str(EventsFiltersModeState.mode)):
return
delete_prev_message_by_peer(bot, event.peer)
raw = str(event.data.value or "").strip()
data = context.get_data() or {}
ctx_name = str(data.get(_CTX_KEY) or "events")
back_offset = _safe_int(data.get(_OFFSET_KEY), 0)
locked_mode = str(data.get(_MODE_LOCKED_KEY) or "").strip() or None
if locked_mode not in ("all", "participant", "organizer"):
locked_mode = None
if raw.startswith("back:"):
_ctx, _off = _split_ctx_offset(raw[len("back:") :])
summary = format_events_filter_summary(get_events_filter(context, ctx_name=_ctx))
context.set_state(None)
context.update_data(
{_CTX_KEY: None, _OFFSET_KEY: None, _MODE_KEY: None, _MODE_LOCKED_KEY: None}
)
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=_ctx, offset=_off),
)
return
if raw.startswith("done:"):
_ctx, _off = _split_ctx_offset(raw[len("done:") :])
flt = get_events_filter(context, ctx_name=_ctx)
if locked_mode in ("participant", "organizer"):
flt.mode = locked_mode # type: ignore[assignement]
else:
sel = str((context.get_data() or {}).get(_MODE_KEY) or "all").strip()
if sel in ("all", "participant", "organizer"):
flt.mode = sel # type: ignore[assignment]
set_events_filter(context, ctx_name=_ctx, flt=flt)
summary = format_events_filter_summary(get_events_filter(context, ctx_name=_ctx))
context.set_state(None)
context.update_data(
{_CTX_KEY: None, _OFFSET_KEY: None, _MODE_KEY: None, _MODE_LOCKED_KEY: None}
)
bot.messaging.send_message(
peer=event.peer,
text=f"🧰 *Фильтры*\n\n{summary}\n\nВыбери, что изменить:",
interactive_media_groups=events_filters_menu_keyboard(ctx_name=_ctx, offset=_off),
)
return
if raw.startswith("set:"):
if locked_mode in ("participant", "organizer"):
# показываем тот же экран с подсказкой, ничего не меняем
_send_mode_picker(
peer=event.peer, ctx_name=ctx_name, back_offset=back_offset, context=context
)
return
sel = raw[len("set:") :].strip()
if sel not in ("all", "participant", "organizer"):
return
context.update_data({_MODE_KEY: sel})
_send_mode_picker(
peer=event.peer, ctx_name=ctx_name, back_offset=back_offset, context=context
)
return
--- 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,
part_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",
),
# дом волонтёра -> мои меропрития
"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",
),
}
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
only_organizer: bool = ctx_name == "moderation"
lock_mode: bool = ctx_name == "moderation"
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,
only_organizer=only_organizer,
filters=flt,
lock_mode=lock_mode,
)
_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_part_events_page(
peer,
*,
offset: int,
event_service: EventService,
context: FSMContext | None = None,
city_id: str = DEFAULT_CITY_UUID,
note: str | None = None,
):
ui = get_ui("part_events")
limit = EVENTS_PAGE_SIZE
flt = get_events_filter(context, ctx_name="part_events") if context is not None else None
summary = format_events_filter_summary(flt) if context is not None else None
if summary:
summary = summary + "\nрежим зафиксирован: участник"
else:
summary = "Фильтр\nрежим зафиксирован: участник"
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,
only_participant=True,
filters=flt,
lock_mode=True,
)
_send_events_list_screen(
peer,
ui=ui,
ctx_name="part_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=(
"Пока нет мероприятий, на которые ты записан.\n\n"
"Как записаться:\n"
'1) Найди подходящие мероприятия во вкладке "Все мероприятия"\n'
'2) Нажми "записаться"\n'
),
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/markups/events_filters.py ---
from dialog_bot_sdk.interactive_media import Button, InteractiveMediaGroup, MediaGroupBuilder
from core.schemas.city import CitySchema
from core.utils.events_filter import EventsMode
def events_filters_button_keyboard(*, ctx_name: str, offset: int) -> list[InteractiveMediaGroup]:
locked_note = ""
if ctx_name == "part_events":
locked_note = "🔒 участник"
return MediaGroupBuilder(
[Button(media_id="events_filters_open", value=f"{ctx_name}|{offset}", label="🧰 Фильтры" + locked_note,
)]
).build()
def events_filters_menu_keyboard(*, ctx_name: str, offset: int) -> list[InteractiveMediaGroup]:
# value всегда "ctx|offset" чтобы назад возвращаться туда же
v = f"{ctx_name}|{offset}"
top = MediaGroupBuilder(
[
Button(media_id="events_filters_tags", value=v, label="🏷️ Теги"),
Button(media_id="events_filters_city", value=v, label="🏙️ Город"),
]
).build()
if ctx_name == "part_events":
mid = MediaGroupBuilder(
[Button(media_id="events_filters_date", value=v, label="🗓️ Дата/время"), ]
).build()
else:
mid = MediaGroupBuilder(
[
Button(media_id="events_filters_date", value=v, label="🗓️ Дата/время"),
Button(media_id="events_filters_mode", value=v, label="🎛️ Режим"),
]
).build()
bottom = MediaGroupBuilder(
[
Button(media_id="events_filters_apply", value=v, label="✅ Применить"),
Button(media_id="events_filters_reset", value=v, label="♻️ Сбросить"),
Button(media_id="events_filters_back", value=v, label="⬅️ Назад"),
]
).build()
return top + mid + bottom
def events_filters_city_keyboard(
*,
ctx_name: str,
back_offset: int,
cities: list[CitySchema],
cities_offset: int,
limit: int,
selected_city_id: str | None,
has_prev: bool,
has_next: bool,
) -> list[InteractiveMediaGroup]:
groups: list[InteractiveMediaGroup] = []
for c in cities or []:
cid = str(getattr(c, "id", "") or "").strip()
name = str(getattr(c, "name", "") or "").strip()
if not cid or not name:
continue
mark = "☑️ " if (selected_city_id and str(selected_city_id) == cid) else ""
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_city_select",
value=f"select:{cid}",
label=mark + name,
)
]
).build()
if has_prev:
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_city_select",
value=f"page:{max(cities_offset - limit, 0)}",
label="⬅️",
)
]
).build()
if has_next:
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_city_select",
value=f"page:{cities_offset + limit}",
label="➡️",
)
]
).build()
groups += MediaGroupBuilder(
[Button(media_id="events_filters_city_select", value="clear", label="Без города")]
).build()
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_city_select",
value=f"done:{ctx_name}|{back_offset}",
label="✅ Готово",
)
]
).build()
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_city_select",
value=f"back:{ctx_name}|{back_offset}",
label="⬅️ Назад",
)
]
).build()
return groups
def events_filters_mode_keyboard(
*,
ctx_name: str,
back_offset: int,
selected_mode: EventsMode,
locked_mode: EventsMode | None,
) -> list[InteractiveMediaGroup]:
groups: list[InteractiveMediaGroup] = []
def _btn(label: str, value: str, marked: bool) -> list[InteractiveMediaGroup]:
mark = "☑️ " if marked else ""
return MediaGroupBuilder(
[Button(media_id="events_filters_mode_select", value=value, label=mark + label)]
).build()
# если locked_mode - показываем эффективный режим вкладки
eff_mode: EventsMode = locked_mode if locked_mode else selected_mode
groups += _btn("Все", "set:all", eff_mode == "all")
groups += _btn("Я участвую", "set:participant", eff_mode == "participant")
groups += _btn("Я организатор", "set:organizer", eff_mode == "organizer")
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_mode_select",
value=f"done:{ctx_name}|{back_offset}",
label="✅ Готово",
)
]
).build()
groups += MediaGroupBuilder(
[
Button(
media_id="events_filters_mode_select",
value=f"back:{ctx_name}|{back_offset}",
label="⬅️ Назад",
)
]
).build()
return groups
Editor is loading...
Leave a Comment