fix: 修复了一些已知问题

This commit is contained in:
xiatianxuan committed 2026-04-30 21:51:46 +08:00
1 parent e433838b43
commit b504e2ffc9
17 files changed
+760 -179

No files matched your search

+26 -10
View File
@@ -73,21 +73,32 @@ sbackup <command>
#### 添加备份策略
```bash
sbackup add
sbackup add <source> <dest> [-i ignore_patterns]
```
运行后,程序会提示你输入:
- **备份文件夹**:需要备份的文件夹路径
- **目标文件夹**:备份文件存放的目标路径
- **需要忽略的文件夹或文件**:用逗号分隔,例如 `.git,__pycache__,node_modules`
参数说明:
- **source**:需要备份的源文件夹路径
- **dest**:备份文件存放的目标路径
- **-i, --ignore**:需要忽略的文件或文件夹名称,使用逗号分隔(默认:`.git,__pycache__`)
示例:
```bash
sbackup add F:/my_folder F:/backup -i node_modules,.git
```
#### 删除备份策略
```bash
sbackup rm
sbackup rm <path>
```
运行后,程序会提示你输入需要删除备份策略的目标文件夹路径。
参数说明:
- **path**:需要删除备份策略的源文件夹路径
示例:
```bash
sbackup rm F:/my_folder
```
#### 查看所有备份策略
@@ -184,13 +195,18 @@ python -m unittest discover -s tests -t .
sbackup/
├── main.py # 程序入口
├── sbackup/
│ ├── __init__.py # 主模块,处理命令行参数
│ ├── _compression.py # 压缩功能实现
│ └── auto_save.py # 备份策略管理
│ ├── __init__.py # CLI 参数解析和命令分发
│ ├── __main__.py # python -m sbackup 入口
│ ├── config.py # 配置加载、语言持久化、数据路径
│ ├── compression.py # ZIP 压缩功能实现
│ ├── auto_save.py # 备份策略管理
│ └── i18n.py # 国际化支持
├── tests/
│ └── sbackup/
│ ├── test_auto_save.py # 备份策略测试
│ ├── test_compression.py # 压缩功能测试
│ ├── test_config.py # 配置加载测试
│ ├── test_i18n.py # 国际化测试
│ └── test_main.py # 主模块测试
├── config.json # 配置文件
└── README.md # 文档
+3
View File
@@ -0,0 +1,3 @@
{
"lang": "en_US"
}
+3 -1
View File
@@ -1,3 +1,4 @@
import sys
from sbackup import run
from sbackup.i18n import t
@@ -5,9 +6,10 @@ from sbackup.i18n import t
def main() -> None:
"""主函数入口"""
try:
run()
sys.exit(run())
except KeyboardInterrupt:
print(t("exit.message"))
sys.exit(130)
if __name__ == "__main__":
+1 -1
View File
@@ -3,7 +3,7 @@ name = "sbackup"
version = "1.0.0"
description = "智能文件夹备份工具,支持增量备份和自定义配置"
readme = "README.md"
requires-python = ">=3.12"
requires-python = ">=3.10"
license = {text = "GPL-3.0"}
authors = [
{name = "xiatianxuan", email = "xiatianxuan2025@163.com"}
+18 -31
View File
@@ -4,33 +4,11 @@ import argparse
import logging
from sbackup.auto_save import BackupManager
from sbackup.i18n import set_locale, t
from sbackup._compression import load_config, save_lang
from sbackup.config import load_config, save_lang
VERSION = "1.0.0"
logger = logging.getLogger(__name__)
def detect_lang() -> str:
"""
优先从命令行参数检测语言,其次从配置文件,最后使用默认英语
"""
# 1. 检查命令行参数
for i, arg in enumerate(sys.argv):
if arg == '--lang' and i + 1 < len(sys.argv):
return sys.argv[i + 1]
if arg.startswith('--lang='):
return arg.split('=', 1)[1]
# 2. 检查配置文件
try:
config = load_config()
if config.lang:
return config.lang
except Exception:
pass
# 3. 默认语言
return "en_US"
EXAMPLES = """示例:
添加备份策略:
sbackup add F:/my_folder F:/backup -i node_modules,.git
@@ -70,7 +48,7 @@ Sbackup 帮助您轻松管理多文件夹的备份任务。它采用增量备份
# 全局参数
parser.add_argument("--debug", action="store_true", help="开启调试模式,输出详细的运行日志和状态信息")
parser.add_argument("-h", "--help", action="help", help="显示此帮助信息并退出")
parser.add_argument("--lang", default=argparse.SUPPRESS, help="设置界面语言: zh_CN (默认) 或 en_US")
parser.add_argument("--lang", default=None, help="设置界面语言: zh_CN (默认) 或 en_US")
subparsers = parser.add_subparsers(dest="command", help="选择要执行的命令")
@@ -103,15 +81,18 @@ def parse_path(path_str: str) -> str:
return os.path.expanduser(path_str.strip())
def run() -> None:
def run() -> int:
"""
主运行函数,返回退出码:0 成功,1 失败
"""
parser = get_parser()
args = parser.parse_args()
# 加载配置以获取默认语言设置
config = load_config()
# 检查是否显式提供了 --lang 参数
if hasattr(args, 'lang'):
if args.lang is not None:
current_lang = args.lang
# 持久化语言设置
save_lang(current_lang)
@@ -131,16 +112,16 @@ def run() -> None:
if args.command is None:
parser.print_help()
return
return 0
if args.command == "version":
print(f"""
Sbackup v{VERSION} — Copyright © 2026 xiatianxuan
Licensed under GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html
""")
return
return 0
manager = BackupManager()
manager = BackupManager(data_file=config.data_file)
if args.command == "add":
source = parse_path(args.source)
@@ -148,12 +129,18 @@ Licensed under GNU GPL v3.0 — https://www.gnu.org/licenses/gpl-3.0.html
success = manager.add_folder(source, dest, args.ignore)
if success:
print(t("cmd.add.success", source=source, dest=dest))
return 0
return 1
elif args.command in ("rm", "remove"):
path = parse_path(args.path)
success = manager.rm_folder(path)
if success:
print(t("cmd.rm.success", path=path))
return 0
return 1
elif args.command == "all":
print(manager.list_folder_table())
return 0
elif args.command == "save":
manager.save_folder()
manager.execute_backups()
return 0
+5
View File
@@ -0,0 +1,5 @@
import sys
from sbackup import run
if __name__ == "__main__":
sys.exit(run())
+113 -39
View File
@@ -1,19 +1,39 @@
import os
import json
import shutil
import logging
from sbackup._compression import Config, ZipfileCompression, load_config
from dataclasses import dataclass
from sbackup.config import Config, load_config, get_default_data_file, DEFAULT_SKIP_PATTERNS
from sbackup.compression import ZipfileCompression
from sbackup.i18n import t
logger = logging.getLogger(__name__)
@dataclass
class BackupEntry:
"""备份策略条目"""
mtime: float
target: str
skip_patterns: list[str]
def to_list(self) -> list:
"""转为 JSON 兼容的列表格式"""
return [self.mtime, self.target, self.skip_patterns]
@staticmethod
def from_list(data: list) -> "BackupEntry":
"""从 JSON 兼容的列表格式创建"""
return BackupEntry(mtime=data[0], target=data[1], skip_patterns=data[2])
class BackupManager:
"""
管理备份策略的类,封装状态和读写操作
"""
def __init__(self, data_file: str = "./sbackup.json"):
self.data_file: str = data_file
self.data: dict = {}
def __init__(self, data_file: str = ""):
self.data_file: str = data_file or get_default_data_file()
self.data: dict[str, list] = {}
self.load()
def load(self):
@@ -31,6 +51,18 @@ class BackupManager:
self.data = json.load(f)
except json.JSONDecodeError:
print(t("warn.json.decode.error", path=self.data_file))
# 备份损坏文件,避免数据丢失
backup_path = self.data_file + ".bak"
try:
shutil.copy2(self.data_file, backup_path)
print(t("warn.json.backup", path=backup_path))
except OSError:
# 备份失败时重命名损坏文件,避免下次再次触发
try:
os.rename(self.data_file, self.data_file + ".corrupted")
print(t("warn.json.renamed", path=self.data_file + ".corrupted"))
except OSError:
pass
self.data = {}
def save(self, initial: bool = False):
@@ -39,14 +71,24 @@ class BackupManager:
"""
if not initial:
logger.debug(f"写入数据文件: {self.data_file}")
data_dir = os.path.dirname(self.data_file)
if data_dir:
os.makedirs(data_dir, exist_ok=True)
with open(self.data_file, "w", encoding="utf-8") as f:
json.dump(self.data, f, ensure_ascii=False, indent=4)
def _get_entry(self, key: str) -> BackupEntry | None:
"""获取指定路径的备份策略条目"""
raw = self.data.get(key)
if raw is None:
return None
return BackupEntry.from_list(raw)
def _set_entry(self, key: str, entry: BackupEntry):
"""设置指定路径的备份策略条目"""
self.data[key] = entry.to_list()
def add_folder(
self,
@@ -58,9 +100,9 @@ class BackupManager:
添加备份策略
"""
if skip_patterns is None:
skip_patterns = ".git,__pycache__"
skip_list = skip_patterns.split(",") if skip_patterns else []
skip_patterns = ",".join(DEFAULT_SKIP_PATTERNS)
skip_list = [s.strip() for s in skip_patterns.split(",") if s.strip()] if skip_patterns else []
if not os.path.isdir(folder_path):
print(t("err.folder.invalid", path=folder_path))
return False
@@ -72,16 +114,20 @@ class BackupManager:
if abs_path in self.data:
print(t("info.already.added", path=abs_path))
return False
self.data[abs_path] = [
os.stat(abs_path).st_mtime,
os.path.abspath(target_folder),
skip_list,
]
try:
entry = BackupEntry(
mtime=os.stat(abs_path).st_mtime,
target=os.path.abspath(target_folder),
skip_patterns=skip_list,
)
except OSError as e:
print(t("err.os", error=e))
return False
self._set_entry(abs_path, entry)
self.save()
return True
def rm_folder(self, folder_path: str) -> bool:
"""
删除备份策略
@@ -95,33 +141,63 @@ class BackupManager:
print(t("warn.no.strategy.found", path=abs_path))
return False
def save_folder(self):
def execute_backups(self):
"""
备份所有文件夹
执行所有备份策略
"""
config = load_config()
for key, value in list(self.data.items()):
backup_count = 0
skip_count = 0
for key, raw in list(self.data.items()):
if not os.path.exists(key):
print(t("warn.source.missing", path=key))
continue
if value[0] != os.stat(key).st_mtime:
# 使用配置文件中的默认值,但允许覆盖特定项
try:
current_mtime = os.stat(key).st_mtime
except OSError as e:
print(t("err.os", error=e))
continue
entry = BackupEntry.from_list(raw)
if entry.mtime != current_mtime:
config_instance = Config(
folder_path=key,
zipfile_path=value[1],
skip_patterns=value[2],
zipfile_path=entry.target,
skip_patterns=entry.skip_patterns,
compression_algorithm=config.compression_algorithm,
compression_level=config.compression_level
)
ZipfileCompression(config_instance).zip_folder()
result = ZipfileCompression(config_instance).zip_folder()
if result["success"]:
entry.mtime = current_mtime
self._set_entry(key, entry)
backup_count += 1
else:
skip_count += 1
if backup_count > 0:
self.save()
print(t("cmd.save.completed", count=backup_count))
elif skip_count > 0:
print(t("cmd.save.uptodate"))
# 向后兼容别名
save_folder = execute_backups
def all_folder(self) -> dict[str, str]:
"""
查看所有备份策略
"""
return {key: value[1] for key, value in self.data.items()}
return {key: BackupEntry.from_list(raw).target for key, raw in self.data.items()}
@staticmethod
def _display_width(s: str) -> int:
"""计算字符串的终端显示宽度(中文字符算2,英文字符算1)"""
width = 0
for ch in s:
if ord(ch) > 0x2E80:
width += 2
else:
width += 1
return width
def list_folder_table(self) -> str:
"""
@@ -129,29 +205,27 @@ class BackupManager:
"""
if not self.data:
return t("cmd.all.empty")
headers = [t("table.header.source"), t("table.header.dest"), t("table.header.ignore")]
rows = []
for path, info in self.data.items():
# 格式化忽略模式
skip = ", ".join(info[2]) if info[2] else t("table.cell.none")
rows.append([path, info[1], skip])
# 计算列宽
col_widths = [len(h) for h in headers]
for path, raw in self.data.items():
entry = BackupEntry.from_list(raw)
skip = ", ".join(entry.skip_patterns) if entry.skip_patterns else t("table.cell.none")
rows.append([path, entry.target, skip])
col_widths = [self._display_width(h) for h in headers]
for row in rows:
for i, cell in enumerate(row):
col_widths[i] = max(col_widths[i], len(cell))
# 构建表格
col_widths[i] = max(col_widths[i], self._display_width(cell))
fmt = " | ".join(["{:<" + str(w) + "}" for w in col_widths])
sep = "-+-".join(["-" * w for w in col_widths])
lines = []
lines.append(fmt.format(*headers))
lines.append(sep)
for row in rows:
lines.append(fmt.format(*row))
return "\n".join(lines)
@@ -1,81 +1,28 @@
"""
@Time: 2025.12.20
@Author: codeseed
压缩模块:ZIP 文件压缩逻辑
"""
import os
import json
import zipfile
from pathlib import Path
from fnmatch import fnmatch
from dataclasses import dataclass, field
from tqdm import tqdm
from sbackup.i18n import t
from sbackup.config import Config
DEFAULT_SKIP_PATTERNS = [".git", "__pycache__"]
# compresslevel 仅对 ZIP_DEFLATED 和 ZIP_BZIP2 有效
_VALID_COMPRESSLEVEL_ALGORITHMS = {zipfile.ZIP_DEFLATED, zipfile.ZIP_BZIP2}
@dataclass
class Config:
folder_path: str = "."
zipfile_path: str | None = None
skip_patterns: list[str] = field(default_factory=lambda: list(DEFAULT_SKIP_PATTERNS))
compression_format: str = "ZIP"
compression_algorithm: str = "ZIP_DEFLATED"
compression_level: int = 6
lang: str = "en_US"
def load_config(config_file: str = "config.json") -> Config:
"""
从配置文件中加载配置
"""
if not os.path.exists(config_file):
return Config()
with open(config_file, "r", encoding="utf-8") as f:
config_data = json.load(f)
compression_config = config_data.get("compression", {})
skip_patterns = config_data.get("skip_patterns", DEFAULT_SKIP_PATTERNS)
data_file = config_data.get("data_file", "sbackup.json")
lang = config_data.get("lang", "en_US")
return Config(
folder_path="",
zipfile_path=None,
skip_patterns=skip_patterns,
compression_format="ZIP",
compression_algorithm=compression_config.get("algorithm", "ZIP_DEFLATED"),
compression_level=compression_config.get("level", 6),
lang=lang
)
def save_lang(lang: str, config_file: str = "config.json") -> None:
"""
将语言偏好保存到配置文件
"""
if os.path.exists(config_file):
with open(config_file, "r", encoding="utf-8") as f:
data = json.load(f)
else:
data = {}
data["lang"] = lang
data_dir = os.path.dirname(config_file)
if data_dir:
os.makedirs(data_dir, exist_ok=True)
with open(config_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
class ZipfileCompression:
def __init__(self, config: Config) -> None:
self.folder_path: Path = Path(config.folder_path)
self.zipfile_path: Path | None = Path(config.zipfile_path) if config.zipfile_path else None
self.skip_patterns: list[str] = config.skip_patterns
self.compression_algorithm: int = self._choose_compression_algorithm(config.compression_algorithm)
self.compression_level: int = config.compression_level
self.compression_level: int | None = self._validate_compresslevel(
config.compression_level, self.compression_algorithm
)
@staticmethod
def _choose_compression_algorithm(compression_algorithm: str) -> int:
@@ -91,12 +38,41 @@ class ZipfileCompression:
case _:
return zipfile.ZIP_DEFLATED
@staticmethod
def _validate_compresslevel(level: int, algorithm: int) -> int | None:
"""校验 compresslevel:仅对 ZIP_DEFLATED 和 ZIP_BZIP2 有效,其他算法不传递该参数"""
if algorithm not in _VALID_COMPRESSLEVEL_ALGORITHMS:
return None # zipfile 不接受 compresslevel 时传 None
if not (0 <= level <= 9):
print(t("warn.invalid.compresslevel", level=level))
return 6
return level
def _should_ignore(self, name: str) -> bool:
for pattern in self.skip_patterns:
if fnmatch(name, pattern):
return True
return False
def _collect_files(self, folder_path: Path) -> list[tuple[str, str]]:
"""遍历文件夹收集需要压缩的文件列表,处理权限错误"""
files = []
try:
for dirpath, dirnames, filenames in os.walk(folder_path):
try:
dirnames[:] = [d for d in dirnames if not self._should_ignore(d)]
for filename in filenames:
if not self._should_ignore(filename):
files.append((dirpath, filename))
except PermissionError as e:
print(t("err.permission", path=dirpath))
continue
except PermissionError as e:
print(t("err.permission", path=folder_path))
except OSError as e:
print(t("err.os", error=e))
return files
def zip_folder(self) -> dict:
"""
压缩指定文件夹到 ZIP 文件
@@ -116,32 +92,40 @@ class ZipfileCompression:
elif zipfile_path.suffix.lower() != ".zip":
zipfile_path = zipfile_path.with_name(zipfile_path.name + ".zip")
# 1. 预先遍历,收集所有需要压缩的文件
files_to_compress = []
for dirpath, dirnames, filenames in os.walk(folder_path):
dirnames[:] = [d for d in dirnames if not self._should_ignore(d)]
for filename in filenames:
if not self._should_ignore(filename):
files_to_compress.append((dirpath, filename))
# ZIP 文件已存在时提示覆盖
if zipfile_path.exists():
print(t("warn.zip.overwrite", path=zipfile_path))
# 1. 收集需要压缩的文件(带异常保护)
files_to_compress = self._collect_files(folder_path)
total_files = len(files_to_compress)
files_count = 0
try:
with zipfile.ZipFile(zipfile_path, "w", self.compression_algorithm, compresslevel=self.compression_level) as zipf:
zip_kwargs = {"mode": "w", "compression": self.compression_algorithm}
if self.compression_level is not None:
zip_kwargs["compresslevel"] = self.compression_level
with zipfile.ZipFile(zipfile_path, **zip_kwargs) as zipf:
# 2. 使用 tqdm 显示进度条
with tqdm(total=total_files, desc=t("compress.progress"), unit=t("compress.unit")) as pbar:
for dirpath, filename in files_to_compress:
file_path = Path(dirpath) / filename
arcname = folder_path.parent / file_path.relative_to(folder_path)
zipf.write(file_path, arcname)
pbar.update(1)
files_count += 1
arcname = str(folder_path.name / file_path.relative_to(folder_path)).replace("\\", "/")
try:
zipf.write(file_path, arcname)
pbar.update(1)
files_count += 1
except (FileNotFoundError, PermissionError):
# 文件在遍历后被删除或权限变更,跳过单个文件
continue
size_mb = zipfile_path.stat().st_size / (1024 * 1024)
# tqdm 清理输出后打印最终结果
print(t("compress.success", path=zipfile_path, size=size_mb, count=files_count))
return {"success": True, "files_count": files_count, "size_mb": size_mb}
except KeyboardInterrupt:
raise
except PermissionError:
print(t("err.permission", path=zipfile_path))
return {"success": False, "files_count": 0, "size_mb": 0.0}
+98
View File
@@ -0,0 +1,98 @@
"""
配置管理模块:配置加载、语言持久化、数据路径
"""
import os
import sys
import json
import logging
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
DEFAULT_SKIP_PATTERNS = [".git", "__pycache__"]
def get_default_data_file() -> str:
"""返回跨平台的默认数据文件路径"""
if sys.platform == "win32":
base = os.environ.get("APPDATA", os.path.expanduser("~"))
elif sys.platform == "darwin":
base = os.path.expanduser("~/Library/Application Support")
else:
base = os.environ.get("XDG_DATA_HOME", os.path.expanduser("~/.local/share"))
return os.path.join(base, "sbackup", "sbackup.json")
@dataclass
class Config:
folder_path: str = "."
zipfile_path: str | None = None
skip_patterns: list[str] = field(default_factory=lambda: list(DEFAULT_SKIP_PATTERNS))
compression_format: str = "ZIP"
compression_algorithm: str = "ZIP_DEFLATED"
compression_level: int = 6
lang: str = "en_US"
data_file: str = field(default_factory=get_default_data_file)
def load_config(config_file: str = "config.json") -> Config:
"""
从配置文件中加载配置
"""
if not os.path.exists(config_file):
return Config()
try:
with open(config_file, "r", encoding="utf-8") as f:
config_data = json.load(f)
except json.JSONDecodeError:
logger.warning("配置文件 %s 格式错误,使用默认配置", config_file)
return Config()
compression_config = config_data.get("compression", {})
skip_patterns = config_data.get("skip_patterns", DEFAULT_SKIP_PATTERNS)
data_file = config_data.get("data_file", get_default_data_file())
lang = config_data.get("lang", "en_US")
return Config(
folder_path="",
zipfile_path=None,
skip_patterns=skip_patterns,
compression_format="ZIP",
compression_algorithm=compression_config.get("algorithm", "ZIP_DEFLATED"),
compression_level=compression_config.get("level", 6),
lang=lang,
data_file=data_file
)
def save_lang(lang: str, config_file: str = "config.json") -> None:
"""
将语言偏好保存到配置文件
"""
if os.path.exists(config_file):
try:
with open(config_file, "r", encoding="utf-8") as f:
data = json.load(f)
except json.JSONDecodeError:
logger.warning("配置文件 %s 格式错误,重置为空", config_file)
data = {}
else:
data = {}
data["lang"] = lang
data_dir = os.path.dirname(config_file)
if data_dir:
try:
os.makedirs(data_dir, exist_ok=True)
except OSError as e:
logger.error("无法创建配置目录 %s: %s", data_dir, e)
return
try:
with open(config_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
except OSError as e:
logger.error("无法写入配置文件 %s: %s", config_file, e)
+24 -9
View File
@@ -1,5 +1,8 @@
import json
import os
import logging
logger = logging.getLogger(__name__)
_current_locale = "zh_CN"
_translations = {}
@@ -7,8 +10,11 @@ _translations = {}
# 初始化默认语言包
_default_file = os.path.join(os.path.dirname(__file__), "locales", "zh_CN.json")
if os.path.exists(_default_file):
with open(_default_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
try:
with open(_default_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning("无法加载默认语言包 %s: %s", _default_file, e)
def set_locale(lang: str) -> None:
"""
@@ -18,13 +24,22 @@ def set_locale(lang: str) -> None:
_current_locale = lang
locale_file = os.path.join(os.path.dirname(__file__), "locales", f"{lang}.json")
if os.path.exists(locale_file):
with open(locale_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
else:
# 如果找不到语言包,回退到默认中文
default_file = os.path.join(os.path.dirname(__file__), "locales", "zh_CN.json")
with open(default_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
try:
with open(locale_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
return
except (json.JSONDecodeError, OSError) as e:
logger.warning("无法加载语言包 %s: %s", locale_file, e)
# 回退到默认中文
default_file = os.path.join(os.path.dirname(__file__), "locales", "zh_CN.json")
if os.path.exists(default_file):
try:
with open(default_file, "r", encoding="utf-8") as f:
_translations = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning("无法加载回退语言包 %s: %s", default_file, e)
_translations = {}
def t(key: str, **kwargs) -> str:
"""
+6
View File
@@ -7,12 +7,18 @@
"warn.no.strategy.found": "Warning: Backup strategy for {path} not found.",
"warn.source.missing": "Warning: Source folder does not exist: {path}",
"warn.json.decode.error": "Warning: Data file {path} is malformed, initializing as empty dict.",
"warn.json.backup": "Warning: Corrupted data file backed up to {path}.",
"warn.json.renamed": "Warning: Corrupted data file renamed to {path}.",
"compress.success": "Backup successful: {path} ({size:.2f} MB, {count} files)",
"compress.progress": "Compressing",
"compress.unit": "files",
"err.permission": "Permission denied: Cannot write to '{path}'",
"err.os": "OS Error: {error}",
"err.unknown": "Unknown error: {error}",
"warn.zip.overwrite": "Warning: Overwriting existing backup file {path}",
"warn.invalid.compresslevel": "Warning: Compression level {level} out of range (0-9), using default 6",
"cmd.save.uptodate": "All backup strategies are up to date, no backup needed.",
"cmd.save.completed": "Backup completed, processed {count} strategy(s).",
"cmd.all.empty": "No backup strategies configured.",
"exit.message": "\nExiting.",
"table.header.source": "Source Folder",
+6
View File
@@ -7,12 +7,18 @@
"warn.no.strategy.found": "警告:未找到 {path} 的备份策略。",
"warn.source.missing": "警告:源文件夹不存在: {path}",
"warn.json.decode.error": "警告:数据文件 {path} 格式错误,初始化为空字典。",
"warn.json.backup": "警告:已备份损坏的数据文件到 {path},",
"warn.json.renamed": "警告:已将损坏的数据文件重命名为 {path}。",
"compress.success": "成功备份: {path} ({size:.2f} MB, {count} 个文件)",
"compress.progress": "正在压缩",
"compress.unit": "文件",
"err.permission": "权限不足:无法写入 '{path}'",
"err.os": "系统错误:{error}",
"err.unknown": "未知错误:{error}",
"warn.zip.overwrite": "警告:将覆盖已有的备份文件 {path}",
"warn.invalid.compresslevel": "警告:压缩级别 {level} 超出范围 (0-9),使用默认值 6",
"cmd.save.uptodate": "所有备份策略均已是最新,无需备份。",
"cmd.save.completed": "备份完成,共处理 {count} 个策略。",
"cmd.all.empty": "没有配置任何备份策略。",
"exit.message": "\n已退出。",
"table.header.source": "源文件夹路径",
+92 -1
View File
@@ -7,7 +7,10 @@ import os
import json
import tempfile
import shutil
import time
from pathlib import Path
from sbackup.auto_save import BackupManager
from sbackup.i18n import t
class TestAutoSave(unittest.TestCase):
@@ -62,12 +65,100 @@ class TestAutoSave(unittest.TestCase):
# 先添加
self.manager.add_folder(self.source_folder, self.target_folder)
self.assertIn(os.path.abspath(self.source_folder), self.manager.data)
# 再删除
result = self.manager.rm_folder(self.source_folder)
self.assertTrue(result)
self.assertNotIn(os.path.abspath(self.source_folder), self.manager.data)
def test_save_folder_updates_mtime(self):
"""测试执行备份后时间戳被更新,实现增量备份"""
# 添加策略
self.manager.add_folder(self.source_folder, self.target_folder, "")
abs_source = os.path.abspath(self.source_folder)
original_mtime = self.manager.data[abs_source][0]
# 修改源文件夹内容,触发备份条件
time.sleep(0.1)
(Path(self.source_folder) / "new_file.txt").write_text("new content")
# 执行备份(使用新方法名 execute_backups)
self.manager.execute_backups()
# 验证 mtime 已更新
new_mtime = self.manager.data[abs_source][0]
self.assertNotEqual(original_mtime, new_mtime, "备份后 mtime 应被更新")
self.assertEqual(new_mtime, os.stat(abs_source).st_mtime)
# 验证数据已持久化
self.manager.load()
self.assertEqual(self.manager.data[abs_source][0], new_mtime)
def test_execute_backups_alias(self):
"""测试 save_folder 向后兼容别名"""
self.assertTrue(hasattr(self.manager, "save_folder"))
# 验证 save_folder 指向 execute_backups 的底层函数
self.assertIs(
self.manager.save_folder.__func__,
self.manager.execute_backups.__func__
)
def test_add_folder_skip_patterns_stripped(self):
"""测试忽略模式中的空格被正确去除"""
result = self.manager.add_folder(
self.source_folder, self.target_folder, " .git , node_modules "
)
self.assertTrue(result)
abs_source = os.path.abspath(self.source_folder)
skip_list = self.manager.data[abs_source][2]
self.assertEqual(skip_list, [".git", "node_modules"])
def test_add_duplicate_folder(self):
"""测试重复添加同一文件夹应失败"""
result1 = self.manager.add_folder(self.source_folder, self.target_folder)
self.assertTrue(result1)
result2 = self.manager.add_folder(self.source_folder, self.target_folder)
self.assertFalse(result2)
abs_source = os.path.abspath(self.source_folder)
self.assertEqual(len(self.manager.data), 1)
def test_rm_nonexistent_folder(self):
"""测试删除不存在的策略应失败"""
result = self.manager.rm_folder("/nonexistent/path")
self.assertFalse(result)
def test_add_invalid_source(self):
"""测试添加无效源目录应失败"""
result = self.manager.add_folder("/nonexistent/path", self.target_folder)
self.assertFalse(result)
def test_add_invalid_dest(self):
"""测试添加无效目标目录应失败"""
result = self.manager.add_folder(self.source_folder, "/nonexistent/path")
self.assertFalse(result)
def test_save_missing_source(self):
"""测试源文件夹不存在时的 save 行为"""
# 手动构造一条源文件夹不存在的记录
self.manager.data["/nonexistent/source"] = [0.0, self.target_folder, []]
self.manager.save()
# save_folder 应打印警告并跳过,不抛出异常
self.manager.save_folder()
# 数据应保持不变(不会被删除)
self.assertIn("/nonexistent/source", self.manager.data)
def test_list_folder_table_empty(self):
"""测试空策略时表格返回提示文本"""
text = self.manager.list_folder_table()
self.assertEqual(text, t("cmd.all.empty"))
def test_all_folder(self):
"""测试 all_folder 返回正确字典"""
self.manager.add_folder(self.source_folder, self.target_folder)
result = self.manager.all_folder()
abs_source = os.path.abspath(self.source_folder)
self.assertEqual(result[abs_source], os.path.abspath(self.target_folder))
if __name__ == "__main__":
unittest.main()
+78 -4
View File
@@ -1,5 +1,5 @@
"""
单元测试 for sbackup._compression 模块
单元测试 for sbackup.compression 模块
"""
import unittest
import os
@@ -7,7 +7,9 @@ import shutil
import zipfile
from pathlib import Path
from unittest.mock import patch
from sbackup._compression import Config, ZipfileCompression
from sbackup.config import Config
from sbackup.compression import ZipfileCompression
from sbackup.i18n import t
class TestCompression(unittest.TestCase):
@@ -18,7 +20,7 @@ class TestCompression(unittest.TestCase):
(Path(self.test_dir) / "file1.txt").write_text("test content 1")
(Path(self.test_dir) / "subdir").mkdir()
(Path(self.test_dir) / "subdir" / "file2.txt").write_text("test content 2")
self.zip_path = "test.zip"
if os.path.exists(self.zip_path):
os.remove(self.zip_path)
@@ -38,9 +40,81 @@ class TestCompression(unittest.TestCase):
)
compressor = ZipfileCompression(config)
compressor.zip_folder()
self.assertTrue(os.path.exists(self.zip_path))
self.assertGreater(os.path.getsize(self.zip_path), 0)
def test_zip_internal_path_structure(self):
"""测试 ZIP 内部路径结构正确(以文件夹名开头,而非绝对路径)"""
config = Config(
folder_path=self.test_dir,
zipfile_path=self.zip_path,
)
compressor = ZipfileCompression(config)
compressor.zip_folder()
with zipfile.ZipFile(self.zip_path, "r") as zf:
namelist = zf.namelist()
for name in namelist:
self.assertTrue(
name.startswith("test_data/"),
f"ZIP 内部路径 '{name}' 应以 'test_data/' 开头"
)
def test_zip_respects_skip_patterns(self):
"""测试压缩时正确忽略匹配的文件"""
(Path(self.test_dir) / ".gitignore").write_text("ignore")
config = Config(
folder_path=self.test_dir,
zipfile_path=self.zip_path,
skip_patterns=[".gitignore"],
)
compressor = ZipfileCompression(config)
result = compressor.zip_folder()
self.assertTrue(result["success"])
with zipfile.ZipFile(self.zip_path, "r") as zf:
namelist = zf.namelist()
self.assertNotIn("test_data/.gitignore", namelist)
@patch("builtins.print")
def test_compresslevel_out_of_range(self, mock_print):
"""测试压缩级别超出范围时回退到默认值"""
config = Config(
folder_path=self.test_dir,
zipfile_path=self.zip_path,
compression_level=99,
)
compressor = ZipfileCompression(config)
# compresslevel 应回退到 6
self.assertEqual(compressor.compression_level, 6)
def test_compresslevel_none_for_stored(self):
"""测试 ZIP_STORED 算法不传递 compresslevel"""
config = Config(
folder_path=self.test_dir,
zipfile_path=self.zip_path,
compression_algorithm="ZIP_STORED",
compression_level=6,
)
compressor = ZipfileCompression(config)
self.assertIsNone(compressor.compression_level)
@patch("builtins.print")
def test_zip_overwrite_warning(self, mock_print):
"""测试 ZIP 文件已存在时输出覆盖警告"""
config = Config(
folder_path=self.test_dir,
zipfile_path=self.zip_path,
)
compressor = ZipfileCompression(config)
# 第一次压缩
compressor.zip_folder()
# 第二次压缩(文件已存在)
compressor.zip_folder()
printed = " ".join(str(call) for call in mock_print.call_args_list)
self.assertIn(t("warn.zip.overwrite").split("{")[0], printed)
if __name__ == "__main__":
unittest.main()
+72 -10
View File
@@ -1,12 +1,12 @@
"""
单元测试 for sbackup._compression.load_config 函数
单元测试 for sbackup.config 模块(load_config / save_lang)
"""
import unittest
import os
import tempfile
import json
from sbackup._compression import load_config, Config
from sbackup.config import load_config, save_lang, Config
class TestConfig(unittest.TestCase):
@@ -55,7 +55,8 @@ class TestConfig(unittest.TestCase):
"""
测试从默认配置文件中加载配置
"""
# 创建默认配置文件
# 在临时目录中创建默认配置文件(而非项目根目录)
default_config = os.path.join(self.test_dir, "config.json")
config_data = {
"compression": {
"algorithm": "ZIP_DEFLATED",
@@ -64,19 +65,14 @@ class TestConfig(unittest.TestCase):
"skip_patterns": [".git", "__pycache__"],
"data_file": "sbackup.json"
}
with open("config.json", "w", encoding="utf-8") as f:
with open(default_config, "w", encoding="utf-8") as f:
json.dump(config_data, f, ensure_ascii=False, indent=4)
# 加载配置
config = load_config()
config = load_config(default_config)
# 验证配置
self.assertEqual(config.compression_algorithm, "ZIP_DEFLATED")
self.assertEqual(config.compression_level, 6)
self.assertEqual(config.skip_patterns, [".git", "__pycache__"])
# 清理默认配置文件
os.remove("config.json")
def test_load_config_from_nonexistent_file(self):
"""
@@ -90,6 +86,72 @@ class TestConfig(unittest.TestCase):
self.assertEqual(config.compression_level, 6)
self.assertEqual(config.skip_patterns, [".git", "__pycache__"])
def test_load_config_malformed_json(self):
"""测试配置文件 JSON 格式错误时使用默认配置"""
malformed_config = os.path.join(self.test_dir, "bad_config.json")
with open(malformed_config, "w", encoding="utf-8") as f:
f.write("{invalid json content!!!")
config = load_config(malformed_config)
self.assertEqual(config.compression_algorithm, "ZIP_DEFLATED")
self.assertEqual(config.compression_level, 6)
class TestSaveLang(unittest.TestCase):
def setUp(self):
self.test_dir = tempfile.mkdtemp()
self.config_file = os.path.join(self.test_dir, "config.json")
def tearDown(self):
if os.path.exists(self.test_dir):
import shutil
shutil.rmtree(self.test_dir)
def test_save_lang_creates_new_file(self):
"""测试 save_lang 在配置文件不存在时创建新文件"""
save_lang("zh_CN", self.config_file)
self.assertTrue(os.path.exists(self.config_file))
with open(self.config_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.assertEqual(data["lang"], "zh_CN")
def test_save_lang_updates_existing_file(self):
"""测试 save_lang 在已有配置文件中更新语言字段"""
# 先写入已有配置
with open(self.config_file, "w", encoding="utf-8") as f:
json.dump({"compression": {"algorithm": "ZIP_STORED", "level": 3}}, f)
save_lang("en_US", self.config_file)
with open(self.config_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.assertEqual(data["lang"], "en_US")
# 原有配置应保留
self.assertEqual(data["compression"]["algorithm"], "ZIP_STORED")
self.assertEqual(data["compression"]["level"], 3)
def test_save_lang_overwrites_existing_lang(self):
"""测试 save_lang 覆盖已有的语言设置"""
with open(self.config_file, "w", encoding="utf-8") as f:
json.dump({"lang": "zh_CN"}, f)
save_lang("en_US", self.config_file)
with open(self.config_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.assertEqual(data["lang"], "en_US")
def test_save_lang_with_subdirectory(self):
"""测试 save_lang 在子目录中创建配置文件"""
subdir_config = os.path.join(self.test_dir, "sub", "config.json")
save_lang("zh_CN", subdir_config)
self.assertTrue(os.path.exists(subdir_config))
with open(subdir_config, "r", encoding="utf-8") as f:
data = json.load(f)
self.assertEqual(data["lang"], "zh_CN")
if __name__ == "__main__":
unittest.main()
+73
View File
@@ -0,0 +1,73 @@
"""
单元测试 for sbackup.i18n 模块
"""
import unittest
import os
import json
import tempfile
from sbackup import i18n
class TestI18n(unittest.TestCase):
def setUp(self):
"""保存原始状态"""
self._orig_locale = i18n._current_locale
self._orig_trans = i18n._translations.copy()
def tearDown(self):
"""恢复原始状态"""
i18n._current_locale = self._orig_locale
i18n._translations = self._orig_trans
def test_set_locale_existing(self):
"""测试设置为存在的语言包"""
i18n.set_locale("zh_CN")
self.assertEqual(i18n._current_locale, "zh_CN")
self.assertIn("cmd.add.success", i18n._translations)
def test_set_locale_fallback(self):
"""测试设置不存在的语言包时回退到中文"""
i18n.set_locale("nonexistent_lang")
self.assertEqual(i18n._current_locale, "nonexistent_lang")
self.assertIn("cmd.add.success", i18n._translations)
def test_t_basic(self):
"""测试基础翻译获取"""
i18n.set_locale("zh_CN")
text = i18n.t("cmd.add.success", source="/a", dest="/b")
self.assertIn("/a", text)
self.assertIn("/b", text)
def test_t_missing_key(self):
"""测试缺失键名时返回键名本身"""
i18n.set_locale("zh_CN")
text = i18n.t("this.key.does.not.exist")
self.assertEqual(text, "this.key.does.not.exist")
def test_t_no_kwargs(self):
"""测试无格式化参数时返回原文"""
i18n.set_locale("en_US")
text = i18n.t("cmd.all.empty")
self.assertEqual(text, "No backup strategies configured.")
def test_t_format_error(self):
"""测试格式化参数不匹配时返回原文"""
i18n.set_locale("zh_CN")
text = i18n.t("cmd.add.success", wrong_key="value")
# 应返回未格式化的原文
self.assertEqual(text, "备份策略添加成功: {source} -> {dest}")
def test_language_switch_returns_correct_text(self):
"""测试切换语言后 t() 返回对应语言的文本"""
i18n.set_locale("zh_CN")
zh_text = i18n.t("cmd.all.empty")
self.assertEqual(zh_text, "没有配置任何备份策略。")
i18n.set_locale("en_US")
en_text = i18n.t("cmd.all.empty")
self.assertEqual(en_text, "No backup strategies configured.")
if __name__ == "__main__":
unittest.main()
+85
View File
@@ -3,10 +3,29 @@
"""
import unittest
import sys
import os
import logging
import tempfile
import shutil
from unittest.mock import patch
from sbackup.i18n import t
class TestMain(unittest.TestCase):
def setUp(self):
self.test_dir = tempfile.mkdtemp()
self.original_argv = sys.argv.copy()
# 保存并重置 logging 状态
self._root_handlers = logging.root.handlers[:]
self._root_level = logging.root.level
def tearDown(self):
sys.argv = self.original_argv
shutil.rmtree(self.test_dir, ignore_errors=True)
# 恢复 logging 状态
logging.root.handlers = self._root_handlers
logging.root.setLevel(self._root_level)
@patch("builtins.print")
def test_version_command(self, mock_print):
"""测试 version 命令"""
@@ -15,5 +34,71 @@ class TestMain(unittest.TestCase):
run()
mock_print.assert_called()
@patch("argparse.ArgumentParser.print_help")
def test_no_command_shows_help(self, mock_help):
"""测试无命令时显示帮助"""
sys.argv = ["sbackup"]
from sbackup import run
run()
mock_help.assert_called_once()
@patch("builtins.print")
def test_add_command_invalid_source(self, mock_print):
"""测试 add 命令传入无效源目录"""
sys.argv = ["sbackup", "--lang", "en_US", "add", "/nonexistent/path", "/tmp"]
from sbackup import run
run()
printed = " ".join(str(call) for call in mock_print.call_args_list)
self.assertIn(t("err.folder.invalid").split("{")[0].strip(), printed)
@patch("builtins.print")
def test_rm_command_nonexistent(self, mock_print):
"""测试 rm 命令删除不存在的策略"""
sys.argv = ["sbackup", "--lang", "en_US", "rm", "/nonexistent/path"]
from sbackup import run
run()
printed = " ".join(str(call) for call in mock_print.call_args_list)
self.assertIn(t("warn.no.strategy.found").split("{")[0].strip(), printed)
@patch("builtins.print")
def test_all_command_empty(self, mock_print):
"""测试 all 命令在无策略时输出提示"""
os.chdir(self.test_dir)
sys.argv = ["sbackup", "--lang", "en_US", "all"]
from sbackup import run
run()
printed = " ".join(str(call) for call in mock_print.call_args_list)
self.assertIn(t("cmd.all.empty"), printed)
@patch("builtins.print")
def test_lang_switch(self, mock_print):
"""测试 --lang 参数切换语言"""
os.chdir(self.test_dir)
sys.argv = ["sbackup", "--lang", "en_US", "all"]
from sbackup import run
run()
printed = " ".join(str(call) for call in mock_print.call_args_list)
self.assertIn("No backup strategies configured", printed)
@patch("builtins.print")
def test_debug_mode_sets_logging(self, mock_print):
"""测试 --debug 参数开启 DEBUG 日志级别"""
os.chdir(self.test_dir)
sys.argv = ["sbackup", "--debug", "--lang", "en_US", "all"]
from sbackup import run
run()
self.assertEqual(logging.root.level, logging.DEBUG)
@patch("builtins.print")
def test_no_debug_keeps_default_logging(self, mock_print):
"""测试无 --debug 时日志级别保持默认"""
os.chdir(self.test_dir)
logging.root.setLevel(logging.WARNING)
sys.argv = ["sbackup", "--lang", "en_US", "all"]
from sbackup import run
run()
self.assertEqual(logging.root.level, logging.WARNING)
if __name__ == "__main__":
unittest.main()