feat: TransferSu 统一文件传输工具

- 多后端: rsync(断点续传) / rclone(并行) / pv+tar
- CLI/TUI 双模式, 未指定 --mode cli 时默认 TUI
- TUI 四栏布局: 进度条/速度曲线/文件日志/错误日志 + 底部交互输入栏
- 冲突策略: --overwrite/--skip/--ask, y/n/a/s/q 交互决策
- 实时日志滚动保留 20 个, 纳秒时间戳排序
- 37 个单元测试 (pytest)
This commit is contained in:
AskaEth committed 2026-08-22 13:25:47 +08:00
commit f7b10f3ee3
4 files changed
+1341

No files matched your search

+3
View File
@@ -0,0 +1,3 @@
__pycache__/
*.pyc
.pytest_cache/
+102
View File
@@ -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
```
+256
View File
@@ -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"]
+980
View File
@@ -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())