Snakemake如何对动态生成的不定数量txt文件批量应用处理规则

Snakemake 默认会在工作流启动的 DAG 构建阶段确定所有要生成的目标文件，你当前的写法有两个核心错误：

• rule all 里的 glob 扫描在工作流刚启动时就执行，这时候 script.py 还没运行，目录里根本没有生成任何 step 系列 txt 文件，最终拿到的目标列表是空的，Snakemake 会判定无任务需要执行。
• 把运行 script.py 的规则输出定义为单个通配符文件 {step}.txt，但该脚本一次运行会生成数量未知的多个 txt，和规则声明的输出不匹配，运行时会触发“规则未生成声明输出”的报错。

这种「某一步运行后才能确定后续待处理文件集合」的场景，必须用 Snakemake 内置的 checkpoint（检查点）机制实现。

import glob

def aggregate_target_pngs(wildcards):
    # 阻塞等待运行script.py的检查点执行完成
    checkpoints.run_script.get()
    # 扫描脚本生成的所有step格式txt文件
    all_step_txt = glob.glob("step_000*.txt")
    # 转换为对应png输出路径，返回给rule all作为最终目标
    return [txt_path.replace(".txt", ".png") for txt_path in all_step_txt]

rule all:
    input:
        aggregate_target_pngs

# 用checkpoint代替普通rule标记运行script.py的步骤
checkpoint run_script:
    output:
        # 生成哨兵文件标记脚本执行完成，避免重复运行
        touch("script_executed.flag")
    script:
        "script.py"

rule process_single_step:
    input:
        "{step_name}.txt"
    output:
        "{step_name}.png"
    shell:
        "process.py -in {input} -out {output}"

• 工作流启动后，Snakemake 解析到 rule all 的输入来自聚合函数，会优先触发函数依赖的 run_script 检查点。
• 检查点执行阶段会完整运行 script.py，等所有 step_000n.txt 全部生成后，创建 script_executed.flag 哨兵文件标记步骤完成。
• 检查点运行结束后，Snakemake 重新执行聚合函数，此时目录下已经存在所有生成的 txt 文件，glob 扫描得到完整文件列表，转换为待生成的 png 目标集合。
• Snakemake 针对每个目标 png 自动匹配 process_single_step 规则，批量调用 process.py 完成所有文件的处理，直到所有 png 生成完毕。

• glob 的匹配规则要和 script.py 实际生成的文件名格式对齐，如果文件存放在特定子目录，同步修改 glob 路径即可，避免漏扫或者扫到无关 txt 文件。
• 哨兵文件是推荐配置，只要该文件存在，Snakemake 就不会重复运行 script.py，如果你的脚本本身是幂等的（重复运行不会出错、结果一致），也可以去掉该输出配置。

内容的提问来源于stack exchange，提问作者ipcamit