55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
"""Метаданные сообщения для событий аналитики.
|
||
|
||
Модуль намеренно **не зависит от aiogram**: работает с любым объектом, у которого
|
||
есть нужные атрибуты (``Message`` в боте или заглушка в тестах). Текст сообщения
|
||
не сохраняется — только метаданные (см. ``docs/events.md``).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
#: Максимальная длина имени команды в payload.
|
||
COMMAND_MAX_LENGTH = 64
|
||
|
||
#: Префикс, разделяющий команду и бота в групповом чате (``/команда@bot``).
|
||
BOT_MENTION_SEPARATOR = "@"
|
||
|
||
|
||
def _text(message: Any) -> str:
|
||
"""Текст сообщения (или подпись) без лишних пробелов."""
|
||
text = getattr(message, "text", None) or getattr(message, "caption", None) or ""
|
||
return str(text).strip()
|
||
|
||
|
||
def _enum_value(value: Any) -> str:
|
||
"""Строковое значение enum-полей aiogram (``ChatType.PRIVATE`` -> ``private``)."""
|
||
return str(getattr(value, "value", value) or "")
|
||
|
||
|
||
def describe_message(message: Any) -> dict[str, Any]:
|
||
"""Метаданные сообщения для события ``message`` (без текста сообщения)."""
|
||
text = _text(message)
|
||
chat = getattr(message, "chat", None)
|
||
return {
|
||
"message_id": int(getattr(message, "message_id", 0) or 0),
|
||
"chat_type": _enum_value(getattr(chat, "type", "")) if chat is not None else "",
|
||
"content_type": _enum_value(getattr(message, "content_type", "")),
|
||
"text_len": len(text),
|
||
"is_command": text.startswith("/"),
|
||
}
|
||
|
||
|
||
def describe_command(message: Any) -> dict[str, Any] | None:
|
||
"""Метаданные команды для события ``command`` (``None``, если это не команда)."""
|
||
text = _text(message)
|
||
if not text.startswith("/"):
|
||
return None
|
||
|
||
parts = text.split()
|
||
name = parts[0].split(BOT_MENTION_SEPARATOR, 1)[0].lstrip("/").lower()[:COMMAND_MAX_LENGTH]
|
||
return {"command": name, "args_count": max(len(parts) - 1, 0)}
|
||
|
||
|
||
__all__ = ["describe_command", "describe_message"]
|