feat: --smart-resume 智能断点续传 (校验最新N个文件, 损坏即重传)
This commit is contained in:
1 parent
15ed4330ae
commit
0f9866a031
3 files changed
+142
-2
No files matched your search
@@ -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` 流程发现处理。
|
||||
|
||||
|
||||
@@ -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
|
||||
+78
-2
@@ -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)
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user