Spaces:
Running
Running
| """Storage service for managing audio files.""" | |
| import os | |
| import shutil | |
| from pathlib import Path | |
| from typing import Optional, BinaryIO | |
| from fastapi import UploadFile | |
| import aiofiles | |
| class StorageService: | |
| """Service for managing file storage.""" | |
| def __init__(self, base_path: str = "/app/storage"): | |
| """Initialize storage service.""" | |
| self.base_path = Path(base_path) | |
| self.audio_path = self.base_path / "audio" | |
| self.text_path = self.base_path / "text" | |
| # Create directories | |
| self.audio_path.mkdir(parents=True, exist_ok=True) | |
| self.text_path.mkdir(parents=True, exist_ok=True) | |
| async def save_audio_file(self, book_id: str, audio_data: bytes) -> str: | |
| """Save audio file to storage.""" | |
| file_path = self.audio_path / f"{book_id}.wav" | |
| async with aiofiles.open(file_path, "wb") as f: | |
| await f.write(audio_data) | |
| return str(file_path) | |
| async def save_text_file(self, book_id: str, text_file: UploadFile) -> str: | |
| """Save text file to storage.""" | |
| file_path = self.text_path / f"{book_id}.txt" | |
| async with aiofiles.open(file_path, "wb") as f: | |
| content = await text_file.read() | |
| await f.write(content) | |
| return str(file_path) | |
| async def save_text_content(self, book_id: str, text_content: str) -> str: | |
| """Save text content to a file.""" | |
| file_path = self.text_path / f"{book_id}.txt" | |
| async with aiofiles.open(file_path, "w", encoding="utf-8") as f: | |
| await f.write(text_content) | |
| return str(file_path) | |
| async def get_audio_file(self, book_id: str) -> Optional[Path]: | |
| """Get audio file path.""" | |
| file_path = self.audio_path / f"{book_id}.wav" | |
| return file_path if file_path.exists() else None | |
| async def get_text_file(self, book_id: str) -> Optional[Path]: | |
| """Get text file path.""" | |
| file_path = self.text_path / f"{book_id}.txt" | |
| return file_path if file_path.exists() else None | |
| async def delete_book_files(self, book_id: str): | |
| """Delete all files associated with a book.""" | |
| # Delete audio file | |
| audio_file = self.audio_path / f"{book_id}.wav" | |
| if audio_file.exists(): | |
| audio_file.unlink() | |
| # Delete text file | |
| text_file = self.text_path / f"{book_id}.txt" | |
| if text_file.exists(): | |
| text_file.unlink() | |
| def cleanup_orphaned_files(self, valid_book_ids: set[str]): | |
| """Clean up files that don't belong to any book.""" | |
| # Clean up audio files | |
| for file_path in self.audio_path.glob("*.wav"): | |
| book_id = file_path.stem | |
| if book_id not in valid_book_ids: | |
| file_path.unlink() | |
| # Clean up text files | |
| for file_path in self.text_path.glob("*.txt"): | |
| book_id = file_path.stem | |
| if book_id not in valid_book_ids: | |
| file_path.unlink() | |
| # Create global instance | |
| storage = StorageService() |