diff --git a/README.md b/README.md index fc8e7ac..32b54a2 100644 --- a/README.md +++ b/README.md @@ -73,21 +73,32 @@ sbackup #### 添加备份策略 ```bash -sbackup add +sbackup add [-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**:需要删除备份策略的源文件夹路径 + +示例: +```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 # 文档 diff --git a/config.json b/config.json new file mode 100644 index 0000000..111f559 --- /dev/null +++ b/config.json @@ -0,0 +1,3 @@ +{ + "lang": "en_US" +} \ No newline at end of file diff --git a/main.py b/main.py index 7abddc5..071b8c4 100644 --- a/main.py +++ b/main.py @@ -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__": diff --git a/pyproject.toml b/pyproject.toml index 544fe59..b48ee80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"} diff --git a/sbackup/__init__.py b/sbackup/__init__.py index 27561ff..a38e8cb 100644 --- a/sbackup/__init__.py +++ b/sbackup/__init__.py @@ -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 diff --git a/sbackup/__main__.py b/sbackup/__main__.py new file mode 100644 index 0000000..1bce02f --- /dev/null +++ b/sbackup/__main__.py @@ -0,0 +1,5 @@ +import sys +from sbackup import run + +if __name__ == "__main__": + sys.exit(run()) diff --git a/sbackup/auto_save.py b/sbackup/auto_save.py index 841437c..2eee2e6 100644 --- a/sbackup/auto_save.py +++ b/sbackup/auto_save.py @@ -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) diff --git a/sbackup/_compression.py b/sbackup/compression.py similarity index 53% rename from sbackup/_compression.py rename to sbackup/compression.py index 850dfd3..f215a22 100644 --- a/sbackup/_compression.py +++ b/sbackup/compression.py @@ -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} diff --git a/sbackup/config.py b/sbackup/config.py new file mode 100644 index 0000000..fb06d82 --- /dev/null +++ b/sbackup/config.py @@ -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) \ No newline at end of file diff --git a/sbackup/i18n.py b/sbackup/i18n.py index 1538a97..a817548 100644 --- a/sbackup/i18n.py +++ b/sbackup/i18n.py @@ -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: """ diff --git a/sbackup/locales/en_US.json b/sbackup/locales/en_US.json index 81e3e0b..4df4549 100644 --- a/sbackup/locales/en_US.json +++ b/sbackup/locales/en_US.json @@ -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", diff --git a/sbackup/locales/zh_CN.json b/sbackup/locales/zh_CN.json index b234d52..9e49d13 100644 --- a/sbackup/locales/zh_CN.json +++ b/sbackup/locales/zh_CN.json @@ -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": "源文件夹路径", diff --git a/tests/sbackup/test_auto_save.py b/tests/sbackup/test_auto_save.py index efd83be..08b6e8e 100644 --- a/tests/sbackup/test_auto_save.py +++ b/tests/sbackup/test_auto_save.py @@ -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() \ No newline at end of file diff --git a/tests/sbackup/test_compression.py b/tests/sbackup/test_compression.py index 2f58620..6dd80c3 100644 --- a/tests/sbackup/test_compression.py +++ b/tests/sbackup/test_compression.py @@ -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() \ No newline at end of file diff --git a/tests/sbackup/test_config.py b/tests/sbackup/test_config.py index f28fae8..267fa73 100644 --- a/tests/sbackup/test_config.py +++ b/tests/sbackup/test_config.py @@ -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() \ No newline at end of file diff --git a/tests/sbackup/test_i18n.py b/tests/sbackup/test_i18n.py new file mode 100644 index 0000000..6e6bc8c --- /dev/null +++ b/tests/sbackup/test_i18n.py @@ -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() diff --git a/tests/sbackup/test_main.py b/tests/sbackup/test_main.py index 15ccf24..08e67fb 100644 --- a/tests/sbackup/test_main.py +++ b/tests/sbackup/test_main.py @@ -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() \ No newline at end of file