243 lines
6.8 KiB
Python
243 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
文件存储
|
|
提供文件读写的统一接口
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
from typing import Any, Optional, Union
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FileStorage:
|
|
"""
|
|
文件存储
|
|
|
|
提供：
|
|
- 文本文件读写
|
|
- JSON 文件读写
|
|
- 二进制文件读写
|
|
- 路径管理
|
|
"""
|
|
|
|
def __init__(self, base_path: str = "data"):
|
|
"""
|
|
初始化文件存储
|
|
|
|
Args:
|
|
base_path: 基础存储路径（相对于项目根目录）
|
|
"""
|
|
self._base_path = Path(base_path)
|
|
self._project_root: Optional[Path] = None
|
|
|
|
def set_project_root(self, project_root: str):
|
|
"""设置项目根目录"""
|
|
self._project_root = Path(project_root)
|
|
|
|
def _get_full_path(self, relative_path: str) -> Path:
|
|
"""获取完整路径"""
|
|
if self._project_root:
|
|
return self._project_root / self._base_path / relative_path
|
|
return self._base_path / relative_path
|
|
|
|
def _ensure_dir(self, file_path: Path):
|
|
"""确保目录存在"""
|
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
def save_text(self, content: str, filename: str, subdir: str = "") -> bool:
|
|
"""
|
|
保存文本文件
|
|
|
|
Args:
|
|
content: 文本内容
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
self._ensure_dir(path)
|
|
path.write_text(content, encoding='utf-8')
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存文本文件失败: {filename}, {e}")
|
|
return False
|
|
|
|
def load_text(self, filename: str, subdir: str = "") -> Optional[str]:
|
|
"""
|
|
加载文本文件
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
文本内容
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
if path.exists():
|
|
return path.read_text(encoding='utf-8')
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"加载文本文件失败: {filename}, {e}")
|
|
return None
|
|
|
|
def save_json(self, data: Any, filename: str, subdir: str = "") -> bool:
|
|
"""
|
|
保存 JSON 文件
|
|
|
|
Args:
|
|
data: 数据
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
self._ensure_dir(path)
|
|
with open(path, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存 JSON 文件失败: {filename}, {e}")
|
|
return False
|
|
|
|
def load_json(self, filename: str, subdir: str = "") -> Optional[Any]:
|
|
"""
|
|
加载 JSON 文件
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
数据
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
if path.exists():
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"加载 JSON 文件失败: {filename}, {e}")
|
|
return None
|
|
|
|
def save_binary(self, data: bytes, filename: str, subdir: str = "") -> bool:
|
|
"""
|
|
保存二进制文件
|
|
|
|
Args:
|
|
data: 二进制数据
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
self._ensure_dir(path)
|
|
path.write_bytes(data)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存二进制文件失败: {filename}, {e}")
|
|
return False
|
|
|
|
def load_binary(self, filename: str, subdir: str = "") -> Optional[bytes]:
|
|
"""
|
|
加载二进制文件
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
二进制数据
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
if path.exists():
|
|
return path.read_bytes()
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"加载二进制文件失败: {filename}, {e}")
|
|
return None
|
|
|
|
def exists(self, filename: str, subdir: str = "") -> bool:
|
|
"""
|
|
检查文件是否存在
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
是否存在
|
|
"""
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
return path.exists()
|
|
|
|
def delete(self, filename: str, subdir: str = "") -> bool:
|
|
"""
|
|
删除文件
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
try:
|
|
path = self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|
|
if path.exists():
|
|
path.unlink()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"删除文件失败: {filename}, {e}")
|
|
return False
|
|
|
|
def list_files(self, subdir: str = "", pattern: str = "*") -> list[str]:
|
|
"""
|
|
列出文件
|
|
|
|
Args:
|
|
subdir: 子目录
|
|
pattern: 匹配模式
|
|
|
|
Returns:
|
|
文件名列表
|
|
"""
|
|
try:
|
|
path = self._get_full_path(subdir) if subdir else self._get_full_path("")
|
|
if path.exists():
|
|
return [f.name for f in path.glob(pattern) if f.is_file()]
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"列出文件失败: {subdir}, {e}")
|
|
return []
|
|
|
|
def get_path(self, filename: str, subdir: str = "") -> Path:
|
|
"""
|
|
获取文件完整路径
|
|
|
|
Args:
|
|
filename: 文件名
|
|
subdir: 子目录
|
|
|
|
Returns:
|
|
完整路径
|
|
"""
|
|
return self._get_full_path(f"{subdir}/{filename}" if subdir else filename)
|