3. 命令行参数解析与任务配置

批处理脚本一旦要处理“多组参数、多批数据、多种运行模式”，就不能再靠脚本里写死的常量。命令行参数负责表达本次运行的临时选择，配置文件负责承载可复用、可版本管理的参数集，两者配合才能让同一个脚本既灵活又可控。本节围绕 Python 标准库 argparse 与配置文件（以 JSON 为主，兼顾 INI）展开，目标是让读者能写出“命令行覆盖配置、配置提供默认值、参数校验清晰”的批处理入口。

一、参数分层的设计原则

先明确三类参数的归属，避免把所有东西都塞进命令行：

• 运行控制参数：输入目录、输出目录、并发数、日志级别、是否试运行（dry-run）。这类参数每次运行可能不同，适合放在命令行。
• 任务参数集：一批待处理的任务列表，例如每个任务的数据文件名、模型阶数、阈值、采样率。数量多、结构复杂，适合放在配置文件。
• 环境与常量：路径前缀、单位换算系数、版本号。可放在配置文件的全局段，也可由环境变量注入。

推荐的优先级是：命令行显式指定 > 配置文件 > 代码内默认值。这样既保证默认可用，又允许临时覆盖，且覆盖行为可追溯。

二、用 argparse 构建入口

下面是一个可直接运行的入口骨架，包含子命令、类型校验和默认值处理：

import argparse, json, pathlib

def build_parser():
    p = argparse.ArgumentParser(description="批量信号处理脚本")
    p.add_argument("-c", "--config", type=pathlib.Path,
                   default=pathlib.Path("config/tasks.json"),
                   help="任务配置文件路径")
    p.add_argument("-i", "--input-dir", type=pathlib.Path, help="覆盖配置中的输入目录")
    p.add_argument("-o", "--output-dir", type=pathlib.Path, help="覆盖配置中的输出目录")
    p.add_argument("-j", "--jobs", type=int, default=None, help="并发进程数")
    p.add_argument("--dry-run", action="store_true", help="只打印任务，不实际执行")
    p.add_argument("--log-level", choices=["DEBUG", "INFO", "WARNING", "ERROR"],
                   default="INFO")
    return p

def load_config(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def merge_args(cfg, args):
    # 命令行非 None 的字段覆盖配置
    if args.input_dir is not None:
        cfg["input_dir"] = str(args.input_dir)
    if args.output_dir is not None:
        cfg["output_dir"] = str(args.output_dir)
    if args.jobs is not None:
        cfg["jobs"] = args.jobs
    cfg["dry_run"] = args.dry_run
    cfg["log_level"] = args.log_level
    return cfg

if __name__ == "__main__":
    args = build_parser().parse_args()
    cfg = merge_args(load_config(args.config), args)
    print(json.dumps(cfg, ensure_ascii=False, indent=2))

关键点有三处：其一，--jobs 的默认值设为 None 而不是具体数字，这样才能区分“用户没传”和“用户传了与配置相同的值”；其二，--dry-run 用 store_true，天然是布尔开关；其三，choices 让非法日志级别在解析阶段就报错，而不是运行到一半才失败。

三、配置文件的结构

配置文件建议分为“全局段 + 任务列表”两层。全局段放目录、并发、日志等；任务列表每项是一个独立参数集，带唯一 id 便于日志定位：

{
  "input_dir": "data/raw",
  "output_dir": "data/out",
  "jobs": 4,
  "log_level": "INFO",
  "tasks": [
    {"id": "sensor_a", "file": "a.csv", "order": 4, "threshold": 0.35},
    {"id": "sensor_b", "file": "b.csv", "order": 6, "threshold": 0.28},
    {"id": "sensor_c", "file": "c.csv", "order": 4, "threshold": 0.40}
  ]
}

如果团队更习惯 INI，可用 configparser 读取，把每个任务写成 [task.sensor_a] 这样的段，但要注意 INI 所有值都是字符串，需要自行转换类型。JSON 在嵌套结构和类型保持上更省事，因此批处理任务集优先选 JSON 或 YAML。

四、参数校验与失败前置

解析完参数不等于参数合法。批处理脚本应在启动阶段集中校验，把错误挡在第一批任务执行之前：

• 输入目录是否存在、是否可读；输出目录不存在则创建。
• 任务 id 是否唯一，file 是否在输入目录下真实存在。
• 数值参数范围，例如 order 为正整数、threshold 在 0 到 1 之间。
• jobs 不超过 CPU 核数或业务允许上限。

校验失败时，用 parser.error() 或抛出带任务 id 的异常，让用户一眼看到是哪一项配置有问题。下面的 SVG 展示了从命令行与配置文件汇合、校验、到任务分发的完整数据流：

五、把参数集落到执行函数

合并与校验完成后，主循环只消费“已确定的配置”，不再回头读命令行：

def run_task(task, cfg):
    src = pathlib.Path(cfg["input_dir"]) / task["file"]
    dst = pathlib.Path(cfg["output_dir"]) / f"{task['id']}.npz"
    if cfg["dry_run"]:
        print(f"[dry-run] {task['id']}: {src} -> {dst}")
        return
    # 实际处理：读取、按 order/threshold 计算、保存
    ...

for task in cfg["tasks"]:
    run_task(task, cfg)

这种写法让 run_task 成为纯函数式的处理单元，便于单元测试：构造一个 task 字典和 cfg 字典即可，无需模拟命令行。并发时，把 cfg["jobs"] 传给 ProcessPoolExecutor(max_workers=cfg["jobs"])，每个任务独立提交，日志中带上 task["id"] 即可定位。

六、常见坑与规避

• 布尔参数误用：--dry-run True 会被当成位置参数报错，布尔开关应使用 store_true。
• 相对路径歧义：配置文件里的相对路径应相对于配置文件所在目录解析，而不是当前工作目录，否则换目录运行就找不到数据。
• 默认值掩盖覆盖：给 --jobs 设默认值 4 会导致无法判断用户是否显式指定，应设 None 再合并。
• 配置与命令行单位不一致：例如配置里阈值是百分比，命令行传小数，需在文档和校验中统一约定。