如何配置Python邮件警报：src目录无Combine.csv时不发送错误邮件

问题 #

现有Python脚本会在执行失败时发送错误警报，但存在误触发场景：当C:/R目录下不存在Combine.csv文件时，脚本会因读取文件失败触发警报，但这并非脚本执行逻辑出错，只是无待处理文件。需要调整邮件警报逻辑，避免这种场景下发送错误邮件。

原代码 #

import pandas as pd
import smtplib
from email.message import EmailMessage
import glob
import os
import shutil
df = pd.read_fwf(r'Combine.csv', header=None)
end_str = '#--- END --'
cols_to_check = ["0"]


def email_alert(subject,body,to):
    msg = EmailMessage()
    msg.set_content(body)
    msg['subject'] = subject
    msg['to'] = to
        
    user = "DataScienceScriptAlerts@afonsfu.com"
    msg['from'] = user
    
        
    server = smtplib.SMTP("smtpray.corp.group.com", 25)
    server.starttls()
    #server.login(user,password)
    server.send_message(msg)
        
    server.quit()
src = r'C:/R'
dest = r'C:/R/Failed Scripts'
if __name__ == '__main__':

    for col in cols_to_check:
        if not df[0].str.contains(end_str).any():
            body = "The Combine.py script in IV had errors on the last execution" + col + "."
            print(body)
            email_alert("Combine failure alert",body,"htvldba@group.com")
        if not df[0].str.contains(end_str).any():
                for file_path in glob.glob(os.path.join(src,'*.Rout'), recursive=True):
                    new_path = os.path.join(dest, os.path.basename(file_path))
                    shutil.copy(file_path, new_path) 

修改后的代码 #

import pandas as pd
import smtplib
from email.message import EmailMessage
import glob
import os
import shutil

end_str = '#--- END --'
cols_to_check = ["0"]
src = r'C:/R'
csv_path = os.path.join(src, 'Combine.csv')  # 明确Combine.csv的完整路径

def email_alert(subject, body, to):
    msg = EmailMessage()
    msg.set_content(body)
    msg['subject'] = subject
    msg['to'] = to
        
    user = "DataScienceScriptAlerts@afonsfu.com"
    msg['from'] = user
    
    server = smtplib.SMTP("smtpray.corp.group.com", 25)
    server.starttls()
    #server.login(user,password)
    server.send_message(msg)
    server.quit()

if __name__ == '__main__':
    dest = r'C:/R/Failed Scripts'
    
    # 先检查Combine.csv是否存在
    if not os.path.exists(csv_path):
        print(f"未找到文件: {csv_path}，无需处理")
        exit()
    
    # 文件存在时再读取
    try:
        df = pd.read_fwf(csv_path, header=None)
    except Exception as e:
        print(f"读取文件失败: {str(e)}")
        exit()
    
    for col in cols_to_check:
        # 仅当文件存在且内容不符合要求时发送警报
        if not df[0].str.contains(end_str).any():
            body = f"The Combine.py script in IV had errors on the last execution{col}."
            print(body)
            email_alert("Combine failure alert", body, "htvldba@group.com")
            
            # 复制Rout文件到失败目录
            for file_path in glob.glob(os.path.join(src, '*.Rout'), recursive=True):
                new_path = os.path.join(dest, os.path.basename(file_path))
                shutil.copy(file_path, new_path)

关键修改点 #

• 提前检查文件存在性：在读取文件前，用os.path.exists()检查Combine.csv是否存在，不存在则直接退出，不触发后续逻辑
• 明确文件路径：将Combine.csv的路径关联到src目录，避免当前工作目录差异导致的路径问题
• 异常捕获：对文件读取过程加异常捕获，处理文件损坏等其他读取失败场景，避免无意义警报
• 逻辑合并：将发送警报和复制文件的逻辑合并到同一个条件判断下，减少重复判断

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