Перейти к содержанию

Package: maxapi.utils

maxapi.utils

inline_keyboard

InlineKeyboardBuilder()

Конструктор инлайн-клавиатур.

Позволяет удобно собирать кнопки в ряды и формировать из них клавиатуру для отправки в сообщениях.

Source code in maxapi/utils/inline_keyboard.py 
18
19
def __init__(self):
    self.payload: List[List[InlineButtonUnion]] = [[]]
row(*buttons)

Добавить новый ряд кнопок в клавиатуру.

Parameters:

Name Type Description Default
*buttons InlineButtonUnion

Произвольное количество кнопок для добавления в ряд.

 ()
Source code in maxapi/utils/inline_keyboard.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def row(self, *buttons: InlineButtonUnion) -> InlineKeyboardBuilder:
    """
    Добавить новый ряд кнопок в клавиатуру.

    Args:
        *buttons: Произвольное количество кнопок для добавления в ряд.
    """

    if not self.payload[-1]:
        self.payload[-1].extend(buttons)
    else:
        self.payload.append([*buttons])
    return self
add(*buttons)

Добавить кнопки в последний ряд клавиатуры.

Parameters:

Name Type Description Default
*buttons InlineButtonUnion

Кнопки для добавления.

 ()
Source code in maxapi/utils/inline_keyboard.py 
35
36
37
38
39
40
41
42
43
44
45
def add(self, *buttons: InlineButtonUnion) -> InlineKeyboardBuilder:
    """
    Добавить кнопки в последний ряд клавиатуры.

    Args:
        *buttons: Кнопки для добавления.
    """

    for button in buttons:
        self.payload[-1].append(button)
    return self
adjust(*sizes)

Перераспределить кнопки по рядам в соответствии с указанными размерами.

Parameters:

Name Type Description Default
*sizes int

Количество кнопок в каждом ряду. Если кнопок больше, чем сумма размеров, размеры повторяются циклично.

 ()

Returns:

Name Type Description
InlineKeyboardBuilder InlineKeyboardBuilder

Текущий объект для цепочки вызовов.
Source code in maxapi/utils/inline_keyboard.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def adjust(self, *sizes: int) -> InlineKeyboardBuilder:
    """
    Перераспределить кнопки по рядам в соответствии с указанными размерами.

    Args:
        *sizes: Количество кнопок в каждом ряду.
               Если кнопок больше, чем сумма размеров, размеры повторяются циклично.

    Returns:
        InlineKeyboardBuilder: Текущий объект для цепочки вызовов.
    """
    if not sizes:
        sizes = (1,)

    flat_buttons = []
    for row in self.payload:
        for button in row:
            flat_buttons.append(button)

    if not flat_buttons:
        return self

    new_payload: List[List[InlineButtonUnion]] = []
    button_index = 0
    size_index = 0

    while button_index < len(flat_buttons):
        size = sizes[size_index % len(sizes)]
        if size <= 0:
            size = 1
        row_buttons = flat_buttons[button_index : button_index + size]
        new_payload.append(row_buttons)
        button_index += size
        size_index += 1

    self.payload = new_payload
    return self
as_markup()

Собрать клавиатуру в объект для отправки.

Returns:

Name Type Description
Attachment Attachment

Объект вложения с типом INLINE_KEYBOARD.
Source code in maxapi/utils/inline_keyboard.py 
85
86
87
88
89
90
91
92
93
94
95
96
def as_markup(self) -> Attachment:
    """
    Собрать клавиатуру в объект для отправки.

    Returns:
        Attachment: Объект вложения с типом INLINE_KEYBOARD.
    """

    return Attachment(
        type=AttachmentType.INLINE_KEYBOARD,
        payload=ButtonsPayload(buttons=self.payload),
    )  # type: ignore

message

process_input_media(base_connection, bot, att) async

Загружает файл вложения и формирует объект AttachmentUpload.

Parameters:

Name Type Description Default
base_connection BaseConnection

Базовое соединение для загрузки файла.

 required
bot Bot

Экземпляр бота.

 required
att InputMedia | InputMediaBuffer

Объект вложения для загрузки.

 required

Returns:

Name Type Description
AttachmentUpload AttachmentUpload

Загруженное вложение с токеном.
Source code in maxapi/utils/message.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def process_input_media(
    base_connection: BaseConnection,
    bot: Bot,
    att: InputMedia | InputMediaBuffer,
) -> AttachmentUpload:
    # очень нестабильный метод независящий от модуля
    # ждем обновлений MAX API

    """
    Загружает файл вложения и формирует объект AttachmentUpload.

    Args:
        base_connection (BaseConnection): Базовое соединение для загрузки файла.
        bot (Bot): Экземпляр бота.
        att (InputMedia | InputMediaBuffer): Объект вложения для загрузки.

    Returns:
        AttachmentUpload: Загруженное вложение с токеном.
    """

    try:
        upload = await bot.get_upload_url(att.type)
    except MaxApiError as e:
        raise MaxUploadFileFailed(
            f"Ошибка при загрузке файла: code={e.code}, raw={e.raw}"
        )

    if isinstance(att, InputMedia):
        upload_file_response = await base_connection.upload_file(
            url=upload.url,
            path=att.path,
            type=att.type,
        )
    elif isinstance(att, InputMediaBuffer):
        upload_file_response = await base_connection.upload_file_buffer(
            filename=att.filename or str(uuid4()),
            url=upload.url,
            buffer=att.buffer,
            type=att.type,
        )

    token = ""

    if att.type in (UploadType.VIDEO, UploadType.AUDIO):
        if upload.token is None:
            if bot.session is not None:
                await bot.session.close()

            raise MaxUploadFileFailed(
                "По неизвестной причине token не был получен"
            )

        token = upload.token

    elif att.type == UploadType.FILE:
        json_r = loads(upload_file_response)
        token = json_r["token"]

    elif att.type == UploadType.IMAGE:
        json_r = loads(upload_file_response)
        json_r_keys = list(json_r["photos"].keys())
        token = json_r["photos"][json_r_keys[0]]["token"]

    return AttachmentUpload(
        type=att.type, payload=AttachmentPayload(token=token)
    )

time

to_ms(value)

Преобразует datetime или числовую метку времени в миллисекунды (int).

Если value — объект datetime, возвращает int(timestamp * 1000). Если value — int или float (уже временная метка), возвращает int(value).

Source code in maxapi/utils/time.py 
 5
 6
 7
 8
 9
10
11
12
13
def to_ms(value: Union[datetime, int, float]) -> int:
    """Преобразует datetime или числовую метку времени в миллисекунды (int).

    Если `value` — объект datetime, возвращает int(timestamp * 1000).
    Если `value` — int или float (уже временная метка), возвращает int(value).
    """
    if isinstance(value, datetime):
        return int(value.timestamp() * 1000)
    return int(value)

from_ms(value)

Преобразует миллисекунды с эпохи в объект datetime.

Если value равен None, возвращает None. Иначе возвращает datetime.fromtimestamp(value / 1000).

Source code in maxapi/utils/time.py 
16
17
18
19
20
21
22
23
24
def from_ms(value: Optional[Union[int, float]]) -> Optional[datetime]:
    """Преобразует миллисекунды с эпохи в объект datetime.

    Если value равен None, возвращает None.
    Иначе возвращает datetime.fromtimestamp(value / 1000).
    """
    if value is None:
        return None
    return datetime.fromtimestamp(int(value) / 1000)

updates

enrich_event(event_object, bot) async

Дополняет объект события данными чата, пользователя и ссылкой на бота.

Parameters:

Name Type Description Default
event_object Any

Событие, которое нужно дополнить.

 required
bot Bot

Экземпляр бота.

 required

Returns:

Name Type Description
Any Any

Обновлённый объект события.
Source code in maxapi/utils/updates.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def enrich_event(event_object: Any, bot: Bot) -> Any:
    """
    Дополняет объект события данными чата, пользователя и ссылкой на бота.

    Args:
        event_object (Any): Событие, которое нужно дополнить.
        bot (Bot): Экземпляр бота.

    Returns:
        Any: Обновлённый объект события.
    """

    if not bot.auto_requests:
        return event_object

    if hasattr(event_object, "chat_id"):
        # Если это удаление бота из канала, мы не сможем получить инфо о чате
        is_bot_removed_from_channel = isinstance(
            event_object, BotRemoved
        ) and getattr(event_object, "is_channel", False)
        if not is_bot_removed_from_channel:
            event_object.chat = await bot.get_chat_by_id(event_object.chat_id)
        else:
            event_object.chat = None

    if isinstance(event_object, (MessageCreated, MessageEdited)):
        if event_object.message.recipient.chat_id is not None:
            if not hasattr(event_object, "chat"):
                event_object.chat = await bot.get_chat_by_id(
                    event_object.message.recipient.chat_id
                )

        event_object.from_user = getattr(event_object.message, "sender", None)

    elif isinstance(event_object, MessageCallback):
        message = event_object.message
        if message is not None and message.recipient.chat_id is not None:
            chat_id = message.recipient.chat_id
            if not hasattr(event_object, "chat"):
                event_object.chat = await bot.get_chat_by_id(chat_id)

        event_object.from_user = getattr(event_object.callback, "user", None)

    elif isinstance(event_object, MessageRemoved):
        if not hasattr(event_object, "chat"):
            event_object.chat = await bot.get_chat_by_id(event_object.chat_id)

        if event_object.chat and event_object.chat.type == ChatType.CHAT:
            event_object.from_user = await bot.get_chat_member(
                chat_id=event_object.chat_id, user_id=event_object.user_id
            )

        elif event_object.chat and event_object.chat.type == ChatType.DIALOG:
            event_object.from_user = event_object.chat  # pyright: ignore[reportAttributeAccessIssue]

    elif isinstance(event_object, UserRemoved):
        if not hasattr(event_object, "chat"):
            event_object.chat = await bot.get_chat_by_id(event_object.chat_id)
        if event_object.admin_id:
            event_object.from_user = await bot.get_chat_member(
                chat_id=event_object.chat_id, user_id=event_object.admin_id
            )

    elif isinstance(event_object, UserAdded):
        if not hasattr(event_object, "chat"):
            event_object.chat = await bot.get_chat_by_id(event_object.chat_id)
        event_object.from_user = event_object.user

    elif isinstance(
        event_object,
        (
            BotAdded,
            BotRemoved,
            BotStarted,
            ChatTitleChanged,
            BotStopped,
            DialogCleared,
            DialogMuted,
            DialogUnmuted,
        ),
    ):
        if not hasattr(event_object, "chat"):
            event_object.chat = await bot.get_chat_by_id(event_object.chat_id)
        event_object.from_user = event_object.user

    if isinstance(
        event_object, (MessageCreated, MessageEdited, MessageCallback)
    ):
        object_message = event_object.message  # pyright: ignore[reportAttributeAccessIssue]

        if object_message is not None:
            object_message.bot = bot
            if object_message.body is not None:
                for att in object_message.body.attachments or []:
                    if hasattr(att, "bot"):
                        att.bot = bot

    if hasattr(event_object, "bot"):
        event_object.bot = bot

    return event_object