From b4c2f15297ae6ef7117e4b036d2082afbfa1355d Mon Sep 17 00:00:00 2001 From: adavyas Date: Wed, 8 Apr 2026 18:49:02 -0400 Subject: [PATCH] Add synchronous audio upload transcription --- .env.template | 10 + Dockerfile | 4 + config.toml.example | 9 + ...026-04-07-audio-upload-ingestion-design.md | 2 +- .../features/advanced/file-uploads.mdx | 9 +- src/config.py | 15 + src/deriver/enqueue.py | 2 - src/models.py | 2 - src/routers/messages.py | 54 +- src/utils/clients.py | 42 +- src/utils/files.py | 467 +++++++++++++++++- tests/conftest.py | 1 + tests/routes/test_files.py | 418 +++++++++++++++- tests/sdk_typescript/conftest.py | 1 + tests/utils/test_audio_processing.py | 383 ++++++++++++++ 15 files changed, 1369 insertions(+), 50 deletions(-) create mode 100644 tests/utils/test_audio_processing.py diff --git a/.env.template b/.env.template index ed456e91..b7517ec2 100644 --- a/.env.template +++ b/.env.template @@ -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 # ============================================================================= diff --git a/Dockerfile b/Dockerfile index 4a68d617..32373259 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/config.toml.example b/config.toml.example index b6b407dc..67bbca7b 100644 --- a/config.toml.example +++ b/config.toml.example @@ -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 diff --git a/docs/superpowers/specs/2026-04-07-audio-upload-ingestion-design.md b/docs/superpowers/specs/2026-04-07-audio-upload-ingestion-design.md index a79594a7..3ffa0390 100644 --- a/docs/superpowers/specs/2026-04-07-audio-upload-ingestion-design.md +++ b/docs/superpowers/specs/2026-04-07-audio-upload-ingestion-design.md @@ -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 diff --git a/docs/v3/documentation/features/advanced/file-uploads.mdx b/docs/v3/documentation/features/advanced/file-uploads.mdx index 73306a3a..5f55aaf5 100644 --- a/docs/v3/documentation/features/advanced/file-uploads.mdx +++ b/docs/v3/documentation/features/advanced/file-uploads.mdx @@ -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 -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. ## 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: diff --git a/src/config.py b/src/config.py index f01c451b..e822ec73 100644 --- a/src/config.py +++ b/src/config.py @@ -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) diff --git a/src/deriver/enqueue.py b/src/deriver/enqueue.py index 71b00a2b..ffd7c346 100644 --- a/src/deriver/enqueue.py +++ b/src/deriver/enqueue.py @@ -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]], diff --git a/src/models.py b/src/models.py index c2aa84dd..769d3f55 100644 --- a/src/models.py +++ b/src/models.py @@ -468,8 +468,6 @@ class Document(Base): "last_sync_at", ), ) - - @final class QueueItem(Base): __tablename__: str = "queue" diff --git a/src/routers/messages.py b/src/routers/messages.py index 3acdc1e5..02838c48 100644 --- a/src/routers/messages.py +++ b/src/routers/messages.py @@ -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 = [ diff --git a/src/utils/clients.py b/src/utils/clients.py index 1c042bff..ca492c75 100644 --- a/src/utils/clients.py +++ b/src/utils/clients.py @@ -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, diff --git a/src/utils/files.py b/src/utils/files.py index cdbac514..4f1f53cb 100644 --- a/src/utils/files.py +++ b/src/utils/files.py @@ -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( { diff --git a/tests/conftest.py b/tests/conftest.py index bd426e89..b8b027dd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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), diff --git a/tests/routes/test_files.py b/tests/routes/test_files.py index 3af9f60b..01d7f933 100644 --- a/tests/routes/test_files.py +++ b/tests/routes/test_files.py @@ -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 diff --git a/tests/sdk_typescript/conftest.py b/tests/sdk_typescript/conftest.py index 15f2c98b..c5cae143 100644 --- a/tests/sdk_typescript/conftest.py +++ b/tests/sdk_typescript/conftest.py @@ -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), diff --git a/tests/utils/test_audio_processing.py b/tests/utils/test_audio_processing.py new file mode 100644 index 00000000..2aae23a1 --- /dev/null +++ b/tests/utils/test_audio_processing.py @@ -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)