Untitled
4ae4d
plain_text
04/15/2026 7:08 AM
179.5 KB
9
Indexable
f=docs/integration-notes.md
f=docs/domain-model.md
f=docs/buisiness-model.md
f=services/backend/internal/domain/event_card/event_card_page.go
f=services/backend/internal/domain/event_card/page_meta.go
f=services/bot/core/handlers/events_pages.py
f=services/bot/core/handlers/events_ui.py
f=services/bot/core/handlers/event.py
f=services/bot/core/markups/event.py
f=services/bot/core/markups/inline.py
f=services/bot/core/schemas/event.py
f=services/bot/core/services/event.py
f=services/bot/core/clients/event.py
f=services/bot/core/utils/events_filter.py
f=services/bot/core/handlers/events_filters.py
--- docs/integration-notes.md ---
# Этот файл описывает процесс перехода с Python backend на Go backend
- файл описывает изменения на ветке `integration/go-backend`
- эту ветку потом планируется вливать в `main`, а текущий оставить как legacy python mvp
- в связи с объёмным кодом, слоёной архитектурой и изменением множества параметров в ТЗ, почти весь старый фронт не работает, за исключением переделанных методов. не рабочие методы в следующем коммите буду помечены комментарием # untested
## рабочие методы и сценарии:
1. `ensure_exists` on /start
- дергает эндпоинт `put /volunteers`
2. `my_profile_handler` при нажатии на кнопку с id `my_profile`
- дергает эндпоинт `get /volunteers/card`
- также был изменён класс `UserSchema`, что наверняка поломает почти всё, что его использует.
3. можно редактировать теги волонтёра: удалён класс `UserTagSchama` и изменён `TagSchema`
--- docs/domain-model.md ---
## роли (от младшего к старшему)
### волонтёр
#### чем является
* базовая роль для всех сотрудников, использующих бота
#### может
* видеть ленту мероприятий (по умолчанию по фактическому рабочему месту)
* взаимодействовать с мероприятием (открывать карточку, записываться/отписываться, вводить код)
* видеть свой прогресс (волонтёро-часы, участие в мероприятиях)
* создавать заявки:
* на разовое мероприятие (вне проекта) → адресат: модератор ТБ
* на мероприятие в рамках проекта → адресат: лидер проекта
* на создание проекта / я — лидер проекта → адресат: модератор ТБ
* если волонтёр назначен организатором конкретного мероприятия, то он получает расширенные права организатора этого мероприятия
* организатор мероприятия — это не отдельная глобальная роль пользователя, а контекстная роль в рамках конкретного мероприятия
* организатор мероприятия может:
* редактировать карточку мероприятия
* удалять мероприятие
* видеть список участников и их статус подтверждения
* создавать отчёт по мероприятию
* если он является автором заявки на создание этого мероприятия и стал его организатором, то он может добавить не более двух дополнительных организаторов
* управлять кодом подтверждения мероприятия:
* код для мероприятия один общий для всех участников
#### доступ к отчётам
* персональная история участия (свои мероприятия/часы)
* отчёты по своим мероприятиям только если он организатор этих мероприятий
#### ответственность
* за себя и свои действия
* за свои мероприятия, в которых он организатор
---
### лидер проекта
#### чем является
* проверенный волонтёр, ведущий долгосрочное направление (проект)
* назначается модератором ТБ
* роль остаётся у человека даже без активных проектов
#### может (помимо волонтёра)
* автономно создавать любые мероприятия (включая разовые) без одобрения модератора
* одобрять заявки на создание мероприятий в рамках своего проекта
* подавать заявку на создание ещё одного проекта (к модератору)
* по мероприятию в рамках проекта, созданному по заявке, является равноправным организатором вместе с подавшим заявку
#### доступ к отчётам
* по своим проектам
#### ответственность
* за проекты, где он назначен лидером
* за обработку заявок на мероприятия в этих проектах
---
### модератор ТБ
#### чем является
* роль на уровне тербанка (ТБ)
* основной куратор в ТБ
#### может (помимо лидера проекта)
* обрабатывать заявки:
* на создание проекта (по одобрению создаётся проект и назначается лидер проекта)
* на создание разовых мероприятий от волонтёров
* на создание проекта от текущих лидеров проектов
* назначать/снимать лидеров проектов
* выгружать статистику/отчёты по своему ТБ
* отвечать за корректное разделение Москва/Среднерусский/ЦА, если относится к его зоне
* иметь доступ к архивным мероприятиям своего ТБ
#### доступ к отчётам
* по своему тербанку
* по архивным мероприятиям своего ТБ
#### ответственность
* за модерацию в своём ТБ
* при наличии нескольких модераторов ТБ ответственность между ними равная
---
### владелец продукта
#### чем является
* высший уровень доступа
#### может (помимо модератора ТБ)
* назначать/снимать модераторов по всем ТБ
#### доступ к отчётам
* по всему банку
#### ответственность
* за иерархию доступов и контроль ролей
---
## термины и сущности (по возможности от “младшего” к “старшего”)
### заявка
#### чем является
* механизм согласования действий, требующих одобрения
#### кто создаёт
* волонтёр:
* создать разовое мероприятие (без проекта) → модератор ТБ
* создать мероприятие в проекте → лидер проекта
* создать проект / я — лидер проекта → модератор ТБ
* лидер проекта:
* создать новый проект → модератор ТБ
#### содержит поля
* тип заявки: создать проект / создать разовое мероприятие / создать мероприятие в проекте
* автор заявки
* адресат (модератор ТБ или лидер проекта)
* данные создаваемой сущности (поля проекта или мероприятия)
* статус: на согласовании / принята / отклонена / ждёт вашего ответа
* комментарий ответственного (опционально)
* дата создания, дата обработки
#### результат обработки
* для заявки на проект: создаётся проект; автору назначается роль лидер проекта; автор становится лидером созданного проекта
* для заявки на мероприятие: создаётся мероприятие; автор становится организатором созданного мероприятия
* для заявки на мероприятие в рамках проекта: лидер проекта и автор заявки становятся равноправными организаторами мероприятия
---
### участие
#### чем является
* связь волонтёр <-> мероприятие, фиксирующая запись и подтверждение участия
#### содержит поля
* волонтёр
* мероприятие
* статус: записан / подтверждено кодом
* использованный код (если подтверждено)
* начисленные часы (из кода; фиксируются при подтверждении)
* даты: запись / подтверждение
#### ограничения
* для пары волонтёр + мероприятие существует максимум одна запись участия
* после подтверждения запрещён ввод любых кодов по этому мероприятию
* доступ к архивному мероприятию для обычного пользователя определяется наличием записи участия в этом мероприятии
---
### код подтверждения
#### чем является
* механизм верификации участия после мероприятия
* содержит стоимость в часах, это источник истины по начислению часов
* для каждого мероприятия используется один общий код подтверждения для всех участников
* код не должен совпадать с кодами, использовавшимися в предыдущие 2 месяца
* при создании кода система должна предлагать сгенерировать код, иначе давать подсказку по его составлению
#### содержит поля
* код
* мероприятие
* часы, начисляемые при вводе этого кода
---
### отчёт
#### чем является
* закрывающий отчёт по мероприятию
* после создания отчёта мероприятие считается завершённым и переводится в архив
#### содержит поля
* количество благополучателей
* описание как прошло
* дата закрытия
* уточнить, какие еще поля нужны
---
### мероприятие
#### чем является
* карточка события, на которое записываются волонтёры
* после создания доступно для отображения в ленте
* после создания отчёта переводится в архив
#### содержит поля
* название
* описание
* дата и время проведения (начало; при необходимости конец)
* место/адрес
* ГОСП
* количество требуемых волонтёров
* список организаторов
* проект (может отсутствовать для разового мероприятия)
* признак архивности
* код подтверждения (1)
#### поведение
* волонтёр: записаться/отписаться; ввести код (один раз на мероприятие)
* организатор: редактировать/удалять; смотреть участников; создавать отчёт
* подавший заявку и ставший организатором может добавить не более двух дополнительных организаторов
* для мероприятия в рамках проекта лидер проекта и подавший заявку являются равноправными организаторами
* модератор имеет доступ к архивным мероприятиям
* остальные имеют доступ к архивным мероприятиям только если у них есть участие в этом мероприятии
---
### проект
#### чем является
* долгосрочное направление, объединяющее серию мероприятий
* принадлежит одному тербанку
* если проект нужен в нескольких тербанках, создаются независимые проекты без связи
#### содержит поля
* название
* описание
* лидер проекта
* тербанк-владелец проекта
* состояние актива
#### правила
* у проекта всегда один текущий лидер
* роль “лидер проекта” как роль пользователя и назначение пользователя лидером конкретного проекта — не одно и то же
* лидер проекта создаёт мероприятия без премодерации
* волонтёры создают мероприятия в проекте через заявку лидеру проекта
* проектная отчётность строится по мероприятиям проекта
* должна поддерживаться архивация, деархивация и передача лидерства проекта
---
### ГОСП
#### чем является
* основная единица привязки мероприятий
* ключевой уровень локальности для ленты
#### содержит поля
* название/код (из справочника)
* привязка к тербанку
* город/адрес (опционально)
---
### тербанк (ТБ)
#### чем является
* территориальная единица банка
#### содержит поля
* название
* территория/регион
* модераторы ТБ (0..\*)
---
### банк + особые сущности Москвы
#### чем является
* совокупность тербанков
* Москва разделена на 3 независимые сущности отчётности: Московский банк, Среднерусский банк, Центральный аппарат (ЦА)
#### статус формализации
* требуется уточнить, являются ли Московский банк / Среднерусский банк / ЦА отдельными сущностями модели или специальным разрезом отчётности
---
## ключевая логика (подтверждение и статистика)
### часы
* единого норматива нет
* часы задаются на уровне кода подтверждения (источник истины)
### верификация
* после мероприятия организатор выдаёт код
* ввод кода:
* подтверждает участие
* начисляет часы в профиль
* запрещает ввод любых других кодов по этому мероприятию
### ключевые метрики (минимум)
* уникальные волонтёры
* суммарные волонтёро-часы
* благополучатели
* количество мероприятий
---
## связи и кратности
* банк (1) — тербанк (1..\*)
* тербанк (1) — ГОСП (1..\*)
* ГОСП (1) — мероприятие (0..\*)
* проект (1) — мероприятие (0..\*)
(у мероприятия проект может отсутствовать; если проект указан, то он один)
* мероприятие (1) — код подтверждения (1)
* мероприятие (1) — отчёт (0..1)
* волонтёр (1) — участие (0..\*)
* мероприятие (1) — участие (0..\*)
* код подтверждения (1) — участие (0..\*)
(код появляется в участии только при подтверждении; один код мероприятия может быть использован многими подтверждёнными участиями этого мероприятия)
* волонтёр (1) — заявка (0..\*)
* проект (1) — лидер проекта (1)
* тербанк (1) — модератор ТБ (0..\*)
* волонтёр (0..\*) — мероприятие (0..\*)
(связь “организатор мероприятия”)
---
## места, которые надо уточнить
* конкретное максимальное количество организаторов мероприятия
* считать ли признак архивности самостоятельным полем или производным признаком от факта создания отчёта
* нужна ли отдельная сущность/журнал истории изменений мероприятия, если в выгрузке должны отражаться все состояния данных
* формальная модель Москва/Среднерусский/ЦА: отдельные сущности модели или специальный разрез отчётности
--- docs/buisiness-model.md ---
# Существующие роли
> [!NOTE]
>
> в документе рассматриваем роли в таком контексте: \
> ***Модератор*** в своём ТБ; \
> ***Организатор*** в мероприятии, которое он организует; \
> вне этих контекстов и ***Модератор*** и ***Организатор*** -- обычные волонтёры.
## 1. Волонтёр
## 2. Организатор
## 3. Лидер проекта
## 4. Модератор ТБ
## 5. Владелец
---
# Описание бизнесс процессов бота `Волонтёрам ПомогAI`
---
## 1. Запись на мероприятие
### Акторы:
#### Основной: Волонтёр
#### Дополнительные: -
### Описание:
***Волонтёр*** выбирает мероприятие из множества доступных, заходит в карточку мероприятия, нажимает "записаться".
### Последующие возможные процессы:
- Подтверждение участия через код
---
## 2. Подтверждение участия через код
### Акторы:
#### Основной: Волонтёр
#### Дополнительные: Организатор
### Описание:
При посещении мероприятия ***Волонтёром*** (state, не контролируемый ботом) может получить код от ***Оргазизатора***, которым ***Волонтёр*** может подтвердить участие в мероприятии.
### Последующие возможные процессы:
- Получение баллов ***Волонтёром*** за соответствующее мероприятие
---
## 3. Создание заявки на разовое мероприятие
### Акторы:
#### Основной: Волонтёр
#### Дополнительные: Модератор
### Описание:
***Волонтёр*** может подать заявку на создание разового мероприятия (т.е. не в рамках проекта), которую будет обрабатывать ***Модератор***
### Последующие возможные процессы:
- Обработка заявки на разовое мероприятие
---
## 4. Обработка заявки на разовое мероприятие
### Акторы:
#### Основной: Модератор
#### Дополнительные: Волонтёр
### Описание:
***Модераторы*** имеют доступ к актуальным заявкам, могут выбрать интересующую заявку из списка, и обработать её.
Возможные состояния заявки на разовое мероприятие: `на согласовании` / `принята` / `отклонена` / `ждёт вашего ответа`; подавший заявку видит её статус
Заявка считается обработанной, когда попадает в состояние `принята` / `отклонена`
### Последующие возможные процессы:
- Создание мероприятия с полями, соответствующими указанным в заявке
- Отказ в создании мероприятия
---
## 5. Создание заявки на мероприятия в рамках проекта
### Акторы:
#### Основной: Волонтёр
#### Дополнительные: Лидер проекта
### Описание:
***Волонтёр*** может подать заявку на создание мероприятия в рамках проекта, которую будет обрабатывать ***Лидер проекта*** (соответствующего)
### Последующие возможные процессы:
- Обработка заявки на мероприятие в рамках проекта
- Смена роли ***Волонтёр*** -> ***Организатор*** в случае одобрения (контекстно)
---
## 6. Обработка заявки на мероприятие в рамках проекта
### Акторы:
#### Основной: Лидер проекта
#### Дополнительные: Волонтёр
### Описание:
***Лидер проекта*** имеет доступ к актуальным заявкам, может выбрать интересующую заявку из списка, и обработать её.
Возможные состояния заявки на мероприятие в рамках проекта: `на согласовании` / `принята` / `отклонена` / `ждёт вашего ответа`; подавший заявку видит её статус
Заявка считается обработанной, когда попадает в состояние `принята` / `отклонена`
В случае одобрения, ***Лидер проекта*** автоматически назначается ***Организатором*** мероприятия (на ряду с заявившим)
### Последующие возможные процессы:
- Создание мероприятия с полями, соответствующими указанным в заявке
- Смена роли ***Волонтёр*** -> ***Организатор*** в случае одобрения (контекстно)
- Отказ в создании мероприятия
---
## 7. Создание заявки на проект
### Акторы:
#### Основной: Волонтёр / Лидер проекта
#### Дополнительные: Модератор
### Описание:
***Волонтёр*** / ***Лидер проекта*** может подать заявку на создание проекта, которую будет обрабатывать ***Модератор***
### Последующие возможные процессы:
- Обработка заявки на создание проекта
---
## 8. Обработка заявки на создание проекта
### Акторы:
#### Основной: Модератор
#### Дополнительные: Волонтёр / Лидер проекта
### Описание:
***Модератор*** имеет доступ к актуальным заявкам, может выбрать интересующую заявку из списка, и обработать её.
Возможные состояния заявки на проект: `на согласовании` / `принята` / `отклонена` / `ждёт вашего ответа`; подавший заявку видит её статус
Заявка считается обработанной, когда попадает в состояние `принята` / `отклонена`
### Последующие возможные процессы:
- Отказ в создании проекта
- Создание проекта с полями, соответствующими указанным в заявке
- Смена роли ***Волонтёр*** -> ***Лидер проекта***
- ЕСЛИ заявка одобрена
- И роль заявителя - ***Волонтёр***
> [!NOTE]
>
> ***Лидер проекта*** - это не контекстная роль. т.е. в таком случае пользователь будет ***Лидером проекта*** всегда и глобально.
---
## 9. Управление мероприятием
### Акторы:
#### Основной: Организатор
#### Дополнительные: -
### Описание:
***Организатор*** может менять все доступные для изменения поля, в соответствии с domain model, также может удалить мероприятие
Например:
- Изменение имени
- Изменение описания
- Изменение состава организаторов
- Создание отчёта по мероприятию
### Последующие возможные процессы:
- Отмена мероприятия
- Изменение мероприятия
- Процесс завершения мероприятия: создание отчёта и помещение его в архив
---
## 10. Управление проектом
### Акторы:
#### Основной: Лидер проекта
#### Дополнительные: -
### Описание:
***Лидер проекта*** может менять все доступные для изменения поля, в соответствии с domain model, также может удалить мероприятие
### Последующие возможные процессы:
Например:
- Изменение имени
- Изменение описания
- Перевод проекта в инактив
---
## 11. Управление составом Лидеров проекта
### Акторы:
#### Основной: Модератор
#### Дополнительные: -
### Описание:
***Модератор*** может пометить проект как неактивный в случае необходимости, например, в случае невозможностью продолжения ведения проекта ***Лидером проекта***
---
## 12. Управление составом Модераторов ТБ
### Акторы:
#### Основной: Владелец
#### Дополнительные: Волонтёр / Модератор
### Описание:
***Владелец*** может назначать и снимать Модераторов ТБ для любой пары:
- ***Волонтёр*** <=> ТБ;
- ***Модератор ТБ*** <=> ТБ.
### Последующие возможные процессы:
Назначение и снятие ***Модератора ТБ***.
---
## 13. Создание мероприятия Лидером проекта
### Акторы:
#### Основной: Лидер проекта
#### Дополнительные: -
### Описание:
***Лидер проекта*** может создавать мероприятия как в рамках проекта, так и вне проекта, без создания заявок и другого рода одобрений.
### Последующие возможные процессы:
- Создание мероприятия в рамках проекта
- Создание мероприятия не в рамках проекта
---
## 14. Доступ к отчётам
### Акторы:
#### Основной: Организатор / Лидер проекта / Модератор / Владелец
#### Дополнительные: -
### Описание:
Пользователь имеет доступ к отчётам по мероприятиям в соответствии со своими правами:
- ***Организатор*** -- к мероприятиям, которые организует / организовал;
- ***Лидер проекта*** -- к мероприятиям, которые организует / организовал;
- ***Модератор*** -- к мероприятиям его ТБ + к мероприятиям, которые организует / организовал;
- ***Владелец*** -- ко всем мероприятиям.
> [!NOTE]
>
> ***Лидер проекта*** является ***Организатором*** во всех мероприятиях своего проекта \
> Доступ к отчетам значит возможность создавать/изменять/просматривать отчеты
### Последующие возможные процессы:
- Получение отчётов по мероприятиям
- В файле .xlsx
- В карточке мероприятия
---
## 14. Доступ к архивным мероприятиям
### Акторы:
#### Основной: Волонтёр / Организатор / Лидер проекта / Модератор / Владелец
#### Дополнительные: -
### Описание:
Пользователь имеет доступ к карточке мероприятия в соответствии со своими правами:
- ***Волонтёр*** -- к мероприятиям, в которых участвовал
- ***Организатор*** -- к мероприятиям, которые организовал + к мероприятиям, в которых участвовал;
- ***Лидер проекта*** -- к мероприятиям, которые организовал + к мероприятиям, в которых участвовал;
- ***Модератор*** -- к мероприятиям его ТБ + к мероприятиям, которые организовал + к мероприятиям, в которых участвовал;
- ***Владелец*** -- ко всем мероприятиям.
> [!NOTE]
>
> ***Лидер проекта*** является ***Организатором*** во всех мероприятиях своего проекта
### Последующие возможные процессы:
- Получение информации по мероприятиям в формате доступа к карточке мероприятия
--- services/backend/internal/domain/event_card/event_card_page.go ---
package event_card
type EventCardPage struct {
EventCards []*EventCard `json:"event_cards"`
PageMeta *PageMeta `json:"page_meta"`
}
--- services/backend/internal/domain/event_card/page_meta.go ---
package event_card
type PageMeta struct {
CurrentPage int `json:"current_page"`
TotalPages int `json:"total_pages"`
}
--- services/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( # untested
*,
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( # untested
*, 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)
--- services/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: # untested
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: # untested
return _UI.get(ctx_name, _UI["events"])
def build_back_keyboard(ctx_name: str) -> list[InteractiveMediaGroup]: # untested
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]: # untested
"""
ожидаем:
- "<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: # untested
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: # untested
# кто может управлять - тот видит код
return can_manage_event(ev, user)
def _send_events_list_screen( # untested
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( # untested
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 = False
lock_mode: bool = False
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( # untested
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, locked_mode="participant")
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,
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( # untested
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, locked_mode="participant")
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( # untested
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, locked_mode="organizer") 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, peer.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]: # untested
ui = get_ui(ctx_name)
return MediaGroupBuilder(
[
Button(
media_id=ui.list_open_media_id,
value=f"{event_id}|{ui.name}",
label="⬅️ К мероприятию",
)
]
).build()
--- services/bot/core/handlers/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,
send_part_events_page,
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): # untested
wait_code = State()
@bot.di
def events_menu_handler( # untested
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( # untested
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 part_events_menu_handler( # untested
event: UpdateInteractiveMediaEvent, context: FSMContext, event_service: EventService
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_part_events_page(
event.peer,
offset=0,
event_service=event_service,
context=context,
)
@bot.di
def part_events_page_handler( # untested
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_part_events_page(
event.peer,
offset=offset,
event_service=event_service,
context=context,
)
@bot.di
def user_events_open_handler( # untested
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( # untested
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( # untested
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
): # untested
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( # untested
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( # untested
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("full_name") 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),
)
--- services/bot/core/markups/event.py ---
from datetime import datetime
from dialog_bot_sdk.interactive_media import Button, InteractiveMediaGroup, MediaGroupBuilder
from core.markups.pagination import pagination_keyboard
from core.schemas.event import EventSchema
def format_event_details( # untested
event: EventSchema,
*,
show_code: bool,
show_organizer: bool,
show_report_status: bool = False,
report_exists: bool | None = None,
) -> str:
"""формирует читаемое описание из EventSchema"""
date_str = (
event.date.strftime("%d.%m.%Y %H:%M")
if isinstance(event.date, datetime)
else str(event.date)
)
lines: list[str] = [
"📅 *Мероприятие*",
"",
f"📝 *Название*: {event.name}",
f"⏰ *Когда*: {date_str}",
f"📍 *Где*: {event.location}",
f"🏷️ *Теги*: {_format_tags(event.tags)}",
f"🏆 *Баллы*: {event.points if event.points is not None else '—'}",
]
if show_code:
lines.append(f"🎟️ *Код*: {event.bonus_code or '—'}")
lines.extend(
[
f"📄 *Описание*: {event.description or '—'}",
"",
"🔧 Техническая информация:",
]
)
if show_report_status:
status = "✅ Создан" if report_exists else "Ещё не создан"
lines.append(f"Отчёт: `{status}`")
if show_organizer:
org = getattr(event, "creator_name", None) or "—"
lines.append(f"Организатор: `{org}`")
lines.append(f"ID: `{event.id}`")
return "\n".join(lines)
def _format_tags(tags: list | None) -> str: # TODO add type to 'tags' and dimplify func depending on type
if not tags:
return "—"
# tags = [{"id": ..., "title": ..., ...}, {...}, ...]
try:
titles = []
for t in tags:
if isinstance(t, dict):
titles.append(t.get("title") or str(t.get("id") or t))
else:
titles.append(t.title)
return ", ".join(titles) if titles else "—"
except Exception:
return str(tags)
def event_actions_keyboard( # untested
event_id: str,
*,
is_participant: bool,
can_manage: bool,
back_value: str,
back_label: str,
enter_code_media_id: str,
sign_up_media_id: str,
sign_out_media_id: str,
edit_media_id: str = "event_menu_edit",
delete_media_id: str = "event_menu_delete",
can_create_report: bool = False,
can_view_report: bool = False,
ctx: str | None = None,
) -> list[InteractiveMediaGroup]:
actions = []
v = f"{event_id}|{ctx}" if ctx else event_id
actions.append(Button(media_id=enter_code_media_id, value=v, label="🏅 Ввести код"))
if is_participant:
actions.append(Button(media_id=sign_out_media_id, value=v, label="❌ Отписаться"))
else:
actions.append(Button(media_id=sign_up_media_id, value=v, label="✅ Записаться"))
if can_manage:
actions.append(Button(media_id=edit_media_id, value=v, label="✏️ Редактировать"))
actions.append(Button(media_id="event_participants_open", value=v, label="👥 Участники"))
actions.append(Button(media_id=delete_media_id, value=v, label="🗑 Удалить"))
if can_create_report:
actions.append(Button(media_id="event_report_create", value=v, label="📄 Создать отчёт"))
if can_view_report:
actions.append(Button(media_id="event_report_view", value=v, label="📄 Посмотреть отчёт"))
actions.append(Button(media_id="leave", value=back_value, label=back_label))
return MediaGroupBuilder(actions).build()
def event_edit_fields_keyboard( # untested
event_id: str, *, leave_value: str = "moderation", leave_label: str = "⬅️ В меню модерации"
) -> list[InteractiveMediaGroup]:
group_builder_1 = MediaGroupBuilder(
[
Button(
media_id="event_menu_edit_name",
value=f"{event_id}|name",
label="📝 Название",
),
Button(
media_id="event_menu_edit_date",
value=f"{event_id}|date",
label="📅 Дата",
),
Button(
media_id="event_menu_edit_time",
value=f"{event_id}|time",
label="⏰ Время",
),
Button(
media_id="event_menu_edit_location",
value=f"{event_id}|location",
label="📍 Локация",
),
Button(
media_id="event_menu_edit_tags",
value=f"{event_id}|tags",
label="🏷️ Теги",
),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button(
media_id="event_menu_edit_creator_id",
value=f"{event_id}|creator_id",
label="👤 Организатор",
),
Button(
media_id="event_menu_edit_points",
value=f"{event_id}|points",
label="🏆 Баллы",
),
Button(
media_id="event_menu_edit_description",
value=f"{event_id}|description",
label="📄 Описание",
),
Button(
media_id="event_menu_edit_bonus_code",
value=f"{event_id}|bonus_code",
label="🎟️ Код",
),
Button(
media_id="leave",
value=leave_value,
label=leave_label,
),
]
)
return group_builder_1.merge([group_builder_2])
def user_events_pagination_keyboard( # untested
*, offset: int, limit: int, has_prev: bool, has_next: bool
) -> list[InteractiveMediaGroup]:
return pagination_keyboard(
offset=offset,
limit=limit,
has_prev=has_prev,
has_next=has_next,
prev_media_id="user_events_prev",
next_media_id="user_events_next",
leave_value="volunteer_home",
leave_label="🏡 В дом волонтёра",
keep_layout=True,
add_view_by_uuid_button=True,
view_by_uuid_value="events",
)
def part_events_pagination_keyboard( # untested
*, offset: int, limit: int, has_prev: bool, has_next: bool
) -> list[InteractiveMediaGroup]:
return pagination_keyboard(
offset=offset,
limit=limit,
has_prev=has_prev,
has_next=has_next,
prev_media_id="part_events_prev",
next_media_id="part_events_next",
leave_value="volunteer_home",
leave_label="🏡 В дом волонтёра",
keep_layout=True,
add_view_by_uuid_button=True,
view_by_uuid_value="part_events",
)
def my_events_pagination_keyboard( # untested
*, offset: int, limit: int, has_prev: bool, has_next: bool
) -> list[InteractiveMediaGroup]:
return pagination_keyboard(
offset=offset,
limit=limit,
has_prev=has_prev,
has_next=has_next,
prev_media_id="my_events_prev",
next_media_id="my_events_next",
leave_value="moderation",
leave_label="🛡️ В меню модерации",
keep_layout=True,
add_view_by_uuid_button=True,
view_by_uuid_value="my_events",
)
def _truncate(text: str, max_len: int = 40) -> str: # untested
text = (text or "").strip()
if len(text) <= max_len:
return text
return text[: max_len - 1] + "..."
def _event_button_label(ev: EventSchema, idx: int) -> str: # untested
date_val = getattr(ev, "date", None)
if isinstance(date_val, datetime): # TODO унифицировать: привести везде к datetime
date_str = date_val.strftime("%d.%m.%Y %H:%M")
else:
date_str = str(date_val or "—")
name = getattr(ev, "name", "—")
location = getattr(ev, "location", "—")
points = getattr(ev, "points", None)
base = f"{idx}. {name} — {date_str} — {location}"
if points is not None:
base += f" — {points} б."
return base
def events_select_keyboard( # untested
events: list[EventSchema],
*,
open_media_id: str,
start_index: int = 1,
per_row: int = 1,
label_max: int = 120,
) -> list[InteractiveMediaGroup]:
"""
строит список кнопок по мероприятиям
нажатие вызывает handler по _id = open_media_id, value = event.id
"""
groups: list[InteractiveMediaGroup] = []
row: list[Button] = []
idx = start_index
for ev in events:
label = _truncate(_event_button_label(ev, idx), max_len=label_max)
row.append(Button(media_id=open_media_id, value=str(getattr(ev, "id", "")), label=label))
idx += 1
if len(row) >= per_row:
groups.extend(MediaGroupBuilder(row).build())
row = []
if row:
groups.extend(MediaGroupBuilder(row).build())
return groups
def view_by_uuid_keyboard(*, media_id: str, label: str) -> list[InteractiveMediaGroup]: # untested
return MediaGroupBuilder([Button(media_id=media_id, value="1", label=label)]).build()
--- services/bot/core/markups/inline.py ---
from dialog_bot_sdk.entities.messaging import InteractiveMediaStyle
from dialog_bot_sdk.interactive_media import Button, MediaGroupBuilder
from core.markups.pagination import pagination_keyboard
# Основные клавиатуры
def main_keyboard_admin(): # untested
group_builder_1 = MediaGroupBuilder(
[
Button(
"1",
"volunteer_home",
"🏡 Дом волонтёра",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button(
"2",
"ai_assistant",
"✨ ИИ Волонтёр",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button(
"3",
"moderation",
"🛡️ Модерация",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button(
"4",
"administration",
"⚙️ Администрация",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
return group_builder_1.merge([group_builder_2])
def main_keyboard_moder(): # untested
group_builder_1 = MediaGroupBuilder(
[
Button(
"1",
"volunteer_home",
"🏡 Дом волонтёра",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button(
"2",
"ai_assistant",
"✨ ИИ Волонтёр",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button(
"3", "moderation", "Модерация", InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT
)
]
)
return group_builder_1.merge([group_builder_2])
def main_keyboard_user(): # untested
builder = MediaGroupBuilder(
[
Button(
"1",
"volunteer_home",
"🏡 Дом волонтёра",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button(
"2",
"ai_assistant",
"✨ ИИ Волонтёр",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
return builder.build()
def back_to_main_menu_keyboard(): # untested
builder = MediaGroupBuilder(
[Button("1", "main_menu", "⬅️ В меню", InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT)]
)
return builder.build()
# Дом волонтёра
def volunteer_home_keyboard(): # untested
group_builder_1 = MediaGroupBuilder(
[
Button(
"1",
"my_profile",
"👤 Мой профиль",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button(
"2", "events", "🌐 Мероприятия", InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT
),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button("3", "points", "🏆 Баллы", InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT),
Button(
"4",
"part_events",
"🎉 Мои мероприятия",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
group_builder_3 = MediaGroupBuilder(
[
Button(
"5", "main_menu", "⬅️ В меню", InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT
),
]
)
return group_builder_1.merge([group_builder_2, group_builder_3])
def back_to_moderation_keyboard(): # untested
builder = MediaGroupBuilder(
[
Button(
"1",
"moderation",
"⬅️ В меню модерации",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
)
return builder.build()
def back_to_volunteer_home_keyboard(): # untested
builder = MediaGroupBuilder(
[
Button(
"1",
"volunteer_home",
"⬅️ В дом волонтёра",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
)
]
)
return builder.build()
# Модерация
def moderation_menu_keyboard(): # untested
group_builder_1 = MediaGroupBuilder(
[
Button("1", "create_event", "🗓️ Создать мероприятие"),
Button("4", "my_events", "📋 Мои мероприятия"),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button(
"5",
"all_events",
"🌐 Все мероприятия",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
Button("7", "download_reports_list", "📥 Выгрузить отчёты"),
]
)
group_builder_3 = MediaGroupBuilder(
[
Button("10", "main_menu", "⬅️ В меню"),
]
)
return group_builder_1.merge([group_builder_2, group_builder_3])
# Администрация
def admin_menu_keyboard(): # untested
group_builder_1 = MediaGroupBuilder(
[
Button("1", "add_admin", "Добавить администратора"),
Button("2", "delete_admin", "Убрать администратора"),
]
)
group_builder_2 = MediaGroupBuilder(
[
Button("3", "add_moderator", "Добавить модератора"),
Button("4", "delete_moderator", "Убрать модератора"),
]
)
group_builder_3 = MediaGroupBuilder(
[
Button("5", "ban_user", "Забанить пользователя"),
Button("6", "unban_user", "Разбанить пользователя"),
]
)
group_builder_4 = MediaGroupBuilder([Button("7", "main_menu", "⬅️ В меню")])
return group_builder_1.merge([group_builder_2, group_builder_3, group_builder_4])
def reports_export_keyboard(): # untested
"""
value везде = 'download_reports_list', обработчик один
выбор различаем по media_id
"""
group_1 = MediaGroupBuilder(
[
Button("rep_day", "download_reports_list", "📅 Последний день"),
Button("rep_week", "download_reports_list", "📆 Неделя"),
]
)
group_2 = MediaGroupBuilder(
[
Button("rep_month", "download_reports_list", "🗓️ Месяц"),
Button("rep_all", "download_reports_list", "🌐 Всё время"),
]
)
group_3 = MediaGroupBuilder(
[
Button("rep_from", "download_reports_list", "📌 От даты до сегодня"),
Button("rep_range", "download_reports_list", "🗓️ От даты до даты"),
]
)
group_4 = MediaGroupBuilder(
[
Button("rep_back", "moderation", "⬅️ В меню модерации"),
]
)
return group_1.merge([group_2, group_3, group_4])
def back_to_reports_export_keyboard(): # untested
return MediaGroupBuilder(
[
Button(
"9",
"download_reports_list",
"⬅️ В меню выгрузки отчётов",
InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
),
]
).build()
def all_events_pagination_keyboard(
*, offset: int, limit: int, has_prev: bool, has_next: bool
): # untested
return pagination_keyboard(
offset=offset,
limit=limit,
has_prev=has_prev,
has_next=has_next,
prev_media_id="all_events_prev",
next_media_id="all_events_next",
leave_value="moderation",
leave_label="🛡️ В меню модерации",
style=InteractiveMediaStyle.INTERACTIVEMEDIASTYLE_DEFAULT,
keep_layout=True,
add_view_by_uuid_button=True,
view_by_uuid_value="moderation",
)
--- services/bot/core/schemas/event.py ---
from datetime import datetime
class EventSchema: # untested
def __init__( # untested
self,
id: str,
name: str,
date: datetime,
description: str,
*,
location: str | None = None,
tags: list | None = None, # TODO add type
bonus_code: str | None = None,
points: int | None = None,
city_id: str | None = None,
creator_id: str | None = None,
creator_name: str | None = None,
):
self.id = id
self.name = name
self.date = date
self.description = description
self.location = location
self.tags = tags
self.bonus_code = bonus_code
self.points = points
self.city_id = city_id
self.creator_id = creator_id
self.creator_name = creator_name
--- services/bot/core/services/event.py ---
from datetime import datetime, timezone
from typing import Any
from uuid import UUID
from requests import HTTPError
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): # untested
"""
сервис мероприятий на стороне бота.
на данный момент:
- ходит в backend через EventClient
- умеет получать список мероприятий по городу
- умеет получать одно мероприятие по id
- создает мероприятие через клиент используя мастера создания
"""
def __init__(self, registry: BaseServiceRegistry, client: EventClient): # untested
# DI может вызвать клиент без аргументов.
super().__init__(registry)
self.client = client
def get_event_participants( # untested
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]: # untested
"""
возвращает список 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: # untested
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( # untested
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( # untested
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( # untested
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, requester_messenger_id: int) -> EventSchema | None:
resp = self.client.get_event_info(id, requester_messenger_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,
)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting user: unauthorized")
case 404:
return None
case 500:
raise HTTPError("Error while getting user: internal server error")
case _:
raise HTTPError("Error while getting user: unexpected status {resp.status_code}")
body = resp.json()
event_body = body.get("event") or {}
event_body["tags"] = body.get("tags") or []
return self._map_event(event_body)
def sign_up(self, event_id: str, messenger_id: int) -> bool: # untested
"""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, requester_messenger_id: int) -> bool: # untested
resp = self.client.delete_event(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
)
return resp.status_code == 204 # in [200, 204]
def update_event( # untested
self,
event_id: str,
payload: dict,
requester_messenger_id: int,
) -> EventSchema | None:
resp = self.client.update_event(
event_id,
payload,
requester_messenger_id=requester_messenger_id,
)
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]: # untested
"""
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( # untested
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: # untested
"""
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( # untested
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: # untested
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): # untested
# TODO: replace with client usage
return code in ["code", "код"]
def create_event( # untested
self,
*,
name: str,
description: str,
bonus_code: str,
points: int,
date_obj: datetime,
city_id: str,
requester_messenger_id: int,
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,
requester_messenger_id=requester_messenger_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]: # untested
"""
обрабатываем данные полученный из мастера создания,
посылаем их в бекенд.
"""
if not user:
raise Exception("Cannot create event: could not find user")
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:
raise Exception("Cannot use '-' for organizer, current user is None")
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,
requester_messenger_id=user.messenger_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: # untested
"""преобразует 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: # untested
if not value:
return datetime.now(timezone.utc)
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(timezone.utc)
--- services/bot/core/clients/event.py ---
from core.clients.base import BaseApiClient
class EventClient(BaseApiClient): # untested
def search_events( # untested
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,
):
"""GET /events/search"""
params: dict = {
"offset": offset,
"limit": limit,
"only_claimed": bool(only_claimed),
"only_organizer": bool(only_organizer),
"only_participant": bool(only_participant),
}
if city_id is not None:
params["city_id"] = city_id
if requester_messenger_id is not None:
params["requester_messenger_id"] = int(requester_messenger_id)
if has_report is not None:
params["has_report"] = bool(has_report)
if tag_ids:
params["tag_ids"] = list(tag_ids)
if cursor_date is not None:
params["cursor_date"] = cursor_date
if cursor_id is not None:
params["cursor_id"] = cursor_id
return self.get("/events/search", params=params)
def get_city_events(self, city_id: str, offset: int = 0, limit: int = 50): # untested
"""GET /events/id/{city_id}"""
return self.get(f"/events/id/{city_id}", params={"offset": offset, "limit": limit})
def get_event_info(self, event_id: str, requester_messenger_id: int):
"""GET /events/{event_id}/card"""
return self.get(
f"GET /events/{event_id}/card",
headers={"Authorization": str(requester_messenger_id)},
)
def get_event_participants( # untested
self,
event_id: str,
requester_messenger_id: int,
offset: int = 0,
limit: int = 50,
):
"""GET /events/{event_id}/participants?requester_messenger_id=..."""
return self.get(
f"/events/{event_id}/participants",
params={
"requester_messenger_id": requester_messenger_id,
"offset": offset,
"limit": limit,
},
)
def get_my_events(self, messenger_id: int): # untested
"""GET /events/my?user_id={messenger_id}"""
return self.get("/events/my", params={"user_id": messenger_id})
def create_event( # untested
self,
*,
name: str,
description: str,
bonus_code: str,
points: int,
date: str,
city_id: str,
location: str | None = None,
app_tag_ids: list[str] | None = None,
creator_id: str | None = None,
requester_messenger_id: int,
):
"""POST /events"""
payload = {
"name": name,
"description": description,
"bonus_code": bonus_code,
"points": points,
"date": date,
"city_id": city_id,
"location": location,
"app_tag_ids": app_tag_ids or [],
}
if creator_id is not None:
payload["creator_id"] = creator_id
return self.post(
"/events",
params={"requester_messenger_id": requester_messenger_id},
json=payload,
)
def delete_event(self, event_id: str, requester_messenger_id: int): # untested
"""DELETE /events/{event_id}"""
return self.delete(
f"/events/{event_id}",
params={"requester_messenger_id": requester_messenger_id},
)
def sign_out(self, participation_id: str): # untested
"""DELETE /events/sign-out"""
return self.delete("/events/sign-out", json={"participation_id": participation_id})
def sign_up(self, event_id: str, messenger_id: int): # untested
"""POST /events/sign-up"""
return self.post(
"/events/sign-up", json={"event_id": event_id, "messenger_id": messenger_id}
)
def enter_code(self, participation_id: str, bonus_code: str): # untested
"""POST /events/code"""
return self.post(
"/events/code", json={"participation_id": participation_id, "bonus_code": bonus_code}
)
def update_event(self, event_id: str, payload: dict, requester_messenger_id: int): # untested
"""PATCH /events/{event_id}"""
return self.patch(
f"/events/{event_id}",
params={"requester_messenger_id": requester_messenger_id},
json=payload,
)
--- services/bot/core/utils/events_filter.py ---
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal, cast
EventsMode = Literal["all", "participant", "organizer"]
FILTERS_CTX_KEY = "events_filters_v1"
@dataclass(slots=True)
class EventsFilter: # untested
"""
фильтры на стороне бота (храним в fsmcontext.data).
- date_from/date_to сейчас не поддержаны backend (/events/search), поэтому
они участвуют только в UI/summary, но не уходят в query params.
"""
city_id: str | None = None
city_name: str | None = None
tag_ids: list[str] = field(default_factory=list)
tag_names: list[str] = field(default_factory=list)
mode: EventsMode = "all"
# UI-only (пока не поддержано backend)
date_from: str | None = None
date_to: str | None = None
@classmethod
def from_dict(cls, raw: Any) -> EventsFilter: # untested
if not isinstance(raw, dict):
return cls()
mode_raw = raw.get("mode") or "all"
if mode_raw not in ("all", "participant", "organizer"):
mode_raw = "all"
mode: EventsMode = cast(EventsMode, mode_raw)
city_id = raw.get("city_id")
city_id = str(city_id) if city_id else None
city_name = raw.get("city_name")
city_name = str(city_name) if city_name else None
tag_ids_raw = raw.get("tag_ids") or []
if not isinstance(tag_ids_raw, list):
tag_ids_raw = []
tag_ids = [str(x) for x in tag_ids_raw if str(x).strip()]
tag_names_raw = raw.get("tag_names") or []
if not isinstance(tag_names_raw, list):
tag_names_raw = []
tag_names = [str(x) for x in tag_names_raw if str(x).strip()]
date_from = raw.get("date_from")
date_from = str(date_from) if date_from else None
date_to = raw.get("date_to")
date_to = str(date_to) if date_to else None
return cls(
city_id=city_id,
city_name=city_name,
tag_ids=tag_ids,
tag_names=tag_names,
mode=mode,
date_from=date_from,
date_to=date_to,
)
def to_dict(self) -> dict[str, Any]: # untested
return {
"city_id": self.city_id,
"city_name": self.city_name,
"tag_ids": list(self.tag_ids or []),
"tag_names": list(self.tag_names or []),
"mode": self.mode,
"date_from": self.date_from,
"date_to": self.date_to,
}
@dataclass(frozen=True, slots=True)
class EventsSearchBase: # untested
"""
базовые параметры конкретного списка (контекст):
- например points всегда only_claimed=True
- my_events (организатор) может фиксировать режим
"""
city_id: str | None
requester_messenger_id: int | None
only_claimed: bool = False
only_organizer: bool = False
only_participant: bool = False
has_report: bool | None = None
tag_ids: list[str] = field(default_factory=list)
# если True — mode из EventsFilter игнорируется
lock_mode: bool = False
def build_events_search_params( # untested
*,
base: EventsSearchBase,
flt: EventsFilter | None,
offset: int,
limit: int,
) -> dict[str, Any]:
"""собираем kwargs для EventService.search_events_with_total(...)"""
f = flt or EventsFilter()
# city/tags: если в фильтре задано — переопределяет базу
city_id = f.city_id if f.city_id else base.city_id
tag_ids = f.tag_ids if f.tag_ids else (base.tag_ids or [])
only_claimed = bool(base.only_claimed)
only_organizer = bool(base.only_organizer)
only_participant = bool(base.only_participant)
# mode: работаем только с organizer/participant
if not base.lock_mode:
if f.mode == "organizer":
only_organizer = True
only_participant = False
elif f.mode == "participant":
only_participant = True
only_organizer = False
else:
# all
# оставляем только то, что задано базой
pass
params: dict[str, Any] = {
"offset": int(offset),
"limit": int(limit),
"only_claimed": bool(only_claimed),
"only_organizer": bool(only_organizer),
"only_participant": bool(only_participant),
}
if city_id is not None:
params["city_id"] = str(city_id)
if base.requester_messenger_id is not None:
params["requester_messenger_id"] = int(base.requester_messenger_id)
if base.has_report is not None:
params["has_report"] = bool(base.has_report)
if tag_ids:
params["tag_ids"] = list(tag_ids)
return params
def format_events_filter_summary( # untested
flt: EventsFilter | None,
*,
locked_mode: EventsMode | None = None,
) -> str:
f = flt or EventsFilter()
parts: list[str] = []
eff_mode: EventsMode = (
locked_mode if locked_mode in ("all", "participant", "organizer") else f.mode
)
suffix = " (заблокирован)" if locked_mode in ("participant", "organizer") else ""
if f.tag_names:
names = [str(x).strip() for x in f.tag_names if str(x).strip()]
parts.append("теги: " + ", ".join(names))
elif f.tag_ids:
parts.append(f"теги: {len(f.tag_ids)}")
if eff_mode == "participant":
parts.append("режим: участвую" + suffix)
elif eff_mode == "organizer":
parts.append("режим: организую" + suffix)
if f.city_id:
if f.city_name:
parts.append(f"город: {f.city_name}...")
else:
parts.append(f"город: {str(f.city_id)[:8]}...")
if f.date_from or f.date_to:
a = f.date_from or "..."
b = f.date_to or "..."
parts.append(f"даты: {a} - {b}")
if not parts:
return "Фильтр\nбез доп. условий"
return "Фильтр\n" + "\n".join(parts)
def get_events_filter(context, *, ctx_name: str) -> EventsFilter: # untested
data = (context.get_data() or {}).get(FILTERS_CTX_KEY) or {}
if not isinstance(data, dict):
return EventsFilter()
return EventsFilter.from_dict(data.get(str(ctx_name)))
def set_events_filter(context, *, ctx_name: str, flt: EventsFilter) -> None: # untested
raw = (context.get_data() or {}).get(FILTERS_CTX_KEY) or {}
data = dict(raw) if isinstance(raw, dict) else {}
data[str(ctx_name)] = flt.to_dict()
context.update_data({FILTERS_CTX_KEY: data})
def reset_events_filter(context, *, ctx_name: str) -> None: # untested
raw = (context.get_data() or {}).get(FILTERS_CTX_KEY) or {}
data = dict(raw) if isinstance(raw, dict) else {}
data.pop(str(ctx_name), None)
context.update_data({FILTERS_CTX_KEY: data})
def clear_context_keep_events_filters(context) -> None: # untested
"""
нужен для мест, где раньше делали context.clear(),
но теперь фильтры должны переживать пагинацию
"""
data = context.get_data() or {}
keep = data.get(FILTERS_CTX_KEY)
context.clear()
if keep is not None:
context.update_data({FILTERS_CTX_KEY: keep})
--- services/bot/core/handlers/events_filters.py ---
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent
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): # untested
tags = State()
class EventsFiltersCityState(StatesGroup): # untested
city = State()
class EventsFiltersModeState(StatesGroup): # untested
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: # untested
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]: # untested
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( # untested
*, 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( # untested
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( # untested
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( # untested
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( # untested
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
): # untested
# 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]: # untested
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( # untested
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( # untested
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: # untested
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( # untested
*,
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( # untested
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( # untested
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: # untested
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: # untested
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( # untested
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( # untested
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[assignment]
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
--- services/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]: # untested
return MediaGroupBuilder(
[Button(media_id="events_filters_open", value=f"{ctx_name}|{offset}", label="🧰 Фильтры")]
).build()
def events_filters_menu_keyboard(
*, ctx_name: str, offset: int
) -> list[InteractiveMediaGroup]: # untested
# 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( # untested
*,
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( # untested
*,
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]: # untested
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