81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""Voice co-creation session storage helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def session_storage_dir(session_id: str) -> Path:
|
|
"""Return the storage directory for one voice session."""
|
|
|
|
return Path(settings.voice_session_storage_dir) / session_id
|
|
|
|
|
|
def build_turn_user_audio_path(session_id: str, turn_index: int, suffix: str) -> Path:
|
|
"""Build the persisted path for one user-uploaded turn audio file."""
|
|
|
|
normalized_suffix = suffix.lstrip(".") or "webm"
|
|
return session_storage_dir(session_id) / f"turn-{turn_index:03d}-user.{normalized_suffix}"
|
|
|
|
|
|
def build_turn_assistant_audio_path(session_id: str, turn_index: int) -> Path:
|
|
"""Build the persisted path for one generated assistant turn audio file."""
|
|
|
|
return session_storage_dir(session_id) / f"turn-{turn_index:03d}-assistant.mp3"
|
|
|
|
|
|
def _normalize_audio_suffix(file_name: str | None, mime_type: str | None) -> str:
|
|
if file_name and "." in file_name:
|
|
return file_name.rsplit(".", 1)[-1].lower()
|
|
|
|
if mime_type == "audio/webm":
|
|
return "webm"
|
|
if mime_type == "audio/wav":
|
|
return "wav"
|
|
if mime_type == "audio/mpeg":
|
|
return "mp3"
|
|
if mime_type == "audio/mp4":
|
|
return "m4a"
|
|
return "bin"
|
|
|
|
|
|
def write_uploaded_user_audio(
|
|
*,
|
|
session_id: str,
|
|
turn_index: int,
|
|
file_name: str | None,
|
|
mime_type: str | None,
|
|
audio_data: bytes,
|
|
) -> str:
|
|
"""Persist one uploaded user-audio turn and return the saved file path."""
|
|
|
|
suffix = _normalize_audio_suffix(file_name, mime_type)
|
|
return write_session_audio(
|
|
build_turn_user_audio_path(session_id, turn_index, suffix),
|
|
audio_data,
|
|
)
|
|
|
|
|
|
def write_session_audio(path: Path, audio_data: bytes) -> str:
|
|
"""Persist session audio bytes atomically and return the saved path."""
|
|
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temp_path = path.with_suffix(f"{path.suffix}.tmp")
|
|
temp_path.write_bytes(audio_data)
|
|
temp_path.replace(path)
|
|
return str(path)
|
|
|
|
|
|
def read_session_audio(audio_path: str) -> bytes:
|
|
"""Read persisted session audio bytes."""
|
|
|
|
return Path(audio_path).read_bytes()
|
|
|
|
|
|
def session_audio_exists(audio_path: str | None) -> bool:
|
|
"""Whether one stored session audio file currently exists."""
|
|
|
|
return bool(audio_path) and Path(audio_path).is_file()
|