feat: --repair 校验发现问题提示重传 + 各等级进度修复

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` 流程发现处理。
 

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

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)