1005 lines
38 KiB
Python
1005 lines
38 KiB
Python
#!/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)]
|
|
cmd = backend.build_cmd(self.cfg.source, self.cfg.dest, **kw)
|
|
if self.cfg.backend == "rsync":
|
|
# stdbuf 行缓冲: rsync progress2 帧逐行输出, 避免管道全缓冲攒帧
|
|
return ["stdbuf", "-oL"] + cmd
|
|
return cmd
|
|
|
|
# -- 输出分发
|
|
def _dispatch(self, line: str) -> None:
|
|
if not line:
|
|
return
|
|
backend = self.cfg.backend
|
|
if backend == "rsync":
|
|
st = RsyncParser.feed_progress2(line)
|
|
if st:
|
|
self._merge_stats(st)
|
|
self._emit_stats()
|
|
return
|
|
if RsyncParser.is_error(line):
|
|
self._error(line)
|
|
return
|
|
f = RsyncParser.feed_file(line)
|
|
if f:
|
|
self._file(f)
|
|
return
|
|
summ = RsyncParser.parse_summary(line)
|
|
if summ:
|
|
self.lm.write(f"summary: sent={summ[0]}B recv={summ[1]}B")
|
|
elif backend == "rclone":
|
|
st = RcloneParser.feed_total(line)
|
|
if st:
|
|
self._merge_stats(st)
|
|
self._emit_stats()
|
|
return
|
|
st2 = RcloneParser.feed_files(line)
|
|
if st2:
|
|
self._merge_files(st2)
|
|
self._emit_stats()
|
|
return
|
|
f = RcloneParser.feed_file(line)
|
|
if f:
|
|
self._file(f)
|
|
return
|
|
if RcloneParser.is_error(line):
|
|
self._error(line)
|
|
return
|
|
else: # tar: pv -n 输出纯百分比数字, tar -v 输出文件名
|
|
stripped = line.strip()
|
|
if stripped.isdigit():
|
|
pct = float(stripped)
|
|
self.stats.pct = pct
|
|
self._emit_stats()
|
|
return
|
|
# 兜底: 普通行写日志文件 + 推送到 TUI 日志栏
|
|
if line.strip():
|
|
self.lm.write(line)
|
|
if self.on_log:
|
|
self.on_log(line)
|
|
|
|
def _merge_stats(self, st: ProgressStats) -> None:
|
|
if st.bytes_done:
|
|
self.stats.bytes_done = st.bytes_done
|
|
if st.bytes_total:
|
|
self.stats.bytes_total = st.bytes_total
|
|
if st.pct:
|
|
self.stats.pct = st.pct
|
|
if st.speed_bps:
|
|
self.stats.speed_bps = st.speed_bps
|
|
if st.files_done:
|
|
self.stats.files_done = st.files_done
|
|
if st.files_total:
|
|
self.stats.files_total = st.files_total
|
|
|
|
def _merge_files(self, st: ProgressStats) -> None:
|
|
if st.files_done:
|
|
self.stats.files_done = st.files_done
|
|
if st.files_total:
|
|
self.stats.files_total = st.files_total
|
|
|
|
def _file(self, f: FileProgress) -> None:
|
|
if self.on_file:
|
|
self.on_file(f)
|
|
if f.pct == 100:
|
|
self.lm.write(f"done: {f.name}")
|
|
|
|
def _error(self, line: str) -> None:
|
|
if self.on_error:
|
|
self.on_error(line)
|
|
self.lm.write_error(line)
|
|
|
|
def _emit_stats(self) -> None:
|
|
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())
|