Untitled
4ae4d
plain_text
a month ago
118 kB
3
Indexable
diff --git a/services/bot/core/config/__init__.py b/services/bot/core/config/__init__.py
index c476155..6901585 100644
--- a/services/bot/core/config/__init__.py
+++ b/services/bot/core/config/__init__.py
@@ -1,9 +1,11 @@
from .bootstrap import BotWithDI, bot
from .settings import bot_config, settings
+from .validation_config import validation_cfg
__all__ = [
"BotWithDI",
"bot",
"bot_config",
"settings",
+ "validation_cfg",
]
diff --git a/services/bot/core/config/validation_config.py b/services/bot/core/config/validation_config.py
index f9c3b23..3a4dd30 100644
--- a/services/bot/core/config/validation_config.py
+++ b/services/bot/core/config/validation_config.py
@@ -1,28 +1,77 @@
import os
+
import yaml
+
class ValidationConfig:
-
def __init__(self, path: str = "/bot/validation.yml"):
if not os.path.exists(path):
raise FileNotFoundError(f"Файл не найден: {path}")
try:
- with open(path, "r", encoding="utf-8") as f:
- self._data = yaml.safe_load(f) or {}
+ with open(path, encoding="utf-8") as f:
+ raw = yaml.safe_load(f) or {}
except yaml.YAMLError as e:
raise ValueError(f"Ошибка парсинга YAML: {e}") from e
+ self._data = self._cast_numeric_rules(raw)
+
+ def _cast_numeric_rules(self, data: dict) -> dict:
+ result = {}
+ for k, v in data.items():
+ if isinstance(v, dict):
+ result[k] = self._cast_numeric_rules(v)
+ elif k in ("min", "max", "min_length", "max_length"):
+ try:
+ result[k] = int(v)
+ except (TypeError, ValueError):
+ result[k] = v
+ else:
+ result[k] = v
+ return result
+
def get(self, key: str, default=None):
+ # Если ключ не начинается с известного корневого раздела — добавляем event
+ if not any(key.startswith(prefix) for prefix in ("event.", "event")):
+ key = f"event.{key}"
+
val = self._data
- for k in key.split("."):
- if isinstance(val, dict) and k in val:
- val = val[k]
+ for part in key.split("."):
+ if isinstance(val, dict) and part in val:
+ val = val[part]
else:
return default
return val
- def __getitem__(self, key):
- return self._data[key]
+ def __getitem__(self, key: str):
+ """
+ - cfg["title.min_length"] → event.title.min_length
+ - cfg["event.hours.min"] → event.hours.min
+ - cfg["event"]["title"] → прямой доступ к вложенному словарю
+ """
+ if "." in key:
+ path = key
+ if not any(path.startswith(p) for p in ("event.", "event")):
+ path = f"event.{path}"
+
+ val = self._data
+ for part in path.split("."):
+ if isinstance(val, dict) and part in val:
+ val = val[part]
+ else:
+ raise KeyError(f"Ключ не найден: {key}")
+ return val
+
+ if key in self._data:
+ return self._data[key]
+ raise KeyError(f"Ключ не найден: {key}")
+
+ def __contains__(self, key: str) -> bool:
+ try:
+ self[key]
+ return True
+ except KeyError:
+ return False
+
- def __contains__(self, key):
- return key in self._data
\ No newline at end of file
+# Глобальный инстанс
+validation_cfg = ValidationConfig()
diff --git a/services/bot/core/handlers/admin.py b/services/bot/core/handlers/admin.py
index 5e29941..0c3f9aa 100644
--- a/services/bot/core/handlers/admin.py
+++ b/services/bot/core/handlers/admin.py
@@ -54,7 +54,7 @@ def set_moderator_value_handler(
try:
moderator_id, tb_id = map(str, text.split(maxsplit=1))
if " " in moderator_id + tb_id:
- raise RuntimeError
+ raise RuntimeError()
moderator_id = int(moderator_id)
except Exception:
bot.messaging.send_message(
@@ -115,7 +115,7 @@ def del_moderator_value_handler(
try:
moderator_id, tb_id = map(str, text.split(maxsplit=1))
if " " in moderator_id + tb_id:
- raise RuntimeError
+ raise RuntimeError()
moderator_id = int(moderator_id)
except Exception:
bot.messaging.send_message(
diff --git a/services/bot/core/handlers/moderation.py b/services/bot/core/handlers/moderation.py
index b8a32e1..f4a15ff 100644
--- a/services/bot/core/handlers/moderation.py
+++ b/services/bot/core/handlers/moderation.py
@@ -1,4 +1,3 @@
-import re
from datetime import datetime
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent, UpdateMessage
@@ -6,7 +5,7 @@ from dialog_bot_sdk.interactive_media import Button
from core.bot_kit.fsm import FSMContext, State, StatesGroup
from core.bot_kit.router import Router
-from core.config import bot
+from core.config import bot, validation_cfg
from core.handlers.events_ui import (
build_back_to_event_editing_keyboard,
build_back_to_event_keyboard,
@@ -17,6 +16,26 @@ from core.handlers.events_ui import (
split_event_value,
)
from core.handlers.request import build_request_actions_keyboard
+from core.handlers.validators import (
+ DescriptionOutOfRange,
+ EventValidationError,
+ HoursInvalidFormat,
+ HoursOutOfRange,
+ ParticipationLimitInvalidFormat,
+ ParticipationLimitOutOfRange,
+ TitleOutOfRange,
+ UnsupportedDateFormat,
+ UnsupportedTimeFormat,
+ VerificationCodeInvalidFormat,
+ VerificationCodeOutOfRange,
+ validate_date,
+ validate_description,
+ validate_hours,
+ validate_participation_limit,
+ validate_time,
+ validate_title,
+ validate_verification_code,
+)
from core.markups import (
back_to_moderation_and_skip_keyboard,
back_to_moderation_keyboard,
@@ -41,7 +60,107 @@ from core.utils import (
TAGS_MEDIA_ID = "event_tags_toggle"
TAGS_TOGGLE_PREFIX = "toggle:"
TAGS_DONE_VALUE = "done"
+
+
# TODO убрать хардкод
+# ─────────────────────────────────────────────────────────────
+# 🔧 Хелперы для адаптивных сообщений (используют validation_cfg["..."])
+# ─────────────────────────────────────────────────────────────
+def _cfg(field: str, rule: str, default=None):
+ """Короткий доступ: _cfg('title', 'min_length', 4) → validation_cfg["title.min_length"] или default"""
+ try:
+ return validation_cfg[f"{field}.{rule}"]
+ except KeyError:
+ return default
+
+
+def _err(field: str, message: str, **kwargs):
+ """Форматирует ошибку, подставляя лимиты из YAML: {min}, {max}, {min_len}, {max_len}"""
+ limits = {
+ "min": _cfg(field, "min"),
+ "max": _cfg(field, "max"),
+ "min_len": _cfg(field, "min_length"),
+ "max_len": _cfg(field, "max_length"),
+ }
+ limits = {k: v for k, v in limits.items() if v is not None}
+ return message.format(**kwargs, **limits)
+
+
+def _validation_error_text(error: EventValidationError) -> str:
+ if isinstance(error, TitleOutOfRange):
+ return _err(
+ "title",
+ "⚠️ Название должно содержать от {min_len} до {max_len} символов. Попробуй ещё раз:",
+ )
+
+ if isinstance(error, DescriptionOutOfRange):
+ return _err(
+ "description",
+ "⚠️ Описание не должно быть длиннее {max_len} символов. Попробуй ещё раз:",
+ )
+
+ if isinstance(error, UnsupportedDateFormat):
+ return (
+ "⚠️ Я не смог распознать дату.\n"
+ "Пожалуйста, введи дату в формате `dd.mm.yyyy`, например: `20.02.2002`:"
+ )
+
+ if isinstance(error, UnsupportedTimeFormat):
+ return (
+ "⚠️ Я не смог распознать время.\n"
+ "Пожалуйста, введи время в формате `HH:MM`, например: `10:30`:"
+ )
+
+ if isinstance(error, (HoursInvalidFormat, HoursOutOfRange)):
+ return _err(
+ "hours",
+ "⚠️ Количество часов должно быть целым числом от {min} до {max}. Попробуй ещё раз:",
+ )
+
+ if isinstance(error, (ParticipationLimitInvalidFormat, ParticipationLimitOutOfRange)):
+ return _err(
+ "participation_limit",
+ "⚠️ Количество участников должно быть целым числом от {min} до {max}. Попробуй ещё раз:",
+ )
+
+ if isinstance(error, VerificationCodeOutOfRange):
+ return _err(
+ "verification_code",
+ "⚠️ Код должен быть длиной от {min_len} до {max_len} символов. Попробуй ещё раз:",
+ )
+
+ if isinstance(error, VerificationCodeInvalidFormat):
+ return (
+ "⚠️ Код должен содержать только русские или латинские буквы и цифры. Попробуй ещё раз:"
+ )
+
+ return "⚠️ Некорректное значение. Попробуй ещё раз:"
+
+
+def _msg_err_create(message, error) -> None:
+ bot.messaging.send_message(
+ peer=message.peer,
+ text=_validation_error_text(error),
+ interactive_media_groups=back_to_moderation_keyboard(),
+ )
+
+
+def _msg_err_edit(
+ message,
+ error,
+ event_id: str | None = None,
+ ctx_name: str | None = None,
+) -> None:
+ if event_id is None or ctx_name is None:
+ raise Exception("event_id and ctx_name must be provided for _msg_err_edit helper function")
+ bot.messaging.send_message(
+ peer=message.peer,
+ text=_validation_error_text(error),
+ interactive_media_groups=build_back_to_event_editing_keyboard(
+ event_id=event_id,
+ ctx_name=ctx_name,
+ ),
+ )
events_rt = Router()
@@ -239,10 +358,12 @@ def create_event_start_handler(
@bot.di
def event_create_name_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
-
- name = message.message.text_message.text.strip()
+ try:
+ name = validate_title(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
+ return
context.update_data({"event_name": name})
-
bot.messaging.send_message(
peer=message.peer,
text="Шаг 2/11: введи, пожалуйста, дату мероприятия например, в формате `20.02.2002`:",
@@ -251,58 +372,23 @@ def event_create_name_step(message: UpdateMessage, context: FSMContext):
context.set_state(EventCreateState.date)
-def _validate_date(text): # throws ValueError
- formats = []
- for first_gap in ". -_":
- for second_gap in ". -_":
- formats.append(f"%d{first_gap}%m{second_gap}%Y")
-
- date_obj = None
- for format in formats:
- try:
- date_obj = datetime.strptime(text, format).date()
- except ValueError:
- continue
-
- if date_obj is None:
- raise ValueError(f"unsupported date format in date {text!r}")
-
- return date_obj
-
-
@events_rt.message(state=EventCreateState.date)
@bot.di
def event_create_date_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
-
- date = message.message.text_message.text.strip()
-
try:
- date_obj = _validate_date(date)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Я не смог распознать дату.\n"
- "Пожалуйста, введи дату, например, в формате `20.02.2002`:"
- ),
- interactive_media_groups=back_to_moderation_keyboard(),
- )
+ date_obj = validate_date(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
return
-
if date_obj <= datetime.now().date():
bot.messaging.send_message(
peer=message.peer,
- text=(
- "⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\n"
- "Пожалуйста, введи дату, например, в формате `20.02.2002`:"
- ),
+ text="⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\nПожалуйста, введи дату, например, в формате `20.02.2002`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
-
context.update_data({"event_date": date_obj})
-
bot.messaging.send_message(
peer=message.peer,
text="Шаг 3/11: введи, пожалуйста, время мероприятия например, в формате `10:30`:",
@@ -311,49 +397,20 @@ def event_create_date_step(message: UpdateMessage, context: FSMContext):
context.set_state(EventCreateState.time)
-def _validate_time(text): # throws ValueError
- time_obj = None
- for format in ["%H:%M", "%H %M", "%H-%M", "%H_%M"]:
- try:
- time_obj = datetime.strptime(text, format).time()
- except ValueError:
- continue
-
- if time_obj is None:
- raise ValueError(f"unsupported time format in time {text!r}", text)
-
- return time_obj
-
-
@events_rt.message(state=EventCreateState.time)
@bot.di
def event_create_time_step(
message: UpdateMessage, context: FSMContext, terbank_service: TerbankService
):
delete_prev_message(bot, message)
-
- time = message.message.text_message.text.strip()
-
try:
- time_obj = _validate_time(time)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Я не смог распознать время.\n"
- "Пожалуйста, введи в формате `HH:MM`, например: `10:30`:"
- ),
- interactive_media_groups=back_to_moderation_keyboard(),
- )
+ time_obj = validate_time(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
return
-
context.update_data({"event_time": time_obj})
-
send_tb_select_menu(
- peer=message.peer,
- terbank_service=terbank_service,
- context=context,
- is_edit=False,
+ peer=message.peer, terbank_service=terbank_service, context=context, is_edit=False
)
@@ -381,12 +438,8 @@ def _tb_select_footer_buttons(
@bot.di
-def event_create_back_to_time_handler(
- event: UpdateInteractiveMediaEvent,
- context: FSMContext,
-):
+def event_create_back_to_time_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
delete_prev_message_by_peer(bot, event.peer)
-
bot.messaging.send_message(
peer=event.peer,
text="Шаг 3/11: введи, пожалуйста, время мероприятия например, в формате `10:30`:",
@@ -619,8 +672,12 @@ def _handle_project_input(peer, raw_text: str, context: FSMContext):
@bot.di
def event_create_description_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
+ try:
+ description = validate_description(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
+ return
- description = message.message.text_message.text.strip()
context.update_data({"event_description": description})
bot.messaging.send_message(
peer=message.peer,
@@ -630,20 +687,6 @@ def event_create_description_step(message: UpdateMessage, context: FSMContext):
context.set_state(EventCreateState.hours)
-def _validate_hours_amount(text) -> int: # throws ValueError
- if not isinstance(text, int) and not isinstance(text, str):
- logger.error("expected type str or int for hours amount, got '%s'", type(text))
- raise ValueError(f"expected type str or int for hours amount, got {type(text)!r}")
- text = str(text)
- if not re.fullmatch(r"^[0-9]{1,2}$", text):
- logger.error("expected <= 2 long digit sequence for hours amount, got '%s'", text)
- raise ValueError(f"expected <= 2 long digit sequence for hours amount, got {text!r}")
- hours = int(text)
- if hours <= 0 or hours >= 19:
- raise ValueError(f"hours value must be positive integer <= 18, got {hours!r}")
- return hours
-
-
@events_rt.message(state=EventCreateState.hours)
@bot.di
def event_create_hours_step(
@@ -652,16 +695,10 @@ def event_create_hours_step(
tag_service: TagService,
):
delete_prev_message(bot, message)
-
- text = message.message.text_message.text.strip()
try:
- hours = _validate_hours_amount(text)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text="⚠️ Количество часов должно быть натуральным числом, меньшим 19. Попробуй ещё раз.",
- interactive_media_groups=back_to_moderation_keyboard(),
- )
+ hours = validate_hours(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
return
context.update_data({"hours": hours})
@@ -719,7 +756,10 @@ def event_tags_toggle_handler(
context.update_data({"event_app_tag_ids": app_tag_ids})
bot.messaging.send_message(
peer=event.peer,
- text="Шаг 10/11: введи код мероприятия:\nБез пробелов, не больше 16 символов. Например: WELCOME2025",
+ text=_err(
+ "verification_code",
+ "Шаг 10/11: введи код мероприятия:\nБез пробелов, от {min_len} до {max_len} символов. Например: WELCOME2025",
+ ),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.code)
@@ -840,40 +880,6 @@ def event_tags_toggle_handler(
return
-def _validate_verification_code(text) -> str: # throws ValueError
- if not isinstance(text, str):
- logger.error("expected type str for verification code, got '%s'", type(text))
- raise ValueError(f"expected type str for verification code, got {type(text)!r}")
- if not re.fullmatch(r"^[A-Za-zА-Яа-я0-9]{1,16}$", text):
- logger.error("unsupported verification code format: '%s'", text)
- raise ValueError(f"unsupported verification code format: {text!r}")
- return text
-
-
-def _validate_participation_limit(value) -> int: # throws ValueError
- try:
- limit = int(value)
- except (TypeError, ValueError) as e:
- logger.error(
- "expected int or numeric string for participation limit, got '%s', which has type '%s'",
- value,
- type(value),
- )
- raise ValueError(
- f"expected int or numeric string for participation limit, got {value!r}, which has type {type(value)!r}"
- ) from e
-
- if limit < 1:
- logger.error("participation limit must be positive integer, got '%d'", limit)
- raise ValueError(f"participation limit must be positive integer, got {limit!r}")
-
- if limit > 10000:
- logger.error("participation limit is too high: %d", limit)
- raise ValueError(f"participation limit is too high: {limit}")
-
- return limit
-
-
@events_rt.message(state=EventCreateState.code)
@bot.di
def event_create_code_step(
@@ -885,25 +891,20 @@ def event_create_code_step(
user: UserSchema,
):
delete_prev_message(bot, message)
-
try:
- code = _validate_verification_code(message.message.text_message.text.strip())
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Код должен быть комбинацией русских и латинских букв и цифр, "
- "длиной не больше 16 символов. Попробуй еще раз:"
- ),
- interactive_media_groups=back_to_moderation_keyboard(),
- )
+ code = validate_verification_code(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
return
context.update_data({"event_verification_code": code})
bot.messaging.send_message(
peer=message.peer,
- text="Шаг 11/11: введи требуемое количество участников:\nПусть это будет число от 1 до 10000",
+ text=_err( # 🔹 адаптивный текст через _err
+ "participation_limit",
+ "Шаг 11/11: введи требуемое количество участников:\nПусть это будет число от {min} до {max}",
+ ),
interactive_media_groups=back_to_moderation_keyboard(),
)
@@ -921,15 +922,10 @@ def event_create_participation_limit_step(
user: UserSchema,
):
delete_prev_message(bot, message)
-
try:
- limit = _validate_participation_limit(message.message.text_message.text.strip())
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=("⚠️ Количество участников должно быть числом от 1 до 10000. Попробуй еще раз:"),
- interactive_media_groups=back_to_moderation_keyboard(),
- )
+ limit = validate_participation_limit(message.message.text_message.text.strip())
+ except EventValidationError as e:
+ _msg_err_create(message, e)
return
context.update_data({"event_participation_limit": limit})
@@ -1415,7 +1411,15 @@ def event_menu_edit_field_handler(
elif field == "time":
hint = "*Введи время в формате* `HH:MM`, например: `10:30`"
elif field == "participation_limit":
- hint = "*Введи максимальное количество участников* от 1 до 10000"
+ hint = _err("participation_limit", "*Введи число от {min} до {max}*")
+ elif field == "hours":
+ hint = _err("hours", "*Введи число от {min} до {max}*")
+ elif field == "title":
+ hint = _err("title", "*Введи от {min_len} до {max_len} символов*")
+ elif field == "code":
+ hint = _err("verification_code", "*От {min_len} до {max_len} символов, буквы и цифры*")
+ elif field == "description":
+ hint = _err("description", "*Максимум {max_len} символов*")
else:
hint = "*Введи новое значение.* Чтобы очистить поле - отправь: «-»"
@@ -1469,65 +1473,50 @@ def event_edit_value_step(
if field == "hours":
try:
- value = _validate_hours_amount(value)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text="⚠️ Количество часов должно быть натуральным числом, меньшим 19. Попробуй ещё раз.",
- interactive_media_groups=build_back_to_event_editing_keyboard(
- event_id=event_id,
- ctx_name=ctx_name,
- ),
- )
+ value = validate_hours(value)
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
return
if field == "participation_limit":
try:
- value = _validate_participation_limit(value)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Количество участников должно быть числом от 1 до 10000. Попробуй еще раз:"
- ),
- interactive_media_groups=build_back_to_event_editing_keyboard(
- event_id=event_id,
- ctx_name=ctx_name,
- ),
- )
+ value = validate_participation_limit(value)
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
return
if field == "code":
try:
- value = _validate_verification_code(value)
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Код должен быть комбинацией русских и латинских букв и цифр, "
- "длиной не больше 16 символов. Попробуй еще раз:"
- ),
- interactive_media_groups=build_back_to_event_editing_keyboard(
- event_id=event_id,
- ctx_name=ctx_name,
- ),
- )
+ value = validate_verification_code(str(value))
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
+ return
+
+ if field == "title":
+ try:
+ value = validate_title(str(value))
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
+ return
+
+ if field == "description":
+ try:
+ value = validate_description(str(value))
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
return
if field == "date":
try:
- date_obj = _validate_date(str(value))
- except ValueError:
+ date_obj = validate_date(str(value))
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
+ return
+ if date_obj <= datetime.now().date():
bot.messaging.send_message(
peer=message.peer,
- text=(
- "⚠️ Я не смог распознать дату.\n"
- "Пожалуйста, введи дату в формате `dd.mm.yyyy`, например: `20.02.2002`:"
- ),
- interactive_media_groups=build_back_to_event_editing_keyboard(
- event_id=event_id,
- ctx_name=ctx_name,
- ),
+ text="⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\nПожалуйста, введи дату, например, в формате `20.02.2002`:",
+ interactive_media_groups=back_to_moderation_keyboard(),
)
return
@@ -1556,19 +1545,9 @@ def event_edit_value_step(
if field == "time":
try:
- time_obj = _validate_time(str(value))
- except ValueError:
- bot.messaging.send_message(
- peer=message.peer,
- text=(
- "⚠️ Я не смог распознать время.\n"
- "Пожалуйста, введи в формате `HH:MM`, например: `10:30`:"
- ),
- interactive_media_groups=build_back_to_event_editing_keyboard(
- event_id=event_id,
- ctx_name=ctx_name,
- ),
- )
+ time_obj = validate_time(str(value))
+ except EventValidationError as e:
+ _msg_err_edit(message, e, event_id, ctx_name)
return
current_event_card = event_service.get_event_by_id(event_id, message.peer.id)
diff --git a/services/bot/core/handlers/report.py b/services/bot/core/handlers/report.py
index 2b35302..5834416 100644
--- a/services/bot/core/handlers/report.py
+++ b/services/bot/core/handlers/report.py
@@ -327,10 +327,10 @@ def report_create_beneficiaries_step(message: UpdateMessage, context: FSMContext
text = message.message.text_message.text.strip()
try:
if not re.fullmatch(r"^[0-9]{1,9}$", text):
- raise ValueError
+ raise ValueError()
val = int(text)
if val < 0 or val >= 2**31:
- raise ValueError
+ raise ValueError()
except ValueError:
bot.messaging.send_message(
peer=message.peer,
diff --git a/services/bot/core/handlers/validators.py b/services/bot/core/handlers/validators.py
new file mode 100644
index 0000000..86ea8de
--- /dev/null
+++ b/services/bot/core/handlers/validators.py
@@ -0,0 +1,142 @@
+import re
+from datetime import datetime
+from typing import TypeVar
+
+from core.config import validation_cfg
+
+T = TypeVar("T")
+
+
+class EventValidationError(ValueError): ...
+
+
+class TitleOutOfRange(EventValidationError): ...
+
+
+class DescriptionOutOfRange(EventValidationError): ...
+
+
+class UnsupportedDateFormat(EventValidationError): ...
+
+
+class UnsupportedTimeFormat(EventValidationError): ...
+
+
+class HoursInvalidFormat(EventValidationError): ...
+
+
+class HoursOutOfRange(EventValidationError): ...
+
+
+class VerificationCodeInvalidFormat(EventValidationError): ...
+
+
+class VerificationCodeOutOfRange(EventValidationError): ...
+
+
+class ParticipationLimitInvalidFormat(EventValidationError): ...
+
+
+class ParticipationLimitOutOfRange(EventValidationError): ...
+
+
+def _get_rule(field: str, rule: str, default: T, excepted_type: type[T]) -> T:
+ value = validation_cfg.get(f"event.{field}.{rule}", default)
+
+ if not isinstance(value, excepted_type):
+ raise RuntimeError(
+ f"validation rule event.{field}.{rule} must be {excepted_type}, got {type(value)}"
+ )
+ return value
+
+
+def validate_title(text: str):
+ min_len = _get_rule("title", "min_length", 4, int)
+ max_len = _get_rule("title", "max_length", 20, int)
+
+ if len(text) < min_len or len(text) > max_len:
+ raise TitleOutOfRange()
+ return text
+
+
+def validate_description(text: str):
+ max_len = _get_rule("description", "max_length", 20, int)
+
+ if len(text) > max_len:
+ raise DescriptionOutOfRange()
+ return text
+
+
+def validate_date(text: str):
+ """Парсинг даты с поддержкой разделителей . - _ . Возвращает date."""
+ formats = []
+ for first_gap in ". -_":
+ for second_gap in ". -_":
+ formats.append(f"%d{first_gap}%m{second_gap}%Y")
+
+ date_obj = None
+ for fmt in formats:
+ try:
+ date_obj = datetime.strptime(text.strip(), fmt).date()
+ break
+ except ValueError:
+ continue
+
+ if date_obj is None:
+ raise UnsupportedDateFormat()
+ return date_obj
+
+
+def validate_time(text: str):
+ time_obj = None
+ for fmt in ["%H:%M", "%H %M", "%H-%M", "%H_%M"]:
+ try:
+ time_obj = datetime.strptime(text.strip(), fmt).time()
+ break
+ except ValueError:
+ continue
+
+ if time_obj is None:
+ raise UnsupportedTimeFormat()
+ return time_obj
+
+
+def validate_hours(text: str | int):
+ """Валидация часов: 1-2 цифры, диапазон 1-18 (из YAML)."""
+ text = str(text).strip()
+ if not re.fullmatch(r"^[0-9]{1,2}$", text):
+ raise HoursInvalidFormat()
+
+ hours = int(text)
+ min_val = _get_rule("hours", "min", 1, int)
+ max_val = _get_rule("hours", "max", 18, int)
+
+ if hours < min_val or hours > max_val:
+ raise HoursOutOfRange()
+ return hours
+
+
+def validate_verification_code(text: str):
+ if not re.fullmatch(r"^[A-Za-zА-Яа-я0-9]+$", text):
+ raise VerificationCodeInvalidFormat()
+
+ min_len = _get_rule("verification_code", "min_length", 4, int)
+ max_len = _get_rule("verification_code", "max_length", 20, int)
+
+ if len(text) < min_len or len(text) > max_len:
+ raise VerificationCodeOutOfRange()
+ return text
+
+
+def validate_participation_limit(value: str | int):
+ try:
+ limit = int(value)
+ except (TypeError, ValueError) as e:
+ raise ParticipationLimitInvalidFormat from e
+
+ min_val = _get_rule("participation_limit", "min", 1, int)
+ max_val = _get_rule("participation_limit", "max", 1000, int)
+
+ if limit < min_val or limit > max_val:
+ raise ParticipationLimitOutOfRange()
+ return limit
а вот файлы, изменённые в этой ветке:
f=services/bot/core/config/__init__.py
f=services/bot/core/config/validation_config.py
f=services/bot/core/handlers/admin.py
f=services/bot/core/handlers/moderation.py
f=services/bot/core/handlers/report.py
--- services/bot/core/config/__init__.py ---
from .bootstrap import BotWithDI, bot
from .settings import bot_config, settings
from .validation_config import validation_cfg
__all__ = [
"BotWithDI",
"bot",
"bot_config",
"settings",
"validation_cfg",
]
--- services/bot/core/config/validation_config.py ---
import os
import yaml
class ValidationConfig:
def __init__(self, path: str = "/bot/validation.yml"):
if not os.path.exists(path):
raise FileNotFoundError(f"Файл не найден: {path}")
try:
with open(path, encoding="utf-8") as f:
raw = yaml.safe_load(f) or {}
except yaml.YAMLError as e:
raise ValueError(f"Ошибка парсинга YAML: {e}") from e
self._data = self._cast_numeric_rules(raw)
def _cast_numeric_rules(self, data: dict) -> dict:
result = {}
for k, v in data.items():
if isinstance(v, dict):
result[k] = self._cast_numeric_rules(v)
elif k in ("min", "max", "min_length", "max_length"):
try:
result[k] = int(v)
except (TypeError, ValueError):
result[k] = v
else:
result[k] = v
return result
def get(self, key: str, default=None):
# Если ключ не начинается с известного корневого раздела — добавляем event
if not any(key.startswith(prefix) for prefix in ("event.", "event")):
key = f"event.{key}"
val = self._data
for part in key.split("."):
if isinstance(val, dict) and part in val:
val = val[part]
else:
return default
return val
def __getitem__(self, key: str):
"""
- cfg["title.min_length"] → event.title.min_length
- cfg["event.hours.min"] → event.hours.min
- cfg["event"]["title"] → прямой доступ к вложенному словарю
"""
if "." in key:
path = key
if not any(path.startswith(p) for p in ("event.", "event")):
path = f"event.{path}"
val = self._data
for part in path.split("."):
if isinstance(val, dict) and part in val:
val = val[part]
else:
raise KeyError(f"Ключ не найден: {key}")
return val
if key in self._data:
return self._data[key]
raise KeyError(f"Ключ не найден: {key}")
def __contains__(self, key: str) -> bool:
try:
self[key]
return True
except KeyError:
return False
# Глобальный инстанс
validation_cfg = ValidationConfig()
--- services/bot/core/handlers/admin.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.markups import admin_menu_keyboard, back_to_admin_menu_keyboard
from core.services import AdminService
from core.utils import delete_prev_message, delete_prev_message_by_peer
admin_rt = Router()
class ManageModeratorState(StatesGroup):
set_wait_value = State()
del_wait_value = State()
def admin_menu_handler(event: UpdateInteractiveMediaEvent):
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Ты в доме администрации!\n\nПожалуйста, выбери, что хочешь сделать 👇",
interactive_media_groups=admin_menu_keyboard(),
)
@bot.di
def set_moderator_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
):
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Введи `messenger_id` модератора, которого хочешь назначить и `tb_id` (через пробел), в который ты хочешь его назначить.",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
context.set_state(ManageModeratorState.set_wait_value)
@admin_rt.message(state=ManageModeratorState.set_wait_value)
@bot.di
def set_moderator_value_handler(
message: UpdateMessage,
context: FSMContext,
admin_service: AdminService,
):
delete_prev_message(bot, message)
text = message.message.text_message.text.strip()
try:
moderator_id, tb_id = map(str, text.split(maxsplit=1))
if " " in moderator_id + tb_id:
raise RuntimeError()
moderator_id = int(moderator_id)
except Exception:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Нужно ввести два значения через пробел: `messenger_id` модератора, которого хочешь назначить и `tb_id`, в который ты хочешь его назначить. Попробуй ещё раз.",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
return
try:
admin_service.set_moderator(
messenger_id=message.peer.id,
tb_id=tb_id,
moderator_id=moderator_id,
)
except Exception as e:
bot.messaging.send_message(
peer=message.peer,
text=f"⚠️ Не удалось назначить модератора. Ошибка: {e}",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
bot.messaging.send_message(
peer=message.peer,
text="✅ Модератор успешно назначен!",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
context.set_state(None)
@bot.di
def del_moderator_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
):
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Введи `messenger_id` модератора, которого хочешь снять с должности и `tb_id` (через пробел), в который в котором ты хочешь его снять с должности.",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
context.set_state(ManageModeratorState.del_wait_value)
@admin_rt.message(state=ManageModeratorState.del_wait_value)
@bot.di
def del_moderator_value_handler(
message: UpdateMessage,
context: FSMContext,
admin_service: AdminService,
):
delete_prev_message(bot, message)
text = message.message.text_message.text.strip()
try:
moderator_id, tb_id = map(str, text.split(maxsplit=1))
if " " in moderator_id + tb_id:
raise RuntimeError()
moderator_id = int(moderator_id)
except Exception:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Нужно ввести два значения через пробел: `messenger_id` модератора, которого хочешь снять с должности и `tb_id`, в который ты хочешь его снять с должности. Попробуй ещё раз.",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
return
try:
admin_service.del_moderator(
messenger_id=message.peer.id,
tb_id=tb_id,
moderator_id=moderator_id,
)
except Exception as e:
bot.messaging.send_message(
peer=message.peer,
text=f"⚠️ Не удалось снять с должности модератора. Ошибка: {e}",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
bot.messaging.send_message(
peer=message.peer,
text="✅ Модератор успешно снят!",
interactive_media_groups=back_to_admin_menu_keyboard(),
)
context.set_state(None)
--- services/bot/core/handlers/moderation.py ---
from datetime import datetime
from dialog_bot_sdk.entities.messaging import UpdateInteractiveMediaEvent, UpdateMessage
from dialog_bot_sdk.interactive_media import Button
from core.bot_kit.fsm import FSMContext, State, StatesGroup
from core.bot_kit.router import Router
from core.config import bot, validation_cfg
from core.handlers.events_ui import (
build_back_to_event_editing_keyboard,
build_back_to_event_keyboard,
get_ui,
send_event_card,
send_event_cards_page,
send_my_events_page,
split_event_value,
)
from core.handlers.request import build_request_actions_keyboard
from core.handlers.validators import (
DescriptionOutOfRange,
EventValidationError,
HoursInvalidFormat,
HoursOutOfRange,
ParticipationLimitInvalidFormat,
ParticipationLimitOutOfRange,
TitleOutOfRange,
UnsupportedDateFormat,
UnsupportedTimeFormat,
VerificationCodeInvalidFormat,
VerificationCodeOutOfRange,
validate_date,
validate_description,
validate_hours,
validate_participation_limit,
validate_time,
validate_title,
validate_verification_code,
)
from core.markups import (
back_to_moderation_and_skip_keyboard,
back_to_moderation_keyboard,
choices_from_backend_tags,
event_edit_fields_keyboard,
format_event_details,
format_request_details,
gosb_select_keyboard,
moderation_menu_keyboard,
tags_toggle_keyboard,
tb_select_keyboard,
)
from core.schemas import EventCardSchema, TagSchema, UserSchema
from core.services import EventService, ReportService, RequestService, TagService, TerbankService
from core.utils import (
clear_context_keep_events_filters,
delete_prev_message,
delete_prev_message_by_peer,
logger,
)
TAGS_MEDIA_ID = "event_tags_toggle"
TAGS_TOGGLE_PREFIX = "toggle:"
TAGS_DONE_VALUE = "done"
# TODO убрать хардкод
# ─────────────────────────────────────────────────────────────
# 🔧 Хелперы для адаптивных сообщений (используют validation_cfg["..."])
# ─────────────────────────────────────────────────────────────
def _cfg(field: str, rule: str, default=None):
"""Короткий доступ: _cfg('title', 'min_length', 4) → validation_cfg["title.min_length"] или default"""
try:
return validation_cfg[f"{field}.{rule}"]
except KeyError:
return default
def _err(field: str, message: str, **kwargs):
"""Форматирует ошибку, подставляя лимиты из YAML: {min}, {max}, {min_len}, {max_len}"""
limits = {
"min": _cfg(field, "min"),
"max": _cfg(field, "max"),
"min_len": _cfg(field, "min_length"),
"max_len": _cfg(field, "max_length"),
}
limits = {k: v for k, v in limits.items() if v is not None}
return message.format(**kwargs, **limits)
def _validation_error_text(error: EventValidationError) -> str:
if isinstance(error, TitleOutOfRange):
return _err(
"title",
"⚠️ Название должно содержать от {min_len} до {max_len} символов. Попробуй ещё раз:",
)
if isinstance(error, DescriptionOutOfRange):
return _err(
"description",
"⚠️ Описание не должно быть длиннее {max_len} символов. Попробуй ещё раз:",
)
if isinstance(error, UnsupportedDateFormat):
return (
"⚠️ Я не смог распознать дату.\n"
"Пожалуйста, введи дату в формате `dd.mm.yyyy`, например: `20.02.2002`:"
)
if isinstance(error, UnsupportedTimeFormat):
return (
"⚠️ Я не смог распознать время.\n"
"Пожалуйста, введи время в формате `HH:MM`, например: `10:30`:"
)
if isinstance(error, (HoursInvalidFormat, HoursOutOfRange)):
return _err(
"hours",
"⚠️ Количество часов должно быть целым числом от {min} до {max}. Попробуй ещё раз:",
)
if isinstance(error, (ParticipationLimitInvalidFormat, ParticipationLimitOutOfRange)):
return _err(
"participation_limit",
"⚠️ Количество участников должно быть целым числом от {min} до {max}. Попробуй ещё раз:",
)
if isinstance(error, VerificationCodeOutOfRange):
return _err(
"verification_code",
"⚠️ Код должен быть длиной от {min_len} до {max_len} символов. Попробуй ещё раз:",
)
if isinstance(error, VerificationCodeInvalidFormat):
return (
"⚠️ Код должен содержать только русские или латинские буквы и цифры. Попробуй ещё раз:"
)
return "⚠️ Некорректное значение. Попробуй ещё раз:"
def _msg_err_create(message, error) -> None:
bot.messaging.send_message(
peer=message.peer,
text=_validation_error_text(error),
interactive_media_groups=back_to_moderation_keyboard(),
)
def _msg_err_edit(
message,
error,
event_id: str | None = None,
ctx_name: str | None = None,
) -> None:
if event_id is None or ctx_name is None:
raise Exception("event_id and ctx_name must be provided for _msg_err_edit helper function")
bot.messaging.send_message(
peer=message.peer,
text=_validation_error_text(error),
interactive_media_groups=build_back_to_event_editing_keyboard(
event_id=event_id,
ctx_name=ctx_name,
),
)
events_rt = Router()
class EventEditState(StatesGroup): # legacy
wait_value = State()
gosb = State()
tags = State()
class EventEnterCodeState(StatesGroup): # legacy
wait_code = State()
class EventViewState(StatesGroup): # legacy
wait_event_id = State()
class EventCreateState(StatesGroup): # legacy
name = State()
date = State()
time = State()
gosb = State()
project = State()
description = State()
hours = State()
tags = State()
code = State()
participation_limit = State()
@bot.di
def moderation_menu_handler(event: UpdateInteractiveMediaEvent, context: FSMContext): # legacy
clear_context_keep_events_filters(context)
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text=("Ты в доме модерации!\n\nПожалуйста, выбери, что хочешь сделать 👇"),
interactive_media_groups=moderation_menu_keyboard(),
)
@bot.di
def all_events_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_event_cards_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def my_events_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
send_my_events_page(
event.peer,
offset=0,
event_service=event_service,
requester_messenger_id=event.peer.id,
ctx_name="my_events",
context=context,
)
@bot.di
def my_events_page_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_my_events_page(
event.peer,
offset=offset,
event_service=event_service,
requester_messenger_id=event.peer.id,
ctx_name="my_events",
context=context,
)
@bot.di
def my_events_open_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx_name = split_event_value(event.data.value)
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name or "my_events",
report_service=report_service,
)
@bot.di
def all_events_page_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
clear_context_keep_events_filters(context)
try:
offset = int(event.data.value)
if offset < 0:
offset = 0
except Exception:
offset = 0
send_event_cards_page(
event.peer,
offset=offset,
event_service=event_service,
ctx_name="moderation",
context=context,
)
@bot.di
def all_events_open_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
user=user,
ctx_name=ctx_name,
report_service=report_service,
)
@bot.di
def create_event_start_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
):
"""старт мастера создания мероприятия"""
delete_prev_message_by_peer(bot, event.peer)
context.clear()
bot.messaging.send_message(
peer=event.peer,
text="Создание мероприятия.\n\nШаг 1/11: введи, пожалуйста, название мероприятия:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.name)
@events_rt.message(state=EventCreateState.name)
@bot.di
def event_create_name_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
try:
name = validate_title(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"event_name": name})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 2/11: введи, пожалуйста, дату мероприятия например, в формате `20.02.2002`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.date)
@events_rt.message(state=EventCreateState.date)
@bot.di
def event_create_date_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
try:
date_obj = validate_date(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
if date_obj <= datetime.now().date():
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\nПожалуйста, введи дату, например, в формате `20.02.2002`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"event_date": date_obj})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 3/11: введи, пожалуйста, время мероприятия например, в формате `10:30`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.time)
@events_rt.message(state=EventCreateState.time)
@bot.di
def event_create_time_step(
message: UpdateMessage, context: FSMContext, terbank_service: TerbankService
):
delete_prev_message(bot, message)
try:
time_obj = validate_time(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"event_time": time_obj})
send_tb_select_menu(
peer=message.peer, terbank_service=terbank_service, context=context, is_edit=False
)
def _tb_select_footer_buttons(
*,
is_edit: bool,
event_id: str | None,
ctx_name: str,
) -> list[Button]:
if is_edit:
return [
Button(
media_id="event_menu_edit",
value=f"{event_id}|{ctx_name}",
label="⬅️ К редактированию мероприятия",
)
]
return [
Button(
media_id="event_create_back_to_time",
value="1",
label="⬅️ Назад",
)
]
@bot.di
def event_create_back_to_time_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Шаг 3/11: введи, пожалуйста, время мероприятия например, в формате `10:30`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.time)
def send_tb_select_menu(
*,
peer,
terbank_service: TerbankService,
context: FSMContext,
is_edit: bool,
event_id: str | None = None,
ctx_name: str = "moderation",
default_tb_id: str | None = None,
default_gosb_id: str | None = None,
):
saved_event_id = event_id or context.get_data().get("edit_event_id")
if is_edit and not saved_event_id:
logger.error("error while editing gosb: edit_event_id is empty before tb select")
bot.messaging.send_message(
peer=peer,
text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново.",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.clear()
return
context.update_data(
{
"tb_id": default_tb_id or "",
"gosb_id": default_gosb_id or "",
"edit_current_tb_id": default_tb_id or "",
"edit_current_gosb_id": default_gosb_id or "",
"edit_event_id": saved_event_id,
"edit_event_ctx": ctx_name,
}
)
if is_edit:
context.set_state(EventEditState.gosb)
title = "Редактирование ГОСБ: сначала выбери тербанк:"
else:
context.set_state(EventCreateState.gosb)
title = "Шаг 4/11: выбери тербанк:"
tbs = terbank_service.get_all_tbs(peer.id)
bot.messaging.send_message(
peer=peer,
text=title,
interactive_media_groups=tb_select_keyboard(
terbanks=tbs,
default_value=default_tb_id,
footer_buttons=_tb_select_footer_buttons(
is_edit=is_edit,
event_id=saved_event_id,
ctx_name=ctx_name,
),
),
)
@bot.di
def tb_reselect_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
terbank_service: TerbankService,
):
delete_prev_message_by_peer(bot, event.peer)
data = context.get_data()
send_tb_select_menu(
peer=event.peer,
terbank_service=terbank_service,
context=context,
is_edit=_is_equal_state(context, EventEditState.gosb),
event_id=data.get("edit_event_id"),
ctx_name=data.get("edit_event_ctx", "moderation"),
)
@bot.di
def tb_update_selected(event: UpdateInteractiveMediaEvent, context: FSMContext):
# keeping what user have selected in "tb_id"
context.update_data({"tb_id": event.data.value})
@bot.di
def tb_approved(
event: UpdateInteractiveMediaEvent,
terbank_service: TerbankService,
context: FSMContext,
):
# tb was selected and approved. now selecting gosb:
delete_prev_message_by_peer(bot, event.peer)
selected_tb_id = context.get_data().get("tb_id")
if not isinstance(selected_tb_id, str) or not selected_tb_id:
logger.error(
"error while creating event: error while selecting tb: selected tb must be a string, got '%s'",
selected_tb_id,
)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Произошла ошибка. Попробуй позже.",
)
context.clear()
return
gosbs = terbank_service.get_gosbs_for_tb(tb_id=selected_tb_id, messenger_id=event.peer.id)
is_edit = _is_equal_state(context, EventEditState.gosb)
data = context.get_data()
default_gosb_id = data.get("edit_current_gosb_id")
bot.messaging.send_message(
peer=event.peer,
text="Редактирование ГОСБ: выбери новый ГОСБ" if is_edit else "Шаг 5/11: Выбери ГОСБ",
interactive_media_groups=gosb_select_keyboard(gosbs, default_value=default_gosb_id),
)
@bot.di
def gosb_update_selected(event: UpdateInteractiveMediaEvent, context: FSMContext):
# keeping what user have selected in "gosb_id"
context.update_data({"gosb_id": event.data.value})
@bot.di
def gosb_approved(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
# gosb was selected and approved.
delete_prev_message_by_peer(bot, event.peer)
selected_gosb_id = context.get_data().get("gosb_id")
if not isinstance(selected_gosb_id, str) or not selected_gosb_id:
logger.error(
"error while creating event: error while selecting gosb: selected gosb must be a string, got '%s'",
selected_gosb_id,
)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Произошла ошибка. Попробуй позже.",
)
context.clear()
return
data = context.get_data()
is_edit = _is_equal_state(context, EventEditState.gosb)
if is_edit:
event_id = str(data.get("edit_event_id") or "")
ctx_name = str(data.get("edit_event_ctx") or "moderation")
if not event_id:
logger.error("error while editing gosb: edit_event_id is empty")
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново.",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.clear()
return
updated = event_service.update_event(
event_id=event_id,
messenger_id=event.peer.id,
payload={"gosb_id": selected_gosb_id},
)
context.set_state(None)
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
note="✅ ГОСБ успешно изменён." if updated else "⚠️ Не удалось изменить ГОСБ.",
event_obj=event_service.get_event_by_id(event_id, event.peer.id),
show_organizer=True,
ctx_name=ctx_name,
)
else:
bot.messaging.send_message(
peer=event.peer,
text=(
"Шаг 6/11: Введи id проекта, если хочешь создать мероприятие в проекте\n"
"Введи «-» или нажми «пропустить», если хочешь создать событийное мероприятие - мероприятие вне проекта."
),
interactive_media_groups=back_to_moderation_and_skip_keyboard(),
)
context.set_state(EventCreateState.project)
@events_rt.message(state=EventCreateState.project)
@bot.di
def event_create_project_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
raw = message.message.text_message.text.strip()
_handle_project_input(message.peer, raw, context)
@bot.di
def project_input_dash_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
delete_prev_message_by_peer(bot, event.peer)
_handle_project_input(event.peer, "-", context)
def _handle_project_input(peer, raw_text: str, context: FSMContext):
project_id = "" if raw_text == "-" else raw_text
context.update_data({"project_id": project_id})
bot.messaging.send_message(
peer=peer,
text="Шаг 7/11: Введи описание:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.description)
@events_rt.message(state=EventCreateState.description)
@bot.di
def event_create_description_step(message: UpdateMessage, context: FSMContext):
delete_prev_message(bot, message)
try:
description = validate_description(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"event_description": description})
bot.messaging.send_message(
peer=message.peer,
text="Шаг 8/11: Введи количество волонтёрских часов, начисляемых за мероприятие:",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.hours)
@events_rt.message(state=EventCreateState.hours)
@bot.di
def event_create_hours_step(
message: UpdateMessage,
context: FSMContext,
tag_service: TagService,
):
delete_prev_message(bot, message)
try:
hours = validate_hours(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"hours": hours})
all_tags = tag_service.get_all_tags()
choices = choices_from_backend_tags(all_tags)
context.update_data(
{
"event_tag_keys": [],
"all_tags": [{"id": t.id, "title": t.title} for t in all_tags],
},
)
_send_tags_toggle_ui(
peer=message.peer,
title="Шаг 9/11: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:",
choices=choices,
selected_keys=set(),
footer_buttons=[Button(media_id="leave", value="moderation", label="⬅️ В меню модерации")],
)
context.set_state(EventCreateState.tags)
@bot.di
def event_tags_toggle_handler(
event: UpdateInteractiveMediaEvent,
event_service: EventService,
context: FSMContext,
):
is_create = _is_equal_state(context, EventCreateState.tags)
is_edit = _is_equal_state(context, EventEditState.tags)
if not (is_create or is_edit):
return
raw = str(event.data.value or "")
data = context.get_data()
ctx_name = str(data.get("edit_event_ctx", "moderation"))
key_field = "event_tag_keys" if is_create else "edit_tag_keys"
selected = set(data.get(key_field) or [])
all_tags = data.get("all_tags", [])
if not all_tags:
logger.info("No tags were taken while editing event tags. Please check seeding file.")
if raw == TAGS_DONE_VALUE:
title_to_id = {t["title"]: t["id"] for t in all_tags if t.get("title") and t.get("id")}
app_tag_ids = [title_to_id[n] for n in selected if n in title_to_id]
delete_prev_message_by_peer(bot, event.peer)
if is_create:
context.update_data({"event_app_tag_ids": app_tag_ids})
bot.messaging.send_message(
peer=event.peer,
text=_err(
"verification_code",
"Шаг 10/11: введи код мероприятия:\nБез пробелов, от {min_len} до {max_len} символов. Например: WELCOME2025",
),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.code)
return
if is_edit:
current = set(data.get("edit_current_tag_keys") or [])
event_id = str(data.get("edit_event_id"))
if not event_id or event_id == "None":
logger.error(
"Error while editing event tags: "
f"no event was found by saved id {event_id!r}, "
"saved in data by the key 'edit_event_id'. "
"exiting editing session"
)
context.set_state(None)
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново.",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
to_add = selected - current
to_remove = current - selected
failed = False
# add
for title in to_add:
tag_id = title_to_id.get(title)
if not tag_id:
logger.error(
"Error while editing event tags: "
f"selected tag with title={title!r} "
f"have no entrance in all_tags={all_tags}. "
"skipping this tag."
)
elif not event_service.add_tag(
event_id=event_id, tag_id=tag_id, messenger_id=event.peer.id
):
failed = True
# remove
for title in to_remove:
tag_id = title_to_id.get(title)
if not tag_id:
logger.error(
"Error while editing event tags: "
f"selected tag with title={title!r} "
f"have no entrance in all_tags={all_tags}. "
"skipping this tag."
)
elif not event_service.remove_tag(
event_id=event_id, tag_id=tag_id, messenger_id=event.peer.id
):
failed = True
context.set_state(None)
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Ошибка изменения тегов! Некоторые теги могли не измениться."
if failed
else "✅ Теги успешно изменены.",
event_obj=event_service.get_event_by_id(event_id, event.peer.id),
show_organizer=True,
ctx_name=ctx_name,
)
return
if raw.startswith(TAGS_TOGGLE_PREFIX):
key = raw[len(TAGS_TOGGLE_PREFIX) :].strip()
if key in selected:
selected.remove(key)
else:
selected.add(key)
context.update_data({key_field: list(selected)})
delete_prev_message_by_peer(bot, event.peer)
if is_create:
footer = [Button(media_id="leave", value="moderation", label="⬅️ В меню модерации")]
title = "Шаг 9/11: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:"
else:
event_id = data.get("edit_event_id") or ""
footer = [
Button(
media_id="event_menu_edit",
value=f"{event_id}|{ctx_name}",
label="⬅️ К редактированию мероприятия",
)
]
title = "Редактирование направления: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:"
normalized_tags: list[TagSchema] = []
for tag in all_tags:
if isinstance(tag, TagSchema):
normalized_tags.append(tag)
elif isinstance(tag, dict):
normalized_tags.append(TagSchema.from_dict(tag))
choices = choices_from_backend_tags(normalized_tags)
_send_tags_toggle_ui(
peer=event.peer,
title=title,
choices=choices,
selected_keys=set(selected),
footer_buttons=footer,
)
return
@events_rt.message(state=EventCreateState.code)
@bot.di
def event_create_code_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
request_service: RequestService,
report_service: ReportService,
user: UserSchema,
):
delete_prev_message(bot, message)
try:
code = validate_verification_code(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"event_verification_code": code})
bot.messaging.send_message(
peer=message.peer,
text=_err( # 🔹 адаптивный текст через _err
"participation_limit",
"Шаг 11/11: введи требуемое количество участников:\nПусть это будет число от {min} до {max}",
),
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventCreateState.participation_limit)
@events_rt.message(state=EventCreateState.participation_limit)
@bot.di
def event_create_participation_limit_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
request_service: RequestService,
report_service: ReportService,
user: UserSchema,
):
delete_prev_message(bot, message)
try:
limit = validate_participation_limit(message.message.text_message.text.strip())
except EventValidationError as e:
_msg_err_create(message, e)
return
context.update_data({"event_participation_limit": limit})
data = context.get_data()
try:
created = event_service.create_from_wizard(data, message.peer)
except Exception as e:
logger.error("Error while creating an event via wizard: %s", e)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Произошла ошибка.",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.clear()
return
context.clear()
note = {
"REQUEST": "✅ Заявка на мероприятие успешно создана.",
"COMPLETELY": "✅ Мероприятие успешно создано.",
"FAILED": "⚠️ Произошла ошибка: не удалось создать заявку/мероприятие.",
}.get(
created.get("TYPE", ""),
"⚠️ Произошла ошибка: не получить статус заявки/мероприятия. Проверь результат в «Своих заявках» или в «Своих мероприятиях».",
)
result_type = created.get("TYPE")
event_id = str(created.get("event_id", ""))
request_id = str(created.get("request_id", ""))
if result_type == "COMPLETELY":
if not event_id:
logger.error("expected event_id, got none whlie creating event")
bot.messaging.send_message(
peer=message.peer,
text="Мероприятие было создано успешно, но у нас не получилось показать его карточку. Попробуй открыть её из меню мероприятий.",
interactive_media_groups=back_to_moderation_keyboard(),
)
else:
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name="my_events", # TODO fix context
note=note,
)
if result_type == "REQUEST":
if not request_id:
logger.error("expected reequest_id, got none whlie creating request")
bot.messaging.send_message(
peer=message.peer,
text="Заявка на мероприятие была создано успешно, но у нас не получилось показать её карточку. Попробуй открыть её из меню заявок.",
interactive_media_groups=back_to_moderation_keyboard(),
)
else:
request_card = request_service.get_request_by_id(request_id, message.peer.id)
if request_card is not None:
bot.messaging.send_message(
peer=message.peer,
text=f"{note}\n\n{format_request_details(request_card)}",
interactive_media_groups=build_request_actions_keyboard(
request_card,
ctx_name="moderation",
),
)
@bot.di
def event_view_by_id_start(event: UpdateInteractiveMediaEvent, context: FSMContext): # legacy
"""старт диалога - просим у модератора UUID мероприятия"""
delete_prev_message_by_peer(bot, event.peer)
bot.messaging.send_message(
peer=event.peer,
text="Введи UUID мероприятия, которое хочешь посмотреть.\n\n",
interactive_media_groups=back_to_moderation_keyboard(),
)
context.set_state(EventViewState.wait_event_id)
@events_rt.message(state=EventViewState.wait_event_id)
@bot.di
def event_view_by_id_step( # legacy
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
user: UserSchema,
):
"""читаем UUID, тянем мероприятие из бэкенда и показываем информацию"""
delete_prev_message(bot, message)
raw_id = message.message.text_message.text.strip()
event = event_service.get_event_by_id(raw_id, message.peer.id)
if event is None:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Мероприятие с таким UUID не найдено.\nПопробуй ещё раз. Пример UUID: `932cf1cb-1ee7-4a45-bc96-41a232f3f538`",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
context.update_data({"current_event_id": event.id})
context.set_state(None)
send_event_card(
message.peer,
event_id=event.id,
event_service=event_service,
user=user,
ctx_name="moderation_menu",
)
@bot.di
def event_menu_enter_code_start_handler(
event: UpdateInteractiveMediaEvent, context: FSMContext
): # legacy
delete_prev_message_by_peer(bot, event.peer)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
context.update_data({"current_event_id": event_id, "current_event_ctx": ctx_name})
context.set_state(EventEnterCodeState.wait_code)
bot.messaging.send_message(
peer=event.peer,
text="🎟️ Введи бонус-код мероприятия (без пробелов):",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
@events_rt.message(state=EventEnterCodeState.wait_code)
@bot.di
def event_menu_enter_code_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema,
):
delete_prev_message(bot, message)
event_id = str(context.get_data().get("current_event_id"))
# если event_id is None, то клиент не найдёт event с id == "None", и всё ок
code = message.message.text_message.text.strip()
note: str
status = event_service.get_participation_status(event_id, message.peer.id)
if status is None:
note = "⚠️ Код не принят, сначала нужно записаться на мероприятие"
elif status.state == "confirmed":
note = "⚠️ Код не принят, баллы уже получены"
else:
ok = event_service.enter_code_by_event(event_id, message.peer.id, code)
if not ok:
note = "⚠️ Код не принят, проверь правильность кода"
else:
note = "✅ Код принят. Часы начислены."
data = context.get_data()
ctx_name = data.get("current_event_ctx") or "moderation"
context.set_state(None)
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note=note,
)
@bot.di
def event_menu_delete_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
report_service: ReportService,
user: UserSchema | None,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx = split_event_value(event.data.value)
ctx_name = ctx or "moderation"
if ctx_name not in ("moderation", "moderation_menu"):
ctx_name = "moderation"
try:
ok = event_service.delete_event(
event_id,
requester_messenger_id=event.peer.id,
)
except Exception:
ok = False
if ok:
context.clear()
if ctx_name == "moderation":
send_event_cards_page(
event.peer,
offset=0,
event_service=event_service,
ctx_name="moderation",
note="Мероприятие удалено.",
)
else:
bot.messaging.send_message(
peer=event.peer,
text="Мероприятие удалено.\n\nТы в доме модерации",
interactive_media_groups=moderation_menu_keyboard(),
)
return
send_event_card(
event.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note="⚠️ Не удалось удалить мероприятие",
)
def _send_event_edit_menu(
peer,
*,
event_id: str,
event_service: EventService,
note: str | None = None,
event_obj=None, # temporary remove type since it was fixed in another branch
show_organizer: bool,
ctx_name: str,
):
ui = get_ui(ctx_name)
refreshed = event_obj or event_service.get_event_by_id(event_id, peer.id)
base = (
format_event_details(refreshed, show_code=True, show_organizer=show_organizer)
if refreshed
else "⚠️ Мероприятие не найдено"
)
prefix = f"{note}\n\n" if note else ""
bot.messaging.send_message(
peer=peer,
text=(f"{prefix}Редактирование мероприятия\n\n{base}\n\nВыбери поле для Редактирования:"),
interactive_media_groups=event_edit_fields_keyboard(
event_id,
leave_media_id=ui.list_open_media_id,
leave_value=f"{event_id}|{ui.name}",
leave_label="⬅️ К мероприятию",
),
)
@bot.di
def event_menu_edit_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
):
delete_prev_message_by_peer(bot, event.peer)
context.set_state(None)
event_id, ctx_name = split_event_value(event.data.value)
ctx_name = "moderation" if ctx_name is None else ctx_name
context.update_data(
{
"edit_event_id": event_id,
"edit_event_ctx": ctx_name,
}
)
_send_event_edit_menu(
event.peer,
event_id=event_id,
event_service=event_service,
show_organizer=True,
ctx_name=ctx_name,
)
def _get_field_label(field: str) -> str:
return {
"title": "Название",
"date": "Дата",
"time": "Время",
"tags": "Теги",
"gosb_id": "ГОСБ",
"event_organizers": "Организаторы",
"participation_limit": "Максимальное количество участников",
"hours": "Часы",
"description": "Описание",
"code": "Код",
}.get(field, field)
def _get_current_value(event_card: EventCardSchema, field: str) -> str:
event_fields = {
# valid fields
"title",
"description",
"gosb_id",
"hours",
"participation_limit",
# TODO rename properly
"verification_code", # from code
"event_organizers", # from creator_id
}
match field:
case "date":
val = event_card.event.date_time.strftime("%d.%m.%Y")
case "time":
val = event_card.event.date_time.strftime("%H:%M")
case "code": # once 'code' is renamed, this will be deleted
val = event_card.event.verification_code
case "creator_id": # once 'creator_id' is renamed, this will be deleted
val = event_card.event.creator_id
case "gosb_id":
val = event_card.gosb.name
case _:
val = (
getattr(event_card.event, field, None)
if field in event_fields
else getattr(event_card, field, None)
)
return (
("\n" if len(str(val)) > 140 else "")
+ "_"
+ ("-" if val is None or val == "" else str(val)).replace("\n", "_\n_")
+ "_"
)
def _is_equal_state(context: FSMContext, state: State) -> bool:
current = context.get_state()
return current == state or str(current) == str(state)
def _send_tags_toggle_ui(
*,
peer,
title: str,
choices,
selected_keys: set[str],
footer_buttons: list[Button],
):
bot.messaging.send_message(
peer=peer,
text=title,
interactive_media_groups=tags_toggle_keyboard(
choices=choices,
selected_keys=selected_keys,
media_id=TAGS_MEDIA_ID,
toggle_value_prefix=TAGS_TOGGLE_PREFIX,
done_value=TAGS_DONE_VALUE,
done_label="✅ Готово",
extra_footer=footer_buttons,
),
)
@bot.di
def event_menu_edit_field_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
event_service: EventService,
tag_service: TagService,
terbank_service: TerbankService,
):
delete_prev_message_by_peer(bot, event.peer)
raw = str(event.data.value or "")
if "|" not in raw:
logger.error("Got unexpected event.data.value='%s' while trying to edit event", raw)
bot.messaging.send_message(peer=event.peer, text="⚠️ Некорректная команда редактирования.")
return
event_id, field = raw.split("|", 1)
field = field.strip()
data = context.get_data()
ctx_name = str(data.get("edit_event_ctx", "moderation"))
edited_event_card = event_service.get_event_by_id(event_id, event.peer.id)
if not edited_event_card:
bot.messaging.send_message(peer=event.peer, text="⚠️ Мероприятие не найдено.")
return
if field == "tags":
# редактируем теги через toggle UI
all_tags = tag_service.get_all_tags()
choices = choices_from_backend_tags(all_tags)
selected_titles: set[str] = set()
for t in edited_event_card.tags or []:
if isinstance(t, dict) and t.get("title"):
selected_titles.add(str(t["title"]))
else:
title = getattr(t, "title", None)
if title:
selected_titles.add(str(title))
context.update_data(
{
"edit_event_id": event_id,
"all_tags": [{"id": t.id, "title": t.title} for t in all_tags],
"edit_tag_keys": list(selected_titles),
"edit_current_tag_keys": list(selected_titles),
}
)
context.set_state(EventEditState.tags)
_send_tags_toggle_ui(
peer=event.peer,
title="Редактирование направления: выбери направления, которым соответствует мероприятие, затем нажми «Готово»:",
choices=choices,
selected_keys=set(selected_titles),
footer_buttons=[
Button(
media_id="event_menu_edit",
value=f"{event_id}|{ctx_name}",
label="⬅️ К редактированию мероприятия",
)
],
)
return
if field == "gosb_id":
# редактируем госб через UI
# вызываем клавиатуру выбора тербанка, дальше она сама вызовет
# выбор ГОСБ и определит, редактируем мы или создаем мероприятие
default_tb_id = edited_event_card.terbank.id
default_gosb_id = edited_event_card.gosb.id
context.update_data(
{
"edit_event_id": event_id,
"edit_event_ctx": ctx_name,
"edit_field": "gosb_id",
"edit_current_tb_id": default_tb_id,
"edit_current_gosb_id": default_gosb_id,
}
)
send_tb_select_menu(
peer=event.peer,
terbank_service=terbank_service,
context=context,
is_edit=True,
event_id=event_id,
ctx_name=ctx_name,
default_tb_id=default_tb_id,
default_gosb_id=default_gosb_id,
)
return
context.update_data({"edit_event_id": event_id, "edit_field": field})
context.set_state(EventEditState.wait_value)
field_label = _get_field_label(field)
field_current_value = _get_current_value(edited_event_card, field)
if field == "event_organizers":
bot.messaging.send_message(
peer=event.peer,
text="Пока не реализовано",
interactive_media_groups=build_back_to_event_editing_keyboard(
event_id=event_id,
ctx_name=ctx_name,
),
)
return
if field == "date":
hint = "*Введи дату в формате* `dd.mm.yyyy`, например: `20.02.2002`"
elif field == "time":
hint = "*Введи время в формате* `HH:MM`, например: `10:30`"
elif field == "participation_limit":
hint = _err("participation_limit", "*Введи число от {min} до {max}*")
elif field == "hours":
hint = _err("hours", "*Введи число от {min} до {max}*")
elif field == "title":
hint = _err("title", "*Введи от {min_len} до {max_len} символов*")
elif field == "code":
hint = _err("verification_code", "*От {min_len} до {max_len} символов, буквы и цифры*")
elif field == "description":
hint = _err("description", "*Максимум {max_len} символов*")
else:
hint = "*Введи новое значение.* Чтобы очистить поле - отправь: «-»"
bot.messaging.send_message(
peer=event.peer,
text=f" ✍️ *Редактирование*: _{field_label}_\n\n*Текущее значение*: {field_current_value}\n\n{hint}",
interactive_media_groups=build_back_to_event_editing_keyboard(
event_id=event_id,
ctx_name=ctx_name,
),
)
@events_rt.message(state=EventEditState.wait_value)
@bot.di
def event_edit_value_step(
message: UpdateMessage,
context: FSMContext,
event_service: EventService,
):
delete_prev_message(bot, message)
data = context.get_data()
event_id: str = str(data.get("edit_event_id", ""))
field: str = str(data.get("edit_field", ""))
ctx_name = str(data.get("edit_event_ctx", "moderation"))
if not event_id or not field:
logger.error("expected values for event_id and field, got '%s' and '%s'", event_id, field)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Сессия редактирования сброшена. Открой мероприятие заново.",
)
context.set_state(None)
return
raw = message.message.text_message.text.strip()
value: str | int = "" if raw == "-" else raw
if field == "event_organizers":
bot.messaging.send_message(
peer=message.peer,
text="Пока не реализовано.",
interactive_media_groups=build_back_to_event_editing_keyboard(
event_id=event_id,
ctx_name=ctx_name,
),
)
context.set_state(None)
return
if field == "hours":
try:
value = validate_hours(value)
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if field == "participation_limit":
try:
value = validate_participation_limit(value)
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if field == "code":
try:
value = validate_verification_code(str(value))
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if field == "title":
try:
value = validate_title(str(value))
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if field == "description":
try:
value = validate_description(str(value))
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if field == "date":
try:
date_obj = validate_date(str(value))
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
if date_obj <= datetime.now().date():
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Ты не можешь создать мероприятие раньше завтрашнего дня.\nПожалуйста, введи дату, например, в формате `20.02.2002`:",
interactive_media_groups=back_to_moderation_keyboard(),
)
return
current_event_card = event_service.get_event_by_id(event_id, message.peer.id)
if current_event_card is None:
logger.error(
"expected event card while editing date, got None for event_id=%s",
event_id,
)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Мероприятие не найдено. Открой мероприятие заново.",
)
context.set_state(None)
return
current_dt = current_event_card.event.date_time
new_dt = current_dt.replace(
year=date_obj.year,
month=date_obj.month,
day=date_obj.day,
)
field = "date_time"
value = new_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
if field == "time":
try:
time_obj = validate_time(str(value))
except EventValidationError as e:
_msg_err_edit(message, e, event_id, ctx_name)
return
current_event_card = event_service.get_event_by_id(event_id, message.peer.id)
if current_event_card is None:
logger.error(
"expected event card while editing time, got None for event_id=%s",
event_id,
)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Мероприятие не найдено. Открой мероприятие заново.",
)
context.set_state(None)
return
current_dt = current_event_card.event.date_time
new_dt = current_dt.replace(
hour=time_obj.hour,
minute=time_obj.minute,
second=0,
microsecond=0,
)
field = "date_time"
value = new_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
payload = {field: value}
updated = event_service.update_event(
event_id=event_id,
messenger_id=message.peer.id,
payload=payload, # type(value) in (int, str)
)
context.set_state(None)
_send_event_edit_menu(
message.peer,
event_id=event_id,
event_service=event_service,
note="⚠️ Не удалось обновить поле." if not updated else "✅ Поле обновлено.",
event_obj=event_service.get_event_by_id(event_id, message.peer.id),
show_organizer=True,
ctx_name=ctx_name,
)
--- services/bot/core/handlers/report.py ---
from __future__ import annotations
import re
from datetime import date, datetime, timedelta
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_event_card, split_event_value
from core.markups import back_to_reports_export_keyboard, reports_export_keyboard
from core.schemas import UserSchema
from core.services import EventService, FileService, ReportService
from core.utils import delete_prev_message, delete_prev_message_by_peer, logger
report_rt = Router()
class ReportCreateState(StatesGroup): # legacy
beneficiaries_count = State()
how_it_went = State()
class ReportExportState(StatesGroup): # legacy
wait_from = State()
wait_to = State()
def _fmt_period(date_from: date | None, date_to: date | None) -> str: # legacy
if date_from is None and date_to is None:
return "всё время"
if date_from is not None and date_to is None:
return f"с {date_from.strftime('%d.%m.%Y')} по сегодня"
if date_from is None and date_to is not None:
return f"по {date_to.strftime('%d.%m.%Y')}"
return f"с {date_from.strftime('%d.%m.%Y')} по {date_to.strftime('%d.%m.%Y')}"
def _get_button_id(event: UpdateInteractiveMediaEvent) -> str: # legacy
return str(getattr(event, "id", "") or "")
@bot.di
def reports_export_handler( # legacy
event: UpdateInteractiveMediaEvent,
context: FSMContext,
report_service: ReportService,
):
peer = event.peer
delete_prev_message_by_peer(bot, peer)
uid = f"reports_export_peer_{peer.id}"
btn_id = _get_button_id(event)
if btn_id not in ("rep_day", "rep_week", "rep_month", "rep_all", "rep_from", "rep_range"):
context.set_state(None)
bot.messaging.send_message(
peer=peer,
text="📥 Выгрузка отчётов\n\nВыбери период:",
interactive_media_groups=reports_export_keyboard(),
)
return
today = date.today()
if btn_id == "rep_day":
_export_and_send(
peer=peer,
report_service=report_service,
date_from=today - timedelta(days=1),
date_to=today,
uid=uid,
)
return
if btn_id == "rep_week":
_export_and_send(
peer=peer,
report_service=report_service,
date_from=today - timedelta(days=7),
date_to=today,
uid=uid,
)
return
if btn_id == "rep_month":
_export_and_send(
peer=peer,
report_service=report_service,
date_from=today - timedelta(days=30),
date_to=today,
uid=uid,
)
return
if btn_id == "rep_all":
_export_and_send(
peer=peer,
report_service=report_service,
date_from=None,
date_to=None,
uid=uid,
)
return
if btn_id == "rep_from":
context.set_state(ReportExportState.wait_from)
context.update_data({"report_export_mode": "from_today"})
bot.messaging.send_message(
peer=peer,
text="📌 Введи дату ОТ в формате `dd.mm.yyyy` (например: `01.02.2026`).",
interactive_media_groups=back_to_reports_export_keyboard(),
)
return
if btn_id == "rep_range":
context.set_state(ReportExportState.wait_from)
context.update_data({"report_export_mode": "range"})
bot.messaging.send_message(
peer=peer,
text="🗓️ Введи дату ОТ в формате `dd.mm.yyyy` (например: `01.02.2026`).",
interactive_media_groups=back_to_reports_export_keyboard(),
)
return
# неизвестная кнопка - показываем выбор периода (без падений)
context.set_state(None)
bot.messaging.send_message(
peer=peer,
text="📥 Выбери период выгрузки:",
interactive_media_groups=reports_export_keyboard(),
)
return
def _parse_user_date(raw: str) -> date | None: # legacy
raw = (raw or "").strip()
try:
return datetime.strptime(raw, "%d.%m.%Y").date()
except Exception:
return None
def _api_date(d: date) -> str: # legacy
return d.isoformat()
def _xlsx_has_data_rows(file_bytes: bytes) -> bool: # legacy
# требование: если отчётов нет - показать предупреждение.
# делаем через чтение xlsx, без изменения backend-контракта.
try:
from io import BytesIO
from openpyxl import load_workbook
wb = load_workbook(BytesIO(file_bytes), read_only=True, data_only=True)
ws = wb.active
for row in ws.iter_rows(min_row=2, values_only=True):
if any(v not in (None, "") for v in row):
return True
return False
except Exception as e:
logger.warning(f"_xlsx_has_data_rows failed: {e}")
return True
def _export_and_send( # legacy
*,
peer,
report_service: ReportService,
date_from: date | None,
date_to: date | None,
uid: str,
) -> None:
try:
period_text = _fmt_period(date_from, date_to)
df = _api_date(date_from) if date_from else None
dt = _api_date(date_to) if date_to else None
file_bytes, filename = report_service.export_reports_xlsx(date_from=df, date_to=dt)
has_rows = _xlsx_has_data_rows(file_bytes)
text = f"✅ Выгрузка готова: {period_text}"
if not has_rows:
text += "\n⚠️ За период отчётов не найдено"
ok = FileService.send_file(
peer,
file_bytes=file_bytes,
filename=filename,
uid=uid,
text=text,
reply_mid=None,
interactive_media_groups=back_to_reports_export_keyboard(),
)
if not ok:
logger.error(
f"_export_and_send: send_file returned False, uid={uid}, filename={filename}"
)
bot.messaging.send_message(
peer=peer,
text="⚠️ Не удалось отправить файл отчётов.",
interactive_media_groups=back_to_reports_export_keyboard(),
)
except Exception as e:
logger.error(f"_export_and_send failed, uid={uid}: {e}")
bot.messaging.send_message(
peer=peer,
text="⚠️ Ошибка экспорта и отправки.",
interactive_media_groups=back_to_reports_export_keyboard(),
)
@report_rt.message(state=ReportExportState.wait_from)
@bot.di
def report_export_wait_from_step( # legacy
message: UpdateMessage, context: FSMContext, report_service: ReportService
):
delete_prev_message(bot, message)
d = _parse_user_date(message.message.text_message.text or "")
if not d:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Не смог распознать дату. Введи в формате `dd.mm.yyyy`, например: `01.02.2026`.",
interactive_media_groups=back_to_reports_export_keyboard(),
)
return
mode = (context.get_data().get("report_export_mode") or "").strip()
context.update_data({"report_export_from": d.isoformat()})
if mode == "from_today":
context.set_state(None)
_export_and_send(
peer=message.peer,
report_service=report_service,
date_from=d,
date_to=date.today(),
uid=f"reports_export_peer_{message.peer.id}",
)
return
# mode == "range"
context.set_state(ReportExportState.wait_to)
bot.messaging.send_message(
peer=message.peer,
text="🗓️ Теперь введи дату ДО в формате `dd.mm.yyyy` (например: `02.02.2026`).",
interactive_media_groups=back_to_reports_export_keyboard(),
)
@report_rt.message(state=ReportExportState.wait_to)
@bot.di
def report_export_wait_to_step( # legacy
message: UpdateMessage, context: FSMContext, report_service: ReportService
):
delete_prev_message(bot, message)
d_to = _parse_user_date(message.message.text_message.text or "")
if not d_to:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Не смог распознать дату. Введи в формате `dd.mm.yyyy`, например: `02.02.2026`.",
interactive_media_groups=back_to_reports_export_keyboard(),
)
return
data = context.get_data()
raw_from = data.get("report_export_from")
d_from = date.fromisoformat(raw_from) if raw_from else None
if d_from and d_to < d_from:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Дата ДО не может быть раньше даты ОТ. Введи дату ДО ещё раз.",
interactive_media_groups=back_to_reports_export_keyboard(),
)
return
context.set_state(None)
_export_and_send(
peer=message.peer,
report_service=report_service,
date_from=d_from,
date_to=d_to,
uid=f"reports_export_peer_{message.peer.id}",
)
@bot.di
def report_create_start_handler(event: UpdateInteractiveMediaEvent, context: FSMContext):
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"
context.update_data(
{
"report_event_id": event_id,
"report_event_ctx": ctx_name,
}
)
bot.messaging.send_message(
peer=event.peer,
text=(
"🧾 Создание отчёта\n\n"
"Шаг 1/2: введи количество благополучателей: целое неотрицательное число."
),
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
context.set_state(ReportCreateState.beneficiaries_count)
@report_rt.message(state=ReportCreateState.beneficiaries_count)
@bot.di
def report_create_beneficiaries_step(message: UpdateMessage, context: FSMContext): # legacy
delete_prev_message(bot, message)
data = context.get_data()
event_id = data.get("report_event_id") or ""
ctx_name = data.get("report_event_ctx") or "events"
text = message.message.text_message.text.strip()
try:
if not re.fullmatch(r"^[0-9]{1,9}$", text):
raise ValueError()
val = int(text)
if val < 0 or val >= 2**31:
raise ValueError()
except ValueError:
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Введи целое число, большее или равное нулю. Попробуй ещё раз:",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
return
context.update_data({"report_beneficiaries_count": val})
bot.messaging.send_message(
peer=message.peer,
text=("Шаг 2/2: напиши одним сообщением, как прошло мероприятие:"),
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
context.set_state(ReportCreateState.how_it_went)
@report_rt.message(state=ReportCreateState.how_it_went)
@bot.di
def report_create_how_it_went_step(
message: UpdateMessage,
context: FSMContext,
report_service: ReportService,
event_service: EventService,
user: UserSchema,
):
delete_prev_message(bot, message)
data = context.get_data()
event_id = data.get("report_event_id") or ""
ctx_name = data.get("report_event_ctx") or "events"
how_it_went = (message.message.text_message.text or "").strip()
beneficiaries_count = data.get("report_beneficiaries_count")
if not isinstance(beneficiaries_count, int):
logger.error(
"beneficiaries_count must be type int, got '%s' of type '%s'. defaulting to zero.",
beneficiaries_count,
type(beneficiaries_count),
)
beneficiaries_count = 0
uid = f"report_create_peer_{message.peer.id}_event_{event_id}"
logger.info(f"report_create start uid={uid}, event_id={event_id}")
try:
report_service.create_report(
event_id=event_id,
requester_messenger_id=message.peer.id,
description=how_it_went,
beneficiaries=beneficiaries_count,
)
except Exception as e:
logger.error(f"report_create failed uid={uid}: {e}")
context.set_state(None)
bot.messaging.send_message(
peer=message.peer,
text="⚠️ Не удалось создать отчёт. Проверь права/наличие отчёта и попробуй ещё раз.",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
return
try:
file_bytes, filename = report_service.get_report(
event_id=event_id,
requester_messenger_id=message.peer.id,
)
except Exception as e:
logger.error(f"fetching after creating event failed uid={uid}: {e}")
context.set_state(None)
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note="⚠️ Отчёт создан, но не удалось выгрузить файл.",
)
return
context.set_state(None)
FileService.send_file(
message.peer,
file_bytes=file_bytes,
filename=filename,
uid=uid,
text="✅ Отчёт создан. Файл выгружен.",
reply_mid=None,
)
send_event_card(
message.peer,
event_id=event_id,
event_service=event_service,
report_service=report_service,
user=user,
ctx_name=ctx_name,
note="✅ Отчёт создан.",
)
@bot.di
def report_view_handler(
event: UpdateInteractiveMediaEvent,
context: FSMContext,
report_service: ReportService,
):
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"
uid = f"report_view_peer_{event.peer.id}_event_{event_id}"
logger.info(f"report_view start uid={uid}, event_id={event_id}")
try:
file_bytes, filename = report_service.get_report(
event_id=event_id,
requester_messenger_id=event.peer.id,
)
except Exception as e:
logger.error(f"report_view failed uid={uid}: {e}")
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Не удалось получить отчёт (возможно, он ещё не создан).",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
return
ok = FileService.send_file(
event.peer,
file_bytes=file_bytes,
filename=filename,
uid=uid,
text="📄 Отчёт по мероприятию",
reply_mid=None,
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
if not ok:
bot.messaging.send_message(
peer=event.peer,
text="⚠️ Не удалось отправить файл отчёта.",
interactive_media_groups=build_back_to_event_keyboard(event_id, ctx_name),
)
--- services/bot/core/handlers/validators.py ---
import re
from datetime import datetime
from typing import TypeVar
from core.config import validation_cfg
T = TypeVar("T")
class EventValidationError(ValueError): ...
class TitleOutOfRange(EventValidationError): ...
class DescriptionOutOfRange(EventValidationError): ...
class UnsupportedDateFormat(EventValidationError): ...
class UnsupportedTimeFormat(EventValidationError): ...
class HoursInvalidFormat(EventValidationError): ...
class HoursOutOfRange(EventValidationError): ...
class VerificationCodeInvalidFormat(EventValidationError): ...
class VerificationCodeOutOfRange(EventValidationError): ...
class ParticipationLimitInvalidFormat(EventValidationError): ...
class ParticipationLimitOutOfRange(EventValidationError): ...
def _get_rule(field: str, rule: str, default: T, excepted_type: type[T]) -> T:
value = validation_cfg.get(f"event.{field}.{rule}", default)
if not isinstance(value, excepted_type):
raise RuntimeError(
f"validation rule event.{field}.{rule} must be {excepted_type}, got {type(value)}"
)
return value
def validate_title(text: str):
min_len = _get_rule("title", "min_length", 4, int)
max_len = _get_rule("title", "max_length", 20, int)
if len(text) < min_len or len(text) > max_len:
raise TitleOutOfRange()
return text
def validate_description(text: str):
max_len = _get_rule("description", "max_length", 20, int)
if len(text) > max_len:
raise DescriptionOutOfRange()
return text
def validate_date(text: str):
"""Парсинг даты с поддержкой разделителей . - _ . Возвращает date."""
formats = []
for first_gap in ". -_":
for second_gap in ". -_":
formats.append(f"%d{first_gap}%m{second_gap}%Y")
date_obj = None
for fmt in formats:
try:
date_obj = datetime.strptime(text.strip(), fmt).date()
break
except ValueError:
continue
if date_obj is None:
raise UnsupportedDateFormat()
return date_obj
def validate_time(text: str):
time_obj = None
for fmt in ["%H:%M", "%H %M", "%H-%M", "%H_%M"]:
try:
time_obj = datetime.strptime(text.strip(), fmt).time()
break
except ValueError:
continue
if time_obj is None:
raise UnsupportedTimeFormat()
return time_obj
def validate_hours(text: str | int):
"""Валидация часов: 1-2 цифры, диапазон 1-18 (из YAML)."""
text = str(text).strip()
if not re.fullmatch(r"^[0-9]{1,2}$", text):
raise HoursInvalidFormat()
hours = int(text)
min_val = _get_rule("hours", "min", 1, int)
max_val = _get_rule("hours", "max", 18, int)
if hours < min_val or hours > max_val:
raise HoursOutOfRange()
return hours
def validate_verification_code(text: str):
if not re.fullmatch(r"^[A-Za-zА-Яа-я0-9]+$", text):
raise VerificationCodeInvalidFormat()
min_len = _get_rule("verification_code", "min_length", 4, int)
max_len = _get_rule("verification_code", "max_length", 20, int)
if len(text) < min_len or len(text) > max_len:
raise VerificationCodeOutOfRange()
return text
def validate_participation_limit(value: str | int):
try:
limit = int(value)
except (TypeError, ValueError) as e:
raise ParticipationLimitInvalidFormat from e
min_val = _get_rule("participation_limit", "min", 1, int)
max_val = _get_rule("participation_limit", "max", 1000, int)
if limit < min_val or limit > max_val:
raise ParticipationLimitOutOfRange()
return limit
Editor is loading...
Leave a Comment