本帖最后由 苏紫方璇 于 2023-8-28 00:55 编辑
[Python] 纯文本查看 复制代码 import smtplib
import csv
import threading
import argparse
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
def send_email(sender_email, sender_password, receiver_email):
# 与SMTP服务器建立连接
smtp_server = 'smtp.qq.com'
server_port = 587
server = smtplib.SMTP(smtp_server, server_port)
server.starttls()
try:
# 登录邮箱账号
server.login(sender_email, sender_password)
# 构造邮件内容
msg_root = MIMEMultipart('related')
msg_root['Subject'] = '测试一下'
msg_root['From'] = sender_email
msg_root['To'] = receiver_email
subject = '测试一下111111'
message = '''
<html>
<body>
<p><img src="cid:image1" alt="升学教育图片" style="width:300px;height:200px;"></p>
<p>班上新来一位学生,名字叫“马騳骉”。</p>
<p>开学点名了,班主任不知怎么念,就说 : 马叉叉到了没?</p>
<p>语文老师有点文学素养,点名道:万马奔腾到了没?</p>
</body>
</html>
'''
# 添加文本内容
html_part = MIMEText(message, 'html')
msg_root.attach(html_part)
# 添加图片附件
with open('F:/data/png/pexels-max-andrey-1366630.jpg', 'rb') as image_file:
image_part = MIMEImage(image_file.read())
image_part.add_header('Content-Disposition', 'attachment; filename=1366630.jpg')
image_part.add_header('Content-ID', '<image1>')
msg_root.attach(image_part)
# 添加附件#
with open('F:/data/Jmeter弱网测试.pdf', 'rb') as file:
attachment_part = MIMEApplication(file.read(), Name='弱网测试.pdf')
attachment_part.add_header('Content-Disposition', 'attachment; filename=弱网测试.pdf')
msg_root.attach(attachment_part)
# 发送邮件
server.sendmail(sender_email, receiver_email, msg_root.as_string())
print(f"邮件发送成功: {receiver_email}")
except smtplib.SMTPException as e:
print(f"邮件发送失败: {receiver_email},错误信息: {str(e)}")
# 关闭连接
server.quit()
def send_emails(sender_email, sender_password, thread_count):
# 读取CSV文件并获取收件人邮箱列表
receiver_emails = []
with open('E:/pythonProject/test/file/email.csv', 'r') as file:
reader = csv.reader(file)
for row in reader:
receiver_emails.append(row[0])
# 创建线程列表
threads = []
# 创建发送邮件的函数
def send_email_thread(receiver_email):
send_email(sender_email, sender_password, receiver_email)
# 创建并启动发送邮件的线程
for receiver_email in receiver_emails:
t = threading.Thread(target=send_email_thread, args=(receiver_email,))
threads.append(t)
t.start()
# 控制最大线程数
if len(threads) >= thread_count:
for thread in threads:
thread.join()
threads = []
# 等待剩余线程完成
for thread in threads:
thread.join()
def main():
# 解析命令行参数
parser = argparse.ArgumentParser(description='发送邮件')
parser.add_argument('-s', '--sender-email', required=True, help='发件人邮箱')
parser.add_argument('-p', '--sender-password', required=True, help='发件人密码')
parser.add_argument('-t', '--thread-count', default=5, type=int, help='发送线程数')
args = parser.parse_args()
# 调用发送邮件函数
send_emails(args.sender_email, args.sender_password, args.thread_count)
if __name__ == '__main__':
main()
PowerShell:python test_email.py -s xxxx@qq.com -p xxxxxxybbbc -t 1 |