Send_message

maxapi.methods.send_message.SendMessage(bot, chat_id=None, user_id=None, text=None, attachments=None, link=None, format=None, parse_mode=None, *, notify=None, disable_link_preview=None, sleep_after_input_media=True)

Bases: BaseConnection

Класс для отправки сообщения в чат или пользователю с поддержкой вложений и форматирования.

https://dev.max.ru/docs-api/methods/POST/messages

Attributes:

Name	Type	Description
bot	Bot	Экземпляр бота для выполнения запроса.
chat_id	Optional[int]	Идентификатор чата, куда отправлять сообщение.
user_id	Optional[int]	Идентификатор пользователя, если нужно отправить личное сообщение.
text	Optional[str]	Текст сообщения.
link	Optional[NewMessageLink]	Связь с другим сообщением (например, ответ или пересылка).
notify	Optional[bool]	Отправлять ли уведомление о сообщении. По умолчанию True.
format	Optional[TextFormat]	Режим форматирования (например, Markdown, HTML).
parse_mode	Optional[ParseMode]	Режим форматирования текста (например, Markdown, HTML).
disable_link_preview	Optional[bool]	Флаг генерации превью.

Source code in maxapi/methods/send_message.py

def __init__(
    self,
    bot: "Bot",
    chat_id: int | None = None,
    user_id: int | None = None,
    text: str | None = None,
    attachments: list[
        Attachment | InputMedia | InputMediaBuffer | AttachmentUpload
    ]
    | list[Attachments]
    | None = None,
    link: NewMessageLink | None = None,
    format: TextFormat | None = None,
    parse_mode: ParseMode | None = None,
    *,
    notify: bool | None = None,
    disable_link_preview: bool | None = None,
    sleep_after_input_media: bool | None = True,
):
    if text is not None and not (len(text) < 4000):
        raise ValueError("text должен быть меньше 4000 символов")

    super().__init__()
    self.bot = bot
    self.chat_id = chat_id
    self.user_id = user_id
    self.text = text
    self.attachments = attachments
    self.link = link
    self.notify = notify
    if parse_mode is not None:
        warnings.warn(
            "Параметр parse_mode устарел, используйте format.",
            DeprecationWarning,
            stacklevel=4,
        )
    self.format = format if format is not None else parse_mode
    self.disable_link_preview = disable_link_preview
    self.sleep_after_input_media = sleep_after_input_media

fetch() async

Отправляет сообщение с вложениями (если есть), с обработкой задержки готовности вложений.

Возвращает результат отправки или ошибку.

Возвращаемое значение

SendedMessage или Error

Source code in maxapi/methods/send_message.py

async def fetch(self) -> SendedMessage | None:
    """
    Отправляет сообщение с вложениями (если есть),
    с обработкой задержки готовности вложений.

    Возвращает результат отправки или ошибку.

    Возвращаемое значение:
        SendedMessage или Error
    """

    bot = self._ensure_bot()

    params = bot.params.copy()

    json: dict[str, Any] = {"attachments": []}

    if self.chat_id:
        params["chat_id"] = self.chat_id
    elif self.user_id:
        params["user_id"] = self.user_id

    if self.disable_link_preview is not None:
        params["disable_link_preview"] = str(
            self.disable_link_preview
        ).lower()

    json["text"] = self.text

    has_input_media = False

    if self.attachments:
        for att in self.attachments:
            if isinstance(att, (InputMedia, InputMediaBuffer)):
                has_input_media = True

                input_media = await process_input_media(
                    base_connection=self, bot=bot, att=att
                )
                json["attachments"].append(input_media.model_dump())
            elif isinstance(att, Attachment) and isinstance(
                att.payload, AttachmentUpload
            ):
                json["attachments"].append(att.payload.model_dump())
            else:
                json["attachments"].append(att.model_dump())

    if self.link is not None:
        json["link"] = self.link.model_dump()

    if self.notify is not None:
        json["notify"] = self.notify

    if self.format is not None:
        json["format"] = self.format.value

    if has_input_media and self.sleep_after_input_media:
        await asyncio.sleep(bot.after_input_media_delay)

    attempts = bot.after_upload_attempts
    retry_delay = bot.after_upload_retry_delay
    give_up_timeout = bot.after_upload_give_up_timeout

    response = None
    start_time = time.monotonic()
    for attempt in range(attempts):
        try:
            response = await super().request(
                method=HTTPMethod.POST,
                path=ApiPath.MESSAGES,
                model=SendedMessage,
                params=params,
                json=json,
            )
        except MaxApiError as e:
            if (
                isinstance(e.raw, dict)
                and e.raw.get("code") == "attachment.not.ready"
            ):
                elapsed = time.monotonic() - start_time
                if (
                    give_up_timeout is not None
                    and elapsed + retry_delay > give_up_timeout
                ):
                    raise RuntimeError(
                        f"Превышено максимальное время ожидания"
                        f" готовности медиа"
                        f" ({give_up_timeout}с),"
                        f" прошло {elapsed:.1f}с"
                    ) from e
                logger_bot.info(
                    f"Ошибка при отправке загруженного медиа,"
                    f" попытка {attempt + 1},"
                    f" жду {retry_delay} секунды"
                )
                await asyncio.sleep(retry_delay)
                continue
            else:
                raise e

        break

    if response is None:
        raise RuntimeError("Не удалось отправить сообщение")

    return cast(SendedMessage | None, response)