zwt13703
44 lines
1.1 KiB
Python
44 lines
1.1 KiB
Python
import os
|
|
import aiofiles
|
|
from pathlib import Path
|
|
from fastapi import UploadFile
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
TEMPLATES_DIR = "templates"
|
|
REF_FILES_DIR = "ref_files"
|
|
RESULTS_DIR = "results"
|
|
|
|
|
|
def _ensure_dir(path: str) -> str:
|
|
os.makedirs(path, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def get_storage_dir(subdir: str) -> str:
|
|
return _ensure_dir(os.path.join(settings.STORAGE_ROOT, subdir))
|
|
|
|
|
|
async def save_upload(upload_file: UploadFile, subdir: str) -> str:
|
|
dir_path = get_storage_dir(subdir)
|
|
file_path = os.path.join(dir_path, upload_file.filename or "unnamed")
|
|
async with aiofiles.open(file_path, "wb") as f:
|
|
content = await upload_file.read()
|
|
await f.write(content)
|
|
return file_path
|
|
|
|
|
|
async def get_file_content(file_path: str) -> bytes:
|
|
async with aiofiles.open(file_path, "rb") as f:
|
|
return await f.read()
|
|
|
|
|
|
def delete_file(file_path: str) -> None:
|
|
if os.path.exists(file_path):
|
|
os.remove(file_path)
|
|
|
|
|
|
def get_absolute_path(rel_path: str) -> str:
|
|
return os.path.abspath(rel_path)
|