49 lines
1.5 KiB
Python
49 lines
1.5 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 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()
|