File size: 1,708 Bytes
ba4a241 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
from __future__ import annotations
import shutil
import uuid
from pathlib import Path
from typing import Tuple
from fastapi import UploadFile
from ..core.config import get_settings
settings = get_settings()
def cleanup_old_audio_files(max_files: int | None = None) -> None:
"""Remove old audio files from the static directory to save space."""
max_files = max_files or settings.max_audio_files
audio_dir = settings.audio_dir
audio_dir.mkdir(parents=True, exist_ok=True)
files = sorted(audio_dir.glob("*") , key=lambda f: f.stat().st_mtime if f.exists() else 0)
if len(files) <= max_files:
return
for old_file in files[:-max_files]:
try:
old_file.unlink()
except OSError:
continue
def save_upload_file(upload: UploadFile) -> Path:
"""Persist an UploadFile to the temporary directory and return its path."""
tmp_dir = settings.tmp_dir
tmp_dir.mkdir(parents=True, exist_ok=True)
suffix = Path(upload.filename or "audio").suffix or ".mp3"
temp_path = tmp_dir / f"upload_{uuid.uuid4().hex}{suffix}"
with temp_path.open("wb") as buffer:
shutil.copyfileobj(upload.file, buffer)
return temp_path
def store_audio_file(audio_path: Path, prefix: str | None = None) -> Tuple[Path, str]:
"""Copy an audio file to the public static folder and return the new path and URL."""
cleanup_old_audio_files()
prefix = prefix or "audio"
suffix = audio_path.suffix or ".mp3"
dest_filename = f"{prefix}_{uuid.uuid4().hex}{suffix}"
dest_path = settings.audio_dir / dest_filename
shutil.copy2(audio_path, dest_path)
url = f"/media/{dest_filename}"
return dest_path, url
|