Python 批处理命令行怎么做进度与安全取消：TTY 降级和检查点实战

一个 Python 批处理脚本要跑两小时，最怕的不是慢，而是终端里一片安静：操作者不知道它还在扫文件、卡在某个目录，还是已经悄悄失败。很多脚本补了一条会刷屏的进度条，又在 CI 日志里留下几万行控制字符；有人按下 Ctrl+C，下一次只能从头再跑。把进度、取消和断点当成命令行交互的一部分，脚本才适合交给别人长期使用。

用户真正需要的是“可判断的等待”

假设有个图片归档脚本，需要把 incoming/ 下的文件复制到归档目录。操作者关心的不是某个炫酷动画，而是四件小事：当前做到了哪里、还会等多久、能不能安全停下、停下后是否会重复处理。

先把这些问题写成可观察的状态，而不要先挑进度条库。一个有用的输出应包含完成数、总数、当前文件的短名称，以及明确的结束状态。失败时要把错误文件写清楚；取消时要告诉用户检查点在哪里。

先辨认终端状态，再决定怎样提示

sys.stderr.isatty() 是个很实用的分界线。它为真时，脚本面对的是可交互终端，可以用 \r 把进度留在一行；为假时，通常意味着日志被收集、重定向，或脚本跑在自动化环境里，此时普通文本更可靠。

这里别把“无终端”理解成次等场景。很多真正重要的批处理都由 CI、容器任务或计划任务触发。那些环境最需要少而稳定的日志，方便检索“处理到第几项”和“哪个文件失败”。

用标准库拼出可取消、可恢复的批处理骨架

下面的示例不依赖第三方库。为了让恢复顺序确定，先把输入路径排序；目标目录应放在源目录之外，避免复制后的文件又被下一轮扫描到。实际项目还应按文件大小、后缀和权限要求补充过滤规则。

from __future__ import annotations

import argparse
import json
import shutil
import signal
import sys
import time
from pathlib import Path

stop_requested = False


def ask_to_stop(signum, frame):
    global stop_requested
    stop_requested = True


def load_completed(checkpoint: Path) -> int:
    if not checkpoint.exists():
        return 0
    data = json.loads(checkpoint.read_text(encoding="utf-8"))
    return int(data.get("completed", 0))


def save_checkpoint(checkpoint: Path, completed: int, total: int) -> None:
    payload = {"completed": completed, "total": total}
    checkpoint.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")


def report(done: int, total: int, name: str, interactive: bool, last: float) -> float:
    now = time.monotonic()
    if interactive:
        if done  int:
    files = sorted(path for path in source.rglob("*") if path.is_file())
    total = len(files)
    completed = min(load_completed(checkpoint), total)
    interactive = sys.stderr.isatty()
    last_report = 0.0

    for index, path in enumerate(files[completed:], start=completed + 1):
        if stop_requested:
            save_checkpoint(checkpoint, index - 1, total)
            print(f"\nstopped; checkpoint={checkpoint}", file=sys.stderr)
            return 130

        relative = path.relative_to(source)
        destination = target / relative
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(path, destination)
        last_report = report(index, total, relative.name, interactive, last_report)

        if index % 100 == 0:
            save_checkpoint(checkpoint, index, total)

    if interactive:
        sys.stderr.write("\n")
    checkpoint.unlink(missing_ok=True)
    print(f"completed={total}", file=sys.stderr)
    return 0


parser = argparse.ArgumentParser(description="copy files with progress and resume support")
parser.add_argument("source", type=Path)
parser.add_argument("target", type=Path)
parser.add_argument("--checkpoint", type=Path, default=Path(".copy-progress.json"))
args = parser.parse_args()

signal.signal(signal.SIGINT, ask_to_stop)
raise SystemExit(copy_batch(args.source, args.target, args.checkpoint))

代码里有两个刻意的取舍。第一，信号处理函数只改一个布尔标记；主循环看到标记后再写检查点。第二，检查点不是每个文件都写一次，而是每 100 个文件写一次，同时在取消时补写一次。频繁落盘能缩小重复范围，却也会让磁盘成为新的瓶颈，具体间隔应按文件体量和可接受的重跑量调整。

让 Ctrl+C 变成可控的中断，而不是半截现场

Python 的信号处理并不适合塞进复制、网络请求或清理逻辑。SIGINT 到来后，主线程会在合适的 Python 执行点运行处理函数；因此更稳的办法是让处理函数只提出停止请求，把收尾工作放回常规控制流。

检查点内容也别只存一个数字。上面的最小示例适合输入清单固定的单次任务；如果输入目录可能在两次运行之间变化，可以在检查点里再写入输入文件数、排序后的前后路径摘要或任务批次号。恢复前对不上，就拒绝续跑，宁可重新扫描也不要错位复制。

性能与可读性要一起验收

进度刷新本身也会消耗资源。百万级小文件场景里，每次循环都刷新终端，I/O 可能比复制逻辑更显眼。示例用 0.1 秒做交互终端节流，并用 100 项作为非交互日志间隔；这两个数是起点，不是固定规则。

验收时不妨同时跑三种方式：直接在终端运行、把 stdout 管道给另一个程序、把 stderr 重定向到日志文件。确认结果数据里没有混入进度文本，确认日志能看懂处理位置，也确认按一次 Ctrl+C 后可以从检查点继续。只有第一种能跑通，还不算完成。

边界状态别留给使用者猜

• 目标磁盘满：保留失败路径和异常信息，不要把检查点推进到未复制成功的文件之后。
• 输入列表为空：明确输出 completed=0，并删除遗留检查点。
• 单个文件无权限：根据业务选择停止或记录后跳过；两种策略都要在参数说明里说清楚。
• 多个任务同时运行：检查点文件应带批次名或锁定策略，不能共享默认文件名。

常见问题

为什么把进度写到 stderr，而不是 stdout？

stdout 往往承载 CSV、JSON 或路径列表等可被下游程序读取的数据。把状态写到 stderr，可以让人看见进度，也不会破坏管道里的结果格式。

按 Ctrl+C 后为什么不立刻退出？

主循环需要一个短暂的机会写检查点、关闭当前文件并输出停止位置。这个窗口应尽量短，不能在取消路径里做长时间网络操作。

检查点能保证完全不重复复制吗？

不能单靠一个完成序号保证。输入目录内容改变、文件顺序改变或复制在写入中断，都可能产生差异。需要严格幂等时，应再记录目标文件校验信息或使用临时文件后原子改名。

什么时候该换用 tqdm 之类的库？

需要多任务进度、嵌套进度或更丰富的交互时，成熟库能减少界面细节工作。无论使用哪种工具，TTY 降级、stderr 分流、取消和恢复策略仍然需要由业务代码决定。

把脚本从“能跑”推进到“能交接”

批处理脚本的体验不靠一条漂亮进度条，而靠状态始终可判断：交互时不刷屏，自动化时不污染日志，取消后有明确检查点，恢复前能发现输入已变化。把这些边界收进同一个命令行入口，脚本交给同事、CI 或定时任务时就少了很多不必要的猜测。