feat: TUI 命令 stop/pause/continue + 单文件进度条改为文件计数轮询

This commit is contained in:
AskaEth committed 2026-08-22 14:03:40 +08:00
1 parent 7ab32bad24
commit 81857d6343
3 files changed
+145 -9

No files matched your search

+7 -3
View File
@@ -37,11 +37,15 @@ python3 transfer.py --mode cli /src/ /dst/ --backend rclone --bwlimit 8M
``` ```
/src /dst --backend rclone --bwlimit 8M --transfers 4 # 启动传输任务 /src /dst --backend rclone --bwlimit 8M --transfers 4 # 启动传输任务
help # 帮助 stop # 停止当前任务
clear # 清空日志面板 pause # 暂停 (SIGSTOP)
quit # 退出 continue # 恢复暂停
help / clear / quit
``` ```
单文件进度条显示**文件计数进度**(目标文件数/源文件总数,每 3 秒刷新),
大文件库下按文件粒度推进。
### 参数 ### 参数
| 参数 | 默认 | 说明 | | 参数 | 默认 | 说明 |
+59
View File
@@ -1,6 +1,7 @@
"""transfer.py 核心逻辑测试:解析器 / 日志轮转 / 参数解析 / 后端命令构建""" """transfer.py 核心逻辑测试:解析器 / 日志轮转 / 参数解析 / 后端命令构建"""
import re import re
import os import os
import subprocess
import sys import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 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(): def test_rsync_verify_cmd_no_bwlimit():
cmd = transfer.RsyncBackend.build_verify_cmd("a/", "b/") cmd = transfer.RsyncBackend.build_verify_cmd("a/", "b/")
assert "--bwlimit" not in " ".join(cmd) 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 # 目录不计
+79 -6
View File
@@ -13,6 +13,7 @@ import argparse
import os import os
import re import re
import shlex import shlex
import signal
import subprocess import subprocess
import sys import sys
import threading import threading
@@ -67,6 +68,18 @@ def dir_size(path: str) -> int:
return 0 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]: def speed_stats(history) -> tuple[float, float, float]:
"""速度统计: (当前, 峰值, 平均) 字节/秒""" """速度统计: (当前, 峰值, 平均) 字节/秒"""
if not history: if not history:
@@ -463,7 +476,10 @@ class TaskRunner:
self.proc: subprocess.Popen | None = None self.proc: subprocess.Popen | None = None
self.stats = ProgressStats() self.stats = ProgressStats()
self._stop = threading.Event() self._stop = threading.Event()
self._paused = False
self.stats_frames = 0 self.stats_frames = 0
self.total_bytes = 0
self.total_files = 0
# -- 命令构建 # -- 命令构建
def build_command(self) -> list[str]: def build_command(self) -> list[str]:
@@ -582,7 +598,7 @@ class TaskRunner:
self.on_log(f"[yellow]补传轮次 {attempt}/{self.cfg.retry}[/]") self.on_log(f"[yellow]补传轮次 {attempt}/{self.cfg.retry}[/]")
time.sleep(3) # 等守护恢复设备 time.sleep(3) # 等守护恢复设备
rc = self._run_once() rc = self._run_once()
if rc == 0: if rc == 0 or self._stop.is_set():
break break
self.lm.write(f"attempt {attempt + 1} failed rc={rc}") self.lm.write(f"attempt {attempt + 1} failed rc={rc}")
if self.on_error and attempt < self.cfg.retry: if self.on_error and attempt < self.cfg.retry:
@@ -600,11 +616,14 @@ class TaskRunner:
f"resume={self.cfg.resume} checksum={self.cfg.checksum}") f"resume={self.cfg.resume} checksum={self.cfg.checksum}")
# 源总大小 (进度分母) # 源总大小 (进度分母)
self.total_bytes = dir_size(self.cfg.source.rstrip("/")) 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 单独读错误 # stdout 丢弃 (进度帧缓冲不可靠), stderr 单独读错误
self.proc = subprocess.Popen( self.proc = subprocess.Popen(
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
text=True, bufsize=1, text=True, bufsize=1, start_new_session=True,
) )
# 轮询线程: 每秒统计目标目录大小驱动进度 # 轮询线程: 每秒统计目标目录大小驱动进度
self._stop.clear() self._stop.clear()
@@ -642,6 +661,8 @@ class TaskRunner:
poll = 0 poll = 0
while not self._stop.is_set(): while not self._stop.is_set():
time.sleep(1) time.sleep(1)
if self._paused:
continue # 暂停期间不轮询
done = dir_size(dst) done = dir_size(dst)
now = time.time() now = time.time()
speed = (done - last) / max(0.1, now - last_t) speed = (done - last) / max(0.1, now - last_t)
@@ -651,6 +672,9 @@ class TaskRunner:
if self.total_bytes: if self.total_bytes:
self.stats.pct = min(100.0, done / self.total_bytes * 100) self.stats.pct = min(100.0, done / self.total_bytes * 100)
last, last_t = done, now 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() self._emit_stats()
poll += 1 poll += 1
# 每 5 秒一条进度快照到日志栏 # 每 5 秒一条进度快照到日志栏
@@ -687,10 +711,39 @@ class TaskRunner:
else: else:
self.on_log(f"[red]✘ 校验发现 {diffs} 个差异 (rc={rc})[/]") 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: def stop(self) -> None:
self._stop.set() self._stop.set()
if self.proc and self.proc.poll() is None: 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="file-log", highlight=True, markup=True, max_lines=200)
yield RichLog(id="error-log", highlight=True, markup=True, max_lines=100) yield RichLog(id="error-log", highlight=True, markup=True, max_lines=100)
with Horizontal(id="cmd-bar"): 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() yield Footer()
def on_mount(self) -> None: def on_mount(self) -> None:
@@ -890,6 +943,9 @@ class TransferApp(App):
if low in ("q", "quit", "exit"): if low in ("q", "quit", "exit"):
self.exit() self.exit()
return return
if low in ("stop", "pause", "continue", "cont", "resume"):
self._handle_transport_cmd(low)
return
if low in ("h", "help", "?"): if low in ("h", "help", "?"):
self._show_help() self._show_help()
return return
@@ -902,6 +958,21 @@ class TransferApp(App):
return return
self._parse_and_start(text) 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: def _handle_ask_input(self, text: str) -> None:
choice = text.strip().lower() choice = text.strip().lower()
ask = self._ask ask = self._ask
@@ -980,7 +1051,7 @@ class TransferApp(App):
ask = self._ask ask = self._ask
self._ask = None self._ask = None
self.query_one("#cmd", Input).placeholder = ( self.query_one("#cmd", Input).placeholder = (
"传输任务: 源路径 目标路径 [--backend rsync|rclone|tar --bwlimit 30000 ...] | help / quit") "传输任务: 源 目标 [参数] | stop / pause / continue / quit / help")
try: try:
skipped = apply_policy(ask["updates"], "ask", ask["answers"]) skipped = apply_policy(ask["updates"], "ask", ask["answers"])
except KeyboardInterrupt: except KeyboardInterrupt:
@@ -1063,6 +1134,8 @@ class TransferApp(App):
parts.append(f"ETA {int(st.eta_seconds // 60)}:{int(st.eta_seconds % 60):02d}") parts.append(f"ETA {int(st.eta_seconds // 60)}:{int(st.eta_seconds % 60):02d}")
if st.files_total: if st.files_total:
parts.append(f"文件 {st.files_done}/{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)) self.query_one("#stats-line", Static).update(" ".join(parts))
def _on_file(self, f: FileProgress) -> None: def _on_file(self, f: FileProgress) -> None: