#!/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 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 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" # ---------------------------------------------------------------- 输出解析器 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", "--info=progress2"] 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 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="传输完成后校验") 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() # -- 命令构建 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)] return backend.build_cmd(self.cfg.source, self.cfg.dest, **kw) # -- 输出分发 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 if stripped.startswith(("./", "/", "-")) or "\t" in stripped: pass # tar 文件条目 self.lm.write(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: if self.on_stats: self.on_stats(self.stats) # -- 执行 def run(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.proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=0, ) # rsync progress2 / rclone --progress 用 \r 刷新帧(不换行), # 故按 \r 与 \n 均作分隔符分帧解析, 避免进度只取到末帧 assert self.proc.stdout is not None buf = "" while True: chunk = self.proc.stdout.read(8192) if not chunk: break buf += chunk while True: m = re.search(r"[\r\n]", buf) if not m: break seg, buf = buf[: m.start()], buf[m.start() + 1:] if self._stop.is_set(): self.proc.kill() break self._dispatch(seg.rstrip()) if buf.strip() and not self._stop.is_set(): self._dispatch(buf.rstrip()) if self._stop.is_set(): self.proc.kill() rc = self.proc.wait() self.lm.write(f"exit code: {rc}") 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 stop(self) -> None: self._stop.set() if self.proc and self.proc.poll() is None: 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) yield RichLog(id="error-log", highlight=True, markup=True) with Horizontal(id="cmd-bar"): yield Input(placeholder="传输任务: 源路径 目标路径 [--backend rsync|rclone|tar --bwlimit 30000 ...] | help / quit", 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]错误日志[/]") if self.preset: self._start_task(self.preset) # -- 输入处理 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 ("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_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, ) 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 = ( "传输任务: 源路径 目标路径 [--backend rsync|rclone|tar --bwlimit 30000 ...] | help / quit") 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("#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) 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) return run_cli(cfg, args) if __name__ == "__main__": sys.exit(main())