diff --git a/README.md b/README.md index 3720c6f..d0a064b 100644 --- a/README.md +++ b/README.md @@ -60,7 +60,9 @@ help / clear / quit | `--transfers` | 4 | rclone 并行传输数 | | `--resume` / `--no-resume` | 开 | 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 = 无限重试** | | `--smart-resume` | 关 | 智能续传: 校验目标最新N个文件, 损坏删除后从断点续传 | | `--smart-check` | 5 | 智能续传校验文件数 (<=100MB 全校验, 更大只查大小) | @@ -91,6 +93,26 @@ help / clear / quit 校验过程有**进度条**(总进度条复用)+ **单文件日志**(`校验 i/N: 文件名 ✓/✘ 损坏`), 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)保证掉卡不会产生半截损坏的正式文件, > 残留临时文件由补传后的 `--verify` 流程发现处理。 diff --git a/tests/test_transfer.py b/tests/test_transfer.py index 6185058..599bf6c 100644 --- a/tests/test_transfer.py +++ b/tests/test_transfer.py @@ -461,3 +461,66 @@ def test_verify_recent_callback_without_progress(tmp_path): (src / "a.txt").write_text("x") (dst / "a.txt").write_text("x") 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"] diff --git a/transfer.py b/transfer.py index 571bbb6..32428b1 100644 --- a/transfer.py +++ b/transfer.py @@ -142,6 +142,55 @@ def verify_recent_files(src_dir: str, dst_dir: str, check_count: int = 5, 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]: """速度统计: (当前, 峰值, 平均) 字节/秒""" if not history: @@ -196,6 +245,8 @@ class TransferConfig: retry: int = 0 smart_resume: bool = False 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", help="rsync 校验模式 (-c)") 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, help="失败自动补传轮数 (-1=无限, 默认 0 不重试)") parser.add_argument("--smart-resume", action="store_true", @@ -658,6 +713,13 @@ class TaskRunner: # -- 执行 def run(self) -> int: """执行传输: 失败自动补传(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 attempt = 0 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)}") 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: - self.on_log("[cyan]校验中 (rsync -c --dry-run)...[/]") - cmd = RsyncBackend.build_verify_cmd(self.cfg.source, self.cfg.dest, - self.cfg.bwlimit) + self.on_log(f"[cyan]校验中 (level={level})...[/]") + if level == "ultrafast": + 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: proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) except OSError as exc: - self.lm.write_error(f"verify: 启动失败 {exc}") + self.lm.write_error(f"verify(fast): 启动失败 {exc}") return assert proc.stdout is not None diffs = 0 @@ -805,6 +921,64 @@ class TaskRunner: diffs += 1 self.lm.write(f"verify diff: {line}") 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}") if self.on_log: if diffs == 0 and rc == 0: @@ -812,6 +986,10 @@ class TaskRunner: else: 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: """暂停: SIGSTOP 整个进程组""" if self._paused: @@ -1109,6 +1287,7 @@ class TransferApp(App): 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, + verify_level=args.verify_level, check_only=args.check_only, ) if cfg.policy == "ask" and cfg.backend == "rsync": self._flash("[yellow]预扫描冲突...[/]") @@ -1275,7 +1454,9 @@ def main() -> int: logdir=args.logdir, policy=args.policy, retry=args.retry, 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() return 0 cfg = TransferConfig(source=args.source, dest=args.dest, @@ -1285,7 +1466,9 @@ def main() -> int: logdir=args.logdir, policy=args.policy, retry=args.retry, 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)