[4133] Тип файлов для аудио

app/common/filetype_ext.py

@@ -1,7 +1,7 @@
 from typing import Final
 
 import filetype  # type: ignore[import-untyped]
-from filetype.types import archive, image  # type: ignore[import-untyped]
+from filetype.types import archive, audio, image  # type: ignore[import-untyped]
 
 FILE_HEADER_SIZE: Final[int] = 8192
 
@@ -21,6 +21,15 @@
     archive.Pdf(),
 ]
 
+SUPPORTED_AUDIO_FORMATS: list[filetype.Type] = [
+    audio.Aac(),
+    audio.Mp3(),
+    audio.M4a(),
+    audio.Ogg(),
+    audio.Flac(),
+    audio.Wav(),
+]
+
 
 def match_filetype(obj: bytes, matchers: list[filetype.Type]) -> filetype.Type | None:
     return filetype.match(obj, matchers)
@@ -32,3 +41,7 @@ def match_image_filetype(obj: bytes) -> filetype.Type | None:
 
 def match_document_filetype(obj: bytes) -> filetype.Type | None:
     return match_filetype(obj, SUPPORTED_DOCUMENT_FORMATS)
+
+
+def match_audio_filetype(obj: bytes) -> filetype.Type | None:
+    return match_filetype(obj, SUPPORTED_AUDIO_FORMATS)

app/storage_v2/dependencies/uploads_dep.py

@@ -1,3 +1,4 @@
+from mimetypes import guess_all_extensions
 from typing import Annotated
 
 from fastapi import Depends, UploadFile
@@ -6,6 +7,7 @@
 from app.common.fastapi_ext import Responses, with_responses
 from app.common.filetype_ext import (
     FILE_HEADER_SIZE,
+    match_audio_filetype,
     match_document_filetype,
     match_image_filetype,
 )
@@ -52,3 +54,23 @@ async def validate_document_upload(upload: UploadFile) -> UploadFile:
 
 
 ValidatedDocumentUpload = Annotated[UploadFile, Depends(validate_document_upload)]
+
+
+@with_responses(FileFormatResponses)
+async def validate_audio_upload(upload: UploadFile) -> UploadFile:
+    upload_header_data = await upload.read(FILE_HEADER_SIZE)
+    audio_type = match_audio_filetype(upload_header_data)
+
+    if audio_type is None:
+        raise FileFormatResponses.WRONG_FORMAT
+
+    audio_extensions = guess_all_extensions(upload.content_type or "")
+
+    if f".{audio_type.extension}" not in audio_extensions:
+        raise FileFormatResponses.CONTENT_TYPE_MISMATCH
+
+    await upload.seek(0)
+    return upload
+
+
+ValidatedAudioUpload = Annotated[UploadFile, Depends(validate_audio_upload)]

app/storage_v2/main.py

@@ -13,6 +13,7 @@
     files_rst,
     ydocs_hocus_int,
 )
+from app.storage_v2.utils.mimetypes import add_missing_mime_to_mimetypes
 
 outside_router = APIRouterExt(prefix="/api/public/storage-service/v2")
 
@@ -40,6 +41,9 @@ async def lifespan(_: Any) -> AsyncIterator[None]:
     settings.storage_path.mkdir(exist_ok=True)
     for sub_folder in FILE_KIND_TO_FOLDER.values():
         (settings.storage_path / sub_folder).mkdir(exist_ok=True)
+
+    add_missing_mime_to_mimetypes()
+
     yield
 
 

app/storage_v2/models/files_db.py

@@ -15,6 +15,7 @@ class FileKind(StrEnum):
     UNCATEGORIZED = "uncategorized"
     IMAGE = "image"
     DOCUMENT = "document"
+    AUDIO = "audio"
 
 
 ContentDisposition = Literal["inline", "attachment"]
@@ -23,16 +24,19 @@ class FileKind(StrEnum):
     FileKind.UNCATEGORIZED: "uncategorized",
     FileKind.IMAGE: "images",
     FileKind.DOCUMENT: "documents",
+    FileKind.AUDIO: "audios",
 }
 FILE_KIND_TO_MEDIA_TYPE: dict[FileKind, str | None] = {
     FileKind.UNCATEGORIZED: None,
     FileKind.IMAGE: "image/webp",
     FileKind.DOCUMENT: "application/pdf",
+    FileKind.AUDIO: None,
 }
 FILE_KIND_TO_CONTENT_DISPOSITION: dict[FileKind, ContentDisposition] = {
     FileKind.UNCATEGORIZED: "attachment",
     FileKind.IMAGE: "inline",
     FileKind.DOCUMENT: "inline",
+    FileKind.AUDIO: "inline",
 }
 
 

app/storage_v2/routers/files_rst.py

@@ -18,6 +18,7 @@
     UploadAllowedStorageTokenPayload,
 )
 from app.storage_v2.dependencies.uploads_dep import (
+    ValidatedAudioUpload,
     ValidatedDocumentUpload,
     ValidatedImageUpload,
 )
@@ -106,6 +107,24 @@ async def upload_document_file(
     )
 
 
+@router.post(
+    "/file-kinds/audio/files/",
+    status_code=status.HTTP_201_CREATED,
+    response_model=File.ResponseSchema,
+    summary="Upload a new audio file",
+)
+async def upload_audio_file(
+    storage_token_payload: UploadAllowedStorageTokenPayload,
+    upload: ValidatedAudioUpload,
+) -> File:
+    return await upload_file(
+        storage_token_payload=storage_token_payload,
+        upload_content=await upload.read(),
+        upload_filename=upload.filename,
+        file_kind=FileKind.AUDIO,
+    )
+
+
 @router.get(
     "/files/{file_id}/meta/",
     response_model=File.ResponseSchema,

app/storage_v2/utils/mimetypes.py

@@ -0,0 +1,28 @@
+import mimetypes
+
+from app.common.filetype_ext import SUPPORTED_AUDIO_FORMATS
+
+SUPPORTED_AUDIO_MIME_TYPES: dict[str, str] = {
+    # taken from https://developer.mozilla.org/en-US/docs/Web/Media/Guides/Formats/Containers
+    "audio/aac": ".aac",
+    "audio/mpeg": ".mp3",
+    "audio/mp3": ".mp3",
+    "audio/mp4": ".m4a",
+    "audio/ogg": ".ogg",
+    "audio/flac": ".flac",
+    "audio/x-flac": ".flac",
+    "audio/wave": ".wav",
+    "audio/wav": ".wav",
+    "audio/x-wav": ".wav",
+    "audio/x-pn-wav": ".wav",
+}
+
+
+def add_missing_mime_to_mimetypes() -> None:
+    for audio_format in SUPPORTED_AUDIO_FORMATS:
+        if mimetypes.guess_extension(audio_format.mime) is None:
+            mimetypes.add_type(audio_format.mime, f".{audio_format.extension}")
+
+    for mime, extension in SUPPORTED_AUDIO_MIME_TYPES.items():
+        if mimetypes.guess_extension(mime) is None:
+            mimetypes.add_type(mime, extension)

migrations/versions/064_audio_files.py

@@ -0,0 +1,57 @@
+"""audio_files
+
+Revision ID: 064
+Revises: 063
+Create Date: 2026-04-17 04:07:30.284265
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+schema_name = "xi_back_2"
+table_name = "files"
+column_name = "kind"
+enum_name = "file_kind"
+tmp_enum_name = f"_{enum_name}"
+
+old_enum = sa.Enum("UNCATEGORIZED", "IMAGE", "DOCUMENT", name=enum_name)
+
+# revision identifiers, used by Alembic.
+revision: str = "064"
+down_revision: Union[str, None] = "063"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(f"ALTER TYPE {enum_name} ADD VALUE 'AUDIO'")
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+
+    # rename new enum
+    op.execute(f"ALTER TYPE {enum_name} RENAME TO {tmp_enum_name}")
+
+    # update old rows
+    metadata = sa.MetaData(schema=schema_name)
+    Files = sa.Table(table_name, metadata, autoload_with=conn)
+
+    conn.execute(
+        sa.update(Files).where(Files.c.kind == "AUDIO").values(kind="UNCATEGORIZED")
+    )
+
+    # remove old members by updating to the new enum
+    old_enum.create(bind=conn)
+    op.execute(
+        f"ALTER TABLE {schema_name}.{table_name}"
+        f" ALTER COLUMN {column_name}"
+        f" TYPE {old_enum.name}"
+        f" USING {column_name}::text::{old_enum.name}"
+    )
+
+    # remove new enum
+    op.execute(f"DROP TYPE {tmp_enum_name}")

tests/storage_v2/conftest.py

@@ -1,3 +1,4 @@
+import wave
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from io import BytesIO
@@ -221,12 +222,37 @@ def pdf_document_file_input_data(
     )
 
 
+@pytest.fixture()
+def wav_audio_file_content(faker: Faker) -> bytes:
+    audio_content = BytesIO()
+    with wave.open(audio_content, "wb") as f:
+        f.setnchannels(1)
+        f.setsampwidth(2)
+        f.setframerate(44100)
+        f.writeframes(faker.binary(44100))
+    return audio_content.getvalue()
+
+
+@pytest.fixture()
+def wav_audio_file_input_data(
+    faker: Faker, wav_audio_file_content: bytes
+) -> FileInputData:
+    return FileInputData(
+        kind=FileKind.AUDIO,
+        name=faker.file_name(extension="wav"),
+        input_content=wav_audio_file_content,
+        processed_content=wav_audio_file_content,
+        content_type="audio/wav",
+    )
+
+
 @pytest.fixture(
     params=[
         pytest.param(lf("uncategorized_file_input_data"), id="uncategorized"),
         pytest.param(lf("webp_image_file_input_data"), id="webp_image"),
         pytest.param(lf("png_image_file_input_data"), id="png_image"),
         pytest.param(lf("pdf_document_file_input_data"), id="pdf_document"),
+        pytest.param(lf("wav_audio_file_input_data"), id="wav_audio"),
     ],
 )
 def parametrized_file_input_data(

tests/storage_v2/functional/test_file_uploads_rst.py

@@ -117,6 +117,12 @@ async def test_file_uploading(
     ("image/tiff", "tiff"),
     ("image/webp", "webp"),
     ("application/pdf", "pdf"),
+    ("audio/aac", "aac"),
+    ("audio/mpeg", "mp3"),
+    ("audio/mp4", "m4a"),
+    ("audio/ogg", "ogg"),
+    ("audio/x-flac", "flac"),
+    ("audio/x-wav", "wav"),
 ]
 
 
@@ -126,9 +132,10 @@ async def test_file_uploading(
         pytest.param(lf("webp_image_file_input_data"), id="webp"),
         pytest.param(lf("png_image_file_input_data"), id="png"),
         pytest.param(lf("pdf_document_file_input_data"), id="pdf"),
+        pytest.param(lf("wav_audio_file_input_data"), id="wav"),
     ],
 )
-async def test_image_file_uploading_content_type_mismatch(
+async def test_file_uploading_content_type_mismatch(
     faker: Faker,
     authorized_client: TestClient,
     file_upload_storage_token: str,
@@ -164,6 +171,7 @@ async def test_image_file_uploading_content_type_mismatch(
     [
         pytest.param(lf("webp_image_file_input_data"), id="image"),
         pytest.param(lf("pdf_document_file_input_data"), id="document"),
+        pytest.param(lf("wav_audio_file_input_data"), id="audio"),
     ],
 )
 async def test_file_uploading_wrong_content_format(

tests/storage_v2/unit/test_adding_missing_mime_to_mimetypes.py

@@ -0,0 +1,35 @@
+import mimetypes
+
+import filetype  # type: ignore[import-untyped]
+import pytest
+
+from app.common.filetype_ext import SUPPORTED_AUDIO_FORMATS
+from app.storage_v2.utils.mimetypes import SUPPORTED_AUDIO_MIME_TYPES
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.mark.parametrize(
+    "audio_format",
+    [pytest.param(audio, id=audio.extension) for audio in SUPPORTED_AUDIO_FORMATS],
+)
+async def test_matching_extension_by_filetype_mime_types(
+    audio_format: filetype.Type,
+) -> None:
+    assert f".{audio_format.extension}" in mimetypes.guess_all_extensions(
+        audio_format.mime
+    )
+
+
+@pytest.mark.parametrize(
+    (
+        "mime",
+        "extension",
+    ),
+    [
+        pytest.param(mime, extension, id=extension)
+        for mime, extension in SUPPORTED_AUDIO_MIME_TYPES.items()
+    ],
+)
+async def test_presence_extension_by_mime_types(mime: str, extension: str) -> None:
+    assert extension in mimetypes.guess_all_extensions(mime)