RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Simplify audio transcription flow

This commit is contained in:
adavyas 2026-04-08 20:39:00 -04:00
parent f008703a6c
commit e1d36e5627
5 changed files with 81 additions and 505 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.env.template
View File

@ -84,10 +84,7 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here
# =============================================================================
# AUDIO_PROVIDER=openai
# AUDIO_MODEL=whisper-1
# AUDIO_MAX_FILE_SIZE_BYTES=9500000
# AUDIO_MAX_CHUNK_DURATION_SECONDS=55
# AUDIO_MAX_CHUNK_BYTES=5242880
# AUDIO_TRANSCRIPTION_CONCURRENCY=4
# AUDIO_MAX_FILE_SIZE_BYTES=25000000
# =============================================================================
# LLM Configuration

5
config.toml.example
View File

@ -71,10 +71,7 @@ MAX_MESSAGE_CONTENT_CHARS = 2000 # Max chars per message in tool results
[audio]
PROVIDER = "openai"
MODEL = "whisper-1"
MAX_FILE_SIZE_BYTES = 9500000 # 9.5MB
MAX_CHUNK_DURATION_SECONDS = 55
MAX_CHUNK_BYTES = 5242880
TRANSCRIPTION_CONCURRENCY = 4
MAX_FILE_SIZE_BYTES = 25000000 # 25MB
# Deriver settings
[deriver]

16
src/config.py
View File

@ -1,3 +1,4 @@
# ruff: noqa: I001
import logging
from pathlib import Path
from typing import Annotated, Any, ClassVar, Literal, Protocol
@ -6,13 +7,7 @@ import tomllib
from dotenv import load_dotenv
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic.fields import FieldInfo
from pydantic_settings import (
BaseSettings,
DotEnvSettingsSource,
EnvSettingsSource,
PydanticBaseSettingsSource,
SettingsConfigDict,
)
from pydantic_settings import BaseSettings, DotEnvSettingsSource, EnvSettingsSource, PydanticBaseSettingsSource, SettingsConfigDict
from src.utils.types import SupportedProviders
@ -238,12 +233,7 @@ class AudioSettings(HonchoSettings):
PROVIDER: Literal["openai"] = "openai"
MODEL: str = "whisper-1"
MAX_FILE_SIZE_BYTES: Annotated[int, Field(default=9_500_000, gt=0)] = 9_500_000
MAX_CHUNK_DURATION_SECONDS: Annotated[int, Field(default=55, gt=0, le=3600)] = (
55
)
MAX_CHUNK_BYTES: Annotated[int, Field(default=5_242_880, gt=0)] = 5_242_880
TRANSCRIPTION_CONCURRENCY: Annotated[int, Field(default=4, gt=0, le=16)] = 4
MAX_FILE_SIZE_BYTES: Annotated[int, Field(default=25_000_000, gt=0)] = 25_000_000
class DeriverSettings(BackupLLMSettingsMixin, HonchoSettings):

284
src/utils/files.py
View File

@ -1,10 +1,9 @@
# ruff: noqa: I001
import asyncio
import datetime
import logging
import math
import subprocess
import tempfile
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
from typing import Any, Protocol
@ -16,11 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src import schemas
from src.config import settings
from src.exceptions import (
FileProcessingError,
UnsupportedFileTypeError,
ValidationException,
)
from src.exceptions import FileProcessingError, UnsupportedFileTypeError, ValidationException
from src.schemas import Message
from src.utils.clients import CLIENTS, transcribe_audio
@ -60,13 +55,6 @@ class ExtractedFileText(str):
return str(self)
@dataclass
class AudioSegment:
index: int
filename: str
content: bytes
class FileProcessor(Protocol):
async def extract_text(self, content: bytes) -> ExtractedFileText: ...
def supports_file_type(self, content_type: str) -> bool: ...
@ -160,193 +148,33 @@ class AudioProcessor:
if not content:
raise ValidationException("Audio upload is empty")
suffix = self.get_output_suffix(filename, content_type)
normalized_filename = self.ensure_audio_filename(filename, suffix)
normalized_content_type = self.normalize_content_type(filename, content_type)
segments = await asyncio.to_thread(
self.split_audio_segments,
self._probe_audio_duration_seconds(
content,
filename,
normalized_content_type,
suffix,
)
return await self.transcribe_segments(
segments,
text = await transcribe_audio(
content,
filename=normalized_filename,
content_type=normalized_content_type,
concurrency=settings.AUDIO.TRANSCRIPTION_CONCURRENCY,
)
async def transcribe_segments(
self,
segments: list[AudioSegment],
*,
content_type: str,
concurrency: int,
) -> ExtractedFileText:
semaphore = asyncio.Semaphore(max(1, concurrency))
async def _transcribe_segment(
segment: AudioSegment,
) -> tuple[int, str]:
async with semaphore:
text = await transcribe_audio(
segment.content,
filename=segment.filename,
content_type=content_type,
)
return segment.index, text.strip()
transcripts = await asyncio.gather(
*[_transcribe_segment(segment) for segment in segments]
)
transcripts.sort(key=lambda item: item[0])
ordered_text = "\n".join(text for _, text in transcripts if text)
return ExtractedFileText(
text=ordered_text,
text=text,
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": len(segments),
"audio_segment_count": 1,
"transcription_provider": settings.AUDIO.PROVIDER,
},
)
def split_audio_segments(
self,
content: bytes,
filename: str,
content_type: str,
) -> list[AudioSegment]:
if not content:
raise ValidationException("Audio upload is empty")
suffix = self.get_output_suffix(filename, content_type)
normalized_filename = self.ensure_audio_filename(filename, suffix)
duration_seconds = self._probe_audio_duration_seconds(content, suffix)
if (
len(content) <= settings.AUDIO.MAX_CHUNK_BYTES
and duration_seconds <= settings.AUDIO.MAX_CHUNK_DURATION_SECONDS
):
return [AudioSegment(index=0, filename=normalized_filename, content=content)]
segment_count = self._estimate_initial_segment_count(
duration_seconds=duration_seconds,
suffix=suffix,
)
with tempfile.TemporaryDirectory(prefix="honcho-audio-") as temp_dir:
temp_path = Path(temp_dir)
input_path = temp_path / f"input{suffix}"
input_path.write_bytes(content)
while True:
segments = self._build_audio_segments(
input_path=input_path,
suffix=suffix,
segment_count=segment_count,
duration_seconds=duration_seconds,
)
if not segments:
raise FileProcessingError("Audio segmentation produced no output")
max_segment_size = max(len(segment.content) for segment in segments)
if max_segment_size <= settings.AUDIO.MAX_CHUNK_BYTES:
return segments
if duration_seconds / segment_count <= 1.0:
if suffix == ".wav":
suffix = ".mp3"
segment_count = self._estimate_initial_segment_count(
duration_seconds=duration_seconds,
suffix=suffix,
)
continue
raise FileProcessingError(
"Audio segmentation could not satisfy max chunk size"
)
next_segment_count = max(
segment_count + 1,
math.ceil(
(max_segment_size * segment_count)
/ settings.AUDIO.MAX_CHUNK_BYTES
),
)
if next_segment_count == segment_count:
next_segment_count += 1
segment_count = next_segment_count
def normalize_content_type(self, filename: str, content_type: str) -> str:
if content_type in SUPPORTED_AUDIO_CONTENT_TYPES:
return content_type
extension = Path(filename).suffix.lower()
return AUDIO_EXTENSION_CONTENT_TYPES.get(extension, content_type)
def _estimate_initial_segment_count(
self,
*,
duration_seconds: float,
suffix: str,
) -> int:
estimated_output_bytes = duration_seconds * self._target_output_bytes_per_second(
suffix
)
return max(
math.ceil(estimated_output_bytes / settings.AUDIO.MAX_CHUNK_BYTES),
math.ceil(duration_seconds / settings.AUDIO.MAX_CHUNK_DURATION_SECONDS),
1,
)
def _target_output_bytes_per_second(self, suffix: str) -> float:
if suffix == ".wav":
return 176_400.0
return 16_000.0
def _build_audio_segments(
self,
*,
input_path: Path,
suffix: str,
segment_count: int,
duration_seconds: float,
) -> list[AudioSegment]:
segment_duration = max(duration_seconds / segment_count, 1.0)
segments: list[AudioSegment] = []
for index in range(segment_count):
output_path = input_path.parent / f"segment_{index:03d}{suffix}"
output_path.unlink(missing_ok=True)
start_time = index * segment_duration
command = [
"ffmpeg",
"-y",
"-hide_banner",
"-loglevel",
"error",
"-ss",
str(start_time),
"-i",
str(input_path),
]
if index < segment_count - 1:
command.extend(["-t", str(segment_duration)])
command.extend(self._segment_encoding_args(suffix))
command.append(str(output_path))
self._run_command(command)
if output_path.exists() and output_path.stat().st_size > 0:
segments.append(
AudioSegment(
index=index,
filename=output_path.name,
content=output_path.read_bytes(),
)
)
return segments
def get_output_suffix(self, filename: str, content_type: str) -> str:
suffix = Path(filename).suffix.lower()
if suffix in SUPPORTED_AUDIO_EXTENSIONS:
@ -397,21 +225,6 @@ class AudioProcessor:
except (subprocess.CalledProcessError, ValueError) as exc:
raise ValidationException("Uploaded audio is invalid or unreadable") from exc
def _segment_encoding_args(self, suffix: str) -> list[str]:
if suffix == ".wav":
return ["-vn", "-acodec", "pcm_s16le"]
return ["-vn", "-acodec", "libmp3lame", "-b:a", "128k"]
def _run_command(self, command: list[str]) -> None:
try:
subprocess.run(command, check=True, capture_output=True, text=True)
except FileNotFoundError as exc:
raise ValidationException(
"Audio uploads require ffmpeg and ffprobe to be installed on the server"
) from exc
except subprocess.CalledProcessError as exc:
raise FileProcessingError(exc.stderr or "Audio processing command failed") from exc
def is_audio_upload(file: UploadFile) -> bool:
return AudioProcessor().supports_upload(file)
@ -465,17 +278,14 @@ class FileProcessingService:
# Add more processors as needed
]
async def extract_text_from_bytes(
self,
content: bytes,
*,
filename: str | None,
content_type: str | None,
) -> ExtractedFileText:
normalized_content_type = content_type or ""
async def extract_text_from_upload(self, file: UploadFile) -> ExtractedFileText:
"""Extract text from uploaded file without saving to disk."""
content = await file.read()
await file.seek(0)
normalized_content_type = file.content_type or ""
if self.audio_processor.supports_content(
filename=filename,
filename=file.filename,
content_type=normalized_content_type,
):
if "openai" not in CLIENTS:
@ -484,31 +294,18 @@ class FileProcessingService:
)
return await self.audio_processor.extract_text(
content,
filename=filename,
filename=file.filename,
content_type=normalized_content_type,
)
processor = self._get_processor(normalized_content_type)
if not processor:
raise UnsupportedFileTypeError(
f"Unsupported file type: {content_type}. Supported types: {[p.__class__.__name__ for p in self.processors]}"
f"Unsupported file type: {file.content_type}. Supported types: {[p.__class__.__name__ for p in self.processors]}"
)
return await processor.extract_text(content)
async def extract_text_from_upload(self, file: UploadFile) -> ExtractedFileText:
"""Extract text from uploaded file without saving to disk."""
content = await file.read()
# Reset file position in case it's needed again
await file.seek(0)
return await self.extract_text_from_bytes(
content,
filename=file.filename,
content_type=file.content_type,
)
def _get_processor(self, content_type: str) -> FileProcessor | None:
for processor in self.processors:
if processor.supports_file_type(content_type):
@ -608,43 +405,10 @@ async def process_file_uploads_for_messages(
HTTPException: If file processing fails
"""
content = await file.read()
await file.seek(0)
return await process_upload_bytes_for_messages(
content,
filename=file.filename,
content_type=file.content_type,
file_size=file.size,
peer_id=peer_id,
max_chars=max_chars,
metadata=metadata,
configuration=configuration,
created_at=created_at,
)
async def process_upload_bytes_for_messages(
content: bytes,
*,
filename: str | None,
content_type: str | None,
file_size: int | None,
peer_id: str,
max_chars: int = settings.MAX_MESSAGE_SIZE,
metadata: dict[str, Any] | None = None,
configuration: schemas.MessageConfiguration | None = None,
created_at: datetime.datetime | None = None,
) -> list[dict[str, Any]]:
"""Process persisted upload bytes and prepare message creation data."""
file_processor = FileProcessingService()
all_message_data: list[dict[str, Any]] = []
extracted = await file_processor.extract_text_from_bytes(
content,
filename=filename,
content_type=content_type,
)
extracted_text = extracted.text
extracted_text = await file_processor.extract_text_from_upload(file)
# Split into chunks and create messages
chunks = split_text_into_chunks(extracted_text, max_chars=max_chars)
@ -666,17 +430,17 @@ async def process_upload_bytes_for_messages(
# Store file metadata separately to add to internal_metadata later
file_metadata = {
"file_id": file_id,
"filename": filename,
"filename": file.filename,
"chunk_index": i,
"total_chunks": len(chunks),
"original_file_size": file_size,
"content_type": content_type,
"original_file_size": file.size,
"content_type": file.content_type,
"chunk_character_range": [
i * max_chars,
min((i + 1) * max_chars, len(extracted_text)),
],
}
file_metadata.update(extracted.metadata)
file_metadata.update(extracted_text.metadata)
all_message_data.append(
{

276
tests/utils/test_audio_processing.py
View File

@ -1,9 +1,5 @@
import asyncio
import io
import math
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
@ -14,38 +10,7 @@ from starlette.datastructures import Headers
from src.config import settings
from src.exceptions import ValidationException
from src.utils.clients import CLIENTS, transcribe_audio
from src.utils.files import (
AudioProcessor,
AudioSegment,
FileProcessingService,
)
def _generate_test_audio_bytes(
audio_format: str,
duration_seconds: int = 1,
*,
audio_bitrate: str | None = None,
) -> bytes:
suffix = f".{audio_format}"
with tempfile.TemporaryDirectory(prefix="honcho-audio-test-") as temp_dir:
output_path = Path(temp_dir) / f"tone{suffix}"
command = [
"ffmpeg",
"-y",
"-hide_banner",
"-loglevel",
"error",
"-f",
"lavfi",
"-i",
f"sine=frequency=440:duration={duration_seconds}",
]
if audio_bitrate is not None:
command.extend(["-b:a", audio_bitrate])
command.append(str(output_path))
subprocess.run(command, check=True, capture_output=True, text=True)
return output_path.read_bytes()
from src.utils.files import AudioProcessor, FileProcessingService
def test_audio_processor_supports_mp3_and_wav_content_types():
@ -153,13 +118,8 @@ async def test_transcribe_audio_raises_when_openai_fails():
@pytest.mark.asyncio
async def test_transcribe_segments_preserves_order_when_tasks_finish_out_of_order():
async def test_audio_processor_extract_text_transcribes_directly():
processor = AudioProcessor()
segments = [
AudioSegment(index=0, filename="seg-0.mp3", content=b"0"),
AudioSegment(index=1, filename="seg-1.mp3", content=b"1"),
AudioSegment(index=2, filename="seg-2.mp3", content=b"2"),
]
async def fake_transcribe(
_content: bytes,
@ -168,37 +128,30 @@ async def test_transcribe_segments_preserves_order_when_tasks_finish_out_of_orde
**_: object,
):
assert content_type == "audio/mpeg"
if filename == "seg-0.mp3":
await asyncio.sleep(0.03)
return "first"
if filename == "seg-1.mp3":
await asyncio.sleep(0.0)
return "second"
assert filename == "seg-0.mp3"
await asyncio.sleep(0.01)
return "third"
return "first"
with patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
extracted = await processor.transcribe_segments(
segments,
with (
patch.object(processor, "_probe_audio_duration_seconds", return_value=1.0),
patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe),
):
extracted = await processor.extract_text(
b"audio-bytes",
filename="seg-0.mp3",
content_type="audio/mpeg",
concurrency=3,
)
assert extracted.text == "first\nsecond\nthird"
assert extracted.text == "first"
assert extracted.metadata["processing_type"] == "audio_transcription"
assert extracted.metadata["audio_segment_count"] == 3
assert extracted.metadata["audio_segment_count"] == 1
assert extracted.metadata["transcription_provider"] == "openai"
assert "transcription_fallback_used" not in extracted.metadata
@pytest.mark.asyncio
async def test_transcribe_segments_ignores_empty_silent_segments():
async def test_audio_processor_extract_text_allows_empty_transcript():
processor = AudioProcessor()
segments = [
AudioSegment(index=0, filename="seg-0.mp3", content=b"0"),
AudioSegment(index=1, filename="seg-1.mp3", content=b"1"),
AudioSegment(index=2, filename="seg-2.mp3", content=b"2"),
]
async def fake_transcribe(
_content: bytes,
@ -207,21 +160,21 @@ async def test_transcribe_segments_ignores_empty_silent_segments():
**_: object,
):
assert content_type == "audio/mpeg"
if filename == "seg-0.mp3":
return "first"
if filename == "seg-1.mp3":
return ""
return "third"
assert filename == "voice-note.mp3"
return ""
with patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
extracted = await processor.transcribe_segments(
segments,
with (
patch.object(processor, "_probe_audio_duration_seconds", return_value=1.0),
patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe),
):
extracted = await processor.extract_text(
b"bytes",
filename="voice-note.mp3",
content_type="audio/mpeg",
concurrency=3,
)
assert extracted.text == "first\nthird"
assert extracted.metadata["audio_segment_count"] == 3
assert extracted.text == ""
assert extracted.metadata["audio_segment_count"] == 1
@pytest.mark.asyncio
@ -238,11 +191,10 @@ async def test_audio_processor_normalizes_octet_stream_mp3_uploads():
assert content_type == "audio/mpeg"
return "normalized"
with patch.object(
processor,
"split_audio_segments",
return_value=[AudioSegment(index=0, filename="voice-note.mp3", content=b"bytes")],
), patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
with (
patch.object(processor, "_probe_audio_duration_seconds", return_value=1.0),
patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe),
):
extracted = await processor.extract_text(
b"bytes",
filename="voice-note.mp3",
@ -256,11 +208,7 @@ async def test_audio_processor_normalizes_octet_stream_mp3_uploads():
async def test_empty_audio_upload_is_rejected_before_transcription():
processor = AudioProcessor()
with patch.object(
processor,
"transcribe_segments",
new=AsyncMock(),
) as mock_transcribe, pytest.raises(
with patch("src.utils.files.transcribe_audio", new=AsyncMock()) as mock_transcribe, pytest.raises(
ValidationException,
match="Audio upload is empty",
):
@ -287,11 +235,10 @@ async def test_audio_wave_mime_is_accepted_for_wav_uploads():
assert content_type == "audio/wave"
return "wav accepted"
with patch.object(
processor,
"split_audio_segments",
return_value=[AudioSegment(index=0, filename="recording.wav", content=b"bytes")],
), patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
with (
patch.object(processor, "_probe_audio_duration_seconds", return_value=1.0),
patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe),
):
extracted = await processor.extract_text(
b"bytes",
filename="recording.wav",
@ -301,148 +248,29 @@ async def test_audio_wave_mime_is_accepted_for_wav_uploads():
assert extracted.text == "wav accepted"
def test_split_audio_segments_raises_validation_when_ffprobe_missing():
@pytest.mark.asyncio
async def test_audio_processor_normalizes_small_mime_only_audio_filename():
processor = AudioProcessor()
async def fake_transcribe(
_content: bytes,
filename: str,
content_type: str,
**_: object,
) -> str:
assert filename == "blob.mp3"
assert content_type == "audio/mpeg"
return "normalized"
with (
patch("src.utils.files.subprocess.run", side_effect=FileNotFoundError("ffprobe")),
pytest.raises(
ValidationException,
match="Audio uploads require ffmpeg and ffprobe to be installed",
),
patch.object(processor, "_probe_audio_duration_seconds", return_value=1.0),
patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe),
):
processor.split_audio_segments(
extracted = await processor.extract_text(
b"audio-bytes",
filename="voice.mp3",
filename="blob",
content_type="audio/mpeg",
)
def test_split_audio_segments_splits_long_wav_when_duration_exceeds_limit():
processor = AudioProcessor()
wav_bytes = _generate_test_audio_bytes("wav", duration_seconds=3)
original_duration_limit = settings.AUDIO.MAX_CHUNK_DURATION_SECONDS
original_chunk_bytes = settings.AUDIO.MAX_CHUNK_BYTES
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = 1
settings.AUDIO.MAX_CHUNK_BYTES = max(len(wav_bytes) * 2, 1)
try:
segments = processor.split_audio_segments(
wav_bytes,
filename="long.wav",
content_type="audio/wav",
)
finally:
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = original_duration_limit
settings.AUDIO.MAX_CHUNK_BYTES = original_chunk_bytes
assert len(segments) >= 3
assert [segment.index for segment in segments] == list(range(len(segments)))
assert all(segment.filename.endswith(".wav") for segment in segments)
assert all(segment.content for segment in segments)
def test_split_audio_segments_normalizes_small_mime_only_audio_filename():
processor = AudioProcessor()
original_duration_limit = settings.AUDIO.MAX_CHUNK_DURATION_SECONDS
original_chunk_bytes = settings.AUDIO.MAX_CHUNK_BYTES
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = 60
settings.AUDIO.MAX_CHUNK_BYTES = 1024
try:
with patch.object(
processor,
"_probe_audio_duration_seconds",
return_value=0.5,
):
segments = processor.split_audio_segments(
b"audio-bytes",
filename="blob",
content_type="audio/mpeg",
)
finally:
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = original_duration_limit
settings.AUDIO.MAX_CHUNK_BYTES = original_chunk_bytes
assert len(segments) == 1
assert segments[0].filename == "blob.mp3"
def test_split_audio_segments_falls_back_to_mp3_when_wav_floor_exceeds_byte_limit():
processor = AudioProcessor()
original_duration_limit = settings.AUDIO.MAX_CHUNK_DURATION_SECONDS
original_chunk_bytes = settings.AUDIO.MAX_CHUNK_BYTES
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = 60
settings.AUDIO.MAX_CHUNK_BYTES = 200_000
build_segments = [
[AudioSegment(index=0, filename="segment_000.wav", content=b"x" * 200_001)],
[AudioSegment(index=0, filename="segment_000.mp3", content=b"x" * 10)],
]
try:
with (
patch.object(
processor,
"_probe_audio_duration_seconds",
return_value=1.0,
),
patch.object(
processor,
"_build_audio_segments",
side_effect=build_segments,
) as mock_build,
):
segments = processor.split_audio_segments(
b"x" * 200_001,
filename="clip.wav",
content_type="audio/wav",
)
finally:
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = original_duration_limit
settings.AUDIO.MAX_CHUNK_BYTES = original_chunk_bytes
assert len(segments) == 1
assert segments[0].filename.endswith(".mp3")
assert mock_build.call_args_list[0].kwargs["suffix"] == ".wav"
assert mock_build.call_args_list[1].kwargs["suffix"] == ".mp3"
def test_split_audio_segments_raises_validation_for_invalid_audio_bytes():
processor = AudioProcessor()
with pytest.raises(
ValidationException,
match="Uploaded audio is invalid or unreadable",
):
processor.split_audio_segments(
b"not-valid-audio",
filename="broken.mp3",
content_type="audio/mpeg",
)
def test_split_audio_segments_keep_low_bitrate_mp3_chunks_under_limit():
processor = AudioProcessor()
mp3_bytes = _generate_test_audio_bytes(
"mp3",
duration_seconds=12,
audio_bitrate="64k",
)
original_duration_limit = settings.AUDIO.MAX_CHUNK_DURATION_SECONDS
original_chunk_bytes = settings.AUDIO.MAX_CHUNK_BYTES
max_chunk_bytes = max(math.ceil(len(mp3_bytes) / 2), 1)
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = 60
settings.AUDIO.MAX_CHUNK_BYTES = max_chunk_bytes
try:
segments = processor.split_audio_segments(
mp3_bytes,
filename="voice-note.mp3",
content_type="audio/mpeg",
)
finally:
settings.AUDIO.MAX_CHUNK_DURATION_SECONDS = original_duration_limit
settings.AUDIO.MAX_CHUNK_BYTES = original_chunk_bytes
assert len(segments) >= 3
assert all(len(segment.content) <= max_chunk_bytes for segment in segments)
assert extracted.text == "normalized"
assert extracted.metadata["audio_segment_count"] == 1