大家好,欢迎来到IT知识分享网。
自动化脚本案例
一、实现功能
1. 邮件附件处理:包含常用文件格式的自动识别,或者自己指定处理方式。 2. 发送邮件 3. 指定时间发送邮件 或 延时发送邮件。 4. 从文件中导入邮件正文。
二、具体实现
1. 附件处理
file_type这个参数就是为了解决这个问题,为None是,函数自动推断文件格式,根据预设的去处理,但是没有包含所有的文件类型。
file_type取0代表文本文件,取1代表二进制文件
""" @func: 返回邮箱附件对象 @param: file_path: 文件路径 file_type: 文件类型 0:文本文件 1:二进制文件 @return: 附件对象 @author: liuhuan @date: 2024年7月26日 """ def create_attachment(file_path, file_type=None): # 获取文件名和扩展名 file_name = file_path.split('/')[-1] file_extension = file_name.split('.')[-1].lower() # 创建邮件对象 message = MIMEMultipart() # 定义 MIME 类型映射 text_types = ['txt', 'md', 'csv', 'json', 'xml'] binary_types = {
'pdf': 'application/pdf', 'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', 'gif': 'image/gif', 'zip': 'application/zip', 'tar': 'application/x-tar', } try: if file_type is None: if file_extension in text_types: with open(file_path, 'r', encoding='utf-8') as file: att = MIMEText(file.read(), _subtype=file_extension, _charset='utf-8') att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) elif file_extension in binary_types: mime_type = binary_types[file_extension] with open(file_path, 'rb') as file: att = MIMEApplication(file.read(), _subtype=mime_type.split('/')[-1]) att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"不支持的文件格式: {
file_extension}") elif file_type == 0: # 文本文件 if file_extension in text_types: with open(file_path, 'r', encoding='utf-8') as file: att = MIMEText(file.read(), _subtype=file_extension, _charset='utf-8') att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"指定的文件类型为文本,但文件格式不支持: {
file_extension}") elif file_type == 1: # 二进制文件 if file_extension in binary_types: mime_type = binary_types[file_extension] with open(file_path, 'rb') as file: att = MIMEApplication(file.read(), _subtype=mime_type.split('/')[-1]) att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"指定的文件类型为二进制,但文件格式不支持: {
file_extension}") else: raise ValueError("无效的 file_type 参数") except Exception as e: print(f"创建附件时发生错误: {
e}") return None return message
2. 定时发送、延时发送
可发送一个以上附件给一名及以上接收者
""" @func: 可发送一个以上附件给一名及以上接收者 @param: send:发送者的邮箱和授权密码,字典格式, recv:接收者的邮箱列表,list email_info:邮件主题和内容,字典格式 filenames:附件文件名列表,list flag:定时类型标志位,0:定时发送,1:延迟发送 time_param:时间参数,定时发送时为发送时间,延迟发送时为延迟时间 @return: 无返回值 @author: liuhuan @date: 2024年7月26日 """ def send_email(send, recv, email_info, filenames, flag, time_param): # 发送者、接受者、邮件数据解析 sender, password = send["sender"], send["password"] recvers = recv # 现在 recvers 是一个列表 subject, content = email_info["subject"], email_info["content"] # 创建邮件对象 message = MIMEMultipart() message.attach(MIMEText(content, "plain", "utf-8")) # 邮件正文 # 添加附件, if filenames: for filename in filenames: file_type = None # 默认为自动判断文件类型 if len(filename) == 1: file_path = filename[0] else: file_path, file_type = filename att = create_attachment(file_path,file_type) # 调用 create_attachment 函数创建附件 if att: message.attach(att) message['Subject'] = subject # 邮件标题 message['From'] = sender # 发件人 def send_now(): try: smtp = smtplib.SMTP_SSL(email_server, email_port) # 实例化 smtp 服务器 smtp.login(sender, password) # 发件人登录 for recver in recvers: message['To'] = recver # 收件人 smtp.sendmail(sender, [recver], message.as_string()) # 发送邮件 print(f"邮件发送成功给 {
recver}") smtp.close() except Exception as e: print(f"发送邮件时发生错误: {
e}") if flag == 0: # 定时发送 schedule.every().day.at(time_param).do(send_now) print(f"定时任务已设置,每天 {
time_param} 发送邮件。") scheduler_thread = threading.Thread(target=start_scheduler) scheduler_thread.start() elif flag == 1: # 延迟发送 def delayed_send(): if type(time_param) == int: time.sleep(time_param) elif type(time_param) == str: if ":" in time_param: h,s,m = map(int,time_param.split(":")) if ":" in time_param: h,s,m = map(int,time_param.split(":")) time.sleep(h*3600+s*60+m) send_now() thread = threading.Thread(target=delayed_send) thread.start() print(f"邮件将在 {
time_param} 秒后发送。") else: print("无效的 flag 参数") def start_scheduler(): while True: schedule.run_pending() time.sleep(1)
3. 从文件初始化邮件附件
""" @ func:从文件初始化邮件内容 @ param: filename:文件名 subject:邮件主题,默认为文件名 @ return:邮件内容字典,包含主题和内容 @ update: 2024年7月26日 """ def inti_email_from_file(filename,subject=None): if not subject: subject = filename.split("/")[-1].split(".")[0] content = open(filename, "r", encoding="utf-8").read() return {
"subject":subject, "content":content}
三、完整代码
import smtplib #smtp服务器 from email.mime.text import MIMEText #邮件文本 from email.mime.multipart import MIMEMultipart #邮件附件 from email.mime.application import MIMEApplication import schedule import time import threading from datetime import datetime, timedelta """ @func: 返回邮箱附件对象 @param: file_path: 文件路径 file_type: 文件类型 0:文本文件 1:二进制文件 @return: 附件对象 @author: liuhuan @date: 2024年7月26日 """ def create_attachment(file_path, file_type=None): # 获取文件名和扩展名 file_name = file_path.split('/')[-1] file_extension = file_name.split('.')[-1].lower() # 创建邮件对象 message = MIMEMultipart() # 定义 MIME 类型映射 text_types = ['txt', 'md', 'csv', 'json', 'xml'] binary_types = {
'pdf': 'application/pdf', 'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', 'gif': 'image/gif', 'zip': 'application/zip', 'tar': 'application/x-tar', } try: if file_type is None: if file_extension in text_types: with open(file_path, 'r', encoding='utf-8') as file: att = MIMEText(file.read(), _subtype=file_extension, _charset='utf-8') att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) elif file_extension in binary_types: mime_type = binary_types[file_extension] with open(file_path, 'rb') as file: att = MIMEApplication(file.read(), _subtype=mime_type.split('/')[-1]) att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"不支持的文件格式: {
file_extension}") elif file_type == 0: # 文本文件 if file_extension in text_types: with open(file_path, 'r', encoding='utf-8') as file: att = MIMEText(file.read(), _subtype=file_extension, _charset='utf-8') att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"指定的文件类型为文本,但文件格式不支持: {
file_extension}") elif file_type == 1: # 二进制文件 if file_extension in binary_types: mime_type = binary_types[file_extension] with open(file_path, 'rb') as file: att = MIMEApplication(file.read(), _subtype=mime_type.split('/')[-1]) att.add_header('Content-Disposition', 'attachment', filename=file_name) message.attach(att) else: raise ValueError(f"指定的文件类型为二进制,但文件格式不支持: {
file_extension}") else: raise ValueError("无效的 file_type 参数") except Exception as e: print(f"创建附件时发生错误: {
e}") return None return message # send:发送者的邮箱和授权密码 # recv:接收者的邮箱列表 # email_info:邮件主题和内容 # filenames:附件文件名列表 # flag:时间标志位,0:定时发送,1:延迟发送 # time_param:时间参数,定时发送时为发送时间,延迟发送时为延迟时间 """ @func: 可发送一个以上附件给一名及以上接收者 @param: end:发送者的邮箱和授权密码,字典格式, recv:接收者的邮箱列表,list email_info:邮件主题和内容,字典格式 filenames:附件文件名列表,list flag:时间标志位,0:定时发送,1:延迟发送 time_param:时间参数,定时发送时为发送时间,延迟发送时为延迟时间 @return: 无返回值 @author: liuhuan @date: 2024年7月26日 """ def send_email(send, recv, email_info, filenames, flag, time_param): sender, password = send["sender"], send["password"] recvers = recv # 现在 recvers 是一个列表 subject, content = email_info["subject"], email_info["content"] # 创建邮件对象 message = MIMEMultipart() message.attach(MIMEText(content, "plain", "utf-8")) # 邮件正文 # 添加附件, if filenames: for filename in filenames: file_type = None # 默认为自动判断文件类型 if len(filename) == 1: file_path = filename[0] else: file_path, file_type = filename att = create_attachment(file_path,file_type) # 调用 create_attachment 函数创建附件 if att: message.attach(att) message['Subject'] = subject # 邮件标题 message['From'] = sender # 发件人 def send_now(): try: smtp = smtplib.SMTP_SSL(email_server, email_port) # 实例化 smtp 服务器 smtp.login(sender, password) # 发件人登录 for recver in recvers: message['To'] = recver # 收件人 smtp.sendmail(sender, [recver], message.as_string()) # 发送邮件 print(f"邮件发送成功给 {
recver}") smtp.close() except Exception as e: print(f"发送邮件时发生错误: {
e}") if flag == 0: # 定时发送 schedule.every().day.at(time_param).do(send_now) print(f"定时任务已设置,每天 {
time_param} 发送邮件。") scheduler_thread = threading.Thread(target=start_scheduler) scheduler_thread.start() elif flag == 1: # 延迟发送 def delayed_send(): if type(time_param) == int: time.sleep(time_param) elif type(time_param) == str: if ":" in time_param: h,s,m = map(int,time_param.split(":")) if ":" in time_param: h,s,m = map(int,time_param.split(":")) time.sleep(h*3600+s*60+m) send_now() thread = threading.Thread(target=delayed_send) thread.start() print(f"邮件将在 {
time_param} 秒后发送。") else: print("无效的 flag 参数") # 不懂 def start_scheduler(): while True: schedule.run_pending() time.sleep(1) """ @ func:从文件初始化邮件内容 @ param: filename:文件名 subject:邮件主题,默认为文件名 @ return:邮件内容字典,包含主题和内容 @ update: 2024年7月26日 """ def inti_email_from_file(filename,subject=None): if not subject: subject = filename.split("/")[-1].split(".")[0] content = open(filename, "r", encoding="utf-8").read() return {
"subject":subject, "content":content} if __name__ == "__main__": # 从字典初始化user_info send_info={
"sender":"发送者@163.com", "password":"xxxxxx" } recv_info=['接受者','接受者'] # 从文件中导入邮件的正文和主题 email_info= inti_email_from_file("test.txt") # 附件文件路径的列表,以字典的形式存放 # 前者是文件路径,后者是文件类型,0:文本类型, 1:二进制类型 filenames = [("用户推荐商品表.xlsx",),("用户推荐商品表.xlsx",1)] # 不同的邮箱使用不同的服务器和端口,默认都是使用SSL的方式。 email_server = "smtp.163.com" email_port = 994 send_type = 1 # 定时,0;延迟,1 send_email(send_info, recv_info, email_info, filenames, send_type, "00:00:00") # 发送邮件
总结
略。
免责声明:本站所有文章内容,图片,视频等均是来源于用户投稿和互联网及文摘转载整编而成,不代表本站观点,不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益,请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权,请联系删除,如若转载,请注明出处:https://haidsoft.com/132573.html