feat: 校验五档等级 (ultrafast/fast/normal/slow/full) + --check-only 纯校验模式

This commit is contained in:
AskaEth committed 2026-08-22 15:31:16 +08:00
1 parent 2e505f6ea3
commit fb59000fd1
3 files changed
+278 -10

No files matched your search

+23 -1
View File
@@ -60,7 +60,9 @@ help / clear / quit
| `--transfers` | 4 | rclone 并行传输数 | | `--transfers` | 4 | rclone 并行传输数 |
| `--resume` / `--no-resume` | 开 | rsync 断点续传 | | `--resume` / `--no-resume` | 开 | rsync 断点续传 |
| `--checksum` | 关 | rsync 校验模式 | | `--checksum` | 关 | rsync 校验模式 |
| `--verify` | 关 | 完成后校验(rsync -c dry-run, 0 差异=一致) | | `--verify` | 关 | 传输完成后校验(等级由 `--verify-level` 决定) |
| `--verify-level` | slow | 校验等级: ultrafast/fast/normal/slow/full |
| `--check-only` | 关 | 纯校验模式: 不传输, 仅按等级校验源/目标 |
| `--retry` | 0 | 失败自动补传轮数, **-1 = 无限重试** | | `--retry` | 0 | 失败自动补传轮数, **-1 = 无限重试** |
| `--smart-resume` | 关 | 智能续传: 校验目标最新N个文件, 损坏删除后从断点续传 | | `--smart-resume` | 关 | 智能续传: 校验目标最新N个文件, 损坏删除后从断点续传 |
| `--smart-check` | 5 | 智能续传校验文件数 (<=100MB 全校验, 更大只查大小) | | `--smart-check` | 5 | 智能续传校验文件数 (<=100MB 全校验, 更大只查大小) |
@@ -91,6 +93,26 @@ help / clear / quit
校验过程有**进度条**(总进度条复用)+ **单文件日志**(`校验 i/N: 文件名 ✓/✘ 损坏`), 校验过程有**进度条**(总进度条复用)+ **单文件日志**(`校验 i/N: 文件名 ✓/✘ 损坏`),
N 调大(如 50)也能实时看到校验进展。 N 调大(如 50)也能实时看到校验进展。
### 校验等级(`--verify-level`)
| 等级 | 校验内容 | 188GB 耗时 | 适用 |
|---|---|---|---|
| `ultrafast` | 仅文件大小 | 秒级 | 快速健康检查 |
| `fast` | 大小+mtime | ~1 分钟 | 常规检查 |
| `normal` | 大小+mtime+最新N个md5 | 分钟级 | 兼顾速度与深度 |
| `slow` | 全部文件 md5 | 40-60 分钟 | 全量可靠(默认) |
| `full` | 全部md5+文件数对比 | 最慢 | 终极确认 |
**纯校验模式(`--check-only`)**: 不传输,只按等级校验源/目标一致性,
用于传输中断后快速确认已传数据是否完好:
```bash
# 快速检查已传数据(秒级)
python3 transfer.py --mode cli --check-only --verify-level ultrafast /src/ /dst/
# 全量校验已传数据(全部 md5)
python3 transfer.py --mode cli --check-only --verify-level slow /src/ /dst/
```
> rsync 原子写(临时文件+rename)保证掉卡不会产生半截损坏的正式文件, > rsync 原子写(临时文件+rename)保证掉卡不会产生半截损坏的正式文件,
> 残留临时文件由补传后的 `--verify` 流程发现处理。 > 残留临时文件由补传后的 `--verify` 流程发现处理。
+63
View File
@@ -461,3 +461,66 @@ def test_verify_recent_callback_without_progress(tmp_path):
(src / "a.txt").write_text("x") (src / "a.txt").write_text("x")
(dst / "a.txt").write_text("x") (dst / "a.txt").write_text("x")
assert transfer.verify_recent_files(str(src), str(dst)) == [] assert transfer.verify_recent_files(str(src), str(dst)) == []
# ---------- 校验进度 (读取量轮询) ----------
def test_read_proc_io_current_process():
"""读取 /proc/self/io 的 read_bytes"""
val = transfer.read_proc_io(os.getpid())
assert isinstance(val, int)
assert val >= 0
def test_read_proc_io_nonexistent_pid():
assert transfer.read_proc_io(99999999) == 0
def test_collect_process_tree_self():
"""自身进程无子进程时 = 自身 read_bytes"""
me = transfer.collect_process_tree_read_bytes(os.getpid())
assert isinstance(me, int) and me >= 0
def test_collect_process_tree_with_child(tmp_path):
"""进程树统计不报错, 返回合理值 (缓存命中时 read_bytes 可能为 0)"""
f = tmp_path / "data.bin"
f.write_bytes(b"x" * (4 * 1024 * 1024))
child = subprocess.Popen(["cat", str(f)], stdout=subprocess.DEVNULL,
start_new_session=True)
time.sleep(0.5)
val = transfer.collect_process_tree_read_bytes(child.pid)
child.kill(); child.wait()
assert isinstance(val, int) and val >= 0
# ---------- 校验等级 + 纯校验模式 ----------
def test_parse_check_only():
args = transfer.parse_args(["--check-only", "s", "d"])
assert args.check_only is True
def test_parse_verify_level_default():
assert transfer.parse_args(["s", "d"]).verify_level == "slow"
assert transfer.parse_args(["--check-only", "s", "d"]).verify_level == "slow"
def test_parse_verify_level_values():
for lv in ("ultrafast", "fast", "normal", "slow", "full"):
args = transfer.parse_args(["--verify-level", lv, "s", "d"])
assert args.verify_level == lv
def test_size_check_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.size_check(str(src), str(dst)) == (0, 0)
def test_size_check_detect_diff_and_missing(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" * 999) # 大小不同
(src / "b.bin").write_bytes(b"y" * 100)
# b.bin 目标缺失
diff, miss = transfer.size_check(str(src), str(dst))
assert diff == 1 and miss == 1
def test_verify_level_order():
"""等级从快到慢的顺序固定"""
assert transfer.VERIFY_LEVELS == ["ultrafast", "fast", "normal", "slow", "full"]
+192 -9
View File
@@ -142,6 +142,55 @@ def verify_recent_files(src_dir: str, dst_dir: str, check_count: int = 5,
return bad return bad
VERIFY_LEVELS = ["ultrafast", "fast", "normal", "slow", "full"]
def size_check(src_dir: str, dst_dir: str) -> tuple[int, int]:
"""超快速校验: 仅比文件大小, 返回 (大小不一致数, 目标缺失数)"""
diff = miss = 0
for root, _dirs, files in os.walk(src_dir):
for name in files:
sp = os.path.join(root, name)
rel = os.path.relpath(sp, src_dir)
tp = os.path.join(dst_dir, rel)
if not os.path.isfile(tp):
miss += 1
elif os.path.getsize(sp) != os.path.getsize(tp):
diff += 1
return diff, miss
def read_proc_io(pid: int) -> int:
"""读取进程累计读字节数 (/proc/pid/io read_bytes)"""
try:
with open(f"/proc/{pid}/io", encoding="utf-8") as fh:
for line in fh:
if line.startswith("read_bytes"):
return int(line.split()[1])
except (OSError, ValueError):
return 0
return 0
def collect_process_tree_read_bytes(root_pid: int) -> int:
"""统计进程树(含子进程)累计读取字节 (校验进度用, 避免统计无关进程)"""
pids: set[int] = {root_pid}
changed = True
while changed:
changed = False
for pid in list(pids):
try:
with open(f"/proc/{pid}/task/{pid}/children", encoding="utf-8") as fh:
for child in fh.read().split():
c = int(child)
if c not in pids:
pids.add(c)
changed = True
except (OSError, ValueError):
pass
return sum(read_proc_io(pid) for pid in pids)
def speed_stats(history) -> tuple[float, float, float]: def speed_stats(history) -> tuple[float, float, float]:
"""速度统计: (当前, 峰值, 平均) 字节/秒""" """速度统计: (当前, 峰值, 平均) 字节/秒"""
if not history: if not history:
@@ -196,6 +245,8 @@ class TransferConfig:
retry: int = 0 retry: int = 0
smart_resume: bool = False smart_resume: bool = False
smart_check: int = 5 smart_check: int = 5
verify_level: str = "slow"
check_only: bool = False
# ---------------------------------------------------------------- 输出解析器 # ---------------------------------------------------------------- 输出解析器
@@ -504,7 +555,11 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument("--checksum", action="store_true", parser.add_argument("--checksum", action="store_true",
help="rsync 校验模式 (-c)") help="rsync 校验模式 (-c)")
parser.add_argument("--verify", action="store_true", parser.add_argument("--verify", action="store_true",
help="传输完成后校验 (-c)") help="传输完成后校验 (等级由 --verify-level 决定)")
parser.add_argument("--verify-level", choices=VERIFY_LEVELS, default="slow",
help=f"校验等级: {VERIFY_LEVELS} (默认 slow 全量md5)")
parser.add_argument("--check-only", action="store_true",
help="纯校验模式: 不传输, 仅按等级校验源/目标")
parser.add_argument("--retry", type=int, default=0, parser.add_argument("--retry", type=int, default=0,
help="失败自动补传轮数 (-1=无限, 默认 0 不重试)") help="失败自动补传轮数 (-1=无限, 默认 0 不重试)")
parser.add_argument("--smart-resume", action="store_true", parser.add_argument("--smart-resume", action="store_true",
@@ -658,6 +713,13 @@ class TaskRunner:
# -- 执行 # -- 执行
def run(self) -> int: def run(self) -> int:
"""执行传输: 失败自动补传(retry) + 完成后校验(verify)""" """执行传输: 失败自动补传(retry) + 完成后校验(verify)"""
# 纯校验模式: 不传输, 直接按等级校验
if self.cfg.check_only:
self.lm.open_session(f"check_{os.path.basename(self.cfg.source.rstrip('/'))}")
self.lm.write(f"cmd: check-only level={self.cfg.verify_level}")
self.lm.write(f"mode: {self.cfg.source} -> {self.cfg.dest}")
self._verify()
return 0
rc = 0 rc = 0
attempt = 0 attempt = 0
retry_label = "∞" if self.cfg.retry < 0 else str(self.cfg.retry) retry_label = "∞" if self.cfg.retry < 0 else str(self.cfg.retry)
@@ -784,17 +846,71 @@ class TaskRunner:
f"({self.stats.pct:.0f}%) {human_speed(self.stats.speed_bps)}") f"({self.stats.pct:.0f}%) {human_speed(self.stats.speed_bps)}")
def _verify(self) -> None: def _verify(self) -> None:
"""完成后校验: rsync -c --dry-run, 统计差异""" """按等级校验源/目标一致性"""
self.lm.write("verify: 开始校验 (-c dry-run)") level = self.cfg.verify_level
self.lm.write(f"verify: 开始校验 level={level}")
if self.on_log: if self.on_log:
self.on_log("[cyan]校验中 (rsync -c --dry-run)...[/]") self.on_log(f"[cyan]校验中 (level={level})...[/]")
cmd = RsyncBackend.build_verify_cmd(self.cfg.source, self.cfg.dest, if level == "ultrafast":
self.cfg.bwlimit) self._verify_size_only()
return
if level == "normal":
self._verify_fast()
# normal: 追加最新 N 个文件 md5
src_dir = self.cfg.source.rstrip("/")
dst_dir = self.cfg.dest.rstrip("/")
n = self.cfg.smart_check or 20
self.lm.write(f"verify(normal): 追加校验最新 {n} 个文件 md5")
bad = verify_recent_files(src_dir, dst_dir, n)
if bad:
self.lm.write(f"verify(normal): {len(bad)} 个文件损坏: {bad}")
if self.on_log:
if bad:
self.on_log(f"[red]校验发现 {len(bad)} 个文件损坏[/]")
else:
self.on_log("[green]✔ 最新文件 md5 校验通过[/]")
return
if level == "full":
self._verify_slow()
# full: 追加文件数对比
src_dir = self.cfg.source.rstrip("/")
dst_dir = self.cfg.dest.rstrip("/")
ns, nd = count_files(src_dir), count_files(dst_dir)
self.lm.write(f"verify(full): 文件数 源={ns} 目标={nd}")
if self.on_log:
if ns == nd:
self.on_log("[green]✔ 文件数一致[/]")
else:
self.on_log(f"[red]✘ 文件数不一致: 源 {ns} vs 目标 {nd}[/]")
return
# fast / slow 共用 rsync --dry-run (fast 无 -c, slow 有 -c)
self._verify_rsync(checksum=(level == "slow"))
def _verify_size_only(self) -> None:
"""ultrafast: 仅比大小"""
src_dir = self.cfg.source.rstrip("/")
dst_dir = self.cfg.dest.rstrip("/")
diff, miss = size_check(src_dir, dst_dir)
self.lm.write(f"verify(ultrafast): 大小不一致={diff} 缺失={miss}")
if self.on_log:
if diff == 0 and miss == 0:
self.on_log("[green]✔ 校验通过: 大小全部一致[/]")
else:
self.on_log(f"[red]✘ 大小不一致 {diff}, 缺失 {miss}[/]")
def _verify_fast(self) -> None:
"""fast: rsync --dry-run 无 -c (size+mtime)"""
src_dir = self.cfg.source.rstrip("/")
dst_dir = self.cfg.dest.rstrip("/")
cmd = ["rsync", "-a", "--dry-run", "--out-format=%i|%n"]
if self.cfg.bwlimit:
cmd += [f"--bwlimit={self.cfg.bwlimit}"]
cmd += [src_dir + "/", dst_dir + "/"]
try: try:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True) stderr=subprocess.STDOUT, text=True)
except OSError as exc: except OSError as exc:
self.lm.write_error(f"verify: 启动失败 {exc}") self.lm.write_error(f"verify(fast): 启动失败 {exc}")
return return
assert proc.stdout is not None assert proc.stdout is not None
diffs = 0 diffs = 0
@@ -805,6 +921,64 @@ class TaskRunner:
diffs += 1 diffs += 1
self.lm.write(f"verify diff: {line}") self.lm.write(f"verify diff: {line}")
rc = proc.wait() rc = proc.wait()
self.lm.write(f"verify(fast): 完成 rc={rc} 差异={diffs}")
if self.on_log:
if diffs == 0 and rc == 0:
self.on_log("[green]✔ 校验通过: 无差异 (size+mtime)[/]")
else:
self.on_log(f"[red]✘ 发现 {diffs} 个差异[/]")
def _verify_rsync(self, checksum: bool) -> None:
"""slow(rsync -c)/fast 共用: dry-run 差异统计 + 进度轮询"""
self.lm.write(f"verify: 开始 ({'checksum' if checksum else 'size+mtime'})")
if self.on_log:
self.on_log("[cyan]校验中 (rsync -c --dry-run)...[/]")
cmd = RsyncBackend.build_verify_cmd(self.cfg.source, self.cfg.dest,
self.cfg.bwlimit)
try:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True)
except OSError as exc:
self.lm.write_error(f"verify: 启动失败 {exc}")
return
# 进度轮询: 校验读取量 = 源+目标各读一遍 (2x), 按 /proc/io 统计
total_read = self.total_bytes * 2 if self.total_bytes else 0
stop_poll = threading.Event()
def poll_verify() -> None:
last = 0
last_t = time.time()
while not stop_poll.is_set():
time.sleep(1)
done = collect_process_tree_read_bytes(proc.pid)
now = time.time()
speed = max(0, (done - last) / max(0.1, now - last_t))
last, last_t = done, now
self.stats.bytes_done = done
self.stats.bytes_total = total_read or done
if total_read:
self.stats.pct = min(100.0, done / total_read * 100)
self.stats.speed_bps = speed
self._emit_stats()
if self.on_log:
label = (f"校验 {self.stats.pct:.0f}% 已读 "
f"{human_size(done)}/{human_size(total_read)} "
f"{human_speed(speed)}") if total_read else f"校验中 已读 {human_size(done)} {human_speed(speed)}"
self.on_log(label)
poll_thread = threading.Thread(target=poll_verify, daemon=True)
poll_thread.start()
assert proc.stdout is not None
diffs = 0
for line in proc.stdout:
line = line.rstrip()
if not line or line.startswith(("sending", "sent ", "total ")):
continue
diffs += 1
self.lm.write(f"verify diff: {line}")
rc = proc.wait()
stop_poll.set()
poll_thread.join(timeout=2)
self.lm.write(f"verify: 完成 rc={rc} 差异文件={diffs}") self.lm.write(f"verify: 完成 rc={rc} 差异文件={diffs}")
if self.on_log: if self.on_log:
if diffs == 0 and rc == 0: if diffs == 0 and rc == 0:
@@ -812,6 +986,10 @@ class TaskRunner:
else: else:
self.on_log(f"[red]✘ 校验发现 {diffs} 个差异 (rc={rc})[/]") self.on_log(f"[red]✘ 校验发现 {diffs} 个差异 (rc={rc})[/]")
def _verify_slow(self) -> None:
"""slow: rsync -c --dry-run 全量 md5 + 进度轮询"""
self._verify_rsync(checksum=True)
def pause(self) -> None: def pause(self) -> None:
"""暂停: SIGSTOP 整个进程组""" """暂停: SIGSTOP 整个进程组"""
if self._paused: if self._paused:
@@ -1109,6 +1287,7 @@ class TransferApp(App):
resume=args.resume, checksum=args.checksum, verify=args.verify, resume=args.resume, checksum=args.checksum, verify=args.verify,
logdir=self.logdir, policy=args.policy, retry=args.retry, logdir=self.logdir, policy=args.policy, retry=args.retry,
smart_resume=args.smart_resume, smart_check=args.smart_check, smart_resume=args.smart_resume, smart_check=args.smart_check,
verify_level=args.verify_level, check_only=args.check_only,
) )
if cfg.policy == "ask" and cfg.backend == "rsync": if cfg.policy == "ask" and cfg.backend == "rsync":
self._flash("[yellow]预扫描冲突...[/]") self._flash("[yellow]预扫描冲突...[/]")
@@ -1275,7 +1454,9 @@ def main() -> int:
logdir=args.logdir, policy=args.policy, logdir=args.logdir, policy=args.policy,
retry=args.retry, retry=args.retry,
smart_resume=args.smart_resume, smart_resume=args.smart_resume,
smart_check=args.smart_check) smart_check=args.smart_check,
verify_level=args.verify_level,
check_only=args.check_only)
TransferApp(preset=cfg, logdir=args.logdir, keep_logs=args.keep_logs).run() TransferApp(preset=cfg, logdir=args.logdir, keep_logs=args.keep_logs).run()
return 0 return 0
cfg = TransferConfig(source=args.source, dest=args.dest, cfg = TransferConfig(source=args.source, dest=args.dest,
@@ -1285,7 +1466,9 @@ def main() -> int:
logdir=args.logdir, policy=args.policy, logdir=args.logdir, policy=args.policy,
retry=args.retry, retry=args.retry,
smart_resume=args.smart_resume, smart_resume=args.smart_resume,
smart_check=args.smart_check) smart_check=args.smart_check,
verify_level=args.verify_level,
check_only=args.check_only)
return run_cli(cfg, args) return run_cli(cfg, args)