能否在Snakemake规则中使用类似try-except的异常处理机制？

要实现指定规则rule a执行成功或失败时发送对应邮件，完全可以借助Snakemake的Python执行能力模拟try-except逻辑——别在shell()指令里硬混Python代码，因为shell()是用来执行系统命令的，里面的Python代码不会被Snakemake的解释器处理。

步骤1：封装邮件发送函数 #

在Snakefile开头定义一个通用的邮件发送函数，用Python原生库或你熟悉的邮件工具实现：

import smtplib
from email.mime.text import MIMEText

def send_email(subject, content):
    # 替换成你的实际邮件配置
    sender = "your_work_email@example.com"
    receiver = "alert_recipient@example.com"
    smtp_server = "smtp.example.com"
    smtp_port = 587
    username = "your_work_email@example.com"
    password = "your_email_app_password"

    msg = MIMEText(content, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = sender
    msg["To"] = receiver

    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(username, password)
        server.sendmail(sender, receiver, msg.as_string())

步骤2：在rule a中用try-except处理执行逻辑 #

放弃shell()，改用run块（纯Python执行环境），通过subprocess执行原shell命令，同时捕获执行状态：

rule a:
    input:
        "raw_data.txt"
    output:
        "processed_result.txt"
    run:
        import subprocess
        try:
            # 执行你的实际任务命令，替换成对应的操作
            subprocess.run(
                ["python", "process_script.py", input[0], output[0]],
                check=True,  # 命令返回非0时抛出异常
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            # 执行成功时发送通知
            send_email(
                subject="规则a执行完成",
                content=f"任务已成功结束，输出文件：{output[0]}"
            )
        except subprocess.CalledProcessError as e:
            # 执行失败时发送错误详情
            error_content = f"规则a执行失败，错误日志：\n{e.stderr}"
            send_email(
                subject="规则a执行失败",
                content=error_content
            )
            # 重新抛出异常，让Snakemake标记规则失败
            raise

补充说明 #

• subprocess.run(check=True)会在命令执行失败（返回非0状态码）时自动抛出CalledProcessError，刚好对应try-except的异常捕获逻辑。
• 捕获异常发送邮件后，必须raise重新抛出异常，否则Snakemake会误认为规则执行成功，导致工作流后续任务不符合预期。
• 如果你的shell命令逻辑复杂，可以写成字符串并用shell=True执行（注意避免命令注入风险）：

  subprocess.run("bash ./complex_process.sh {input} {output}", check=True, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  

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