#!/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 # ---------------------------------------------------------------- 工具函数 _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 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 # ---------------------------------------------------------------- 输出解析器 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="传输完成后校验 (-c)") parser.add_argument("--retry", type=int, default=0, help="失败自动补传轮数 (-1=无限, 默认 0 不重试)") 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)""" 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}") # 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 _verify(self) -> None: """完成后校验: rsync -c --dry-run, 统计差异""" self.lm.write("verify: 开始校验 (-c dry-run)") 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 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: 完成 rc={rc} 差异文件={diffs}") if self.on_log: if diffs == 0 and rc == 0: self.on_log("[green]✔ 校验通过: 源与目标完全一致[/]") else: self.on_log(f"[red]✘ 校验发现 {diffs} 个差异 (rc={rc})[/]") 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() # 冲突策略: 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, 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, ) 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) 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) return run_cli(cfg, args) if __name__ == "__main__": sys.exit(main())