36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import UploadFile
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
class StorageClient:
|
|
def save_upload(self, file: UploadFile) -> str:
|
|
return self.save_bytes(file.file.read(), suffix=Path(file.filename or "upload.bin").suffix)
|
|
|
|
def save_bytes(self, raw: bytes, suffix: str = "") -> str:
|
|
base = self._base_dir()
|
|
date_seg = datetime.now().strftime("%Y/%m/%d")
|
|
target_dir = base / date_seg
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
filename = f"{uuid4().hex}{suffix}"
|
|
(target_dir / filename).write_bytes(raw)
|
|
return f"{date_seg}/{filename}"
|
|
|
|
def temporary_url(self, key: str) -> str:
|
|
return f"/api/files/{key}"
|
|
|
|
def resolve(self, key: str) -> Path:
|
|
base = self._base_dir().resolve()
|
|
target = (base / key).resolve()
|
|
if base != target and base not in target.parents:
|
|
raise ValueError("非法存储路径")
|
|
return target
|
|
|
|
@staticmethod
|
|
def _base_dir() -> Path:
|
|
return Path(get_settings().local_storage_dir)
|