import os import uuid from pathlib import Path from fastapi import UploadFile from app.core.config import get_settings settings = get_settings() class FileStorageService: """文件存储服务""" @staticmethod def get_upload_dir() -> Path: """获取上传目录""" upload_dir = Path(settings.upload_dir) upload_dir.mkdir(parents=True, exist_ok=True) return upload_dir @staticmethod async def save_file(file: UploadFile, company_id: int) -> tuple[str, int]: """ 保存上传的文件 Args: file: 上传的文件 company_id: 企业ID Returns: (存储文件名, 文件大小) """ # 创建企业专属目录 company_dir = FileStorageService.get_upload_dir() / str(company_id) company_dir.mkdir(parents=True, exist_ok=True) # 生成唯一文件名 file_ext = Path(file.filename or "file").suffix stored_filename = f"{uuid.uuid4().hex}{file_ext}" file_path = company_dir / stored_filename # 保存文件 content = await file.read() file_size = len(content) with open(file_path, "wb") as f: f.write(content) return f"{company_id}/{stored_filename}", file_size @staticmethod def get_file_path(stored_filename: str) -> Path: """ 获取文件完整路径 Args: stored_filename: 存储文件名(格式: company_id/filename) Returns: 文件路径 """ return FileStorageService.get_upload_dir() / stored_filename @staticmethod def delete_file(stored_filename: str) -> bool: """ 删除文件 Args: stored_filename: 存储文件名 Returns: 是否删除成功 """ try: file_path = FileStorageService.get_file_path(stored_filename) if file_path.exists(): file_path.unlink() return True return False except Exception: return False # 单例实例 file_storage_service = FileStorageService()