diff --git a/README.md b/README.md index cebe172..59454b6 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ help / clear / quit | `--verify` | 关 | 传输完成后校验（等级由 `--verify-level` 决定） | | `--verify-level` | slow | 校验等级: ultrafast/fast/normal/slow/full | | `--check-only` | 关 | 纯校验模式: 不传输, 仅按等级校验源/目标 | +| `--repair` | 关 | 校验发现问题后提示并重传问题文件（先删目标避免增量盲区） | | `--retry` | 0 | 失败自动补传轮数, **-1 = 无限重试** | | `--smart-resume` | 关 | 智能续传: 校验目标最新N个文件, 损坏删除后从断点续传 | | `--smart-check` | 5 | 智能续传校验文件数 (<=100MB 全校验, 更大只查大小) | @@ -118,8 +119,14 @@ N 调大（如 50）也能实时看到校验进展。 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/ +# 校验发现问题自动提示重传（--repair） +python3 transfer.py --mode cli --check-only --verify-level slow --repair /src/ /dst/ ``` +> `--repair`: 校验发现问题文件后提示 [y/N]，确认后用 rsync 重传问题文件。 +> 重传前**先删除目标问题文件**——避免 rsync 按 size+mtime 判定相同而跳过 +> （内容损坏但大小/时间戳一致的场景，增量检测有盲区）。 + > rsync 原子写（临时文件+rename）保证掉卡不会产生半截损坏的正式文件， > 残留临时文件由补传后的 `--verify` 流程发现处理。 diff --git a/tests/test_transfer.py b/tests/test_transfer.py index 599bf6c..363dc8d 100644 --- a/tests/test_transfer.py +++ b/tests/test_transfer.py @@ -509,7 +509,7 @@ def test_size_check_all_ok(tmp_path): 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) + assert transfer.size_check(str(src), str(dst)) == ([], []) def test_size_check_detect_diff_and_missing(tmp_path): src = tmp_path / "s"; dst = tmp_path / "d" @@ -518,9 +518,78 @@ def test_size_check_detect_diff_and_missing(tmp_path): (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 + diffs, missing = transfer.size_check(str(src), str(dst)) + assert diffs == ["a.bin"] and missing == ["b.bin"] def test_verify_level_order(): """等级从快到慢的顺序固定""" assert transfer.VERIFY_LEVELS == ["ultrafast", "fast", "normal", "slow", "full"] + +# ---------- 各等级校验进度 ---------- + +def test_stat_check_detects_size_and_mtime(tmp_path): + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + (src / "ok.txt").write_text("good") + (dst / "ok.txt").write_text("good") + (src / "diff.txt").write_text("longer content here") + (dst / "diff.txt").write_text("short") + (src / "gone.txt").write_text("x") + # gone.txt 目标缺失 + diffs, missing = transfer.stat_check(str(src), str(dst)) + assert "diff.txt" in diffs # 大小不同 + assert "gone.txt" in missing # 缺失 + assert "ok.txt" not in diffs + +def test_stat_check_mtime_diff(tmp_path): + """mtime 不同但大小相同 → 记为差异""" + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + (src / "a.txt").write_text("same") + (dst / "a.txt").write_text("same") + import time as _t + old = _t.time() - 500 + os.utime(str(dst / "a.txt"), (old, old)) + diffs, _missing = transfer.stat_check(str(src), str(dst)) + assert "a.txt" in diffs + +def test_stat_check_progress_callback(tmp_path): + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + for i in range(50): + (src / f"f{i}.txt").write_text("x") + (dst / f"f{i}.txt").write_text("x") + events = [] + transfer.stat_check(str(src), str(dst), + on_progress=lambda i, n: events.append((i, n))) + assert len(events) > 0 # 有进度回调 + assert events[-1] == (50, 50) # 最终回调 + +def test_size_check_progress_callback(tmp_path): + src = tmp_path / "s"; dst = tmp_path / "d" + src.mkdir(); dst.mkdir() + for i in range(30): + (src / f"f{i}.txt").write_text("x") + (dst / f"f{i}.txt").write_text("x") + events = [] + transfer.size_check(str(src), str(dst), + on_progress=lambda i, n: events.append((i, n))) + assert events and events[-1][0] == events[-1][1] + +# ---------- 校验后修复 (--repair) ---------- + +def test_parse_repair(): + assert transfer.parse_args(["s", "d"]).repair is False + assert transfer.parse_args(["--repair", "s", "d"]).repair is True + +def test_build_repair_cmd(tmp_path): + list_file = str(tmp_path / "list.txt") + with open(list_file, "w") as f: + f.write("a.flac\nsub/b.lrc\n") + cmd, used = transfer.build_repair_cmd("src/", "dst/", list_file, "30000") + joined = " ".join(cmd) + assert cmd[0] == "rsync" + assert f"--files-from={list_file}" in joined + assert "--mkpath" in joined + assert "--bwlimit=30000" in joined + assert "src/" in joined and "dst/" in joined diff --git a/transfer.py b/transfer.py index 32428b1..f1b92e0 100644 --- a/transfer.py +++ b/transfer.py @@ -145,19 +145,68 @@ def verify_recent_files(src_dir: str, dst_dir: str, check_count: int = 5, 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 _walk_files(root_dir: str) -> list[str]: + """收集目录下所有文件完整路径""" + files: list[str] = [] + for root, _dirs, names in os.walk(root_dir): + for name in names: + files.append(os.path.join(root, name)) + return files + + +def size_check(src_dir: str, dst_dir: str, on_progress=None) -> tuple[list[str], list[str]]: + # 超快速校验: 仅比文件大小, 返回 (大小不一致文件, 目标缺失文件) 列表 + diffs: list[str] = [] + missing: list[str] = [] + files = _walk_files(src_dir) + total = len(files) + for i, sp in enumerate(files, 1): + rel = os.path.relpath(sp, src_dir) + tp = os.path.join(dst_dir, rel) + if not os.path.isfile(tp): + missing.append(rel) + elif os.path.getsize(sp) != os.path.getsize(tp): + diffs.append(rel) + if on_progress and (i % 200 == 0 or i == total): + on_progress(i, total) + return diffs, missing + + +def stat_check(src_dir: str, dst_dir: str, on_progress=None) -> tuple[list[str], list[str]]: + """fast 校验: 大小+mtime 对比, 返回 (差异文件, 缺失文件) 列表 + + on_progress(i, total): 进度回调 + """ + diffs: list[str] = [] + missing: list[str] = [] + files = _walk_files(src_dir) + total = len(files) + for i, sp in enumerate(files, 1): + rel = os.path.relpath(sp, src_dir) + tp = os.path.join(dst_dir, rel) + if not os.path.isfile(tp): + missing.append(rel) + else: + sst = os.stat(sp) + tst = os.stat(tp) + if (sst.st_size != tst.st_size + or int(sst.st_mtime) != int(tst.st_mtime)): + diffs.append(rel) + if on_progress and (i % 200 == 0 or i == total): + on_progress(i, total) + return diffs, missing + + +def build_repair_cmd(src: str, dst: str, list_file: str, + bwlimit: str = "") -> tuple[list[str], str]: + # 构建重传问题文件的 rsync 命令 (--files-from) + src = src.rstrip("/") + "/" + dst = dst.rstrip("/") + "/" + cmd = ["rsync", "-a", "--files-from=" + list_file, "--mkpath"] + if bwlimit: + cmd += [f"--bwlimit={bwlimit}"] + cmd += [src, dst] + return cmd, list_file def read_proc_io(pid: int) -> int: @@ -247,6 +296,7 @@ class TransferConfig: smart_check: int = 5 verify_level: str = "slow" check_only: bool = False + repair: bool = False # ---------------------------------------------------------------- 输出解析器 @@ -560,6 +610,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: help=f"校验等级: {VERIFY_LEVELS} (默认 slow 全量md5)") parser.add_argument("--check-only", action="store_true", help="纯校验模式: 不传输, 仅按等级校验源/目标") + parser.add_argument("--repair", action="store_true", + help="校验发现问题后提示并重传问题文件") parser.add_argument("--retry", type=int, default=0, help="失败自动补传轮数 (-1=无限, 默认 0 不重试)") parser.add_argument("--smart-resume", action="store_true", @@ -845,6 +897,73 @@ class TaskRunner: self.on_log(f"已传 {human_size(done)} / {human_size(self.total_bytes)} " f"({self.stats.pct:.0f}%) {human_speed(self.stats.speed_bps)}") + def _repair_if_requested(self, issues: list[str]) -> None: + """校验发现问题且开启 --repair: 询问后重传问题文件""" + if not issues or not self.cfg.repair or self._stop.is_set(): + return + self.lm.write(f"repair: 发现 {len(issues)} 个问题文件, 等待用户确认") + if self.on_error: + self.on_error(f"校验发现 {len(issues)} 个问题文件, 是否重新传输? [y/N]") + # CLI input 交互 (TUI 通过输入栏另行处理) + try: + answer = input(f"发现 {len(issues)} 个问题文件, 重新传输? [y/N]: ").strip().lower() + except EOFError: + answer = "n" + if answer not in ("y", "yes"): + self.lm.write("repair: 用户跳过修复") + if self.on_log: + self.on_log("[yellow]跳过问题文件修复[/]") + return + self._do_repair(issues) + + def _do_repair(self, issues: list[str]) -> None: + """用 rsync --files-from 重传问题文件""" + # 先删除目标问题文件: 避免 rsync 按 size+mtime 判定相同而跳过 + # (内容损坏但大小/mtime 一致的场景, 增量检测不出来) + dst_dir = self.cfg.dest.rstrip("/") + deleted = 0 + for rel in issues: + tp = os.path.join(dst_dir, rel) + if os.path.exists(tp): + try: + os.remove(tp) + deleted += 1 + except OSError as exc: + self.lm.write_error(f"repair: 删除 {rel} 失败: {exc}") + if deleted: + self.lm.write(f"repair: 已删除 {deleted} 个损坏目标文件") + list_path = os.path.join(tempfile.gettempdir(), "tsu_repair.txt") + with open(list_path, "w", encoding="utf-8") as fh: + fh.write("\n".join(issues) + "\n") + cmd, _ = build_repair_cmd(self.cfg.source, self.cfg.dest, + list_path, self.cfg.bwlimit) + self.lm.write(f"repair: 重传 {len(issues)} 个文件: {' '.join(cmd)}") + if self.on_log: + self.on_log(f"[cyan]重传 {len(issues)} 个问题文件...[/]") + try: + proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, text=True) + errs = "".join(proc.stderr.readlines()) if proc.stderr else "" + rc = proc.wait() + except OSError as exc: + self.lm.write_error(f"repair: 启动失败 {exc}") + return + self.lm.write(f"repair: 完成 rc={rc}") + if self.on_log: + if rc == 0: + self.on_log("[green]✔ 问题文件已重传[/]") + else: + self.on_log(f"[red]✘ 重传失败 rc={rc}: {errs.strip()[:200]}[/]") + + def _emit_verify_progress(self, i: int, total: int) -> None: + """校验进度: 复用 stats 驱动进度条 + 每 5% 打日志""" + self.stats.bytes_done = i + self.stats.bytes_total = total + self.stats.pct = i / max(1, total) * 100 + self._emit_stats() + if self.on_log and (i % max(1, total // 20) == 0 or i == total): + self.on_log(f"校验 {i}/{total} ({self.stats.pct:.0f}%)") + def _verify(self) -> None: """按等级校验源/目标一致性""" level = self.cfg.verify_level @@ -869,6 +988,7 @@ class TaskRunner: self.on_log(f"[red]校验发现 {len(bad)} 个文件损坏[/]") else: self.on_log("[green]✔ 最新文件 md5 校验通过[/]") + self._repair_if_requested(bad) return if level == "full": self._verify_slow() @@ -887,46 +1007,42 @@ class TaskRunner: self._verify_rsync(checksum=(level == "slow")) def _verify_size_only(self) -> None: - """ultrafast: 仅比大小""" + """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}") + + def on_progress(i: int, total: int) -> None: + self._emit_verify_progress(i, total) + + diffs, missing = size_check(src_dir, dst_dir, on_progress=on_progress) + issues = diffs + missing + self.lm.write(f"verify(ultrafast): 大小不一致={len(diffs)} 缺失={len(missing)}") if self.on_log: - if diff == 0 and miss == 0: + if not issues: self.on_log("[green]✔ 校验通过: 大小全部一致[/]") else: - self.on_log(f"[red]✘ 大小不一致 {diff}, 缺失 {miss}[/]") + self.on_log(f"[red]✘ 发现 {len(issues)} 个问题文件[/]") + self._repair_if_requested(issues) def _verify_fast(self) -> None: - """fast: rsync --dry-run 无 -c (size+mtime)""" + """fast: 大小+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(fast): 启动失败 {exc}") - return - 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() - self.lm.write(f"verify(fast): 完成 rc={rc} 差异={diffs}") + + def on_progress(i: int, total: int) -> None: + self._emit_verify_progress(i, total) + + diffs, missing = stat_check(src_dir, dst_dir, on_progress=on_progress) + issues = diffs + missing + for rel in issues: + self.lm.write(f"verify diff: {rel}") + self.lm.write(f"verify(fast): 完成 差异={len(diffs)} 缺失={len(missing)}") if self.on_log: - if diffs == 0 and rc == 0: + if not issues: self.on_log("[green]✔ 校验通过: 无差异 (size+mtime)[/]") else: - self.on_log(f"[red]✘ 发现 {diffs} 个差异[/]") + self.on_log(f"[red]✘ 发现 {len(issues)} 个差异/缺失[/]") + self._repair_if_requested(issues) def _verify_rsync(self, checksum: bool) -> None: """slow(rsync -c)/fast 共用: dry-run 差异统计 + 进度轮询""" @@ -970,16 +1086,21 @@ class TaskRunner: poll_thread.start() assert proc.stdout is not None diffs = 0 + issues: list[str] = [] for line in proc.stdout: line = line.rstrip() if not line or line.startswith(("sending", "sent ", "total ")): continue diffs += 1 + # out-format=%i|%n, 提取文件名 + if "|" in line: + issues.append(line.rsplit("|", 1)[-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._repair_if_requested(issues) if self.on_log: if diffs == 0 and rc == 0: self.on_log("[green]✔ 校验通过: 源与目标完全一致[/]") @@ -1091,6 +1212,10 @@ def run_cli(cfg: TransferConfig, args: argparse.Namespace) -> int: sys.stderr.write(f"\r\033[K[ERROR] {line}\n") sys.stderr.flush() + def on_log(line: str) -> None: + sys.stderr.write(f"\r\033[K{line}\n") + sys.stderr.flush() + # 冲突策略: ask 需要预扫描+交互; skip 由后端 --ignore-existing 处理 exclude_from = "" if args.policy == "ask" and cfg.backend == "rsync": @@ -1109,7 +1234,8 @@ def run_cli(cfg: TransferConfig, args: argparse.Namespace) -> int: print("[TransferSu] rclone 使用原生 --interactive 交互") runner = TaskRunner(cfg, lm, on_stats=on_stats, on_file=on_file, - on_error=on_error, exclude_from=exclude_from) + on_error=on_error, on_log=on_log, + exclude_from=exclude_from) print(f"[TransferSu] {cfg.backend}: {cfg.source} -> {cfg.dest}") print(f"[TransferSu] 策略: {args.policy} 日志: {lm.logdir} (保留 {args.keep_logs} 个)") try: @@ -1288,6 +1414,7 @@ class TransferApp(App): 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, + repair=args.repair, ) if cfg.policy == "ask" and cfg.backend == "rsync": self._flash("[yellow]预扫描冲突...[/]") @@ -1456,7 +1583,8 @@ def main() -> int: smart_resume=args.smart_resume, smart_check=args.smart_check, verify_level=args.verify_level, - check_only=args.check_only) + check_only=args.check_only, + repair=args.repair) TransferApp(preset=cfg, logdir=args.logdir, keep_logs=args.keep_logs).run() return 0 cfg = TransferConfig(source=args.source, dest=args.dest, @@ -1468,7 +1596,8 @@ def main() -> int: smart_resume=args.smart_resume, smart_check=args.smart_check, verify_level=args.verify_level, - check_only=args.check_only) + check_only=args.check_only, + repair=args.repair) return run_cli(cfg, args)