From dd1bdb5e37c0b5538ffc8b17e0023d1c3a1e2f77 Mon Sep 17 00:00:00 2001 From: Denis Date: Sat, 19 Jul 2025 13:41:03 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B7=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=BA=D0=B0?= =?UTF-8?q?=20=D1=84=D0=B0=D0=B9=D0=BB=D0=BE=D0=B2=20=D0=B8=D0=B7=20=D0=B1?= =?UTF-8?q?=D1=83=D1=84=D0=B5=D1=80=D0=B0,=20InputMediaBuffer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- maxapi/connection/base.py | 48 ++++++++++++++++++++++ maxapi/methods/send_message.py | 23 +++++++---- maxapi/types/__init__.py | 1 + maxapi/types/input_media.py | 74 +++++++++++++++++++++++++++++----- 4 files changed, 128 insertions(+), 18 deletions(-) diff --git a/maxapi/connection/base.py b/maxapi/connection/base.py index 7951521..33214d1 100644 --- a/maxapi/connection/base.py +++ b/maxapi/connection/base.py @@ -1,10 +1,13 @@ import os +import mimetypes from typing import TYPE_CHECKING +from uuid import uuid4 import aiofiles import aiohttp +import puremagic from pydantic import BaseModel from ..exceptions.invalid_token import InvalidToken @@ -135,6 +138,51 @@ class BaseConnection: return await response.text() + async def upload_file_buffer( + self, + url: str, + buffer: bytes, + type: UploadType + ): + """ + Загружает файл из буфера. + + :param url: Конечная точка загрузки файла + :param buffer: Буфер (bytes) + :param type: Тип файла (video, image, audio, file) + + :return: Сырой .text() ответ от сервера после загрузки файла + """ + + try: + matches = puremagic.magic_string(buffer[:4096]) + if matches: + mime_type = matches[0][1] + ext = mimetypes.guess_extension(mime_type) or '' + else: + mime_type = f"{type.value}/*" + ext = '' + except Exception: + mime_type = f"{type.value}/*" + ext = '' + + basename = f'{uuid4()}{ext}' + + form = aiohttp.FormData() + form.add_field( + name='data', + value=buffer, + filename=basename, + content_type=mime_type + ) + + async with aiohttp.ClientSession() as session: + response = await session.post( + url=url, + data=form + ) + return await response.text() + async def download_file( self, path: str, diff --git a/maxapi/methods/send_message.py b/maxapi/methods/send_message.py index d9ecfb6..8d3333b 100644 --- a/maxapi/methods/send_message.py +++ b/maxapi/methods/send_message.py @@ -9,7 +9,7 @@ from .types.sended_message import SendedMessage from ..types.attachments.upload import AttachmentPayload, AttachmentUpload from ..types.errors import Error from ..types.message import NewMessageLink -from ..types.input_media import InputMedia +from ..types.input_media import InputMedia, InputMediaBuffer from ..types.attachments.attachment import Attachment from ..enums.upload_type import UploadType @@ -67,7 +67,7 @@ class SendMessage(BaseConnection): async def __process_input_media( self, - att: InputMedia + att: InputMedia | InputMediaBuffer ): # очень нестабильный метод независящий от модуля @@ -85,11 +85,18 @@ class SendMessage(BaseConnection): upload = await self.bot.get_upload_url(att.type) - upload_file_response = await self.upload_file( - url=upload.url, - path=att.path, - type=att.type - ) + if isinstance(att, InputMedia): + upload_file_response = await self.upload_file( + url=upload.url, + path=att.path, + type=att.type, + ) + elif isinstance(att, InputMediaBuffer): + upload_file_response = await self.upload_file_buffer( + url=upload.url, + buffer=att.buffer, + type=att.type, + ) if att.type in (UploadType.VIDEO, UploadType.AUDIO): token = upload.token @@ -134,7 +141,7 @@ class SendMessage(BaseConnection): for att in self.attachments: - if isinstance(att, InputMedia): + if isinstance(att, InputMedia) or isinstance(att, InputMediaBuffer): input_media = await self.__process_input_media(att) json['attachments'].append( input_media.model_dump() diff --git a/maxapi/types/__init__.py b/maxapi/types/__init__.py index b6e4de7..7d81881 100644 --- a/maxapi/types/__init__.py +++ b/maxapi/types/__init__.py @@ -26,6 +26,7 @@ from ..types.message import Message from ..types.command import Command, BotCommand from .input_media import InputMedia +from .input_media import InputMediaBuffer __all__ = [ UpdateUnion, diff --git a/maxapi/types/input_media.py b/maxapi/types/input_media.py index 602fdef..8e30080 100644 --- a/maxapi/types/input_media.py +++ b/maxapi/types/input_media.py @@ -1,34 +1,38 @@ -import mimetypes +from __future__ import annotations + +from typing import TYPE_CHECKING + +import puremagic from ..enums.upload_type import UploadType +if TYPE_CHECKING: + from io import BytesIO + + class InputMedia: - """ Класс для представления медиафайла. Attributes: path (str): Путь к файлу. - type (UploadType): Тип файла, определенный на основе MIME-типа. + type (UploadType): Тип файла, определенный на основе содержимого (MIME-типа). """ - + def __init__(self, path: str): - """ Инициализирует объект медиафайла. Args: path (str): Путь к файлу. """ - self.path = path self.type = self.__detect_file_type(path) def __detect_file_type(self, path: str) -> UploadType: - """ - Определяет тип файла на основе его MIME-типа. + Определяет тип файла на основе его содержимого (MIME-типа). Args: path (str): Путь к файлу. @@ -36,12 +40,62 @@ class InputMedia: Returns: UploadType: Тип файла (VIDEO, IMAGE, AUDIO или FILE). """ - - mime_type, _ = mimetypes.guess_type(path) + with open(path, 'rb') as f: + sample = f.read(4096) + + try: + matches = puremagic.magic_string(sample) + if matches: + mime_type = matches[0].mime_type + else: + mime_type = None + except Exception: + mime_type = None if mime_type is None: return UploadType.FILE + if mime_type.startswith('video/'): + return UploadType.VIDEO + elif mime_type.startswith('image/'): + return UploadType.IMAGE + elif mime_type.startswith('audio/'): + return UploadType.AUDIO + else: + return UploadType.FILE + + +class InputMediaBuffer: + """ + Класс для представления медиафайла из буфера. + + Attributes: + buffer (BytesIO): Буфер с содержимым файла. + type (UploadType): Тип файла, определенный по содержимому. + """ + + def __init__(self, buffer: BytesIO): + """ + Инициализирует объект медиафайла из буфера. + + Args: + buffer (IO): Буфер с содержимым файла. + """ + self.buffer = buffer + self.type = self.__detect_file_type(buffer) + + def __detect_file_type(self, buffer: BytesIO) -> UploadType: + try: + matches = puremagic.magic_string(buffer) + if matches: + mime_type = matches[0].mime_type + else: + mime_type = None + except Exception: + mime_type = None + + if mime_type is None: + return UploadType.FILE if mime_type.startswith('video/'): return UploadType.VIDEO elif mime_type.startswith('image/'):