#!/usr/bin/env python3 """transfer.py - 统一文件传输工具（rsync/rclone/pv+tar 后端，CLI/TUI 双模式） 功能: - 多后端支持: rsync / rclone / pv+tar - CLI/TUI 双模式, TUI 底部可交互输入栏 - 实时日志写入本地目录, 滚动保留最近 N 个 (默认 20) - 进度/性能可视化 (总进度、单文件、速度曲线) """ from __future__ import annotations import argparse import os import re import shlex import signal import subprocess import sys import threading import time import tempfile from dataclasses import dataclass, field from datetime import datetime from glob import glob import hashlib # ---------------------------------------------------------------- 工具函数 _ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]|\x1b\][^\x07]*\x07|\r") _SPEED_RE = re.compile(r"([\d.]+)\s*([KMGTP]?)i?B?/s", re.IGNORECASE) _UNIT_MULT = {"": 1, "K": 1024, "M": 1024**2, "G": 1024**3, "T": 1024**4, "P": 1024**5} def clean_ansi(text: str) -> str: """移除 ANSI 转义序列与回车符""" return _ANSI_RE.sub("", text) def parse_speed(text: str) -> float: """解析速度字符串如 '356.77MB/s' '2.715 MiB/s' -> 字节/秒""" m = _SPEED_RE.search(text) if not m: return 0.0 return float(m.group(1)) * _UNIT_MULT[m.group(2).upper()] def human_size(num: float) -> str: """字节数 -> 人类可读字符串""" for unit in ("B", "KB", "MB", "GB", "TB"): if abs(num) < 1024 or unit == "TB": return f"{num:.1f}{unit}" if unit != "B" else f"{int(num)}B" num /= 1024 return f"{num:.1f}TB" def human_speed(bps: float) -> str: return f"{human_size(bps)}/s" def dir_size(path: str) -> int: """du 目录字节数 (轮询进度用, 绕过 rsync 输出缓冲)""" if not os.path.isdir(path): return 0 try: proc = subprocess.run(["du", "-sb", path], capture_output=True, text=True, timeout=30) return int(proc.stdout.split()[0]) except (ValueError, OSError, subprocess.TimeoutExpired): return 0 def count_files(path: str) -> int: """统计目录内文件数 (单文件进度条数据源)""" if not os.path.isdir(path): return 0 try: proc = subprocess.run(["find", path, "-type", "f"], capture_output=True, text=True, timeout=60) return len([l for l in proc.stdout.splitlines() if l.strip()]) except (OSError, subprocess.TimeoutExpired): 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, on_file=None) -> list[str]: """智能续传: 校验目标最新 N 个文件与源一致性, 返回损坏/缺失列表并删除目标坏副本 - <=max_size 的文件: 大小 + 内容 hash 全校验 - >max_size 的文件: 只查大小 (内容由后续 rsync --append 兜底) - on_file(i, total, rel, ok): 每文件进度回调 (i 从 1 开始) - 返回需重传的相对路径列表 (目标坏副本已删除) """ 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) # 选择最新 N 个 (按 mtime), 校验/显示顺序按路径字典序 (直观连续) selected = sorted(rel for _mtime, rel in recent[:check_count]) total = len(selected) for idx, rel in enumerate(selected, 1): sp = os.path.join(src_dir, rel) tp = os.path.join(dst_dir, rel) if not os.path.isfile(sp): if on_file: on_file(idx, total, rel, True) # 源无此文件, 跳过 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) if on_file: on_file(idx, total, rel, False) continue if os.path.getsize(tp) <= max_size and not files_identical(sp, tp): bad.append(rel) os.remove(tp) if on_file: on_file(idx, total, rel, False) elif on_file: on_file(idx, total, rel, True) return bad VERIFY_LEVELS = ["ultrafast", "fast", "normal", "slow", "full"] 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 verify_files_list(src_dir: str, dst_dir: str, rels: list[str], level: str = "slow", on_progress=None) -> tuple[list[str], list[str]]: """复核指定文件列表: 按校验等级验证源/目标一致性 - ultrafast: 仅大小 - fast: 大小+mtime - normal/slow/full: md5 内容校验 返回 (通过列表, 失败列表); on_progress(i, total) 进度回调 """ ok: list[str] = [] fail: list[str] = [] total = len(rels) for i, rel in enumerate(rels, 1): 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): fail.append(rel) elif level == "ultrafast": (ok if os.path.getsize(sp) == os.path.getsize(tp) else fail).append(rel) elif level == "fast": sst, tst = os.stat(sp), os.stat(tp) good = (sst.st_size == tst.st_size and int(sst.st_mtime) == int(tst.st_mtime)) (ok if good else fail).append(rel) else: (ok if files_identical(sp, tp) else fail).append(rel) if on_progress: on_progress(i, total) return ok, fail 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: return 0.0, 0.0, 0.0 cur = float(history[-1]) peak = max(history) avg = sum(history) / len(history) return cur, peak, avg # ---------------------------------------------------------------- 数据结构 @dataclass class ProgressStats: """总体进度""" bytes_done: int = 0 bytes_total: int = 0 pct: float = 0.0 speed_bps: float = 0.0 files_done: int = 0 files_total: int = 0 @property def eta_seconds(self) -> float: if self.speed_bps <= 0 or self.bytes_total <= 0: return 0.0 return max(0.0, (self.bytes_total - self.bytes_done) / self.speed_bps) @dataclass class FileProgress: """单文件进度""" name: str = "" pct: float = 0.0 bytes_done: int = 0 bytes_total: int = 0 @dataclass class TransferConfig: """传输任务配置""" source: str dest: str backend: str = "rsync" bwlimit: str = "30000" transfers: int = 4 resume: bool = True checksum: bool = False verify: bool = False logdir: str = "" policy: str = "overwrite" retry: int = 0 smart_resume: bool = False smart_check: int = 5 verify_level: str = "slow" check_only: bool = False repair: bool = False # ---------------------------------------------------------------- 输出解析器 class RsyncParser: """解析 rsync --info=progress1/2 输出""" PROGRESS2_RE = re.compile( r"^\s*([\d,]+)\s+(\d+)%\s+([\d.]+[KMGTP]?i?B?/s)\s+[\d:]+\s*" r"(?:\(xfr#(\d+),\s*to-chk=(\d+)/(\d+)\))?" ) FILE_RE = re.compile(r"^\s*(\S+)\s+([\d,]+)\s+(\d+)%\s+([\d.]+[KMGTP]?i?B?/s)") SUMMARY_RE = re.compile(r"^sent\s+([\d,]+)\s+bytes\s+received\s+([\d,]+)\s+bytes") @classmethod def feed_progress2(cls, line: str) -> ProgressStats | None: line = clean_ansi(line).strip() m = cls.PROGRESS2_RE.match(line) if not m: return None files_done = int(m.group(4)) if m.group(4) else 0 files_total = int(m.group(6)) if m.group(6) else 0 return ProgressStats( bytes_done=int(m.group(1).replace(",", "")), pct=float(m.group(2)), speed_bps=parse_speed(m.group(3)), files_done=files_done, files_total=files_total, ) @classmethod def feed_file(cls, line: str) -> FileProgress | None: line = clean_ansi(line).strip() m = cls.FILE_RE.match(line) if not m: return None return FileProgress(name=m.group(1), pct=float(m.group(3))) @classmethod def is_error(cls, line: str) -> bool: return line.startswith("rsync:") or line.startswith("rsync error") @classmethod def parse_summary(cls, line: str) -> tuple[int, int] | None: m = cls.SUMMARY_RE.search(clean_ansi(line)) if not m: return None return (int(m.group(1).replace(",", "")), int(m.group(2).replace(",", ""))) class RcloneParser: """解析 rclone --progress 输出""" FILE_RE = re.compile(r"(.+?):\s*(\d+)%\s*/\s*([\d.]+)\s+[KMGTP]?i?B") TOTAL_RE = re.compile( r"Transferred:\s+([\d.]+)\s+[KMGTP]?i?B\s*/\s*([\d.]+)\s+[KMGTP]?i?B,\s*(\d+)%" ) FILES_RE = re.compile(r"Transferred:\s+(\d+)\s*/\s*(\d+),\s*(\d+)%") @classmethod def feed_file(cls, line: str) -> FileProgress | None: line = clean_ansi(line).strip() m = cls.FILE_RE.search(line) if not m: return None name = m.group(1).split()[-1] if m.group(1).split() else m.group(1) return FileProgress(name=name.strip("* "), pct=float(m.group(2))) @classmethod def feed_total(cls, line: str) -> ProgressStats | None: line = clean_ansi(line).strip() m = cls.TOTAL_RE.search(line) if not m: return None done, total = float(m.group(1)), float(m.group(2)) return ProgressStats( bytes_done=int(done * 1024 * 1024), bytes_total=int(total * 1024 * 1024), pct=float(m.group(3)), speed_bps=parse_speed(line), ) @classmethod def feed_files(cls, line: str) -> ProgressStats | None: line = clean_ansi(line).strip() m = cls.FILES_RE.search(line) if not m: return None return ProgressStats( files_done=int(m.group(1)), files_total=int(m.group(2)), pct=float(m.group(3)), ) @classmethod def is_error(cls, line: str) -> bool: low = line.lower() return "error" in low or "failed" in low # ---------------------------------------------------------------- 日志轮转 class LogManager: """日志管理: 会话级日志文件, 滚动保留最近 keep 个""" def __init__(self, logdir: str, keep: int = 20): self.logdir = logdir self.keep = max(1, keep) os.makedirs(logdir, exist_ok=True) self.current_path: str = "" def open_session(self, name: str) -> str: """创建新会话日志文件, 返回路径 (纳秒时间戳保证唯一有序)""" stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + f"_{time.time_ns() % 10**9:09d}" safe = re.sub(r"[^\w.-]", "_", name) self.current_path = os.path.join(self.logdir, f"{stamp}_{safe}.log") with open(self.current_path, "a", encoding="utf-8") as fh: fh.write(f"=== session {name} started at {stamp} ===\n") return self.current_path def write(self, text: str, level: str = "INFO") -> None: """实时追加一行 (带时间戳)""" if not self.current_path: return stamp = datetime.now().strftime("%H:%M:%S") with open(self.current_path, "a", encoding="utf-8") as fh: fh.write(f"[{stamp}] [{level}] {text}\n") def write_error(self, text: str) -> None: self.write(text, "ERROR") def prune(self) -> list[str]: """删除超出 keep 的最旧日志, 返回被删文件""" files = sorted(glob(os.path.join(self.logdir, "*.log"))) removed = [] for f in files[:-self.keep]: try: os.remove(f) removed.append(f) except OSError: pass return removed def list_logs(self) -> list[str]: return sorted(glob(os.path.join(self.logdir, "*.log"))) # ---------------------------------------------------------------- 后端 class RsyncBackend: name = "rsync" @staticmethod def build_cmd(src: str, dst: str, bwlimit: str = "30000", resume: bool = True, checksum: bool = False, exclude_from: str = "", skip_existing: bool = False, append: bool = True, **_) -> list[str]: src = src.rstrip("/") + "/" cmd = ["rsync", "-a", "-v", "--info=progress2", "--mkpath"] if checksum: cmd.append("--checksum") if skip_existing: cmd.append("--ignore-existing") if bwlimit: cmd += [f"--bwlimit={bwlimit}"] if resume: cmd += ["--partial"] if append: cmd += ["--append"] if exclude_from: cmd += [f"--exclude-from={exclude_from}"] cmd += [src, dst] return cmd @staticmethod def build_verify_cmd(src: str, dst: str, bwlimit: str = "") -> list[str]: """校验命令: rsync -c --dry-run, 无输出=一致""" src = src.rstrip("/") + "/" cmd = ["rsync", "-a", "-c", "--dry-run", "--out-format=%i|%n"] if bwlimit: cmd += [f"--bwlimit={bwlimit}"] cmd += [src, dst] return cmd class RcloneBackend: name = "rclone" @staticmethod def build_cmd(src: str, dst: str, bwlimit: str = "8M", transfers: int = 4, skip_existing: bool = False, interactive: bool = False, **_) -> list[str]: src = src.rstrip("/") + "/" cmd = ["rclone", "copy", src, dst, "--progress", "--log-level", "INFO"] if bwlimit: cmd += ["--bwlimit", bwlimit] if skip_existing: cmd.append("--ignore-existing") if interactive: cmd.append("--interactive") cmd += ["--transfers", str(transfers)] return cmd class TarPvBackend: name = "tar" @staticmethod def build_cmd(src: str, dst: str, bwlimit: str | None = None, **_) -> str: src = src.rstrip("/") dst = dst.rstrip("/") pv_opts = f"-n -s $(du -sb {shlex.quote(src)} | cut -f1)" if bwlimit: # pv -L 单位是字节/秒, 参数按 KB/s 传入换算 pv_opts += f" -L {int(bwlimit) * 1024}" return (f"tar -C {shlex.quote(src)} -vcf - . " f"| pv {pv_opts} " f"| tar -C {shlex.quote(dst)} -xf -") BACKENDS = {"rsync": RsyncBackend, "rclone": RcloneBackend, "tar": TarPvBackend} # ---------------------------------------------------------------- 冲突策略 class ConflictScanner: """基于 rsync --dry-run 预扫描冲突清单""" @staticmethod def parse_rsync_line(line: str) -> tuple[str, str] | None: """解析 dry-run 行 '>f.st......|123|file.txt' -> (name, 'new'|'update')""" parts = line.split("|") if len(parts) < 3: return None op = parts[0] if not op.startswith((">f", "cf")): return None action = "new" if "+" in op else "update" return (parts[2], action) @classmethod def scan_rsync(cls, src: str, dst: str, checksum: bool = False) -> list[tuple[str, str]]: """执行 dry-run, 返回 (name, action) 冲突清单""" src = src.rstrip("/") + "/" cmd = ["rsync", "-a", "--dry-run", "--out-format=%i|%l|%n"] if checksum: cmd.append("--checksum") cmd += [src, dst] proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600) return [r for line in proc.stdout.splitlines() if (r := cls.parse_rsync_line(line))] def apply_policy(conflicts: list[tuple[str, str]], policy: str = "overwrite", answers: dict[str, str] | None = None) -> list[str]: """按策略返回需要排除(跳过)的文件名列表 answers: {文件名: 'y'|'n'} 或 {'*': 'a'(全部覆盖) | 's'(全部跳过) | 'q'(中止)} """ answers = answers or {} updates = [name for name, act in conflicts if act == "update"] if policy == "overwrite": return [] if policy == "skip": return updates # ask: y/n 只影响当前项; 全局 a/s 作用于"后续"(已明确决策的保留) global_choice = answers.get("*") if global_choice == "a": # 全部覆盖: 仅跳过此前已明确选 n 的 return [n for n in updates if answers.get(n) == "n"] if global_choice == "s": # 全部跳过: 此前已明确选 y 的仍覆盖 return [n for n in updates if answers.get(n) != "y"] if global_choice == "q": raise KeyboardInterrupt("用户中止传输") return [n for n in updates if answers.get(n) != "y"] # ---------------------------------------------------------------- 参数解析 def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="transfer", description="统一文件传输工具: rsync/rclone/pv+tar, CLI/TUI 双模式", ) parser.add_argument("source", nargs="?", help="源路径") parser.add_argument("dest", nargs="?", help="目标路径") parser.add_argument("--backend", choices=list(BACKENDS), default="rsync", help="传输后端 (默认 rsync)") parser.add_argument("--bwlimit", default="30000", help="限速 (rsync 单位 KB/s; rclone 如 8M)") parser.add_argument("--transfers", type=int, default=4, help="rclone 并行传输数 (默认 4)") parser.add_argument("--resume", action="store_true", default=True, help="断点续传 (默认开启)") parser.add_argument("--no-resume", dest="resume", action="store_false", help="关闭断点续传") parser.add_argument("--policy", choices=["overwrite", "skip", "ask"], default=None, help="冲突策略: 覆盖/跳过/询问") parser.add_argument("--overwrite", dest="policy", action="store_const", const="overwrite", help="覆盖已存在的文件") parser.add_argument("--skip", dest="policy", action="store_const", const="skip", help="跳过已存在的文件") parser.add_argument("--ask", dest="policy", action="store_const", const="ask", help="遇冲突时询问") parser.add_argument("--checksum", action="store_true", help="rsync 校验模式 (-c)") parser.add_argument("--verify", action="store_true", 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("--repair", action="store_true", help="校验发现问题后提示并重传问题文件") 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, help="日志保留数量 (默认 20)") parser.add_argument("--logdir", default=os.path.expanduser("~/logs"), help="日志目录 (默认 ~/logs)") args = parser.parse_args(argv) if args.policy is None: args.policy = "overwrite" if args.mode is None: args.mode = "tui" # 未显式指定 --mode cli 时, 一律 TUI 启动 return args # ---------------------------------------------------------------- 执行引擎 class TaskRunner: """传输任务执行器: 启动子进程, 解析输出, 实时写日志, 回调更新""" def __init__(self, cfg: TransferConfig, lm: LogManager, on_stats=None, on_file=None, on_log=None, on_error=None, on_done=None, exclude_from: str = ""): self.cfg = cfg self.lm = lm self.exclude_from = exclude_from self.on_stats = on_stats # callable(ProgressStats) self.on_file = on_file # callable(FileProgress) self.on_log = on_log # callable(str) 普通日志行 self.on_error = on_error # callable(str) 错误行 self.on_done = on_done # callable(int rc) self.proc: subprocess.Popen | None = None self.stats = ProgressStats() self._stop = threading.Event() self._paused = False self.stats_frames = 0 self.total_bytes = 0 self.total_files = 0 # -- 命令构建 def build_command(self) -> list[str]: backend = BACKENDS[self.cfg.backend] kw = dict(bwlimit=self.cfg.bwlimit or None, resume=self.cfg.resume, checksum=self.cfg.checksum, transfers=self.cfg.transfers, exclude_from=self.exclude_from, skip_existing=(self.cfg.policy == "skip"), interactive=(self.cfg.policy == "ask"), append=(self.cfg.policy != "ask")) if self.cfg.backend == "tar": return ["bash", "-c", backend.build_cmd(self.cfg.source, self.cfg.dest, **kw)] cmd = backend.build_cmd(self.cfg.source, self.cfg.dest, **kw) if self.cfg.backend == "rsync": # stdbuf 行缓冲: rsync progress2 帧逐行输出, 避免管道全缓冲攒帧 return ["stdbuf", "-oL"] + cmd return cmd # -- 输出分发 def _dispatch(self, line: str) -> None: if not line: return backend = self.cfg.backend if backend == "rsync": st = RsyncParser.feed_progress2(line) if st: self._merge_stats(st) self._emit_stats() return if RsyncParser.is_error(line): self._error(line) return f = RsyncParser.feed_file(line) if f: self._file(f) return summ = RsyncParser.parse_summary(line) if summ: self.lm.write(f"summary: sent={summ[0]}B recv={summ[1]}B") elif backend == "rclone": st = RcloneParser.feed_total(line) if st: self._merge_stats(st) self._emit_stats() return st2 = RcloneParser.feed_files(line) if st2: self._merge_files(st2) self._emit_stats() return f = RcloneParser.feed_file(line) if f: self._file(f) return if RcloneParser.is_error(line): self._error(line) return else: # tar: pv -n 输出纯百分比数字, tar -v 输出文件名 stripped = line.strip() if stripped.isdigit(): pct = float(stripped) self.stats.pct = pct self._emit_stats() return # 兜底: 普通行写日志文件 + 推送到 TUI 日志栏 if line.strip(): self.lm.write(line) if self.on_log: self.on_log(line) def _merge_stats(self, st: ProgressStats) -> None: if st.bytes_done: self.stats.bytes_done = st.bytes_done if st.bytes_total: self.stats.bytes_total = st.bytes_total if st.pct: self.stats.pct = st.pct if st.speed_bps: self.stats.speed_bps = st.speed_bps if st.files_done: self.stats.files_done = st.files_done if st.files_total: self.stats.files_total = st.files_total def _merge_files(self, st: ProgressStats) -> None: if st.files_done: self.stats.files_done = st.files_done if st.files_total: self.stats.files_total = st.files_total def _file(self, f: FileProgress) -> None: if self.on_file: self.on_file(f) if f.pct == 100: self.lm.write(f"done: {f.name}") def _error(self, line: str) -> None: if self.on_error: self.on_error(line) self.lm.write_error(line) def _emit_stats(self) -> None: self.stats_frames += 1 if self.on_stats: self.on_stats(self.stats) # -- 执行 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) while True: if attempt > 0: self.lm.write(f"=== 补传轮次 {attempt}/{retry_label} ===") if self.on_log: self.on_log(f"[yellow]补传轮次 {attempt}/{retry_label}[/]") time.sleep(3) # 等守护恢复设备 rc = self._run_once() if rc == 0 or self._stop.is_set(): break attempt += 1 # retry=-1 无限重试; retry=N 重试 N 轮; retry=0 不重试 if self.cfg.retry >= 0 and attempt > self.cfg.retry: break self.lm.write(f"attempt {attempt} failed rc={rc}") if self.on_error: self.on_error(f"本轮退出码 {rc}, 将自动补传 ({attempt}/{retry_label})") if rc == 0 and self.cfg.verify and self.cfg.backend in ("rsync",): self._verify() return rc def _run_once(self) -> int: """单次传输, 返回退出码 (进度由目录轮询驱动, 绕过后端输出缓冲)""" cmd = self.build_command() self.lm.open_session(f"{self.cfg.backend}_{os.path.basename(self.cfg.source.rstrip('/'))}") self.lm.write(f"cmd: {' '.join(cmd)}") self.lm.write(f"mode: {self.cfg.backend} bwlimit={self.cfg.bwlimit or 'off'} " f"resume={self.cfg.resume} checksum={self.cfg.checksum}") # 源总大小 (进度分母) self.total_bytes = dir_size(self.cfg.source.rstrip("/")) self.total_files = count_files(self.cfg.source.rstrip("/")) 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} 个文件...[/]") def on_verify_file(idx: int, total: int, rel: str, ok: bool) -> None: # 校验进度: 复用 stats 驱动进度条 + 日志栏每文件输出 self.stats.bytes_done = idx self.stats.bytes_total = total self.stats.pct = idx / max(1, total) * 100 self._emit_stats() if self.on_log: mark = "✓" if ok else "✘ 损坏" self.on_log(f"校验 {idx}/{total}: {os.path.basename(rel)} {mark}") bad = verify_recent_files(src_dir, dst_dir, self.cfg.smart_check, on_file=on_verify_file) 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, text=True, bufsize=1, start_new_session=True, ) # 轮询线程: 每秒统计目标目录大小驱动进度 self._stop.clear() poll = threading.Thread(target=self._poll_loop, daemon=True) poll.start() assert self.proc.stderr is not None for line in self.proc.stderr: if self._stop.is_set(): break line = line.rstrip() if line: self._error(line) rc = self.proc.wait() self._stop.set() poll.join(timeout=3) # 最终统计 final = dir_size(self.cfg.dest.rstrip("/")) self.stats.bytes_done = final if self.total_bytes: self.stats.pct = min(100.0, final / self.total_bytes * 100) self.lm.write(f"exit code: {rc} (帧 {self.stats_frames} 次, " f"目标 {human_size(final)}/{human_size(self.total_bytes)})") removed = self.lm.prune() if removed: self.lm.write(f"pruned old logs: {len(removed)}") if self.on_done: self.on_done(rc) return rc def _poll_loop(self) -> None: """每秒轮询目标目录大小, 更新进度 (独立线程)""" dst = self.cfg.dest.rstrip("/") last = dir_size(dst) last_t = time.time() poll = 0 while not self._stop.is_set(): time.sleep(1) if self._paused: continue # 暂停期间不轮询 done = dir_size(dst) now = time.time() speed = (done - last) / max(0.1, now - last_t) self.stats.bytes_done = done self.stats.bytes_total = self.total_bytes self.stats.speed_bps = max(0, speed) if self.total_bytes: self.stats.pct = min(100.0, done / self.total_bytes * 100) last, last_t = done, now # 每 3 秒更新文件计数 (find 开销控制) if self.total_files and poll % 3 == 0: self.stats.files_done = count_files(dst) self._emit_stats() poll += 1 # 每 5 秒一条进度快照到日志栏 if self.on_log and poll % 5 == 0: 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]}[/]") # 重传后复核: 用前置校验等级再验一遍重传的文件 if rc == 0 and issues: self._recheck_repaired(issues) def _recheck_repaired(self, issues: list[str]) -> None: """重传后用前置校验等级复核问题文件""" src_dir = self.cfg.source.rstrip("/") dst_dir = self.cfg.dest.rstrip("/") self.lm.write(f"repair: 复核 {len(issues)} 个重传文件 " f"(level={self.cfg.verify_level})") if self.on_log: self.on_log(f"[cyan]复核重传文件 ({self.cfg.verify_level})...[/]") def on_progress(i: int, total: int) -> None: self._emit_verify_progress(i, total) ok, fail = verify_files_list(src_dir, dst_dir, issues, self.cfg.verify_level, on_progress=on_progress) self.lm.write(f"repair: 复核完成 通过={len(ok)} 失败={len(fail)}") if self.on_log: if not fail: self.on_log(f"[green]✔ 复核通过: {len(ok)} 个重传文件完整[/]") else: self.on_log(f"[red]✘ 复核发现 {len(fail)} 个文件仍未修复: " f"{', '.join(os.path.basename(f) for f in fail[:5])}[/]") 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 self.lm.write(f"verify: 开始校验 level={level}") if self.on_log: 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 校验通过[/]") self._repair_if_requested(bad) 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("/") 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 not issues: self.on_log("[green]✔ 校验通过: 大小全部一致[/]") else: self.on_log(f"[red]✘ 发现 {len(issues)} 个问题文件[/]") self._repair_if_requested(issues) def _verify_fast(self) -> None: """fast: 大小+mtime 对比 (自实现, 带进度)""" src_dir = self.cfg.source.rstrip("/") dst_dir = self.cfg.dest.rstrip("/") 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 not issues: self.on_log("[green]✔ 校验通过: 无差异 (size+mtime)[/]") else: 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 差异统计 + 进度轮询""" 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 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]✔ 校验通过: 源与目标完全一致[/]") 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: return proc = self.proc if proc and proc.poll() is None: try: os.killpg(os.getpgid(proc.pid), signal.SIGSTOP) except (ProcessLookupError, PermissionError, OSError): proc.send_signal(signal.SIGSTOP) self._paused = True def resume(self) -> None: """继续: SIGCONT 整个进程组""" if not self._paused: return proc = self.proc if proc and proc.poll() is None: try: os.killpg(os.getpgid(proc.pid), signal.SIGCONT) except (ProcessLookupError, PermissionError, OSError): proc.send_signal(signal.SIGCONT) self._paused = False def stop(self) -> None: self._stop.set() if self.proc and self.proc.poll() is None: if self._paused: self.resume() try: os.killpg(os.getpgid(self.proc.pid), signal.SIGTERM) except (ProcessLookupError, PermissionError, OSError): self.proc.terminate() # ---------------------------------------------------------------- 冲突交互 def interactive_ask(updates: list[tuple[str, str]]) -> dict[str, str]: """CLI 逐项询问: y=覆盖当前 n=跳过当前 a=全部覆盖 s=全部跳过 q=中止""" print(f"发现 {len(updates)} 个已存在文件将被覆盖:") answers: dict[str, str] = {} for name, _ in updates: while True: choice = input(f" {name} [y]覆盖 [n]跳过 [a]全部覆盖 [s]全部跳过 [q]中止 (默认 n): ").strip().lower() if choice in ("y", "n", "a", "s", "q"): break if choice == "": choice = "n" break if choice == "a": answers["*"] = "a" break if choice == "s": answers["*"] = "s" break if choice == "q": answers["*"] = "q" break answers[name] = "y" if choice == "y" else "n" return answers def write_exclude_file(skipped: list[str]) -> str: """将跳过的文件写入 exclude-from 临时文件""" path = os.path.join(tempfile.gettempdir(), "tsu_exclude.txt") with open(path, "w", encoding="utf-8") as fh: fh.write("\n".join(skipped) + "\n") return path # ---------------------------------------------------------------- CLI 模式 def run_cli(cfg: TransferConfig, args: argparse.Namespace) -> int: """CLI 模式: 进度刷新到 stderr, 日志写文件""" lm = LogManager(args.logdir, args.keep_logs) def on_stats(st: ProgressStats) -> None: parts = [] if st.pct: parts.append(f"{st.pct:5.1f}%") if st.bytes_total: parts.append(f"{human_size(st.bytes_done)}/{human_size(st.bytes_total)}") elif st.bytes_done: parts.append(human_size(st.bytes_done)) if st.speed_bps: parts.append(human_speed(st.speed_bps)) if st.eta_seconds: parts.append(f"ETA {int(st.eta_seconds // 60)}:{int(st.eta_seconds % 60):02d}") if st.files_total: parts.append(f"files {st.files_done}/{st.files_total}") sys.stderr.write("\r\033[K" + " ".join(parts)) sys.stderr.flush() def on_file(f: FileProgress) -> None: sys.stderr.write(f"\r\033[K[{f.pct:5.1f}%] {f.name}\n") sys.stderr.flush() def on_error(line: str) -> None: 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": print("[TransferSu] 预扫描冲突...") conflicts = ConflictScanner.scan_rsync(cfg.source, cfg.dest, args.checksum) updates = [c for c in conflicts if c[1] == "update"] if updates: answers = interactive_ask(updates) skipped = apply_policy(updates, "ask", answers) if skipped: exclude_from = write_exclude_file(skipped) print(f"[TransferSu] 将跳过 {len(skipped)} 个文件") else: print("[TransferSu] 无冲突") elif args.policy == "ask" and cfg.backend == "rclone": print("[TransferSu] rclone 使用原生 --interactive 交互") runner = TaskRunner(cfg, lm, on_stats=on_stats, on_file=on_file, 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: rc = runner.run() except KeyboardInterrupt: print("\n[TransferSu] 用户中断") runner.stop() rc = 130 finally: sys.stderr.write("\n") print(f"[TransferSu] 完成, 退出码 {rc}") return rc # ---------------------------------------------------------------- TUI 模式 from collections import deque from textual.app import App, ComposeResult from textual.message import Message from textual.containers import Horizontal, Vertical from textual.widgets import (Footer, Header, Input, ProgressBar, RichLog, Sparkline, Static) TUI_CSS = """ Screen { layout: vertical; } #top { height: 7; border: round $primary; padding: 0 1; } #total-bar, #file-bar { width: 100%; } #total-bar Bar, #file-bar Bar { width: 1fr; } #stats-line { height: 1; content-align: left middle; } #chart-box { height: 6; border: round $secondary; padding: 0 1; } #speed-stats { height: 1; color: $text-muted; } #logs { height: 1fr; } #file-log { width: 1fr; border: round $success; } #error-log { width: 1fr; border: round $error; } #cmd-bar { height: 3; border-top: solid $accent; padding: 0 1; } Input { width: 100%; } """ class ScanDone(Message): """冲突扫描完成消息 (跨线程通知)""" def __init__(self, cfg: TransferConfig, updates: list): super().__init__() self.cfg = cfg self.updates = updates class TransferApp(App): """TransferSu TUI: 上栏进度 / 中栏性能 / 下栏日志 / 底部输入栏""" TITLE = "TransferSu" CSS = TUI_CSS BINDINGS = [("ctrl+c", "quit", "退出"), ("f1", "help", "帮助")] def __init__(self, preset: TransferConfig | None = None, logdir: str = "", keep_logs: int = 20): super().__init__() self.preset = preset self.logdir = logdir or os.path.expanduser("~/logs") self.keep_logs = keep_logs self.lm = LogManager(self.logdir, self.keep_logs) self.runner: TaskRunner | None = None self.speed_history: deque[float] = deque(maxlen=80) self._task_running = False self._ask: dict | None = None # -- 布局 def compose(self) -> ComposeResult: yield Header() with Vertical(id="top"): yield Static("就绪: 在底部输入栏输入任务 (源 目标 [参数]) 或 help", id="stats-line") yield ProgressBar(total=100, show_eta=True, id="total-bar") yield ProgressBar(total=100, id="file-bar") with Vertical(id="chart-box"): yield Static("速度曲线 (最近 80 秒)", id="speed-title") yield Sparkline([0], summary_function=max, id="speed-spark") yield Static("", id="speed-stats") with Horizontal(id="logs"): yield RichLog(id="file-log", highlight=True, markup=True, max_lines=200) yield RichLog(id="error-log", highlight=True, markup=True, max_lines=100) with Horizontal(id="cmd-bar"): yield Input(placeholder="传输任务: 源 目标 [参数] | stop / pause / continue / quit / help", id="cmd") yield Footer() def on_mount(self) -> None: self.query_one("#file-log", RichLog).write("[green]文件日志[/]") self.query_one("#error-log", RichLog).write("[red]错误日志[/]") # 每秒定时器: 强制重绘 UI (不依赖帧回调) self.set_interval(1.0, self._tick) if self.preset: self._start_task(self.preset) def _tick(self) -> None: """每秒重绘: 基于 runner 缓存的 stats 刷新进度/文字""" if self.runner and self._task_running: self._on_stats(self.runner.stats) # -- 输入处理 def on_input_submitted(self, event: Input.Submitted) -> None: text = event.value.strip() event.input.value = "" if not text: return if self._ask is not None: self._handle_ask_input(text) return low = text.lower() if low in ("q", "quit", "exit"): self.exit() return if low in ("stop", "pause", "continue", "cont", "resume"): self._handle_transport_cmd(low) return if low in ("h", "help", "?"): self._show_help() return if low == "clear": self.query_one("#file-log", RichLog).clear() self.query_one("#error-log", RichLog).clear() return if self._task_running: self._flash("[yellow]任务运行中, 请先等待完成或输入 quit 退出[/]") return self._parse_and_start(text) def _handle_transport_cmd(self, cmd: str) -> None: """传输控制命令: stop / pause / continue""" if not self.runner: self._flash("[yellow]当前无任务[/]") return if cmd == "stop": self.runner.stop() self._flash("[red]已发送停止信号[/]") elif cmd == "pause": self.runner.pause() self._flash("[yellow]已暂停 (输入 continue 恢复)[/]") elif cmd in ("continue", "cont", "resume"): self.runner.resume() self._flash("[green]已恢复[/]") def _handle_ask_input(self, text: str) -> None: choice = text.strip().lower() ask = self._ask if choice not in ("y", "n", "a", "s", "q"): self._flash("[yellow]请输入 y/n/a/s/q[/]") return if choice in ("a", "s", "q"): ask["answers"]["*"] = choice self._ask_finish() return name, _ = ask["updates"][ask["idx"]] ask["answers"][name] = "y" if choice == "y" else "n" ask["idx"] += 1 if ask["idx"] >= len(ask["updates"]): self._ask_finish() else: self._ask_next() def _parse_and_start(self, text: str) -> None: try: parts = shlex.split(text) args = parse_args(parts) except SystemExit: self._flash("[red]参数错误, 示例: /src /dst --backend rclone --bwlimit 8M[/]") return if not (args.source and args.dest): self._flash("[red]需要源路径和目标路径[/]") return args.mode = "tui" cfg = TransferConfig( source=args.source, dest=args.dest, backend=args.backend, 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, verify_level=args.verify_level, check_only=args.check_only, repair=args.repair, ) if cfg.policy == "ask" and cfg.backend == "rsync": self._flash("[yellow]预扫描冲突...[/]") threading.Thread(target=self._scan_thread, args=(cfg,), daemon=True).start() return self._start_task(cfg) def _scan_thread(self, cfg: TransferConfig) -> None: try: conflicts = ConflictScanner.scan_rsync(cfg.source, cfg.dest, cfg.checksum) updates = [c for c in conflicts if c[1] == "update"] except Exception as exc: # noqa: BLE001 self.call_from_thread(self._flash, f"[red]冲突扫描失败: {exc}[/]") return # 用消息机制回主线程 (post_message 线程安全, 处理在 app 上下文) self.post_message(ScanDone(cfg, updates)) def on_scan_done(self, message: "ScanDone") -> None: self._show_ask(message.cfg, message.updates) def _show_ask(self, cfg: TransferConfig, updates: list) -> None: if not updates: self._flash("[green]无冲突[/]") self._start_task(cfg) return # 冲突交互走底部输入栏: 逐项 y/n/a/s/q self._ask = {"cfg": cfg, "updates": updates, "idx": 0, "answers": {}} self._ask_next() def _ask_next(self) -> None: ask = self._ask name, _ = ask["updates"][ask["idx"]] self.query_one("#stats-line", Static).update( f"[bold yellow]冲突 {ask['idx'] + 1}/{len(ask['updates'])}: {name}[/]") self.query_one("#cmd", Input).placeholder = ( f"[y]覆盖 [n]跳过 [a]全部覆盖 [s]全部跳过 [q]中止 -> {name}") def _ask_finish(self) -> None: ask = self._ask self._ask = None self.query_one("#cmd", Input).placeholder = ( "传输任务: 源 目标 [参数] | stop / pause / continue / quit / help") try: skipped = apply_policy(ask["updates"], "ask", ask["answers"]) except KeyboardInterrupt: self._flash("[red]已中止传输[/]") return exclude_from = write_exclude_file(skipped) if skipped else "" if skipped: self._flash(f"[yellow]将跳过 {len(skipped)} 个文件[/]") self._start_task(ask["cfg"], exclude_from=exclude_from) def _on_ask_result(self, cfg: TransferConfig, updates: list, answers: dict) -> None: try: skipped = apply_policy(updates, "ask", answers) except KeyboardInterrupt: self._flash("[red]已中止传输[/]") return exclude_from = write_exclude_file(skipped) if skipped else "" if skipped: self._flash(f"[yellow]将跳过 {len(skipped)} 个文件[/]") self._start_task(cfg, exclude_from=exclude_from) def _show_help(self) -> None: self._flash( "[cyan]用法: 源 目标 [--backend rsync|rclone|tar] [--bwlimit 30000] " "[--resume/--no-resume] [--checksum] [--verify] [--transfers 4] | quit/clear[/]" ) def _flash(self, msg: str) -> None: self.query_one("#stats-line", Static).update(msg) # -- 任务执行 (独立线程, 不依赖 textual worker 机制) def _start_task(self, cfg: TransferConfig, exclude_from: str = "") -> None: self._task_running = True self._flash(f"[bold]{cfg.backend}[/] {cfg.source} -> {cfg.dest}") threading.Thread(target=self._run_task_thread, args=(cfg, exclude_from), daemon=True).start() def _run_task_thread(self, cfg: TransferConfig, exclude_from: str = "") -> None: lm = self.lm lm.open_session(f"{cfg.backend}_{os.path.basename(cfg.source.rstrip('/'))}") self.call_from_thread(self._flash, f"日志: {lm.current_path}") runner = TaskRunner( cfg, lm, on_stats=lambda s: self.call_from_thread(self._on_stats, s), on_file=lambda f: self.call_from_thread(self._on_file, f), on_log=lambda l: self.call_from_thread(self._on_log, l), on_error=lambda l: self.call_from_thread(self._on_error, l), on_done=lambda rc: self.call_from_thread(self._on_done, rc), exclude_from=exclude_from, ) self.runner = runner try: rc = runner.run() except Exception as exc: # noqa: BLE001 self.call_from_thread(self._on_error, f"传输异常: {exc}") rc = 1 if rc != 0: self.call_from_thread(self._on_error, f"传输失败, 退出码 {rc}") # -- 回调 (主线程执行) def _on_stats(self, st: ProgressStats) -> None: self.query_one("#total-bar", ProgressBar).update( progress=min(st.pct, 100.0)) if st.speed_bps: self.speed_history.append(st.speed_bps) self.query_one("#speed-spark", Sparkline).data = list(self.speed_history) cur, peak, avg = speed_stats(self.speed_history) self.query_one("#speed-stats", Static).update( f"当前 {human_speed(cur)} | 峰值 {human_speed(peak)} | " f"平均 {human_speed(avg)} ({len(self.speed_history)} 样本)") parts = [f"总进度 {st.pct:5.1f}%"] if st.bytes_total: parts.append(f"{human_size(st.bytes_done)}/{human_size(st.bytes_total)}") elif st.bytes_done: parts.append(human_size(st.bytes_done)) if st.speed_bps: parts.append(human_speed(st.speed_bps)) if st.eta_seconds: parts.append(f"ETA {int(st.eta_seconds // 60)}:{int(st.eta_seconds % 60):02d}") if st.files_total: parts.append(f"文件 {st.files_done}/{st.files_total}") self.query_one("#file-bar", ProgressBar).update( progress=min(100.0, st.files_done / st.files_total * 100)) self.query_one("#stats-line", Static).update(" ".join(parts)) def _on_file(self, f: FileProgress) -> None: self.query_one("#file-bar", ProgressBar).update(progress=min(f.pct, 100.0)) self.query_one("#file-log", RichLog).write(f"{f.pct:5.1f}% {f.name}") def _on_log(self, line: str) -> None: self.query_one("#file-log", RichLog).write(line) def _on_error(self, line: str) -> None: self.query_one("#error-log", RichLog).write(f"[red]{line}[/]") def _on_done(self, rc: int) -> None: self._task_running = False bar = self.query_one("#total-bar", ProgressBar) if rc == 0: bar.update(progress=100.0) # 小文件无进度帧, 完成时强制满格 self._flash("[green]✔ 传输完成[/]") else: self._flash(f"[red]✘ 传输失败 (退出码 {rc}), 日志见 {self.lm.current_path}[/]") self.query_one("#file-bar", ProgressBar).update(progress=100 if rc == 0 else 0) # ---------------------------------------------------------------- 入口 def main() -> int: args = parse_args() if args.mode == "tui" or not (args.source and args.dest): cfg = None if args.source and args.dest: cfg = TransferConfig(source=args.source, dest=args.dest, backend=args.backend, bwlimit=args.bwlimit, transfers=args.transfers, resume=args.resume, checksum=args.checksum, verify=args.verify, logdir=args.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) TransferApp(preset=cfg, logdir=args.logdir, keep_logs=args.keep_logs).run() return 0 cfg = TransferConfig(source=args.source, dest=args.dest, backend=args.backend, bwlimit=args.bwlimit, transfers=args.transfers, resume=args.resume, checksum=args.checksum, verify=args.verify, logdir=args.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) return run_cli(cfg, args) if __name__ == "__main__": sys.exit(main())