|
|
from __future__ import annotations
|
|
|
|
|
|
import os
|
|
|
from typing import BinaryIO, Dict, Optional
|
|
|
|
|
|
from .interface import StorageInterface
|
|
|
|
|
|
|
|
|
class LocalStorage(StorageInterface):
|
|
|
"""本地文件存储适配器"""
|
|
|
|
|
|
storage_type = "local"
|
|
|
|
|
|
def __init__(self, config: dict):
|
|
|
self.base_path = config.get("base_path")
|
|
|
if not self.base_path:
|
|
|
raise ValueError("LocalStorage 需要 base_path 配置")
|
|
|
os.makedirs(self.base_path, exist_ok=True)
|
|
|
|
|
|
def _abs_path(self, key: str) -> str:
|
|
|
"""获取文件绝对路径"""
|
|
|
return os.path.join(self.base_path, key)
|
|
|
|
|
|
def upload(self, file_stream: BinaryIO, key: str, mime_type: Optional[str] = None) -> Dict:
|
|
|
"""上传文件到本地（流式写入，避免内存溢出）"""
|
|
|
# 本地文件存储需要移除前缀
|
|
|
key = key.lstrip(self.storage_type + ":")
|
|
|
abs_path = self._abs_path(key)
|
|
|
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
|
|
|
|
|
|
# 流式写入，每次读取 8MB 缓冲区
|
|
|
BUFFER_SIZE = 8 * 1024 * 1024 # 8MB
|
|
|
with open(abs_path, "wb") as f:
|
|
|
while True:
|
|
|
buffer = file_stream.read(BUFFER_SIZE)
|
|
|
if not buffer:
|
|
|
break
|
|
|
f.write(buffer)
|
|
|
|
|
|
file_size = os.path.getsize(abs_path)
|
|
|
return {
|
|
|
"path": abs_path,
|
|
|
"size": file_size,
|
|
|
}
|
|
|
|
|
|
def append_chunk(self, key: str, chunk_stream: BinaryIO, offset: int) -> None:
|
|
|
"""追加写入数据块"""
|
|
|
abs_path = self._abs_path(key)
|
|
|
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
|
|
|
|
|
|
mode = "r+b" if os.path.exists(abs_path) else "wb"
|
|
|
with open(abs_path, mode) as f:
|
|
|
f.seek(offset)
|
|
|
f.write(chunk_stream.read())
|
|
|
|
|
|
def download(self, key: str) -> BinaryIO:
|
|
|
"""下载文件"""
|
|
|
# 本地文件存储需要移除前缀
|
|
|
key = key.lstrip(self.storage_type + ":")
|
|
|
abs_path = self._abs_path(key)
|
|
|
if not os.path.exists(abs_path):
|
|
|
raise FileNotFoundError(f"文件不存在: {key}")
|
|
|
return open(abs_path, "rb")
|
|
|
|
|
|
def delete(self, key: str) -> None:
|
|
|
"""删除文件"""
|
|
|
# 本地文件存储需要移除前缀
|
|
|
key = key.lstrip(self.storage_type + ":")
|
|
|
abs_path = self._abs_path(key)
|
|
|
if os.path.exists(abs_path):
|
|
|
os.remove(abs_path)
|
|
|
|
|
|
def exists(self, key: str) -> bool:
|
|
|
"""检查文件是否存在"""
|
|
|
# 本地文件存储需要移除前缀
|
|
|
key = key.lstrip(self.storage_type + ":")
|
|
|
abs_path = self._abs_path(key)
|
|
|
return os.path.exists(abs_path)
|
|
|
|
|
|
def get_url(self, key: str, expires: int = 3600) -> str:
|
|
|
"""获取文件访问URL（本地存储返回下载路由）"""
|
|
|
# 本地存储通过后端路由提供下载，无需签名
|
|
|
return key
|
|
|
|