From 0f9866a0314b35202419fab915da8af6958d356d Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 22 Aug 2026 14:51:09 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20--smart-resume=20=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E6=96=AD=E7=82=B9=E7=BB=AD=E4=BC=A0=20(=E6=A0=A1=E9=AA=8C?= =?UTF-8?q?=E6=9C=80=E6=96=B0N=E4=B8=AA=E6=96=87=E4=BB=B6,=20=E6=8D=9F?= =?UTF-8?q?=E5=9D=8F=E5=8D=B3=E9=87=8D=E4=BC=A0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 6 ++++ tests/test_transfer.py | 58 ++++++++++++++++++++++++++++++ transfer.py | 80 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 142 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 3ef5e18..9ddb5f5 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,8 @@ help / clear / quit | `--checksum` | 关 | rsync 校验模式 | | `--verify` | 关 | 完成后校验（rsync -c dry-run, 0 差异=一致） | | `--retry` | 0 | 失败自动补传轮数, **-1 = 无限重试** | +| `--smart-resume` | 关 | 智能续传: 校验目标最新N个文件, 损坏删除后从断点续传 | +| `--smart-check` | 5 | 智能续传校验文件数 (<=100MB 全校验, 更大只查大小) | | `--policy` | overwrite | 冲突策略: overwrite / skip / ask | | `--mode` | tui | cli / tui | | `--keep-logs` | 20 | 日志滚动保留数 | @@ -83,6 +85,10 @@ help / clear / quit 3. 掉卡时: 守护恢复设备 → TransferSu 自动重跑补传（`--partial` 只补缺失）→ 直到成功 → `rsync -c` 全量校验（0 差异 = 源目标完全一致） +**智能续传（`--smart-resume`）**: 中断后重跑时，只校验目标**最新 N 个文件** +（`--smart-check N`，默认 5；≤100MB 做 md5 全校验，更大只查大小）， +发现损坏立即删除该文件并从断点续传——避免对已传的 100+GB 全量重扫。 + > rsync 原子写（临时文件+rename）保证掉卡不会产生半截损坏的正式文件， > 残留临时文件由补传后的 `--verify` 流程发现处理。 diff --git a/tests/test_transfer.py b/tests/test_transfer.py index 3551aa9..00de157 100644 --- a/tests/test_transfer.py +++ b/tests/test_transfer.py @@ -370,3 +370,61 @@ def test_count_files_only_files(tmp_path): def test_parse_args_retry_infinite(): args = transfer.parse_args(["--retry", "-1", "s", "d"]) assert args.retry == -1 # 无限重试 + +# ---------- 智能断点续传 (校验最新N个文件) ---------- + +def test_verify_recent_all_ok(tmp_path): + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + (src / "a.bin").write_bytes(b"x" * 1000) + (dst / "a.bin").write_bytes(b"x" * 1000) + assert transfer.verify_recent_files(str(src), str(dst)) == [] + +def test_verify_recent_detects_corrupt_and_deletes(tmp_path): + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + (src / "a.bin").write_bytes(b"hello world") + (dst / "a.bin").write_bytes(b"hello worlX") # 内容损坏 + bad = transfer.verify_recent_files(str(src), str(dst)) + assert bad == ["a.bin"] + assert not (dst / "a.bin").exists() # 坏副本已删除 + +def test_verify_recent_empty_dst_no_check(tmp_path): + """目标无已传文件时无可校验 (缺失文件由 rsync 正常补传)""" + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + (src / "a.bin").write_bytes(b"data") + assert transfer.verify_recent_files(str(src), str(dst)) == [] + +def test_verify_recent_large_file_checks_size_only(tmp_path): + """>100MB 只查大小, 不做内容 hash""" + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + big = b"a" * (101 * 1024 * 1024) # 101MB + (src / "big.bin").write_bytes(big) + (dst / "big.bin").write_bytes(b"b" * len(big)) # 同大小不同内容 + assert transfer.verify_recent_files(str(src), str(dst)) == [] # 只查大小, 不报 + +def test_verify_recent_count_limit(tmp_path): + """只校验最新 N 个, 更早的损坏不报""" + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + for i in range(3): + (src / f"f{i}.txt").write_text("ok" * 100) + (dst / f"f{i}.txt").write_text("ok" * 100) + import time + # 让 f0 是最早的 + old = time.time() - 100 + os.utime(str(dst / "f0.txt"), (old, old)) + os.utime(str(src / "f0.txt"), (old, old)) + # 损坏最早的文件 (不在最新2个内) + (src / "f0.txt").write_text("changed") + (dst / "f0.txt").write_text("changed-too") + bad = transfer.verify_recent_files(str(src), str(dst), check_count=2) + assert "f0.txt" not in bad # 只查最新2个 + +def test_parse_smart_resume(): + assert transfer.parse_args(["s", "d"]).smart_resume is False + assert transfer.parse_args(["--smart-resume", "s", "d"]).smart_resume is True + assert transfer.parse_args(["--smart-check", "3", "s", "d"]).smart_check == 3 + assert transfer.parse_args(["s", "d"]).smart_check == 5 diff --git a/transfer.py b/transfer.py index abe9a7d..00206dc 100644 --- a/transfer.py +++ b/transfer.py @@ -22,6 +22,7 @@ import tempfile from dataclasses import dataclass, field from datetime import datetime from glob import glob +import hashlib # ---------------------------------------------------------------- 工具函数 @@ -80,6 +81,54 @@ def count_files(path: str) -> int: return 0 +def files_identical(path_a: str, path_b: str, chunk: int = 1 << 20) -> bool: + """分块 md5 对比 (大文件内存安全)""" + hash_a, hash_b = hashlib.md5(), hashlib.md5() + with open(path_a, "rb") as fa, open(path_b, "rb") as fb: + while True: + ca, cb = fa.read(chunk), fb.read(chunk) + if not ca and not cb: + break + if ca != cb: + return False + hash_a.update(ca) + hash_b.update(cb) + return hash_a.digest() == hash_b.digest() + + +def verify_recent_files(src_dir: str, dst_dir: str, check_count: int = 5, + max_size: int = 100 * 1024 * 1024) -> list[str]: + """智能续传: 校验目标最新 N 个文件与源一致性, 返回损坏/缺失列表并删除目标坏副本 + + - <=max_size 的文件: 大小 + 内容 hash 全校验 + - >max_size 的文件: 只查大小 (内容由后续 rsync --append 兜底) + - 返回需重传的相对路径列表 (目标坏副本已删除) + """ + bad: list[str] = [] + if check_count <= 0 or not os.path.isdir(dst_dir): + return bad + recent: list[tuple[float, str]] = [] + for root, _dirs, files in os.walk(dst_dir): + for name in files: + full = os.path.join(root, name) + recent.append((os.path.getmtime(full), os.path.relpath(full, dst_dir))) + recent.sort(reverse=True) + for _mtime, rel in recent[:check_count]: + sp = os.path.join(src_dir, rel) + tp = os.path.join(dst_dir, rel) + if not os.path.isfile(sp): + continue # 源无此文件(目标多余), 不处理 + if not os.path.isfile(tp) or os.path.getsize(sp) != os.path.getsize(tp): + bad.append(rel) + if os.path.exists(tp): + os.remove(tp) + continue + if os.path.getsize(tp) <= max_size and not files_identical(sp, tp): + bad.append(rel) + os.remove(tp) + return bad + + def speed_stats(history) -> tuple[float, float, float]: """速度统计: (当前, 峰值, 平均) 字节/秒""" if not history: @@ -132,6 +181,8 @@ class TransferConfig: logdir: str = "" policy: str = "overwrite" retry: int = 0 + smart_resume: bool = False + smart_check: int = 5 # ---------------------------------------------------------------- 输出解析器 @@ -443,6 +494,10 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: help="传输完成后校验 (-c)") parser.add_argument("--retry", type=int, default=0, help="失败自动补传轮数 (-1=无限, 默认 0 不重试)") + parser.add_argument("--smart-resume", action="store_true", + help="智能续传: 校验目标最新N个文件, 损坏删除后从问题文件续传") + parser.add_argument("--smart-check", type=int, default=5, metavar="N", + help="智能续传校验文件数 (默认 5, <=100MB 全校验)") parser.add_argument("--mode", choices=["cli", "tui"], default=None, help="运行模式 (默认: 有任务=cli, 无任务=tui)") parser.add_argument("--keep-logs", type=int, default=20, @@ -626,6 +681,22 @@ class TaskRunner: self.stats.files_total = self.total_files self.lm.write(f"source size: {human_size(self.total_bytes)}, " f"files: {self.total_files}") + # 智能续传: 校验目标最新 N 个文件, 坏文件删除后由 rsync 重传 + if self.cfg.smart_resume and not self._stop.is_set(): + src_dir = self.cfg.source.rstrip("/") + dst_dir = self.cfg.dest.rstrip("/") + self.lm.write(f"smart-resume: 校验目标最新 {self.cfg.smart_check} 个文件...") + if self.on_log: + self.on_log(f"[cyan]智能续传: 校验最近 {self.cfg.smart_check} 个文件...[/]") + bad = verify_recent_files(src_dir, dst_dir, self.cfg.smart_check) + if bad: + self.lm.write(f"smart-resume: 发现 {len(bad)} 个损坏/缺失, 将重传: {bad}") + if self.on_log: + self.on_log(f"[yellow]发现 {len(bad)} 个损坏文件, 从断点重传: " + f"{', '.join(os.path.basename(b) for b in bad[:5])}[/]") + else: + if self.on_log: + self.on_log("[green]智能续传: 最近文件校验通过[/]") # stdout 丢弃 (进度帧缓冲不可靠), stderr 单独读错误 self.proc = subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, @@ -1013,6 +1084,7 @@ class TransferApp(App): bwlimit=args.bwlimit, transfers=args.transfers, resume=args.resume, checksum=args.checksum, verify=args.verify, logdir=self.logdir, policy=args.policy, retry=args.retry, + smart_resume=args.smart_resume, smart_check=args.smart_check, ) if cfg.policy == "ask" and cfg.backend == "rsync": self._flash("[yellow]预扫描冲突...[/]") @@ -1177,7 +1249,9 @@ def main() -> int: transfers=args.transfers, resume=args.resume, checksum=args.checksum, verify=args.verify, logdir=args.logdir, policy=args.policy, - retry=args.retry) + retry=args.retry, + smart_resume=args.smart_resume, + smart_check=args.smart_check) TransferApp(preset=cfg, logdir=args.logdir, keep_logs=args.keep_logs).run() return 0 cfg = TransferConfig(source=args.source, dest=args.dest, @@ -1185,7 +1259,9 @@ def main() -> int: transfers=args.transfers, resume=args.resume, checksum=args.checksum, verify=args.verify, logdir=args.logdir, policy=args.policy, - retry=args.retry) + retry=args.retry, + smart_resume=args.smart_resume, + smart_check=args.smart_check) return run_cli(cfg, args)