Python實現(xiàn)批量發(fā)送郵件腳本
簡介
閑來無事,幫朋友寫個郵件發(fā)送腳本,順便拿來寫篇技術(shù)文,一舉兩得。腳本從指定的 excel 文件讀取客戶姓名、郵箱等信息,使用指定的郵件模版批量發(fā)送郵件。由于朋友使用文本模版就能達成要求,所以腳本雖然支持 Html 郵件模版,但并未測試。有自定義樣式的朋友還需自行調(diào)試代碼。 備注:本人測試是使用 163 郵箱
依賴
- yagmail==0.15.293:用來發(fā)送郵件
- pandas==2.3.1:用來讀取 Excel 文件
- jinja2==3.1.6:用來編寫 Html 模版
配置信息
郵箱配置
- password:部分郵箱(如163、QQ)需要使用授權(quán)碼而不是登錄密碼,請大家按需從郵箱設(shè)置中開啟SMTP服務(wù)并獲取授權(quán)碼。
- host:對不同的郵箱請使用不同的郵箱服務(wù)器,記得選擇 SMTP 郵箱服務(wù)器
- port:端口號可使用 465 端口(SSL加密)和 587 端口(STARTTLS),避免使用 25 端口(無加密,不安全)
class EmailConfig:
"""
郵箱配置
"""
# 你的郵箱
user: str = '138xxxx8888@163.com'
# 郵箱授權(quán)碼
password: str = 'TXYuclse932VzHJ'
# 郵箱服務(wù)器
host: str = 'smtp.163.com'
# 端口
port: int = '587'
發(fā)件人信息
- 發(fā)件人信息是用在郵件模版中的,可按需添加、刪除
- get_subject:在這里配置通用的郵件主題,client 是從 excel 文件中讀取出的客戶信息,使用該信息請和 excel 文件中的標(biāo)題一致
- to_json:返回的數(shù)據(jù)是用在 Html 模版中的數(shù)據(jù),可按需添加。
class SenderInfo:
"""
發(fā)件人信息
"""
# 顯示給客戶的發(fā)件人名稱
name: str = '起風(fēng)了-dev'
# 職位
position: str = '開發(fā)人員'
# 公司
company: str = 'xxx科技有限責(zé)任公司'
# 郵箱
email: str = '138xxxx8888@163.com'
# 簽名
signature: str = '簽名測試'
def get_subject(self, client: dict):
"""
獲取郵件主題
:param client: 客戶信息, 與客戶信息文件內(nèi)一致
:return: 郵件主題
"""
return f"{self.name}致{client['name']}的郵件"
def to_json(self):
return {
'name': self.name,
'position': self.position,
'company': self.company,
'email': self.email,
'signature': self.signature
}
發(fā)送設(shè)置
- 發(fā)送配置可以調(diào)整郵件的發(fā)送速度,避免被郵件服務(wù)器限制,將郵件當(dāng)作垃圾郵件
- delay_between_emails:每封郵件之間的間隔時間
- test_mode:在模版測試時可以先將郵件發(fā)送給自己,檢查完畢后再給客戶發(fā)送
- max_emails_per_batch:發(fā)送郵件每次達到該數(shù)目時暫停 1 分鐘,有其他需求的朋友可以自行修改代碼
class SendSettings:
"""
發(fā)送設(shè)置
"""
# 郵件間隔(秒),避免被限制
delay_between_emails: int = 2
# 測試模式(只發(fā)給自己)
test_mode: bool = False
# 每批最多發(fā)送量
max_emails_per_batch: int = 20
加載客戶數(shù)據(jù)
excel 文件格式
| index | name | |
|---|---|---|
| 1 | 起風(fēng)了 | xxxx@qq.com |
加載客戶數(shù)據(jù)時會檢驗客戶姓名和郵箱是否存在
返回數(shù)據(jù)結(jié)構(gòu): [{index: 1,name:'起風(fēng)了', email: 'xxxx@qq.com'},{index: 2, name:'', email: ''}]
def load_clients(self) -> list[dict]:
"""加載客戶數(shù)據(jù)"""
df = pd.read_excel(self.client_data_file)
client_list = df.to_dict(orient='records')
if not client_list:
raise ValueError("客戶數(shù)據(jù)為空")
# 客戶數(shù)據(jù)校驗并清洗
for client in client_list:
if not client['email']:
raise ValueError(f"客戶 {json.dumps(client, ensure_ascii=False)} 的郵箱為空")
if not client['name']:
raise ValueError(f"客戶 {json.dumps(client, ensure_ascii=False)} 的姓名為空")
client['name'] = client['name'].strip()
client['email'] = client['email'].strip()
print(f"已加載 {len(client_list)} 個客戶數(shù)據(jù)")
return client_list
加載模版數(shù)據(jù)
模版從文件內(nèi)加載
def load_template(self):
"""加載郵件模板"""
template = None
if self.template_file:
with open(self.template_file, 'r', encoding='utf-8') as f:
template = Template(f.read())
txt_template = None
if self.txt_template_file:
with open(self.txt_template_file, 'r', encoding='utf-8') as f:
txt_template = f.read()
return template, txt_template
轉(zhuǎn)換 HTML 到純文本
如果只指定了 HTML 模版,會使用這個函數(shù)轉(zhuǎn)換一份純文本郵件出來,HTML 模版本人并未測試,有相應(yīng)需求的朋友可自行調(diào)試代碼
def _convert_html_to_text(self, html):
"""
智能轉(zhuǎn)換HTML到純文本
保留重要格式(列表、段落、強調(diào)等)
"""
# 自定義轉(zhuǎn)換規(guī)則
replacements = [
(r'<br\s*/?>', '\n'),
(r'<p.*?>', '\n\n'),
(r'</p>', ''),
(r'<li.*?>', '? '),
(r'</li>', '\n'),
(r'<strong.*?>', '*'),
(r'</strong>', '*'),
(r'<em.*?>', '_'),
(r'</em>', '_'),
(r'<h[1-6].*?>', '\n\n'),
(r'</h[1-6]>', '\n\n'),
(r'<[^>]+>', ''), # 移除其他所有HTML標(biāo)簽
]
text = html
for pattern, replacement in replacements:
text = re.sub(pattern, replacement, text)
# 清理多余的空行
text = re.sub(r'\n\s*\n\s*\n', '\n\n', text)
print(f'_convert_html_to_text:{text}')
return text.strip()
郵件內(nèi)容組織
在這個函數(shù)中將模版中的占位符替換為指定數(shù)據(jù)
- 純文本模版使用 {client_key} 結(jié)構(gòu)的參數(shù)作為客戶信息占位符
- {sender_key}為發(fā)送者信息占位符
- 其中 key 是客戶 excel 文件標(biāo)題和配置中 SenderInfo 對象 to_json 函數(shù)返回的對象 key
Html 模版內(nèi)容替換方式請移步學(xué)習(xí) jinja2 庫
def personalize_content(self, client, template, txt_template: str):
"""個性化郵件內(nèi)容"""
html_content = template.render(client=client, sender=self.sender_info.to_json()) if template else None
if txt_template:
txt_content = txt_template
for key in client.keys():
txt_content = txt_content.replace('{client_' + key + '}', str(client[key]))
sender_json = self.sender_info.to_json()
for key in sender_json:
txt_content = txt_content.replace('{sender_' + key + '}', str(sender_json[key]))
else:
txt_content = self._convert_html_to_text(html_content)
return html_content, txt_content
發(fā)送郵件
這部分是郵件發(fā)送的核心代碼,郵件主題從 SenderInfo 配置中獲取,實現(xiàn)統(tǒng)一配置。 發(fā)送成功或失敗都要記錄日志,已備后續(xù)查看、核對
def send_to_client(self, client, content):
"""發(fā)送給單個客戶"""
try:
# 構(gòu)建郵件
subject = self.sender_info.get_subject(client=client)
# 如果是測試模式,發(fā)給自己
to_email = client['email']
if self.send_settings.test_mode:
to_email = self.email_config.user
subject = f"[測試] {subject}"
# 發(fā)送
self.yag.send(
to=to_email,
subject=subject,
contents=content,
# attachments=['attachments/sample.pdf'], # 可選附件
headers={
# 'X-Priority': '1', # 高優(yōu)先級 ??(顯示感嘆號)
'X-Priority': '3', # 普通優(yōu)先級(默認,不顯示特殊標(biāo)記)
# 'X-Priority': '5', # 低優(yōu)先級
'X-Mailer': 'PersonalEmailSender'
}
)
# 記錄日志
log_entry = {
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': '成功'
}
self.sent_log.append(log_entry)
print(f"? 已發(fā)送給 {client['name']} <{client['email']}>")
return True
except Exception as e:
print(f"? 發(fā)送失敗 {client['name']}: {e}")
self.sent_log.append({
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': f'失敗: {str(e)}'
})
return False
存儲日志
運行完畢將日志寫入文件
def save_log(self):
"""保存發(fā)送記錄"""
if not self.sent_log:
return
log_df = pd.DataFrame(self.sent_log)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
log_file = f'sent_logs/sent_{timestamp}.csv'
# 確保目錄存在
os.makedirs('sent_logs', exist_ok=True)
log_df.to_csv(log_file, index=False, encoding='utf-8-sig')
print(f"\n?? 發(fā)送記錄已保存: {log_file}")
功能組合后的主運行函數(shù)
在此處整合以上介紹的各部分功能。
流程:加載客戶 --> 加載模版 --> 分別為客戶生成郵件內(nèi)容 --> 內(nèi)容校驗 --> 發(fā)送郵件 --> 發(fā)送延遲 --> 保存日志 --> 發(fā)送結(jié)束
def run(self):
"""主運行函數(shù)"""
print("開始發(fā)送個性化郵件...\n")
# 加載數(shù)據(jù)
client_list = self.load_clients()
template, txt_template = self.load_template()
sent_count = 0
total_clients = len(client_list)
for i, client in enumerate(client_list, 1):
print(f"\n[{i}/{total_clients}] 處理: {client['name']}")
# 個性化內(nèi)容
html_content, text_content = self.personalize_content(client, template, txt_template)
if not text_content:
print(f"? 無內(nèi)容可發(fā)送給 {client['name']}")
# 記錄日志
log_entry = {
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': '失敗',
'原因': '無內(nèi)容'
}
self.sent_log.append(log_entry)
continue
content = [text_content, html_content] if html_content else text_content
# 發(fā)送
if self.send_to_client(client, content):
sent_count += 1
# 延遲,避免被限制
if i < total_clients:
delay = self.send_settings.delay_between_emails
print(f"等待 {delay} 秒...")
time.sleep(delay)
# 分批暫停(如果發(fā)送量大)
if i % self.send_settings.max_emails_per_batch == 0:
print(f"\n已發(fā)送 {i} 封,暫停 60 秒...")
time.sleep(60)
# 保存日志
self.save_log()
print(f"\n? 完成!成功發(fā)送 {sent_count}/{total_clients} 封郵件")
# 關(guān)閉連接
if self.yag:
self.yag.close()
運行示例
if __name__ == "__main__":
# 客戶數(shù)據(jù)文件
client_data_file = './clients/clients.xlsx'
# 郵件模板文件
txt_template_file = './templates/christmas.txt'
sender = EmailSender(client_data_file=client_data_file,
txt_template_file=txt_template_file)
sender.run()
附源碼
# 配置信息
class EmailConfig:
"""
郵箱配置
"""
# 你的郵箱
user: str = '138xxxx8888@163.com'
# 郵箱授權(quán)碼
password: str = 'TXYuclse932VzHJ'
# 郵箱服務(wù)器
host: str = 'smtp.163.com'
# 端口
port: int = '587'
class SenderInfo:
"""
發(fā)件人信息
"""
# 顯示給客戶的發(fā)件人名稱
name: str = '起風(fēng)了-dev'
# 職位
position: str = '開發(fā)人員'
# 公司
company: str = 'xxx科技有限責(zé)任公司'
# 郵箱
email: str = '138xxxx8888@163.com'
# 簽名
signature: str = '簽名測試'
def get_subject(self, client: dict):
"""
獲取郵件主題
:param client: 客戶信息, 與客戶信息文件內(nèi)一致
:return: 郵件主題
"""
return f"{self.name}致{client['name']}的郵件"
def to_json(self):
return {
'name': self.name,
'position': self.position,
'company': self.company,
'email': self.email,
'signature': self.signature
}
class SendSettings:
"""
發(fā)送設(shè)置
"""
# 郵件間隔(秒),避免被限制
delay_between_emails: int = 2
# 測試模式(只發(fā)給自己)
test_mode: bool = False
# 每批最多發(fā)送量
max_emails_per_batch: int = 20
# 主功能代碼
import json
import re
import yagmail
import pandas as pd
import time
import os
from datetime import datetime
from jinja2 import Template
from config.config import EmailConfig, SenderInfo, SendSettings
class EmailSender:
"""郵件發(fā)送器"""
def __init__(self, client_data_file: str, txt_template_file: str | None):
"""初始化"""
# 客戶數(shù)據(jù)
self.client_data_file = client_data_file
# 郵件模板
self.template_file = None
self.txt_template_file = txt_template_file
# 郵件配置
self.email_config = EmailConfig()
self.sender_info = SenderInfo()
self.send_settings = SendSettings()
self.yag = yagmail.SMTP(
user=self.email_config.user,
password=self.email_config.password,
host=self.email_config.host,
port=self.email_config.port
)
self.sent_log = []
def load_clients(self) -> list[dict]:
"""加載客戶數(shù)據(jù)"""
df = pd.read_excel(self.client_data_file)
client_list = df.to_dict(orient='records')
if not client_list:
raise ValueError("客戶數(shù)據(jù)為空")
# 客戶數(shù)據(jù)校驗并清洗
for client in client_list:
if not client['email']:
raise ValueError(f"客戶 {json.dumps(client, ensure_ascii=False)} 的郵箱為空")
if not client['name']:
raise ValueError(f"客戶 {json.dumps(client, ensure_ascii=False)} 的姓名為空")
client['name'] = client['name'].strip()
client['email'] = client['email'].strip()
print(f"已加載 {len(client_list)} 個客戶數(shù)據(jù)")
return client_list
def load_template(self):
"""加載郵件模板"""
template = None
if self.template_file:
with open(self.template_file, 'r', encoding='utf-8') as f:
template = Template(f.read())
txt_template = None
if self.txt_template_file:
with open(self.txt_template_file, 'r', encoding='utf-8') as f:
txt_template = f.read()
return template, txt_template
def _convert_html_to_text(self, html):
"""
智能轉(zhuǎn)換HTML到純文本
保留重要格式(列表、段落、強調(diào)等)
"""
# 自定義轉(zhuǎn)換規(guī)則
replacements = [
(r'<br\s*/?>', '\n'),
(r'<p.*?>', '\n\n'),
(r'</p>', ''),
(r'<li.*?>', '? '),
(r'</li>', '\n'),
(r'<strong.*?>', '*'),
(r'</strong>', '*'),
(r'<em.*?>', '_'),
(r'</em>', '_'),
(r'<h[1-6].*?>', '\n\n'),
(r'</h[1-6]>', '\n\n'),
(r'<[^>]+>', ''), # 移除其他所有HTML標(biāo)簽
]
text = html
for pattern, replacement in replacements:
text = re.sub(pattern, replacement, text)
# 清理多余的空行
text = re.sub(r'\n\s*\n\s*\n', '\n\n', text)
print(f'_convert_html_to_text:{text}')
return text.strip()
def personalize_content(self, client, template, txt_template: str):
"""個性化郵件內(nèi)容"""
html_content = template.render(client=client, sender=self.sender_info.to_json()) if template else None
if txt_template:
txt_content = txt_template
for key in client.keys():
txt_content = txt_content.replace('{client_' + key + '}', str(client[key]))
sender_json = self.sender_info.to_json()
for key in sender_json:
txt_content = txt_content.replace('{sender_' + key + '}', str(sender_json[key]))
else:
txt_content = self._convert_html_to_text(html_content)
return html_content, txt_content
def send_to_client(self, client, content):
"""發(fā)送給單個客戶"""
try:
# 構(gòu)建郵件
subject = self.sender_info.get_subject(client=client)
# 如果是測試模式,發(fā)給自己
to_email = client['email']
if self.send_settings.test_mode:
to_email = self.email_config.user
subject = f"[測試] {subject}"
# 發(fā)送
self.yag.send(
to=to_email,
subject=subject,
contents=content,
# attachments=['attachments/sample.pdf'], # 可選附件
headers={
# 'X-Priority': '1', # 高優(yōu)先級 ??(顯示感嘆號)
'X-Priority': '3', # 普通優(yōu)先級(默認,不顯示特殊標(biāo)記)
# 'X-Priority': '5', # 低優(yōu)先級
'X-Mailer': 'PersonalEmailSender'
}
)
# 記錄日志
log_entry = {
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': '成功'
}
self.sent_log.append(log_entry)
print(f"? 已發(fā)送給 {client['name']} <{client['email']}>")
return True
except Exception as e:
print(f"? 發(fā)送失敗 {client['name']}: {e}")
self.sent_log.append({
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': f'失敗: {str(e)}'
})
return False
def save_log(self):
"""保存發(fā)送記錄"""
if not self.sent_log:
return
log_df = pd.DataFrame(self.sent_log)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
log_file = f'sent_logs/sent_{timestamp}.csv'
# 確保目錄存在
os.makedirs('sent_logs', exist_ok=True)
log_df.to_csv(log_file, index=False, encoding='utf-8-sig')
print(f"\n?? 發(fā)送記錄已保存: {log_file}")
def run(self):
"""主運行函數(shù)"""
print("開始發(fā)送個性化郵件...\n")
# 加載數(shù)據(jù)
client_list = self.load_clients()
template, txt_template = self.load_template()
sent_count = 0
total_clients = len(client_list)
for i, client in enumerate(client_list, 1):
print(f"\n[{i}/{total_clients}] 處理: {client['name']}")
# 個性化內(nèi)容
html_content, text_content = self.personalize_content(client, template, txt_template)
if not text_content:
print(f"? 無內(nèi)容可發(fā)送給 {client['name']}")
# 記錄日志
log_entry = {
'client': client['name'],
'email': client['email'],
'time': datetime.now().isoformat(),
'status': '失敗',
'原因': '無內(nèi)容'
}
self.sent_log.append(log_entry)
continue
content = [text_content, html_content] if html_content else text_content
# 發(fā)送
if self.send_to_client(client, content):
sent_count += 1
# 延遲,避免被限制
if i < total_clients:
delay = self.send_settings.delay_between_emails
print(f"等待 {delay} 秒...")
time.sleep(delay)
# 分批暫停(如果發(fā)送量大)
if i % self.send_settings.max_emails_per_batch == 0:
print(f"\n已發(fā)送 {i} 封,暫停 60 秒...")
time.sleep(60)
# 保存日志
self.save_log()
print(f"\n? 完成!成功發(fā)送 {sent_count}/{total_clients} 封郵件")
# 關(guān)閉連接
if self.yag:
self.yag.close()
if __name__ == "__main__":
# 客戶數(shù)據(jù)文件
client_data_file = './clients/clients.xlsx'
# 郵件模板文件
txt_template_file = './templates/christmas.txt'
sender = EmailSender(client_data_file=client_data_file,
txt_template_file=txt_template_file)
sender.run()
到此這篇關(guān)于Python實現(xiàn)批量發(fā)送郵件腳本的文章就介紹到這了,更多相關(guān)Python發(fā)送郵件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- Python實現(xiàn)Excel批量處理+郵件自動發(fā)送的全流程
- Python3實現(xiàn)SMTP發(fā)送郵件的實戰(zhàn)指南
- 基于Python實現(xiàn)自動化郵件發(fā)送系統(tǒng)的完整指南
- 基于Python編寫自動化郵件發(fā)送程序(進階版)
- Python辦公自動化實戰(zhàn)之打造智能郵件發(fā)送工具
- Python編寫郵件自動發(fā)送工具的完整指南
- Python使用smtplib庫開發(fā)一個郵件自動發(fā)送工具
- Python利用SMTP發(fā)送郵件的常見問題與解決方案詳解
- 使用Python構(gòu)建帶GUI的郵件自動發(fā)送工具
相關(guān)文章
一起來學(xué)習(xí)一下python的數(shù)據(jù)類型
這篇文章主要為大家詳細介紹了python的數(shù)據(jù)類型,文中示例代碼介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們可以參考一下希望能夠給你帶來幫助2022-01-01
pandas 相關(guān)性和正態(tài)性分析的實踐
當(dāng)我們談?wù)撜龖B(tài)性(Normality)和相關(guān)性(Correlation)時,我們實際上在嘗試理解數(shù)據(jù)的分布模式和不同變量之間的關(guān)系,本文就來介紹一下pandas 相關(guān)性和正態(tài)性的實踐,感興趣的可以了解一下2024-07-07
解決keras+flask模型的重復(fù)調(diào)用出錯ValueError: Tensor is n
這篇文章主要介紹了解決keras+flask模型的重復(fù)調(diào)用出錯ValueError: Tensor is not an element of this graph問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-01-01

