如何在Snakemake规则中嵌入自定义Python函数及处理多场景需求

步骤1：读取配置与样本表，构建有效文件映射 #

先在 Snakefile 开头完成数据加载与文件校验逻辑，用 pandas 处理样本表，同时检查目标 fastq 文件的存在性：

import pandas as pd
import os

# 加载配置与样本TSV
configfile: "config.yml"
samples_df = pd.read_csv(config["sample_tsv"], sep="\t")

# 校验并获取有效fastq文件路径
def get_valid_fastq(sample_row):
    # 假设TSV包含sample_name（样本名）和fastq_filename（文件名）两列，按需调整列名
    fq_name = sample_row["fastq_filename"]
    # 生成data/A、data/B下的可能路径
    candidate_paths = [
        os.path.join("data", "A", fq_name),
        os.path.join("data", "B", fq_name)
    ]
    # 筛选实际存在的文件
    valid_files = [path for path in candidate_paths if os.path.exists(path)]
    if not valid_files:
        raise ValueError(f"样本 {sample_row['sample_name']} 的fastq文件 {fq_name} 在data/A/B中均不存在")
    # 返回第一个有效路径（若允许多文件可返回列表）
    return valid_files[0]

# 构建「样本名 -> 有效fastq路径」的映射字典
sample_fastq_map = {
    row["sample_name"]: get_valid_fastq(row)
    for _, row in samples_df.iterrows()
}

# 提取所有样本名，作为规则的遍历目标
ALL_SAMPLES = list(sample_fastq_map.keys())

步骤2：定义规则实现样本名输出与文件调用 #

通过通配符和映射字典，实现输出按样本名命名，同时调用对应fastq文件：

# 总规则：指定所有要生成的输出文件
rule all:
    input:
        expand("{sample}_output.txt", sample=ALL_SAMPLES)

# 核心处理规则
rule process_sample:
    input:
        # 根据通配符中的样本名，从映射字典中获取对应fastq路径
        fastq=lambda wildcards: sample_fastq_map[wildcards.sample]
    output:
        "{sample}_output.txt"
    shell:
        """
        # 替换为你的实际处理命令，示例为打印文件路径到输出
        echo "处理文件: {input.fastq}" > {output}
        # 实际场景示例：fastqc {input.fastq} -o {output}.dir
        """

关键说明 #

• 列名适配：如果你的TSV中样本名、fastq文件名的列名不是sample_name/fastq_filename，直接修改代码中对应的索引即可。
• 多文件支持：若单个样本对应R1/R2两个fastq文件，可修改get_valid_fastq返回路径列表，规则输入改为fastq=lambda wildcards: sample_fastq_map[wildcards.sample]，shell中用{input.fastq[0]}、{input.fastq[1]}调用。
• 调试提示：可在get_valid_fastq中添加print(f"找到有效文件: {valid_files[0]}")，方便验证文件匹配是否正确。

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