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

Add synchronous audio upload transcription

This commit is contained in:
adavyas 2026-04-08 18:49:02 -04:00
parent 6c5a9c29f6
commit b4c2f15297
15 changed files with 1369 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
.env.template
View File

@ -79,6 +79,16 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here
# LLM_VLLM_API_KEY=
# LLM_VLLM_BASE_URL=
# =============================================================================
# Audio Transcription Settings
# =============================================================================
# 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
# =============================================================================
# LLM Configuration
# =============================================================================

4
Dockerfile
View File

@ -4,6 +4,10 @@ FROM python:3.13-slim-bookworm
COPY --from=ghcr.io/astral-sh/uv:0.9.24 /uv /bin/uv
RUN apt-get update \
&& apt-get install -y --no-install-recommends ffmpeg \
&& rm -rf /var/lib/apt/lists/*
# Set Working directory
WORKDIR /app

9
config.toml.example
View File

@ -67,6 +67,15 @@ MAX_MESSAGE_CONTENT_CHARS = 2000 # Max chars per message in tool results
# VLLM_API_KEY = "your-api-key"
# VLLM_BASE_URL = "your-base-url"
# Audio transcription settings
[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
# Deriver settings
[deriver]
ENABLED = true

2
docs/superpowers/specs/2026-04-07-audio-upload-ingestion-design.md
View File

@ -4,7 +4,7 @@
Add audio-only ingestion to the existing `/v3/workspaces/{workspace_id}/sessions/{session_id}/messages/upload` endpoint so `.mp3` and `.wav` uploads are accepted and converted into ordinary Honcho message rows. The endpoint remains one file per request. Audio uploads are transcribed during upload using an external provider, normalized into transcript text, chunked into message-sized text blocks, stored as regular messages, and then enqueued into the existing deriver path without changing the deriver message format.
This design is intended to support migrations into Honcho without introducing a new ingestion API surface. Throughput comes from two places: clients may issue many upload requests concurrently, and the server may transcribe large single files by splitting them into provider-safe audio segments and processing those segments with bounded parallelism.
This design is intended to support migrat Pions into Honcho without introducing a new ingestion API surface. Throughput comes from two places: clients may issue many upload requests concurrently, and the server may transcribe large single files by splitting them into provider-safe audio segments and processing those segments with bounded parallelism.
## Goals

9
docs/v3/documentation/features/advanced/file-uploads.mdx
View File

@ -1,10 +1,10 @@
---
title: 'File Uploads'
description: 'Upload PDFs, text files, and JSON documents to create messages in Honcho'
description: 'Upload PDFs, text files, JSON documents, and audio files to create messages in Honcho'
icon: 'upload'
---
Honcho's file upload feature allows you to convert documents into messages automatically. Upload PDFs, text files, or JSON documents, and Honcho will extract the text content, split it into appropriately sized chunks, and create messages that become part of your peer's representation or session context.
Honcho's file upload feature allows you to convert documents and audio into messages automatically. Upload PDFs, text files, JSON documents, or supported audio files, and Honcho will extract or transcribe the content, split it into appropriately sized chunks, and create messages that become part of your peer's representation or session context.
This feature is perfect for ingesting documents, reports, research papers, or any text-based content that you want your AI agents to understand and reference.
@ -25,9 +25,10 @@ Honcho currently supports the following file types with more to come:
- **PDF files** (`application/pdf`) - Text extraction with page numbers
- **Text files** (`text/*`) - Plain text, markdown, code files, etc.
- **JSON files** (`application/json`) - Structured data converted to readable format
- **Audio files** (`audio/mpeg`, `audio/wav`, `audio/x-wav`) - Server-side transcription for `.mp3` and `.wav` uploads
<Note>
Files are processed in memory and not stored on disk. Only the extracted text content is preserved in Honcho's message system.
Files are processed during upload and only the extracted or transcribed text content is preserved in Honcho's message system.
</Note>
## Basic Usage
@ -102,6 +103,8 @@ Our approach involves...
**JSON Files**: Structured data is converted to string format.
**Audio Files**: `.mp3` and `.wav` uploads are transcribed by OpenAI during upload using `whisper-1` by default. Large audio files may be split into multiple audio segments before transcription and then merged back into ordered transcript text before message chunking.
### Chunking Strategy
Large files are automatically split into chunks of ~49,500 characters. The system seeks to break at natural boundaries if present:

15
src/config.py
View File

@ -61,6 +61,7 @@ class TomlConfigSettingsSource(PydanticBaseSettingsSource):
"SENTRY": "sentry",
"CACHE": "cache",
"LLM": "llm",
"AUDIO": "audio",
"DERIVER": "deriver",
"PEER_CARD": "peer_card",
"DIALECTIC": "dialectic",
@ -232,6 +233,19 @@ class LLMSettings(HonchoSettings):
)
class AudioSettings(HonchoSettings):
model_config = SettingsConfigDict(env_prefix="AUDIO_", extra="ignore") # pyright: ignore
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
class DeriverSettings(BackupLLMSettingsMixin, HonchoSettings):
model_config = SettingsConfigDict(env_prefix="DERIVER_", extra="ignore") # pyright: ignore
@ -651,6 +665,7 @@ class AppSettings(HonchoSettings):
AUTH: AuthSettings = Field(default_factory=AuthSettings)
SENTRY: SentrySettings = Field(default_factory=SentrySettings)
LLM: LLMSettings = Field(default_factory=LLMSettings)
AUDIO: AudioSettings = Field(default_factory=AudioSettings)
DERIVER: DeriverSettings = Field(default_factory=DeriverSettings)
DIALECTIC: DialecticSettings = Field(default_factory=DialecticSettings)
PEER_CARD: PeerCardSettings = Field(default_factory=PeerCardSettings)

2
src/deriver/enqueue.py
View File

@ -77,8 +77,6 @@ async def enqueue(payload: list[dict[str, Any]]) -> None:
import sentry_sdk
sentry_sdk.capture_exception(e)
async def handle_session(
db_session: AsyncSession,
payload: list[dict[str, Any]],

2
src/models.py
View File

@ -468,8 +468,6 @@ class Document(Base):
"last_sync_at",
),
)
@final
class QueueItem(Base):
__tablename__: str = "queue"

54
src/routers/messages.py
View File

@ -18,12 +18,17 @@ from sqlalchemy.orm.attributes import flag_modified
from src import crud, schemas
from src.config import settings
from src.dependencies import db
from src.dependencies import db, tracked_db
from src.deriver import enqueue
from src.exceptions import FileTooLargeError, ResourceNotFoundException
from src.security import require_auth
from src.telemetry import prometheus_metrics
from src.utils.files import process_file_uploads_for_messages
from src.utils.files import (
is_audio_transcription_enabled,
is_audio_upload,
is_validated_audio_upload,
process_file_uploads_for_messages,
)
logger = logging.getLogger(__name__)
@ -141,14 +146,24 @@ async def create_messages_with_file(
session_id: str = Path(...),
form_data: schemas.MessageUploadCreate = Depends(parse_upload_form),
file: UploadFile = File(...),
db: AsyncSession = db,
):
"""Create messages from uploaded files. Files are converted to text and split into multiple messages."""
# Validate file size
if file.size and file.size > settings.MAX_FILE_SIZE:
max_file_size = settings.MAX_FILE_SIZE
if (
file.size
and file.size > settings.MAX_FILE_SIZE
and is_audio_transcription_enabled()
and is_audio_upload(file)
and file.size <= settings.AUDIO.MAX_FILE_SIZE_BYTES
and await is_validated_audio_upload(file)
):
max_file_size = settings.AUDIO.MAX_FILE_SIZE_BYTES
if file.size and file.size > max_file_size:
raise FileTooLargeError(
f"File size ({file.size} bytes) exceeds maximum allowed size ({settings.MAX_FILE_SIZE} bytes)",
f"File size ({file.size} bytes) exceeds maximum allowed size ({max_file_size} bytes)",
)
# Process files using shared utility function
@ -160,22 +175,23 @@ async def create_messages_with_file(
created_at=form_data.created_at,
)
# Create messages
message_creates = [item["message_create"] for item in all_message_data]
created_messages = await crud.create_messages(
db,
messages=message_creates,
workspace_name=workspace_id,
session_name=session_id,
)
async with tracked_db("messages.upload") as db_session:
# Create messages
message_creates = [item["message_create"] for item in all_message_data]
created_messages = await crud.create_messages(
db_session,
messages=message_creates,
workspace_name=workspace_id,
session_name=session_id,
)
# Update internal_metadata for file-related messages
for i, message in enumerate(created_messages):
file_metadata = all_message_data[i]["file_metadata"]
message.internal_metadata.update(file_metadata)
flag_modified(message, "internal_metadata")
# Update internal_metadata for file-related messages
for i, message in enumerate(created_messages):
file_metadata = all_message_data[i]["file_metadata"]
message.internal_metadata.update(file_metadata)
flag_modified(message, "internal_metadata")
await db.commit()
await db_session.commit()
# Enqueue for processing (same as regular messages)
payloads = [

42
src/utils/clients.py
View File

@ -1,3 +1,4 @@
import io
import json
import logging
from collections.abc import AsyncIterator, Callable
@ -71,7 +72,6 @@ T = TypeVar("T")
ReasoningEffortType = Literal["low", "medium", "high", "minimal"] | None
VerbosityType = Literal["low", "medium", "high"] | None
def count_message_tokens(messages: list[dict[str, Any]]) -> int:
"""Count tokens in a list of messages using tiktoken."""
total = 0
@ -317,6 +317,46 @@ for component_name, backup_provider in BACKUP_PROVIDERS:
)
async def _transcribe_audio_once(
content: bytes,
*,
filename: str,
content_type: str,
model: str,
) -> str:
if not content_type.startswith("audio/"):
raise LLMError(f"Unsupported audio content type: {content_type}")
if "openai" not in CLIENTS:
raise LLMError("Audio transcription provider 'openai' is not initialized")
client = cast(AsyncOpenAI, CLIENTS["openai"])
audio_buffer = io.BytesIO(content)
audio_buffer.name = filename
response = cast(
Any,
await client.audio.transcriptions.create(
file=audio_buffer,
model=model,
response_format="text",
),
)
return response.strip() if isinstance(response, str) else str(response).strip()
async def transcribe_audio(
content: bytes,
*,
filename: str,
content_type: str,
model: str | None = None,
) -> str:
return await _transcribe_audio_once(
content,
filename=filename,
content_type=content_type,
model=model or settings.AUDIO.MODEL,
)
def convert_tools_for_provider(
tools: list[dict[str, Any]],
provider: SupportedProviders,

467
src/utils/files.py
View File

@ -1,6 +1,12 @@
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
from fastapi import UploadFile
@ -16,12 +22,53 @@ from src.exceptions import (
ValidationException,
)
from src.schemas import Message
from src.utils.clients import CLIENTS, transcribe_audio
logger = logging.getLogger(__name__)
SUPPORTED_AUDIO_CONTENT_TYPES = {
"audio/mpeg",
"audio/mp3",
"audio/wave",
"audio/wav",
"audio/x-wav",
}
SUPPORTED_AUDIO_EXTENSIONS = {".mp3", ".wav"}
AUDIO_EXTENSION_CONTENT_TYPES = {
".mp3": "audio/mpeg",
".wav": "audio/wav",
}
UPLOAD_VALIDATION_CHUNK_BYTES = 1024 * 1024
GENERIC_CONTENT_TYPES = {
"",
"application/octet-stream",
}
class ExtractedFileText(str):
metadata: dict[str, Any]
def __new__(
cls, text: str, metadata: dict[str, Any] | None = None
) -> "ExtractedFileText":
obj = str.__new__(cls, text)
obj.metadata = metadata or {}
return obj
@property
def text(self) -> str:
return str(self)
@dataclass
class AudioSegment:
index: int
filename: str
content: bytes
class FileProcessor(Protocol):
async def extract_text(self, content: bytes) -> str: ...
async def extract_text(self, content: bytes) -> ExtractedFileText: ...
def supports_file_type(self, content_type: str) -> bool: ...
@ -29,7 +76,7 @@ class PDFProcessor:
def supports_file_type(self, content_type: str) -> bool:
return content_type == "application/pdf"
async def extract_text(self, content: bytes) -> str:
async def extract_text(self, content: bytes) -> ExtractedFileText:
import pdfplumber
with pdfplumber.open(BytesIO(content)) as pdf_reader:
@ -38,18 +85,18 @@ class PDFProcessor:
text = page.extract_text()
if text and text.strip():
text_parts.append(f"[Page {page_num + 1}]\n{text}")
return "\n\n".join(text_parts)
return ExtractedFileText(text="\n\n".join(text_parts))
class TextProcessor:
def supports_file_type(self, content_type: str) -> bool:
return content_type.startswith("text/")
async def extract_text(self, content: bytes) -> str:
async def extract_text(self, content: bytes) -> ExtractedFileText:
# Try different encodings
for encoding in ["utf-8", "utf-16", "latin-1"]:
try:
return content.decode(encoding)
return ExtractedFileText(text=content.decode(encoding))
except UnicodeDecodeError:
continue
raise ValueError("Could not decode text file")
@ -59,7 +106,7 @@ class JSONProcessor:
def supports_file_type(self, content_type: str) -> bool:
return content_type == "application/json"
async def extract_text(self, content: bytes) -> str:
async def extract_text(self, content: bytes) -> ExtractedFileText:
import json
try:
@ -68,7 +115,7 @@ class JSONProcessor:
raise ValidationException("JSON uploads must be UTF-8 encoded") from exc
if not decoded_content.strip():
return ""
return ExtractedFileText(text="")
try:
data = json.loads(decoded_content)
@ -76,11 +123,327 @@ class JSONProcessor:
raise ValidationException("Uploaded JSON is invalid") from exc
# Convert JSON to readable text format
return json.dumps(data, ensure_ascii=False)
return ExtractedFileText(text=json.dumps(data, ensure_ascii=False))
class AudioProcessor:
def supports_file_type(self, content_type: str) -> bool:
return content_type in SUPPORTED_AUDIO_CONTENT_TYPES
def supports_filename(self, filename: str | None) -> bool:
if not filename:
return False
return Path(filename).suffix.lower() in SUPPORTED_AUDIO_EXTENSIONS
def supports_content(self, *, filename: str | None, content_type: str) -> bool:
if self.supports_file_type(content_type):
return True
if content_type not in GENERIC_CONTENT_TYPES:
return False
return self.supports_filename(filename)
def supports_upload(self, file: UploadFile) -> bool:
return self.supports_content(
filename=file.filename,
content_type=file.content_type or "",
)
async def extract_text(
self,
content: bytes,
*,
filename: str | None,
content_type: str,
) -> ExtractedFileText:
if not filename:
raise ValidationException("Audio upload requires a filename")
if not content:
raise ValidationException("Audio upload is empty")
normalized_content_type = self.normalize_content_type(filename, content_type)
segments = await asyncio.to_thread(
self.split_audio_segments,
content,
filename,
normalized_content_type,
)
return await self.transcribe_segments(
segments,
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,
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": len(segments),
"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)
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=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:
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:
return suffix
if content_type in {"audio/wave", "audio/wav", "audio/x-wav"}:
return ".wav"
return ".mp3"
def _probe_audio_duration_seconds(self, content: bytes, suffix: str) -> float:
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
temp_file.write(content)
temp_path = Path(temp_file.name)
try:
return self.probe_audio_duration_seconds_from_path(temp_path)
finally:
temp_path.unlink(missing_ok=True)
def probe_audio_duration_seconds_from_path(self, path: Path) -> float:
try:
command = [
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(path),
]
result = subprocess.run(
command,
capture_output=True,
text=True,
check=True,
)
return max(float(result.stdout.strip()), 0.0)
except FileNotFoundError as exc:
raise ValidationException(
"Audio uploads require ffmpeg and ffprobe to be installed on the server"
) from exc
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)
def is_audio_transcription_enabled() -> bool:
return settings.AUDIO.PROVIDER == "openai" and "openai" in CLIENTS
async def is_validated_audio_upload(file: UploadFile) -> bool:
processor = AudioProcessor()
if not processor.supports_upload(file):
return False
filename = file.filename
if not filename:
return False
content_type = processor.normalize_content_type(filename, file.content_type or "")
suffix = processor.get_output_suffix(filename, content_type)
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
temp_path = Path(temp_file.name)
try:
while chunk := await file.read(UPLOAD_VALIDATION_CHUNK_BYTES):
temp_file.write(chunk)
finally:
await file.seek(0)
try:
await asyncio.to_thread(
processor.probe_audio_duration_seconds_from_path,
temp_path,
)
return True
except ValidationException as exc:
if str(exc) == "Uploaded audio is invalid or unreadable":
return False
raise
finally:
temp_path.unlink(missing_ok=True)
class FileProcessingService:
def __init__(self):
self.audio_processor: AudioProcessor = AudioProcessor()
self.processors: list[FileProcessor] = [
PDFProcessor(),
TextProcessor(),
@ -88,20 +451,49 @@ class FileProcessingService:
# Add more processors as needed
]
async def extract_text_from_upload(self, file: UploadFile) -> str:
async def extract_text_from_bytes(
self,
content: bytes,
*,
filename: str | None,
content_type: str | None,
) -> ExtractedFileText:
normalized_content_type = content_type or ""
if self.audio_processor.supports_content(
filename=filename,
content_type=normalized_content_type,
):
if "openai" not in CLIENTS:
raise ValidationException(
"Audio uploads require OpenAI transcription credentials"
)
return await self.audio_processor.extract_text(
content,
filename=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]}"
)
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)
processor = self._get_processor(file.content_type or "")
if not processor:
raise UnsupportedFileTypeError(
f"Unsupported file type: {file.content_type}. Supported types: {[p.__class__.__name__ for p in self.processors]}"
)
return await processor.extract_text(content)
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:
@ -202,11 +594,43 @@ 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]] = []
# Process the uploaded file
extracted_text = await file_processor.extract_text_from_upload(file)
extracted = await file_processor.extract_text_from_bytes(
content,
filename=filename,
content_type=content_type,
)
extracted_text = extracted.text
# Split into chunks and create messages
chunks = split_text_into_chunks(extracted_text, max_chars=max_chars)
@ -228,16 +652,17 @@ async def process_file_uploads_for_messages(
# Store file metadata separately to add to internal_metadata later
file_metadata = {
"file_id": file_id,
"filename": file.filename,
"filename": filename,
"chunk_index": i,
"total_chunks": len(chunks),
"original_file_size": file.size,
"content_type": file.content_type,
"original_file_size": file_size,
"content_type": content_type,
"chunk_character_range": [
i * max_chars,
min((i + 1) * max_chars, len(extracted_text)),
],
}
file_metadata.update(extracted.metadata)
all_message_data.append(
{

1
tests/conftest.py
View File

@ -746,6 +746,7 @@ def mock_tracked_db(db_engine: AsyncEngine, request: pytest.FixtureRequest):
patch("src.deriver.consumer.tracked_db", mock_tracked_db_context),
patch("src.deriver.enqueue.tracked_db", mock_tracked_db_context),
patch("src.routers.peers.tracked_db", mock_tracked_db_context),
patch("src.routers.messages.tracked_db", mock_tracked_db_context),
patch("src.crud.representation.tracked_db", mock_tracked_db_context),
patch("src.dreamer.orchestrator.tracked_db", mock_tracked_db_context),
patch("src.dreamer.dream_scheduler.tracked_db", mock_tracked_db_context),

418
tests/routes/test_files.py
View File

@ -1,17 +1,24 @@
# File upload tests for session endpoints
import io
import json
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import BackgroundTasks, UploadFile
from fastapi.testclient import TestClient
from nanoid import generate as generate_nanoid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.datastructures import Headers
from src import models
from src import models, schemas
from src.config import settings
from src.models import Peer, Workspace
from src.routers.messages import create_messages_with_file
from src.utils.files import ExtractedFileText
async def _create_test_session(
@ -613,3 +620,412 @@ async def test_large_file_upload_with_metadata(
assert message["peer_id"] == test_peer.name
assert message["session_id"] == session_name
assert message["metadata"] == metadata
@pytest.mark.asyncio
async def test_create_messages_with_mp3_file(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
"""Test creating messages with an MP3 upload."""
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"fake mp3 bytes")
files = {"file": ("call.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
extracted = ExtractedFileText(
text="First sentence.\nSecond sentence.",
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": 1,
"transcription_provider": "openai",
},
)
with (
patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False),
patch(
"src.utils.files.AudioProcessor.extract_text",
new=AsyncMock(return_value=extracted),
),
):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == extracted.text
assert data[0]["peer_id"] == test_peer.name
assert data[0]["session_id"] == session_name
stmt = select(models.Message).where(models.Message.public_id == data[0]["id"])
result = await db_session.execute(stmt)
db_message = result.scalar_one()
assert db_message.internal_metadata["processing_type"] == "audio_transcription"
assert db_message.internal_metadata["audio_segment_count"] == 1
assert db_message.internal_metadata["transcription_provider"] == "openai"
assert "transcription_fallback_used" not in db_message.internal_metadata
@pytest.mark.asyncio
async def test_audio_upload_over_generic_limit_uses_audio_size_limit_when_validated(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
original_generic_max = settings.MAX_FILE_SIZE
original_audio_max = settings.AUDIO.MAX_FILE_SIZE_BYTES
settings.MAX_FILE_SIZE = 5
settings.AUDIO.MAX_FILE_SIZE_BYTES = 10
extracted = ExtractedFileText(
text="Transcribed audio",
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": 1,
"transcription_provider": "openai",
},
)
try:
file_data = io.BytesIO(b"123456")
files = {"file": ("call.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
with (
patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False),
patch(
"src.routers.messages.is_validated_audio_upload",
new=AsyncMock(return_value=True),
),
patch(
"src.utils.files.AudioProcessor.extract_text",
new=AsyncMock(return_value=extracted),
),
):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
finally:
settings.MAX_FILE_SIZE = original_generic_max
settings.AUDIO.MAX_FILE_SIZE_BYTES = original_audio_max
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == extracted.text
@pytest.mark.asyncio
async def test_extension_only_audio_upload_over_generic_limit_uses_audio_size_limit_when_validated(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
original_generic_max = settings.MAX_FILE_SIZE
original_audio_max = settings.AUDIO.MAX_FILE_SIZE_BYTES
settings.MAX_FILE_SIZE = 5
settings.AUDIO.MAX_FILE_SIZE_BYTES = 10
extracted = ExtractedFileText(
text="Transcribed extension-only audio",
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": 1,
"transcription_provider": "openai",
},
)
try:
file_data = io.BytesIO(b"123456")
files = {"file": ("renamed.mp3", file_data, "application/octet-stream")}
form_data = {"peer_id": test_peer.name}
with (
patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False),
patch(
"src.routers.messages.is_validated_audio_upload",
new=AsyncMock(return_value=True),
),
patch(
"src.utils.files.AudioProcessor.extract_text",
new=AsyncMock(return_value=extracted),
),
):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
finally:
settings.MAX_FILE_SIZE = original_generic_max
settings.AUDIO.MAX_FILE_SIZE_BYTES = original_audio_max
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == extracted.text
@pytest.mark.asyncio
async def test_audio_upload_over_generic_limit_keeps_generic_limit_without_transcription_credentials(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
original_generic_max = settings.MAX_FILE_SIZE
original_audio_max = settings.AUDIO.MAX_FILE_SIZE_BYTES
settings.MAX_FILE_SIZE = 5
settings.AUDIO.MAX_FILE_SIZE_BYTES = 10
try:
file_data = io.BytesIO(b"123456")
files = {"file": ("call.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
with patch.dict("src.utils.files.CLIENTS", {}, clear=True):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
finally:
settings.MAX_FILE_SIZE = original_generic_max
settings.AUDIO.MAX_FILE_SIZE_BYTES = original_audio_max
assert response.status_code == 413
@pytest.mark.asyncio
async def test_create_messages_with_file_opens_tracked_db_after_file_processing():
background_tasks = BackgroundTasks()
form_data = schemas.MessageUploadCreate(peer_id="peer")
file = UploadFile(
file=io.BytesIO(b"ID3\x03\x00\x00"),
filename="call.mp3",
headers=Headers({"content-type": "audio/mpeg"}),
)
tracked_db_entered = False
db_session = AsyncMock()
created_message = models.Message(
session_name="session",
peer_name="peer",
workspace_name="workspace",
content="transcribed text",
public_id=generate_nanoid(),
token_count=2,
seq_in_session=1,
created_at=datetime.now(UTC),
h_metadata={},
internal_metadata={},
)
@asynccontextmanager
async def fake_tracked_db(_operation_name: str | None = None):
nonlocal tracked_db_entered
tracked_db_entered = True
yield db_session
async def fake_process_file_uploads_for_messages(*_args: Any, **_kwargs: Any):
assert not tracked_db_entered
return [
{
"message_create": schemas.MessageCreate(
content="transcribed text",
peer_id="peer",
),
"file_metadata": {"processing_type": "audio_transcription"},
}
]
with (
patch(
"src.routers.messages.process_file_uploads_for_messages",
side_effect=fake_process_file_uploads_for_messages,
),
patch("src.routers.messages.tracked_db", fake_tracked_db),
patch(
"src.routers.messages.crud.create_messages",
new=AsyncMock(return_value=[created_message]),
),
patch("src.routers.messages.flag_modified"),
):
result = await create_messages_with_file(
background_tasks=background_tasks,
workspace_id="workspace",
session_id="session",
form_data=form_data,
file=file,
)
assert tracked_db_entered
db_session.commit.assert_awaited_once()
assert result == [created_message]
@pytest.mark.asyncio
async def test_filename_audio_extension_with_text_plain_mime_uses_text_processor(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"plain text, not audio")
files = {"file": ("notes.mp3", file_data, "text/plain")}
form_data = {"peer_id": test_peer.name}
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == "plain text, not audio"
@pytest.mark.asyncio
async def test_create_messages_with_wav_file_accepts_audio_wave_mime(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"fake wav bytes")
files = {"file": ("call.wav", file_data, "audio/wave")}
form_data = {"peer_id": test_peer.name}
extracted = ExtractedFileText(
text="WAV transcript",
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": 1,
"transcription_provider": "openai",
},
)
with (
patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False),
patch(
"src.utils.files.AudioProcessor.extract_text",
new=AsyncMock(return_value=extracted),
),
):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == "WAV transcript"
@pytest.mark.asyncio
async def test_malformed_audio_upload_returns_validation_error(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"not-valid-audio")
files = {"file": ("broken.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
url = _get_upload_url(test_workspace.name, session_name)
with patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False):
response = client.post(url, files=files, data=form_data)
assert response.status_code == 422
assert "Uploaded audio is invalid or unreadable" in response.json()["detail"]
@pytest.mark.asyncio
async def test_empty_audio_upload_returns_validation_error(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"")
files = {"file": ("empty.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
url = _get_upload_url(test_workspace.name, session_name)
with patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False):
response = client.post(url, files=files, data=form_data)
assert response.status_code == 422
assert "Audio upload is empty" in response.json()["detail"]
@pytest.mark.asyncio
async def test_large_audio_upload_applies_audio_metadata_to_all_message_chunks(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
"""Large audio transcripts should chunk into multiple messages with shared audio metadata."""
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"fake long mp3 bytes")
files = {"file": ("lecture.mp3", file_data, "audio/mpeg")}
form_data = {"peer_id": test_peer.name}
long_text = "Segment line. " * 4000
extracted = ExtractedFileText(
text=long_text,
metadata={
"processing_type": "audio_transcription",
"audio_segment_count": 3,
"transcription_provider": "openai",
},
)
with (
patch.dict("src.utils.files.CLIENTS", {"openai": object()}, clear=False),
patch(
"src.utils.files.AudioProcessor.extract_text",
new=AsyncMock(return_value=extracted),
),
):
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
assert response.status_code == 201
data = response.json()
assert len(data) > 1
stmt = select(models.Message).where(
models.Message.session_name == session_name,
models.Message.peer_name == test_peer.name,
)
result = await db_session.execute(stmt)
db_messages = list(result.scalars().all())
assert len(db_messages) == len(data)
for db_message in db_messages:
assert db_message.internal_metadata["processing_type"] == "audio_transcription"
assert db_message.internal_metadata["audio_segment_count"] == 3
assert db_message.internal_metadata["transcription_provider"] == "openai"
assert "transcription_fallback_used" not in db_message.internal_metadata
assert "chunk_index" in db_message.internal_metadata
assert "total_chunks" in db_message.internal_metadata

1
tests/sdk_typescript/conftest.py
View File

@ -130,6 +130,7 @@ def mock_tracked_db(ts_db_session: async_sessionmaker[AsyncSession]):
patch("src.deriver.consumer.tracked_db", ts_tracked_db),
patch("src.deriver.enqueue.tracked_db", ts_tracked_db),
patch("src.routers.peers.tracked_db", ts_tracked_db),
patch("src.routers.messages.tracked_db", ts_tracked_db),
patch("src.crud.representation.tracked_db", ts_tracked_db),
patch("src.dreamer.dream_scheduler.tracked_db", ts_tracked_db),
patch("src.dreamer.orchestrator.tracked_db", ts_tracked_db),

383
tests/utils/test_audio_processing.py Normal file
View File

@ -0,0 +1,383 @@
import asyncio
import io
import math
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import UploadFile
from openai import AsyncOpenAI
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()
def test_audio_processor_supports_mp3_and_wav_content_types():
processor = AudioProcessor()
assert processor.supports_file_type("audio/mpeg")
assert processor.supports_file_type("audio/wave")
assert processor.supports_file_type("audio/wav")
assert processor.supports_file_type("audio/x-wav")
assert not processor.supports_file_type("text/plain")
def test_audio_defaults_use_openai_whisper_without_backup():
assert settings.AUDIO.PROVIDER == "openai"
assert settings.AUDIO.MODEL == "whisper-1"
@pytest.mark.asyncio
async def test_audio_upload_requires_openai_client_before_processing():
file = UploadFile(
file=io.BytesIO(b"audio-bytes"),
filename="voice.mp3",
headers=Headers({"content-type": "audio/mpeg"}),
)
service = FileProcessingService()
with (
patch.dict(CLIENTS, {}, clear=True),
patch.object(service.audio_processor, "extract_text", new=AsyncMock()) as mock_extract,
pytest.raises(
ValidationException,
match="Audio uploads require OpenAI transcription credentials",
),
):
await service.extract_text_from_upload(file)
mock_extract.assert_not_awaited()
@pytest.mark.asyncio
async def test_filename_audio_extension_does_not_override_explicit_text_plain_mime():
file = UploadFile(
file=io.BytesIO(b"plain text body"),
filename="notes.mp3",
headers=Headers({"content-type": "text/plain"}),
)
service = FileProcessingService()
with patch.object(service.audio_processor, "extract_text", new=AsyncMock()) as mock_extract:
extracted = await service.extract_text_from_upload(file)
assert extracted.text == "plain text body"
mock_extract.assert_not_awaited()
@pytest.mark.asyncio
async def test_transcribe_audio_uses_openai_whisper():
mock_openai = AsyncMock(spec=AsyncOpenAI)
mock_openai.audio.transcriptions.create = AsyncMock(return_value="hello from whisper")
with patch.dict(CLIENTS, {"openai": mock_openai}, clear=False):
text = await transcribe_audio(
b"audio-bytes",
filename="clip.mp3",
content_type="audio/mpeg",
)
assert text == "hello from whisper"
mock_openai.audio.transcriptions.create.assert_awaited_once()
call = mock_openai.audio.transcriptions.create.await_args
assert call is not None
assert call.kwargs["model"] == "whisper-1"
assert call.kwargs["response_format"] == "text"
@pytest.mark.asyncio
async def test_transcribe_audio_allows_empty_transcript_for_silence():
mock_openai = AsyncMock(spec=AsyncOpenAI)
mock_openai.audio.transcriptions.create = AsyncMock(return_value="")
with patch.dict(CLIENTS, {"openai": mock_openai}, clear=False):
text = await transcribe_audio(
b"audio-bytes",
filename="clip.mp3",
content_type="audio/mpeg",
)
assert text == ""
@pytest.mark.asyncio
async def test_transcribe_audio_raises_when_openai_fails():
mock_openai = AsyncMock(spec=AsyncOpenAI)
mock_openai.audio.transcriptions.create = AsyncMock(side_effect=RuntimeError("openai failed"))
with (
patch.dict(CLIENTS, {"openai": mock_openai}, clear=False),
pytest.raises(RuntimeError, match="openai failed"),
):
await transcribe_audio(
b"audio-bytes",
filename="clip.mp3",
content_type="audio/mpeg",
)
@pytest.mark.asyncio
async def test_transcribe_segments_preserves_order_when_tasks_finish_out_of_order():
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,
filename: str,
content_type: str,
**_: 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"
await asyncio.sleep(0.01)
return "third"
with patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
extracted = await processor.transcribe_segments(
segments,
content_type="audio/mpeg",
concurrency=3,
)
assert extracted.text == "first\nsecond\nthird"
assert extracted.metadata["processing_type"] == "audio_transcription"
assert extracted.metadata["audio_segment_count"] == 3
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():
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,
filename: str,
content_type: str,
**_: object,
):
assert content_type == "audio/mpeg"
if filename == "seg-0.mp3":
return "first"
if filename == "seg-1.mp3":
return ""
return "third"
with patch("src.utils.files.transcribe_audio", side_effect=fake_transcribe):
extracted = await processor.transcribe_segments(
segments,
content_type="audio/mpeg",
concurrency=3,
)
assert extracted.text == "first\nthird"
assert extracted.metadata["audio_segment_count"] == 3
@pytest.mark.asyncio
async def test_audio_processor_normalizes_octet_stream_mp3_uploads():
processor = AudioProcessor()
async def fake_transcribe(
_content: bytes,
filename: str,
content_type: str,
**_: object,
) -> str:
assert filename == "voice-note.mp3"
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):
extracted = await processor.extract_text(
b"bytes",
filename="voice-note.mp3",
content_type="application/octet-stream",
)
assert extracted.text == "normalized"
@pytest.mark.asyncio
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(
ValidationException,
match="Audio upload is empty",
):
await processor.extract_text(
b"",
filename="empty.mp3",
content_type="audio/mpeg",
)
mock_transcribe.assert_not_awaited()
@pytest.mark.asyncio
async def test_audio_wave_mime_is_accepted_for_wav_uploads():
processor = AudioProcessor()
async def fake_transcribe(
_content: bytes,
filename: str,
content_type: str,
**_: object,
) -> str:
assert filename == "recording.wav"
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):
extracted = await processor.extract_text(
b"bytes",
filename="recording.wav",
content_type="audio/wave",
)
assert extracted.text == "wav accepted"
def test_split_audio_segments_raises_validation_when_ffprobe_missing():
processor = AudioProcessor()
with (
patch("src.utils.files.subprocess.run", side_effect=FileNotFoundError("ffprobe")),
pytest.raises(
ValidationException,
match="Audio uploads require ffmpeg and ffprobe to be installed",
),
):
processor.split_audio_segments(
b"audio-bytes",
filename="voice.mp3",
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_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)