geMoldInsight/src/utils/file_handler.py

# utils/file_handler.py
import aiofiles
from pathlib import Path
from fastapi import UploadFile
from typing import Tuple


class FileHandler:
    def __init__(self, upload_dir: str = "uploads"):
        self.upload_dir = Path(upload_dir)
        self.upload_dir.mkdir(exist_ok=True)

    async def save_uploaded_file(self, file: UploadFile) -> Tuple[Path, int]:
        """保存上传的文件"""
        file_path = self.upload_dir / file.filename

        content = await file.read()
        async with aiofiles.open(file_path, 'wb') as f:
            await f.write(content)

        return file_path, len(content)

    def cleanup_file(self, file_path: Path):
        """清理文件"""
        try:
            if file_path.exists():
                file_path.unlink()
        except Exception as e:
            print(f"文件清理失败: {e}")