diff --git a/README.md b/README.md index 84e8bc8..24e6bcd 100644 --- a/README.md +++ b/README.md @@ -37,11 +37,15 @@ python3 transfer.py --mode cli /src/ /dst/ --backend rclone --bwlimit 8M ``` /src /dst --backend rclone --bwlimit 8M --transfers 4 # 启动传输任务 -help # 帮助 -clear # 清空日志面板 -quit # 退出 +stop # 停止当前任务 +pause # 暂停 (SIGSTOP) +continue # 恢复暂停 +help / clear / quit ``` +单文件进度条显示**文件计数进度**(目标文件数/源文件总数,每 3 秒刷新), +大文件库下按文件粒度推进。 + ### 参数 | 参数 | 默认 | 说明 | diff --git a/tests/test_transfer.py b/tests/test_transfer.py index 9ce005a..c1257b4 100644 --- a/tests/test_transfer.py +++ b/tests/test_transfer.py @@ -1,6 +1,7 @@ """transfer.py 核心逻辑测试:解析器 / 日志轮转 / 参数解析 / 后端命令构建""" import re import os +import subprocess import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -307,3 +308,61 @@ def test_rsync_verify_cmd(): def test_rsync_verify_cmd_no_bwlimit(): cmd = transfer.RsyncBackend.build_verify_cmd("a/", "b/") assert "--bwlimit" not in " ".join(cmd) + +# ---------- 暂停/继续/停止 (TUI 命令) ---------- + +def _proc_state(pid: int) -> str: + with open(f"/proc/{pid}/stat") as fh: + return fh.read().split()[2] # 状态字段: R/S/T/D/Z + + +def _make_runner(tmp_path) -> "transfer.TaskRunner": + cfg = transfer.TransferConfig("s", "d") + lm = transfer.LogManager(str(tmp_path), keep=20) + return transfer.TaskRunner(cfg, lm) + + +def test_pause_sends_sigstop(tmp_path): + r = _make_runner(tmp_path) + p = subprocess.Popen(["sleep", "30"], start_new_session=True) + r.proc = p + r.pause() + assert r._paused is True + assert _proc_state(p.pid) == "T" # T = stopped + p.kill() + p.wait() + + +def test_resume_sends_sigcont(tmp_path): + r = _make_runner(tmp_path) + p = subprocess.Popen(["sleep", "30"], start_new_session=True) + r.proc = p + r.pause() + r.resume() + assert r._paused is False + assert _proc_state(p.pid) in ("S", "R") # 已恢复 + p.kill() + p.wait() + + +def test_pause_idempotent_when_no_proc(tmp_path): + r = _make_runner(tmp_path) + r.pause() # 无进程不报错 + assert r._paused is False + +# ---------- 文件计数 (单文件进度条数据源) ---------- + +def test_count_files_recursive(tmp_path): + (tmp_path / "a.flac").write_text("x") + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.lrc").write_text("y") + (tmp_path / "sub" / "c.flac").write_text("z") + assert transfer.count_files(str(tmp_path)) == 3 + +def test_count_files_missing_returns_zero(): + assert transfer.count_files("/nonexistent/xyz") == 0 + +def test_count_files_only_files(tmp_path): + (tmp_path / "a.txt").write_text("x") + (tmp_path / "dir1").mkdir() + assert transfer.count_files(str(tmp_path)) == 1 # 目录不计 diff --git a/transfer.py b/transfer.py index 4d0e900..f2d9176 100644 --- a/transfer.py +++ b/transfer.py @@ -13,6 +13,7 @@ import argparse import os import re import shlex +import signal import subprocess import sys import threading @@ -67,6 +68,18 @@ def dir_size(path: str) -> int: return 0 +def count_files(path: str) -> int: + """统计目录内文件数 (单文件进度条数据源)""" + if not os.path.isdir(path): + return 0 + try: + proc = subprocess.run(["find", path, "-type", "f"], + capture_output=True, text=True, timeout=60) + return len([l for l in proc.stdout.splitlines() if l.strip()]) + except (OSError, subprocess.TimeoutExpired): + return 0 + + def speed_stats(history) -> tuple[float, float, float]: """速度统计: (当前, 峰值, 平均) 字节/秒""" if not history: @@ -463,7 +476,10 @@ class TaskRunner: self.proc: subprocess.Popen | None = None self.stats = ProgressStats() self._stop = threading.Event() + self._paused = False self.stats_frames = 0 + self.total_bytes = 0 + self.total_files = 0 # -- 命令构建 def build_command(self) -> list[str]: @@ -582,7 +598,7 @@ class TaskRunner: self.on_log(f"[yellow]补传轮次 {attempt}/{self.cfg.retry}[/]") time.sleep(3) # 等守护恢复设备 rc = self._run_once() - if rc == 0: + if rc == 0 or self._stop.is_set(): break self.lm.write(f"attempt {attempt + 1} failed rc={rc}") if self.on_error and attempt < self.cfg.retry: @@ -600,11 +616,14 @@ class TaskRunner: f"resume={self.cfg.resume} checksum={self.cfg.checksum}") # 源总大小 (进度分母) self.total_bytes = dir_size(self.cfg.source.rstrip("/")) - self.lm.write(f"source size: {human_size(self.total_bytes)}") + self.total_files = count_files(self.cfg.source.rstrip("/")) + self.stats.files_total = self.total_files + self.lm.write(f"source size: {human_size(self.total_bytes)}, " + f"files: {self.total_files}") # stdout 丢弃 (进度帧缓冲不可靠), stderr 单独读错误 self.proc = subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - text=True, bufsize=1, + text=True, bufsize=1, start_new_session=True, ) # 轮询线程: 每秒统计目标目录大小驱动进度 self._stop.clear() @@ -642,6 +661,8 @@ class TaskRunner: poll = 0 while not self._stop.is_set(): time.sleep(1) + if self._paused: + continue # 暂停期间不轮询 done = dir_size(dst) now = time.time() speed = (done - last) / max(0.1, now - last_t) @@ -651,6 +672,9 @@ class TaskRunner: if self.total_bytes: self.stats.pct = min(100.0, done / self.total_bytes * 100) last, last_t = done, now + # 每 3 秒更新文件计数 (find 开销控制) + if self.total_files and poll % 3 == 0: + self.stats.files_done = count_files(dst) self._emit_stats() poll += 1 # 每 5 秒一条进度快照到日志栏 @@ -687,10 +711,39 @@ class TaskRunner: else: self.on_log(f"[red]✘ 校验发现 {diffs} 个差异 (rc={rc})[/]") + def pause(self) -> None: + """暂停: SIGSTOP 整个进程组""" + if self._paused: + return + proc = self.proc + if proc and proc.poll() is None: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGSTOP) + except (ProcessLookupError, PermissionError, OSError): + proc.send_signal(signal.SIGSTOP) + self._paused = True + + def resume(self) -> None: + """继续: SIGCONT 整个进程组""" + if not self._paused: + return + proc = self.proc + if proc and proc.poll() is None: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGCONT) + except (ProcessLookupError, PermissionError, OSError): + proc.send_signal(signal.SIGCONT) + self._paused = False + def stop(self) -> None: self._stop.set() if self.proc and self.proc.poll() is None: - self.proc.terminate() + if self._paused: + self.resume() + try: + os.killpg(os.getpgid(self.proc.pid), signal.SIGTERM) + except (ProcessLookupError, PermissionError, OSError): + self.proc.terminate() # ---------------------------------------------------------------- 冲突交互 @@ -861,7 +914,7 @@ class TransferApp(App): yield RichLog(id="file-log", highlight=True, markup=True, max_lines=200) yield RichLog(id="error-log", highlight=True, markup=True, max_lines=100) with Horizontal(id="cmd-bar"): - yield Input(placeholder="传输任务: 源路径 目标路径 [--backend rsync|rclone|tar --bwlimit 30000 ...] | help / quit", id="cmd") + yield Input(placeholder="传输任务: 源 目标 [参数] | stop / pause / continue / quit / help", id="cmd") yield Footer() def on_mount(self) -> None: @@ -890,6 +943,9 @@ class TransferApp(App): if low in ("q", "quit", "exit"): self.exit() return + if low in ("stop", "pause", "continue", "cont", "resume"): + self._handle_transport_cmd(low) + return if low in ("h", "help", "?"): self._show_help() return @@ -902,6 +958,21 @@ class TransferApp(App): return self._parse_and_start(text) + def _handle_transport_cmd(self, cmd: str) -> None: + """传输控制命令: stop / pause / continue""" + if not self.runner: + self._flash("[yellow]当前无任务[/]") + return + if cmd == "stop": + self.runner.stop() + self._flash("[red]已发送停止信号[/]") + elif cmd == "pause": + self.runner.pause() + self._flash("[yellow]已暂停 (输入 continue 恢复)[/]") + elif cmd in ("continue", "cont", "resume"): + self.runner.resume() + self._flash("[green]已恢复[/]") + def _handle_ask_input(self, text: str) -> None: choice = text.strip().lower() ask = self._ask @@ -980,7 +1051,7 @@ class TransferApp(App): ask = self._ask self._ask = None self.query_one("#cmd", Input).placeholder = ( - "传输任务: 源路径 目标路径 [--backend rsync|rclone|tar --bwlimit 30000 ...] | help / quit") + "传输任务: 源 目标 [参数] | stop / pause / continue / quit / help") try: skipped = apply_policy(ask["updates"], "ask", ask["answers"]) except KeyboardInterrupt: @@ -1063,6 +1134,8 @@ class TransferApp(App): parts.append(f"ETA {int(st.eta_seconds // 60)}:{int(st.eta_seconds % 60):02d}") if st.files_total: parts.append(f"文件 {st.files_done}/{st.files_total}") + self.query_one("#file-bar", ProgressBar).update( + progress=min(100.0, st.files_done / st.files_total * 100)) self.query_one("#stats-line", Static).update(" ".join(parts)) def _on_file(self, f: FileProgress) -> None: