diff --git a/maxapi/connection/base.py b/maxapi/connection/base.py index c8615ae..54cd7fd 100644 --- a/maxapi/connection/base.py +++ b/maxapi/connection/base.py @@ -264,7 +264,7 @@ async def upload_file_buffer( try: matches = puremagic.magic_string(buffer[:4096]) if matches: - mime_type = matches[0][1] + mime_type = matches[0].mime_type ext = mimetypes.guess_extension(mime_type) or "" else: mime_type = f"{type.value}/*" diff --git a/maxapi/types/input_media.py b/maxapi/types/input_media.py index 8aa151c..6577cb1 100644 --- a/maxapi/types/input_media.py +++ b/maxapi/types/input_media.py @@ -6,6 +6,53 @@ from ..enums.upload_type import UploadType +READ_FILE_CHUNK_SIZE = 4096 + + +def detect_file_type(data: bytes) -> UploadType: + """ + Определяет тип файла на основе его содержимого (MIME-типа). + + Args: + data (bytes): Буфер с содержимым файла. + Returns: + UploadType: Определенный тип файла. Если MIME-тип не удалось + определить или при определении произошла ошибка, + возвращается ``UploadType.FILE``. + """ + try: + matches = puremagic.magic_string(data) + 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 + + +def validate_uploading_type(type: UploadType | str) -> UploadType: + if not isinstance(type, UploadType): + try: + return UploadType(type) + except ValueError as e: + allowed = ", ".join(item.value for item in UploadType) + raise ValueError( + f"Неверный тип загружаемого файла: {type!r}. Ожидается: {allowed}" # noqa: E501 + ) from e + + return type + class InputMedia: """ @@ -14,56 +61,32 @@ class InputMedia: Attributes: path (str): Путь к файлу. type (UploadType): Тип файла, определенный на основе содержимого - (MIME-типа). + (MIME-типа) или указанный вручную. """ - def __init__(self, path: str, type: UploadType | None = None): + def __init__(self, path: str, type: UploadType | str | None = None): """ Инициализирует объект медиафайла. Args: path (str): Путь к файлу. - type (UploadType, optional): Тип файла. Если не указан, + type (UploadType | str | None): Тип файла. Если не указан, определяется автоматически. """ self.path = path - self.type = type or self.__detect_file_type(path) - - def __detect_file_type(self, path: str) -> UploadType: - """ - Определяет тип файла на основе его содержимого (MIME-типа). - - Args: - path (str): Путь к файлу. - Returns: - UploadType: Тип файла (VIDEO, IMAGE, AUDIO или FILE). - """ + if type is not None: + self.type = validate_uploading_type(type) + else: + self.type = detect_file_type(InputMedia._read_file_sample(path)) + @staticmethod + def _read_file_sample( + path: str, size: int = READ_FILE_CHUNK_SIZE + ) -> bytes: with Path(path).open("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 + return f.read(size) class InputMediaBuffer: @@ -72,14 +95,15 @@ class InputMediaBuffer: Attributes: buffer (bytes): Буфер с содержимым файла. - type (UploadType): Тип файла, определенный по содержимому. + type (UploadType): Тип файла, определенный на основе содержимого + (MIME-типа) или указанный вручную. """ def __init__( self, buffer: bytes, filename: str | None = None, - type: UploadType | None = None, + type: UploadType | str | None = None, ): """ Инициализирует объект медиафайла из буфера. @@ -88,31 +112,14 @@ def __init__( buffer (bytes): Буфер с содержимым файла. filename (str, optional): Название файла (по умолчанию присваивается uuid4). - type (UploadType, optional): Тип файла. Если не указан, + type (UploadType | str | None): Тип файла. Если не указан, определяется автоматически. """ self.filename = filename self.buffer = buffer - self.type = type or self.__detect_file_type(buffer) - - def __detect_file_type(self, buffer: bytes) -> 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/"): - return UploadType.IMAGE - elif mime_type.startswith("audio/"): - return UploadType.AUDIO + if type is not None: + self.type = validate_uploading_type(type) else: - return UploadType.FILE + self.type = detect_file_type(buffer) diff --git a/tests/test_upload_file.py b/tests/test_upload_file.py index 8312300..3b57481 100644 --- a/tests/test_upload_file.py +++ b/tests/test_upload_file.py @@ -7,6 +7,7 @@ from maxapi.client.default import DefaultConnectionProperties from maxapi.connection.base import BaseConnection from maxapi.enums.upload_type import UploadType +from maxapi.types.input_media import InputMedia, InputMediaBuffer def _make_connection_with_bot(*, session=None): @@ -170,3 +171,188 @@ async def test_uses_existing_session_when_open(self, tmp_path): mock_cs_cls.assert_not_called() mock_session.post.assert_awaited_once() + + +def assert_invalid_type_error(exc_info, invalid_value: str) -> None: + """Проверяет текст ошибки для невалидного type.""" + message = str(exc_info.value) + assert "Неверный тип загружаемого файла" in message + assert repr(invalid_value) in message + assert "file" in message + assert "image" in message + assert "video" in message + assert "audio" in message + + +class TestInputMediaTypeValidation: + """Тесты валидации type в InputMedia.""" + + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("file", UploadType.FILE), + ("image", UploadType.IMAGE), + ("video", UploadType.VIDEO), + ("audio", UploadType.AUDIO), + ], + ) + def test_accepts_valid_string_type( + self, tmp_path, monkeypatch, value, expected + ): + """Явно переданный строковый type валидируется без autodetect.""" + test_file = tmp_path / "sample.bin" + test_file.write_bytes(b"fake-data") + + mock_detect = Mock(return_value=UploadType.FILE) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMedia(path=str(test_file), type=value) + + assert media.path == str(test_file) + assert media.type == expected + mock_detect.assert_not_called() + + def test_invalid_string_type_raises_value_error(self, tmp_path): + """Невалидный строковый type вызывает ValueError со списком значений.""" # noqa: E501 + test_file = tmp_path / "sample.bin" + test_file.write_bytes(b"fake-data") + + with pytest.raises(ValueError) as exc_info: # noqa: PT011 + InputMedia(path=str(test_file), type="document") + + assert_invalid_type_error(exc_info, "document") + + def test_none_type_detects_from_file(self, tmp_path, monkeypatch): + """Если type не передан, тип определяется автоматически.""" + test_file = tmp_path / "sample.bin" + test_file.write_bytes(b"fake-data") + + mock_detect = Mock(return_value=UploadType.VIDEO) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMedia(path=str(test_file)) + + assert media.path == str(test_file) + assert media.type == UploadType.VIDEO + mock_detect.assert_called_once() + + def test_accepts_enum_type_without_autodetect(self, tmp_path, monkeypatch): + """Явно переданный UploadType используется без autodetect.""" + test_file = tmp_path / "sample.bin" + test_file.write_bytes(b"fake-data") + + mock_detect = Mock(return_value=UploadType.FILE) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMedia(path=str(test_file), type=UploadType.IMAGE) + + assert media.type == UploadType.IMAGE + mock_detect.assert_not_called() + + +class TestInputMediaBufferTypeValidation: + """Тесты валидации type в InputMediaBuffer.""" + + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("file", UploadType.FILE), + ("image", UploadType.IMAGE), + ("video", UploadType.VIDEO), + ("audio", UploadType.AUDIO), + ], + ) + def test_accepts_valid_string_type(self, monkeypatch, value, expected): + """Явно переданный строковый type валидируется без autodetect.""" + mock_detect = Mock(return_value=UploadType.FILE) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMediaBuffer( + buffer=b"fake-bytes", + filename="sample.bin", + type=value, + ) + + assert media.filename == "sample.bin" + assert media.buffer == b"fake-bytes" + assert media.type == expected + mock_detect.assert_not_called() + + def test_invalid_string_type_raises_value_error(self): + """Невалидный строковый type вызывает ValueError со списком значений.""" # noqa: E501 + with pytest.raises(ValueError) as exc_info: # noqa: PT011 + InputMediaBuffer( + buffer=b"fake-bytes", + filename="sample.bin", + type="document", + ) + + assert_invalid_type_error(exc_info, "document") + + def test_none_type_detects_from_buffer(self, monkeypatch): + """Если type не передан, тип определяется автоматически.""" + mock_detect = Mock(return_value=UploadType.IMAGE) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMediaBuffer(buffer=b"fake-bytes") + + assert media.filename is None + assert media.buffer == b"fake-bytes" + assert media.type == UploadType.IMAGE + mock_detect.assert_called_once_with(b"fake-bytes") + + def test_accepts_enum_type_without_autodetect(self, monkeypatch): + """Явно переданный UploadType используется без autodetect.""" + mock_detect = Mock(return_value=UploadType.FILE) + monkeypatch.setattr( + "maxapi.types.input_media.detect_file_type", + mock_detect, + ) + + media = InputMediaBuffer( + buffer=b"fake-bytes", + filename="sample.bin", + type=UploadType.AUDIO, + ) + + assert media.type == UploadType.AUDIO + mock_detect.assert_not_called() + + def test_default_upload_type_input_media_buffer(self, tmp_path): + """ + Если mimetype не определился (None), + для файла должен вернуться тип UploadType.FILE + """ + media = InputMediaBuffer( + buffer=b"fake-bytes", + filename="sample.bin", + ) + + assert media.type == UploadType.FILE + + def test_default_upload_type_input_media(self, tmp_path): + """ + Если mimetype не определился (None), + для файла должен вернуться тип UploadType.FILE + """ + test_file = tmp_path / "sample.bin" + test_file.write_bytes(b"fake-data") + + media = InputMedia(path=test_file) + + assert media.type == UploadType.FILE