From f7b10f3ee3a1ace903208ae61e175a96f2225966 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 22 Aug 2026 13:25:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20TransferSu=20=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E4=BC=A0=E8=BE=93=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 多后端: rsync(断点续传) / rclone(并行) / pv+tar - CLI/TUI 双模式, 未指定 --mode cli 时默认 TUI - TUI 四栏布局: 进度条/速度曲线/文件日志/错误日志 + 底部交互输入栏 - 冲突策略: --overwrite/--skip/--ask, y/n/a/s/q 交互决策 - 实时日志滚动保留 20 个, 纳秒时间戳排序 - 37 个单元测试 (pytest) --- .gitignore | 3 + README.md | 102 +++++ tests/test_transfer.py | 256 +++++++++++ transfer.py | 980 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1341 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 tests/test_transfer.py create mode 100644 transfer.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..75c6182 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..f2c5617 --- /dev/null +++ b/README.md @@ -0,0 +1,102 @@ +# TransferSu - 统一文件传输工具 + +基于 rsync / rclone / pv+tar 的统一拷贝管理工具，CLI/TUI 双模式，实时日志轮转。 + +## 特性 + +- **多后端**: rsync（推荐, 断点续传）/ rclone（并行）/ pv+tar（零依赖） +- **CLI/TUI 双模式**: 未显式 `--mode cli` 时默认 TUI 启动 +- **TUI 四栏布局**: + - 上栏: 总进度条 + 单文件进度条 + - 中栏: 速度曲线可视化（最近 80 秒采样） + - 下栏: 文件日志 + 错误日志 双面板 + - 底部: 交互输入栏（可空启动后设定任务） +- **实时日志**: 写入本地目录, 滚动保留最近 N 个（默认 20）, 文件名含纳秒时间戳 +- **可控性**: 限速 / 断点续传 / 校验 / 并行数 全部可配 + +## 安装 + +```bash +pip install -i https://mirrors.aliyun.com/pypi/simple/ pytest textual +``` + +## 用法 + +```bash +# TUI 模式 (默认): 空启动后在底部输入栏设定任务 +python3 transfer.py + +# TUI 模式带预设任务 +python3 transfer.py /src /dst --bwlimit 30000 + +# CLI 模式 (显式指定) +python3 transfer.py --mode cli /src/ /dst/ --backend rclone --bwlimit 8M +``` + +### TUI 输入栏命令 + +``` +/src /dst --backend rclone --bwlimit 8M --transfers 4 # 启动传输任务 +help # 帮助 +clear # 清空日志面板 +quit # 退出 +``` + +### 参数 + +| 参数 | 默认 | 说明 | +|---|---|---| +| `--backend` | rsync | rsync / rclone / tar | +| `--bwlimit` | 30000 | 限速 (rsync: KB/s, rclone: 如 8M) | +| `--transfers` | 4 | rclone 并行传输数 | +| `--resume` / `--no-resume` | 开 | rsync 断点续传 | +| `--checksum` | 关 | rsync 校验模式 | +| `--verify` | 关 | 完成后校验 | +| `--mode` | tui | cli / tui | +| `--keep-logs` | 20 | 日志滚动保留数 | +| `--logdir` | ~/logs | 日志目录 | + +## 日志 + +- 目录: `--logdir` 指定（默认 `~/logs`） +- 文件名: `{时间戳}_{后端}_{源名}.log`（时间戳开头保证排序即时间序） +- 滚动: 超过 `--keep-logs` 个自动删除最旧 + +## 测试 + +```bash +python3 -m pytest tests/ -q +``` + +## 冲突策略（覆盖/跳过） + +传输前可用参数预设策略，或中途逐项决策： + +| 参数 | 行为 | +|---|---| +| `--overwrite` | 覆盖已存在文件（默认） | +| `--skip` | 跳过已存在文件（`--ignore-existing`） | +| `--ask` | 逐项询问（扫描冲突后交互） | + +**ask 交互选项**（CLI 与 TUI 一致）： + +``` +y = 覆盖当前项 n = 跳过当前项 +a = 全部覆盖 s = 全部跳过 +q = 中止传输 +``` + +- **CLI**：预扫描后逐项提示，输入 y/n/a/s/q +- **TUI**：扫描完成后在底部输入栏逐项交互（stats 栏显示当前冲突文件） +- **仅当前/后续同步**：y/n 只作用于当前文件；a/s 一次决定后续所有冲突 +- rsync 覆盖场景自动禁用 `--append`（追加模式会阻止重写），保留 `--partial` +- rclone 的 `--ask` 使用原生 `--interactive`（y/n/s）；tar 后端不支持 ask/skip + +```bash +# CLI 示例: ask 交互 +python3 transfer.py --mode cli --policy ask /src/ /dst/ + +# TUI 输入栏示例 +/src /dst --policy ask +/src /dst --skip +``` diff --git a/tests/test_transfer.py b/tests/test_transfer.py new file mode 100644 index 0000000..08b8dad --- /dev/null +++ b/tests/test_transfer.py @@ -0,0 +1,256 @@ +"""transfer.py 核心逻辑测试：解析器 / 日志轮转 / 参数解析 / 后端命令构建""" +import re +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import transfer + +# ---------- 解析器: rsync ---------- + +def test_rsync_progress2_line(): + line = " 31,457,280 12% 356.77MB/s 0:00:00 (xfr#1, to-chk=51/53)" + s = transfer.RsyncParser.feed_progress2(line) + assert s is not None + assert s.pct == 12 + assert s.bytes_done == 31457280 + assert s.files_done == 1 + assert s.files_total == 53 + assert s.speed_bps > 0 + +def test_rsync_progress2_complete(): + line = " 243,401,728 100% 305.39MB/s 0:00:00 (xfr#52, to-chk=0/53)" + s = transfer.RsyncParser.feed_progress2(line) + assert s is not None + assert s.pct == 100 + assert s.files_done == 52 + assert s.files_total == 53 + +def test_rsync_progress2_unparseable_returns_none(): + assert transfer.RsyncParser.feed_progress2("building file list") is None + assert transfer.RsyncParser.feed_progress2("") is None + +def test_rsync_single_file_line(): + line = "file_1.bin 31,457,280 100% 356.77MB/s 0:00:00" + f = transfer.RsyncParser.feed_file(line) + assert f is not None + assert f.name == "file_1.bin" + assert f.pct == 100 + +def test_rsync_error_detection(): + assert transfer.RsyncParser.is_error("rsync: [sender] write error: Broken pipe (32)") + assert transfer.RsyncParser.is_error("rsync error: some files/attrs were not transferred (code 23)") + assert not transfer.RsyncParser.is_error("file_1.bin 100%") + assert not transfer.RsyncParser.is_error("") + +# ---------- 解析器: rclone ---------- + +def test_rclone_file_line(): + line = " * file_7.bin: 98% / 7.023 MiB, 2.715 MiB/s" + f = transfer.RcloneParser.feed_file(line) + assert f is not None + assert f.name == "file_7.bin" + assert f.pct == 98 + +def test_rclone_total_line(): + line = "Transferred: 232.126 MiB / 232.126 MiB, 100%, 7.997 MiB/s, ETA 0s" + s = transfer.RcloneParser.feed_total(line) + assert s is not None + assert s.pct == 100 + +def test_rclone_files_count_line(): + line = "Transferred: 52 / 52, 100%" + s = transfer.RcloneParser.feed_files(line) + assert s is not None + assert s.files_done == 52 + assert s.files_total == 52 + +# ---------- 日志轮转 ---------- + +def test_logmanager_creates_and_appends(tmp_path): + lm = transfer.LogManager(tmp_path, keep=20) + path = lm.open_session("test") + assert os.path.exists(path) + lm.write("hello") + lm.write("world") + content = open(path).read() + assert "hello" in content + assert "world" in content + assert content.index("hello") < content.index("world") # 追加顺序正确 + +def test_logmanager_keeps_only_latest_20(tmp_path): + lm = transfer.LogManager(tmp_path, keep=20) + for i in range(25): + lm.open_session(f"run{i}") + lm.write(f"data{i}") + lm.prune() + files = sorted(f for f in os.listdir(tmp_path) if f.endswith(".log")) + assert len(files) == 20 + run_nums = [int(re.search(r"run(\d+)", f).group(1)) for f in files] + assert 5 in run_nums + assert 4 not in run_nums + +def test_logmanager_filenames_sorted_chronologically(tmp_path): + lm = transfer.LogManager(tmp_path, keep=20) + p1 = lm.open_session("a") + p2 = lm.open_session("b") + assert p1 != p2 + assert p2 > p1 # 时间戳递增 + +# ---------- 参数解析 ---------- + +def test_parse_args_with_positional(): + args = transfer.parse_args(["src", "dst"]) + assert args.source == "src" + assert args.dest == "dst" + assert args.backend == "rsync" + assert args.keep_logs == 20 + assert args.mode == "tui" # 未显式指定 cli 时默认 TUI + +def test_parse_args_explicit_cli(): + args = transfer.parse_args(["--mode", "cli", "src", "dst"]) + assert args.mode == "cli" + +def test_parse_args_no_positional_goes_tui(): + args = transfer.parse_args([]) + assert args.mode == "tui" + assert args.source is None + +def test_parse_args_full_options(): + args = transfer.parse_args([ + "--backend", "rclone", "--bwlimit", "30M", "--resume", + "--checksum", "--mode", "cli", "--keep-logs", "5", + "--logdir", "/tmp/x", "src", "dst", + ]) + assert args.backend == "rclone" + assert args.bwlimit == "30M" + assert args.resume is True + assert args.checksum is True + assert args.keep_logs == 5 + assert args.logdir == "/tmp/x" + +# ---------- 后端命令构建 ---------- + +def test_rsync_backend_cmd(): + cmd = transfer.RsyncBackend.build_cmd("a/", "b/", bwlimit="30000", resume=True, checksum=False) + assert cmd[0] == "rsync" + joined = " ".join(cmd) + assert "--info=progress2" in joined + assert "--bwlimit=30000" in joined + assert "--partial" in joined + assert "--append" in joined + assert "-a" in joined + +def test_rsync_backend_checksum_flag(): + cmd = transfer.RsyncBackend.build_cmd("a/", "b/", bwlimit="", resume=False, checksum=True) + assert "--checksum" in " ".join(cmd) + +def test_rclone_backend_cmd(): + cmd = transfer.RcloneBackend.build_cmd("a/", "b/", bwlimit="8M", transfers=4) + assert cmd[0] == "rclone" + joined = " ".join(cmd) + assert "--bwlimit" in joined + assert "--transfers" in joined + assert "--progress" in joined + +def test_tar_pv_backend_cmd(): + cmd = transfer.TarPvBackend.build_cmd("a/", "b/") + assert "tar" in cmd + assert "pv" in cmd + assert "|" in cmd # 管道串接 + +# ---------- 冲突策略: 参数 ---------- + +def test_policy_default_overwrite(): + args = transfer.parse_args(["src", "dst"]) + assert args.policy == "overwrite" + +def test_policy_skip_flag(): + args = transfer.parse_args(["--policy", "skip", "src", "dst"]) + assert args.policy == "skip" + +def test_policy_ask_flag(): + args = transfer.parse_args(["--policy", "ask", "src", "dst"]) + assert args.policy == "ask" + +def test_policy_synonyms(): + assert transfer.parse_args(["--overwrite", "s", "d"]).policy == "overwrite" + assert transfer.parse_args(["--skip", "s", "d"]).policy == "skip" + assert transfer.parse_args(["--ask", "s", "d"]).policy == "ask" + +# ---------- 冲突策略: 扫描解析 ---------- + +def test_scan_rsync_new_file(): + conflicts = transfer.ConflictScanner.parse_rsync_line(">f+++++++++|4|new.txt") + assert conflicts == ("new.txt", "new") + +def test_scan_rsync_update(): + conflicts = transfer.ConflictScanner.parse_rsync_line(">f.st......|123|f.txt") + assert conflicts == ("f.txt", "update") + +def test_scan_rsync_metadata_only_update(): + conflicts = transfer.ConflictScanner.parse_rsync_line(">f..t......|0|same.txt") + assert conflicts == ("same.txt", "update") + +def test_scan_rsync_ignore_non_file_lines(): + assert transfer.ConflictScanner.parse_rsync_line("cd+++++++++|0|dir/") is None + assert transfer.ConflictScanner.parse_rsync_line("*deleting |0|old.txt") is None + assert transfer.ConflictScanner.parse_rsync_line("") is None + +# ---------- 冲突策略: 决策 ---------- + +def test_apply_policy_overwrite(): + conflicts = [("a.txt", "update"), ("b.txt", "update")] + skipped = transfer.apply_policy(conflicts, "overwrite", {}) + assert skipped == [] + +def test_apply_policy_skip(): + conflicts = [("a.txt", "update"), ("b.txt", "update")] + skipped = transfer.apply_policy(conflicts, "skip", {}) + assert set(skipped) == {"a.txt", "b.txt"} + +def test_apply_policy_ask_mixed_answers(): + conflicts = [("a.txt", "update"), ("b.txt", "update")] + answers = {"a.txt": "y", "b.txt": "n"} + skipped = transfer.apply_policy(conflicts, "ask", answers) + assert skipped == ["b.txt"] # 仅跳过选 n 的 + +def test_apply_policy_ask_all_overwrite(): + conflicts = [("a.txt", "update")] + skipped = transfer.apply_policy(conflicts, "ask", {"*": "a"}) + assert skipped == [] + +def test_apply_policy_ask_all_skip(): + conflicts = [("a.txt", "update"), ("b.txt", "update")] + skipped = transfer.apply_policy(conflicts, "ask", {"*": "s"}) + assert set(skipped) == {"a.txt", "b.txt"} + +# ---------- 后端 exclude 集成 ---------- + +def test_rsync_backend_exclude_from(): + cmd = transfer.RsyncBackend.build_cmd("a/", "b/", exclude_from="/tmp/excl.txt") + assert "--exclude-from=/tmp/excl.txt" in cmd + +def test_rclone_backend_skip_policy(): + cmd = transfer.RcloneBackend.build_cmd("a/", "b/", skip_existing=True) + assert "--ignore-existing" in " ".join(cmd) + +def test_rsync_backend_no_append_when_overwriting(): + """覆盖场景禁用 --append (追加模式会阻止重写)""" + cmd = transfer.RsyncBackend.build_cmd("a/", "b/", resume=True, append=False) + assert "--append" not in cmd + assert "--partial" in cmd + +def test_apply_policy_global_skip_keeps_explicit_yes(): + """先选 y 再选 s: 已确认覆盖的保留, 其余跳过""" + conflicts = [("a.txt", "update"), ("b.txt", "update")] + answers = {"a.txt": "y", "*": "s"} + skipped = transfer.apply_policy(conflicts, "ask", answers) + assert skipped == ["b.txt"] + +def test_apply_policy_global_overwrite_keeps_explicit_no(): + """先选 n 再选 a: 已确认跳过的保留, 其余覆盖""" + conflicts = [("a.txt", "update"), ("b.txt", "update")] + answers = {"a.txt": "n", "*": "a"} + skipped = transfer.apply_policy(conflicts, "ask", answers) + assert skipped == ["a.txt"] diff --git a/transfer.py b/transfer.py new file mode 100644 index 0000000..7327542 --- /dev/null +++ b/transfer.py @@ -0,0 +1,980 @@ +#!/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" + + +# ---------------------------------------------------------------- 数据结构 + +@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; } +#stats-line { height: 1; content-align: left middle; } +#chart-box { height: 5; border: round $secondary; padding: 0 1; } +#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 秒)") + yield Sparkline([0], summary_function=max, id="speed-spark") + 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) + 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())