python自动发送邮件
哎我跟你们说个真事啊,上周五晚上我都准备关电脑了,群里突然有人@我,说“东哥,那个日报邮件怎么又没发出去”,我当时脑子一空,心里想的是:不就发个邮件嘛,咋还能发丢……结果你越觉得简单的东西,越容易在你不注意的地方狠狠干你一巴掌。
我们那玩意儿其实很土,就一个定时任务,到了点把当天的统计表拼一下,然后发到几个邮箱。平时都好好的,偏偏那天老板临时加了两个收件人,其中一个邮箱是外部的,安全网关一拦,SMTP就开始各种奇奇怪怪的报错:有时候是握手慢,有时候是 535 认证失败(其实是频率限制),还有一次更离谱,连上了,发一半断了,你日志里看着像成功,收件人那边没收到……你说气不气。
我后来干脆把“发邮件”这事当成一个小型“消息投递”来做了,不要想着一把梭哈就完事。你看我写的这个小脚本,核心思路就三个:配置别写死、超时要有、失败要能重试还得有记录(不然你不知道到底发没发出去)。
先来个最小可用的,别整花里胡哨的库,就用标准库
smtplib
+
email.message
,够用。
import os
import smtplib
import socket
from email.message import EmailMessage
from datetime import datetime
def build_msg(subject: str, body: str, sender: str, to_list: list[str]) -> EmailMessage:
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = sender
msg["To"] = ", ".join(to_list)
msg.set_content(body)
return msg
def send_mail_once(msg: EmailMessage) -> None:
host = os.environ.get("SMTP_HOST", "")
port = int(os.environ.get("SMTP_PORT", "587"))
user = os.environ.get("SMTP_USER", "")
pwd = os.environ.get("SMTP_PASS", "")
timeout = float(os.environ.get("SMTP_TIMEOUT", "15"))
if not host or not user or not pwd:
raise RuntimeError("SMTP 配置没给全…你别笑,这个我真踩过")
socket.setdefaulttimeout(timeout)
with smtplib.SMTP(host, port, timeout=timeout) as smtp:
smtp.ehlo()
# 587 一般是 starttls;如果你们是 465,那就得用 SMTP_SSL
smtp.starttls()
smtp.ehlo()
smtp.login(user, pwd)
smtp.send_message(msg)
if __name__ == "__main__":
sender = os.environ.get("MAIL_SENDER", os.environ.get("SMTP_USER", ""))
to_list = ["[email protected]", "[email protected]"] # 真实项目你从配置/数据库读
now = datetime.now().strftime("%Y-%m-%d %H:%M")
subject = f"日报 {now}"
body = "今天数据都在附件里哈…(其实这里你拼接正文就行)"
msg = build_msg(subject, body, sender, to_list)
send_mail_once(msg)
print("ok")
上面这个跑通了就算“能发”。但是啊,真实场景你肯定会遇到:网络抖一下、网关抽风、对方服务器灰度、你们自己的账号触发限制。你要是还用“发一次就算了”,那就等着被人@吧,真的。
我一般会加个很朴素的重试,而且要注意一点:
别无限重试
,不然你定时任务变成“永动机”,一晚上发几千封,第二天账号直接被封你就好玩了。
import time
import json
from pathlib import Path
SENT_LOG = Path("mail_sent_log.jsonl")
def log_event(event: dict) -> None:
event["ts"] = time.time()
SENT_LOG.parent.mkdir(parents=True, exist_ok=True)
with SENT_LOG.open("a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
def send_with_retry(msg, max_retry: int = 3, base_sleep: float = 1.5) -> None:
last_err = None
for i in range(max_retry + 1):
try:
send_mail_once(msg)
log_event({"status": "success", "subject": msg["Subject"], "to": msg["To"]})
return
except Exception as e:
last_err = e
log_event({"status": "fail", "try": i + 1, "err": repr(e), "subject": msg["Subject"], "to": msg["To"]})
if i < max_retry:
time.sleep(base_sleep * (2 ** i)) # 简单指数退避
raise last_err
这里面我故意写成 jsonl,一行一条,后面你拿去 grep、或者随便写个脚本统计都行。因为你要回答一个很现实的问题:到底是没发,还是发了对方没收到,还是你发重复了。你别看我说得轻松哈,我之前就因为“重试没做幂等”,同一封报警邮件给老板发了 7 次……我当时人都麻了,手机震得像电钻。
所以幂等咋搞?很简单粗暴:给每封邮件做一个“业务唯一键”,比如
report_date + 收件人列表 + subject
之类,落盘,发之前查一下有没有发过。你要是怕文件锁,你就用 sqlite 也行,但咱今天别扩大话题,我先给你个文件版的。
import hashlib
ID_STORE = Path("mail_id_store.txt")
def make_mail_id(msg: EmailMessage) -> str:
raw = f"{msg['Subject']}|{msg['To']}|{msg.get_content()[:80]}"
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def already_sent(mail_id: str) -> bool:
if not ID_STORE.exists():
return False
s = set(ID_STORE.read_text(encoding="utf-8").splitlines())
return mail_id in s
def mark_sent(mail_id: str) -> None:
with ID_STORE.open("a", encoding="utf-8") as f:
f.write(mail_id + "\n")
def safe_send(msg: EmailMessage) -> None:
mail_id = make_mail_id(msg)
if already_sent(mail_id):
log_event({"status": "skip", "reason": "already_sent", "subject": msg["Subject"], "to": msg["To"]})
return
send_with_retry(msg, max_retry=3)
mark_sent(mail_id)
对了,还有附件这个事,很多人一上来就 MIMEBase 搞得很绕,其实
EmailMessage
自带
add_attachment
,你只要把二进制读出来,给个文件名就行。尤其你发 excel、pdf 这种,别手写边界,太容易写错。
import mimetypes
def attach_file(msg: EmailMessage, file_path: str) -> None:
p = Path(file_path)
data = p.read_bytes()
ctype, encoding = mimetypes.guess_type(p.name)
if ctype is None:
ctype = "application/octet-stream"
maintype, subtype = ctype.split("/", 1)
msg.add_attachment(data, maintype=maintype, subtype=subtype, filename=p.name)
# 用法
# attach_file(msg, "report.xlsx")
然后你们可能还会遇到一个“看着很玄学”的问题:公司邮箱一般要走网关,网关偶尔会把连接挂着不动,你这边不设置 timeout,程序就卡死在那……你以为任务没跑,其实它在那“沉思”。我反正现在是宁愿失败,也不要卡住。上面我把
SMTP_TIMEOUT
做成环境变量,就是为了这事,改起来不用发版,临时救火用得上。
最后我再插一句啊,你要是真搞“自动发邮件”,别把收件人写死在代码里,真的,太容易出事故。你想想人事调岗、离职、外部邮箱加白名单,都是天天变的。最稳的方式是:收件人放配置中心/数据库/甚至一个 yaml 文件都行,脚本只负责读。
行了我先说到这,我现在脑子里还在回放那晚老板被我邮件轰炸的表情…算了不提了,反正你照着这个思路写,基本不会太难看。你要是想再狠一点,下次我给你们整一个“队列+失败告警+邮件模板”那种,别急,等我先把这周的会开完再说。