如何在任务计划程序中Python脚本执行失败时获取邮件/WhatsApp通知？

一、邮件通知（文件层面全覆盖） #

直接在Python脚本中用try-except包裹核心执行逻辑，捕获所有未处理的异常并触发邮件发送。也可以封装成装饰器，方便复用在多个任务脚本里。

示例1：基础脚本内直接实现 #

import smtplib
from email.mime.text import MIMEText
from email.header import Header

def send_failure_email(error_msg):
    # 配置邮箱参数
    sender = "your_email@example.com"
    receiver = "alert_email@example.com"
    smtp_server = "smtp.example.com"
    smtp_port = 587
    username = "your_email@example.com"
    password = "your_email_auth_code"  # 建议用邮箱授权码而非登录密码

    # 构造邮件内容
    msg = MIMEText(f"Python任务执行失败，错误信息：\n{error_msg}", "plain", "utf-8")
    msg["From"] = Header("任务监控系统", "utf-8")
    msg["To"] = Header("运维人员", "utf-8")
    msg["Subject"] = Header("【紧急】Python任务执行失败通知", "utf-8")

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as server:
            server.starttls()
            server.login(username, password)
            server.sendmail(sender, receiver, msg.as_string())
    except Exception as e:
        print(f"发送通知邮件失败: {e}")

# 核心任务逻辑
if __name__ == "__main__":
    try:
        # 这里编写你的任务代码（数据处理、接口调用等）
        print("开始执行任务...")
        # 模拟执行失败场景
        raise ValueError("模拟数据格式校验失败")
        print("任务执行完成")  # 若前面报错，该行不会执行
    except Exception as e:
        send_failure_email(str(e))
        raise  # 可选：重新抛出异常，让调度器捕获任务失败状态

示例2：复用装饰器实现 #

如果有多个任务脚本，把通知逻辑封装成装饰器，每个脚本的主函数直接装饰即可：

import smtplib
from email.mime.text import MIMEText
from email.header import Header
from functools import wraps

def notify_on_failure(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 发送邮件逻辑
            sender = "your_email@example.com"
            receiver = "alert_email@example.com"
            smtp_server = "smtp.example.com"
            smtp_port = 587
            username = "your_email@example.com"
            password = "your_email_auth_code"

            msg = MIMEText(f"任务 {func.__name__} 执行失败，错误信息：\n{str(e)}", "plain", "utf-8")
            msg["From"] = Header("任务监控系统", "utf-8")
            msg["To"] = Header("运维人员", "utf-8")
            msg["Subject"] = Header(f"【紧急】任务 {func.__name__} 执行失败", "utf-8")

            try:
                with smtplib.SMTP(smtp_server, smtp_port) as server:
                    server.starttls()
                    server.login(username, password)
                    server.sendmail(sender, receiver, msg.as_string())
            except Exception as notify_e:
                print(f"发送通知邮件失败: {notify_e}")
            raise
    return wrapper

# 任务脚本中使用装饰器
@notify_on_failure
def main_task():
    print("开始执行任务...")
    # 模拟失败场景
    raise IOError("模拟文件读取权限不足")
    print("任务完成")

if __name__ == "__main__":
    main_task()

二、WhatsApp通知方案 #

利用第三方工具的API实现WhatsBot消息推送，以Twilio为例：

示例：Twilio实现WhatsApp通知 #

from twilio.rest import Client
from functools import wraps

def send_whatsapp_alert(error_msg):
    # 配置Twilio参数
    account_sid = "your_twilio_account_sid"
    auth_token = "your_twilio_auth_token"
    from_whatsapp_number = "whatsapp:+14155238886"  # Twilio官方测试号码
    to_whatsapp_number = "whatsapp:+86你的手机号"

    client = Client(account_sid, auth_token)
    try:
        message = client.messages.create(
            body=f"Python任务执行失败，错误信息：\n{error_msg}",
            from_=from_whatsapp_number,
            to=to_whatsapp_number
        )
        print(f"通知已发送，消息ID: {message.sid}")
    except Exception as e:
        print(f"发送WhatsApp通知失败: {e}")

# 封装成装饰器复用
def notify_whatsapp_on_failure(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            send_whatsapp_alert(str(e))
            raise
    return wrapper

# 任务中使用装饰器
@notify_whatsapp_on_failure
def data_process_task():
    print("开始数据处理任务...")
    # 模拟失败场景
    raise ZeroDivisionError("模拟除数为零错误")
    print("数据处理完成")

if __name__ == "__main__":
    data_process_task()

三、任务调度层面的补充（可选） #

如果使用cron、Celery或Airflow等调度工具，可结合工具本身的失败回调机制：

• Cron：在调度命令后追加错误捕获逻辑，比如python your_task.py || python send_alert.py "任务执行失败"
• Airflow：给DAG的Task设置on_failure_callback函数，直接调用通知逻辑

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