Python技术迷

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 文件都行,脚本只负责读。 行了我先说到这,我现在脑子里还在回放那晚老板被我邮件轰炸的表情…算了不提了,反正你照着这个思路写,基本不会太难看。你要是想再狠一点,下次我给你们整一个“队列+失败告警+邮件模板”那种,别急,等我先把这周的会开完再说。