Python自动化办公项目实例：Excel处理与邮件批量发送

一、业务场景与系统全貌设计

邮件群发在企业办公场景中存在一个核心限制：不能使用抄送或密送将所有人放在同一封邮件中。每个接收者只能看到属于自己的私密数据（例如应发工资、扣税明细、银行账户尾号或专属附件），并且发信过程需要具备完备的容错与状态追踪能力，防止遇到网络闪断或邮箱地址失效时整个任务崩溃。

为了实现这一目标，我们将整个自动化办公系统划分为清晰的四个层次：

• 数据源与配置层：存储受众数据的 Excel 表格、独立附件资源池以及独立存放的发件服务器授权配置。
• 数据清洗与校验层：利用 pandas 对表格进行结构化读取，过滤空白行、检验收件人邮箱的格式合法性，并识别“已发送”的历史记录以实现断点续传。
• MIME 组装与调度层：利用 Python 内置的 email.mime 模块为每位收件人单独生成多部件（Multipart）报文，内嵌格式规范的 HTML 排版表格与经过 RFC 2231 规范编码的附件。
• 安全投递与状态回写层：建立基于 SSL 加密的 SMTP 会话，按照预设的随机时间间隔安全投递，单条异常独立隔离，并利用 openpyxl 实时将发送结果（成功或报错原因）写回原 Excel 文件中。

二、环境准备与发信服务配置

在开始编写代码前，需要准备两项前置工作：安装 Python 依赖库，以及获取邮箱服务商的 SMTP 客户端发信授权码。

1. 安装核心处理依赖

Python 标准库自带了网络通信模块 smtplib 与邮件报文构造模块 email，无需额外安装。我们只需要安装用于高性能数据读取的 pandas 以及支持 Excel 无损读写与样式保留的 openpyxl：

# 终端执行安装命令
pip install pandas openpyxl

2. 获取 SMTP 专属发件授权码

主流邮件服务商（如 QQ 邮箱、163 网易邮箱、企业微信邮箱、Gmail、Office 365 等）出于安全保护策略，默认禁止直接使用邮箱账号的登录密码发起 SMTP 连接。用户必须登录邮箱网页端，进入“账户设置”开启 POP3/SMTP 或 IMAP/SMTP 服务，并生成一串专用的客户端授权码（通常为 16 位英文字符）。

3. 数据源 Excel 表格设计

我们以企业月度考勤绩效发放为例，准备一个名为 staff_payroll.xlsx 的工作簿。表格包含两类字段：受众属性字段（工号、姓名、邮箱）、业务明细字段（基本工资、绩效奖金、扣减金额、实发工资、附件文件名）以及自动化运维字段（发送状态、发信时间、备注日志）。

三、核心操作步骤与实现原理

整个自动化工作流严格按照五步工序展开：配置装载 ➔ 清洗预审 ➔ MIME 编码装配 ➔ 受控安全投递 ➔ 原子状态回写。

步骤一：基于 pandas 读取数据与断点续发判定

在实际批量任务中，经常会遇到网络闪断、触发临时限流或部分邮箱地址拼写错误的情况。如果在失败后重新执行脚本，最忌讳的是给已经发送成功的受众再次重发一封邮件。

因此，我们的程序首先读取 Excel，并在循环迭代前建立幂等性检查机制：只要该行的“发送状态”列已经被标记为 "SUCCESS"，就自动跳过。同时，对缺失收件人邮箱的无效行记录错误并直接跳过：

import re
import pandas as pd

def load_and_validate_data(excel_path):
    df = pd.read_excel(excel_path)
    # 确保运维字段存在
    for col in ['发送状态', '发信时间', '备注日志']:
        if col not in df.columns:
            df[col] = ''
    
    # 简易正则过滤合规邮箱
    email_regex = re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')
    valid_tasks = []
    
    for idx, row in df.iterrows():
        # 1. 幂等性：已成功的不重复发送
        if str(row['发送状态']).strip().upper() == 'SUCCESS':
            continue
            
        email = str(row['邮箱']).strip() if pd.notna(row['邮箱']) else ''
        if not email or not email_regex.match(email):
            df.at[idx, '发送状态'] = 'FAILED'
            df.at[idx, '备注日志'] = '邮箱地址为空或格式不合法'
            continue
            
        valid_tasks.append((idx, row))
        
    return df, valid_tasks

步骤二：组装个性化 HTML 报文与中文附件

邮件正文如果采用纯文本，可读性和专业度较差。使用 HTML 内联表格可以呈现出与 Excel 同样规整的视觉排版。邮件协议要求使用 MIME 格式对图文及附件进行封装，具体规范如下：

• 使用 MIMEMultipart('mixed') 作为最外层报文容器；
• 将 HTML 字符串包裹为 MIMEText(html_content, 'html', 'utf-8') 并挂载到容器中；
• 对于附件，使用 MIMEBase('application', 'octet-stream') 以二进制模式读取，并执行 base64 编码；
• 关键细节：为了防止收件人在 Outlook、Foxmail 或网页端查看附件时出现中文乱码，必须使用 email.header.Header 并按照 RFC 2231 / RFC 2047 规范格式化文件名。

import os
from email.header import Header
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email import encoders

def build_custom_email(sender_email, row_data, template_html):
    msg = MIMEMultipart('mixed')
    
    # 个性化标题
    name = row_data['姓名']
    subject_text = f"【保密】{name} 先生/女士 2026年10月薪资对账明细"
    msg['Subject'] = Header(subject_text, 'utf-8')
    msg['From'] = f"财务运营部 <{sender_email}>"
    msg['To'] = row_data['邮箱']
    
    # 填充个性化正文
    rendered_body = template_html.format(
        name=name,
        emp_id=row_data['工号'],
        base_salary=row_data['基本工资'],
        bonus=row_data['绩效奖金'],
        deduction=row_data['扣减金额'],
        net_salary=row_data['实发工资']
    )
    msg.attach(MIMEText(rendered_body, 'html', 'utf-8'))
    
    # 检查并挂载个性化附件
    attachment_name = str(row_data.get('附件文件名', '')).strip()
    if attachment_name and pd.notna(row_data.get('附件文件名')):
        attachment_path = os.path.join("attachments", attachment_name)
        if os.path.exists(attachment_path):
            with open(attachment_path, 'rb') as f:
                part = MIMEBase('application', 'octet-stream')
                part.set_payload(f.read())
            encoders.encode_base64(part)
            # 解决客户端中文文件名乱码的核心写法
            part.add_header('Content-Disposition', 'attachment', filename=('utf-8', '', attachment_name))
            msg.attach(part)
        else:
            print(f"[警告] 工号 {row_data['工号']} 的指定附件 {attachment_name} 不存在，已仅发送正文")
            
    return msg

步骤三：建立 SSL 安全通道与发信限流调度

为了避免频繁创建和销毁 TCP/SSL 连接被服务商判定为异常嗅探，应采用单个长连接复用机制：在批量任务启动时建立一次 smtplib.SMTP_SSL 连接并登录，在连接内遍历发送所有待投递邮件，最后通过 quit() 优雅关闭。

同时，在每封邮件发送之后，加入 1.5 到 3 秒的随机延迟，模拟正常用户的发信间隔，避免被触发频控阈值：

import time
import random
import smtplib

def dispatch_batch_emails(smtp_cfg, valid_tasks, df, template_html):
    host = smtp_cfg['host']
    port = smtp_cfg['port']
    user = smtp_cfg['user']
    auth_code = smtp_cfg['auth_code']
    
    # 建立 SSL 加密通道连接
    with smtplib.SMTP_SSL(host, port, timeout=15) as server:
        # 客户端鉴权
        server.login(user, auth_code)
        print(f"成功登录发信服务器：{host}:{port}")
        
        for idx, row in valid_tasks:
            recipient = row['邮箱']
            name = row['姓名']
            try:
                # 1. 组装报文
                msg = build_custom_email(user, row, template_html)
                
                # 2. 投递邮件
                server.sendmail(user, [recipient], msg.as_string())
                
                # 3. 记录内存状态
                now_str = time.strftime('%Y-%m-%d %H:%M:%S')
                df.at[idx, '发送状态'] = 'SUCCESS'
                df.at[idx, '发信时间'] = now_str
                df.at[idx, '备注日志'] = '投递成功'
                print(f"[成功] 索引 {idx} -> {name} ({recipient}) 发送完毕")
                
            except smtplib.SMTPRecipientsRefused:
                df.at[idx, '发送状态'] = 'FAILED'
                df.at[idx, '备注日志'] = '邮箱地址被对方服务器拒绝(550)'
                print(f"[失败] 索引 {idx} -> {name} 收件人地址被拒")
            except Exception as e:
                df.at[idx, '发送状态'] = 'FAILED'
                df.at[idx, '备注日志'] = f"异常: {str(e)[:50]}"
                print(f"[异常] 索引 {idx} -> {name} 失败: {e}")
                
            # 4. 防风控限流休眠
            delay = round(random.uniform(1.5, 3.0), 2)
            time.sleep(delay)

步骤四：使用 openpyxl 进行结果原子回写

如果直接使用 df.to_excel('staff_payroll.xlsx') 进行全量覆盖保存，会导致原 Excel 文件中原本配置的单元格背景色、边框、字体、公式及列宽完全丢失。

为了保护原表格的格式完整性，我们利用 openpyxl 加载原文件，只针对“发送状态”、“发信时间”和“备注日志”所在的特定单元格进行逐格赋值更新：

import openpyxl

def write_results_back(excel_path, df):
    wb = openpyxl.load_workbook(excel_path)
    ws = wb.active
    
    # 获取表头映射 {列名: 列索引(从1开始)}
    header_map = {}
    for col_idx in range(1, ws.max_column + 1):
        cell_val = ws.cell(row=1, column=col_idx).value
        if cell_val:
            header_map[str(cell_val).strip()] = col_idx
            
    # 如果原表不存在回写列，自动追加列头
    target_cols = ['发送状态', '发信时间', '备注日志']
    for t_col in target_cols:
        if t_col not in header_map:
            new_idx = ws.max_column + 1
            ws.cell(row=new_idx, column=new_idx, value=t_col)
            header_map[t_col] = new_idx
            
    # 逐行写回（Excel数据行从第2行开始，与df索引对应）
    for idx, row in df.iterrows():
        excel_row_num = idx + 2
        for t_col in target_cols:
            col_idx = header_map[t_col]
            val = str(row[t_col]) if pd.notna(row[t_col]) else ''
            ws.cell(row=excel_row_num, column=col_idx, value=val)
            
    wb.save(excel_path)
    print(f"执行状态已成功无损写回工作簿：{excel_path}")

四、SMTP 通信协议深度剖析与异常隔离

理解底层 SMTP（Simple Mail Transfer Protocol）交互时序，有助于在出现异常时迅速定位问题根源。客户端与服务器之间基于命令-响应模式运行。

在批量邮件分发过程中，常见的三种错误层级必须进行差异化分流处理：

1. 致命认证错误（535 Authentication Failed）：通常由于授权码写错、授权码过期或发信账号被锁导致。此时后续所有发信操作都必然失败，程序应当捕获 SMTPAuthenticationError 后立即中断执行并打印报警信息，避免徒劳重试导致 IP 被服务商短期封禁。
2. 单用户地址失效（550 User Not Found / Mailbox Unavailable）：对方邮箱账号已注销或拼写有误。此类错误仅属于当前行的个案，程序应当捕获 SMTPRecipientsRefused，将当前行记为失败，继续执行队列中的下一封邮件。
3. 内容或频率拦截（554 Transaction Failed / Spam Detected）：邮件正文包含过多营销诱导词、缺少规范 Header，或当前发信速率超限。此时应记录日志并适当动态增加休眠时间（如延长至 10 秒以上）。

五、完整工程可执行源码

以下是集成上述所有工业级特性的完整脚本 batch_email_sender.py。代码包含配置抽取、数据校验、模板渲染、发信控制与安全回写全流程：

# -*- coding: utf-8 -*-
"""
Python 自动化办公实战：Excel 数据处理与邮件批量发送
包含功能：断点续跑检测、个性化 HTML 渲染、带附件安全发送、异常隔离、原表无损状态回写
"""

import os
import re
import time
import random
import smtplib
import openpyxl
import pandas as pd
from email.header import Header
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email import encoders

# 1. 基础配置信息
CONFIG = {
    "excel_path": "staff_payroll.xlsx",
    "attachments_dir": "attachments",
    "smtp_server": "smtp.qq.com",
    "smtp_port": 465,
    "sender_email": os.environ.get("SMTP_USER", "hr_service@example.com"),
    "auth_code": os.environ.get("SMTP_AUTH_CODE", "abcdefghijklmnop")
}

# 2. 内联 HTML 邮件模板（样式美观，适配主流桌面及移动邮件客户端）
EMAIL_HTML_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <style>
    .mail-box { font-family: 'PingFang SC', 'Microsoft YaHei', sans-serif; max-width: 600px; margin: 0 auto; border: 1px solid #D9E2EC; border-radius: 8px; overflow: hidden; }
    .header { background-color: #6B88A6; color: #FFFFFF; padding: 20px; text-align: center; font-size: 18px; font-weight: bold; }
    .content { padding: 24px; color: #334E68; font-size: 14px; line-height: 1.6; }
    .data-table { width: 100%; border-collapse: collapse; margin-top: 16px; margin-bottom: 20px; }
    .data-table th, .data-table td { border: 1px solid #E2E8F0; padding: 10px 14px; text-align: left; }
    .data-table th { background-color: #F0F4F8; color: #486581; }
    .highlight { color: #71A397; font-weight: bold; }
    .footer { background-color: #F8FAFC; padding: 14px; text-align: center; font-size: 12px; color: #829AB1; border-top: 1px solid #E2E8F0; }
  </style>
</head>
<body>
  <div class="mail-box">
    <div class="header">月度薪资与绩效核算明细通知</div>
    <div class="content">
      <p>尊敬的 <strong>{name}</strong> 同事（工号：{emp_id}）：</p>
      <p>您好！本月个人绩效核算已完成，个人薪酬对账明细如下，请您查阅核对：</p>
      <table class="data-table">
        <tr><th>核算项目</th><th>金额 (元)</th></tr>
        <tr><td>基础工资</td><td>¥ {base_salary:.2f}</td></tr>
        <tr><td>绩效考核奖金</td><td>¥ {bonus:.2f}</td></tr>
        <tr><td>代扣五险一金/税款</td><td>- ¥ {deduction:.2f}</td></tr>
        <tr><td><strong>本月实发金额</strong></td><td class="highlight"><strong>¥ {net_salary:.2f}</strong></td></tr>
      </table>
      <p>注：如附带个人明细对账单，已作为加密附件添加于本邮件中，请查收核实。</p>
    </div>
    <div class="footer">本邮件由财务自动化系统自动发出，内容涉及个人隐私，请勿向他人转发泄露。</div>
  </div>
</body>
</html>
"""

def main():
    excel_file = CONFIG["excel_path"]
    if not os.path.exists(excel_file):
        print(f"[错误] 数据文件未找到：{excel_file}")
        return

    # 1. 结构化读取与前置校验
    df = pd.read_excel(excel_file)
    for col in ['发送状态', '发信时间', '备注日志']:
        if col not in df.columns:
            df[col] = ''

    email_pattern = re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')
    task_queue = []

    for idx, row in df.iterrows():
        # 幂等性过滤已发送成功的行
        if str(row['发送状态']).strip().upper() == 'SUCCESS':
            continue
        
        email = str(row.get('邮箱', '')).strip()
        if not email or not email_pattern.match(email):
            df.at[idx, '发送状态'] = 'FAILED'
            df.at[idx, '备注日志'] = '邮箱为空或格式错误'
            continue
        
        task_queue.append((idx, row))

    total_tasks = len(task_queue)
    print(f"成功加载待投递任务：共 {total_tasks} 封邮件需要处理")
    if total_tasks == 0:
        print("没有需要发送的邮件（所有记录已成功或无合规数据），程序结束。")
        return

    # 2. 建立 SMTP 会话并执行批量发送
    success_count = 0
    fail_count = 0

    try:
        server = smtplib.SMTP_SSL(CONFIG["smtp_server"], CONFIG["smtp_port"], timeout=15)
        server.login(CONFIG["sender_email"], CONFIG["auth_code"])
        print("SMTP 服务器鉴权成功，开始逐封发送...\n")
    except smtplib.SMTPAuthenticationError:
        print("[致命错误] 授权码验证失败(535)，请检查发信邮箱与授权码。任务终止。")
        return
    except Exception as e:
        print(f"[致命错误] 无法连接发件服务器：{e}。任务终止。")
        return

    try:
        for idx, row in task_queue:
            recipient = str(row['邮箱']).strip()
            name = str(row['姓名']).strip()
            emp_id = str(row['工号']).strip()
            
            # 组装邮件报文
            msg = MIMEMultipart('mixed')
            msg['Subject'] = Header(f"【对账通知】{name} 先生/女士 月度核算明细", 'utf-8')
            msg['From'] = f"人力资源与财务部 <{CONFIG['sender_email']}>"
            msg['To'] = recipient

            # 填充正文变量
            body_html = EMAIL_HTML_TEMPLATE.format(
                name=name,
                emp_id=emp_id,
                base_salary=float(row.get('基本工资', 0)),
                bonus=float(row.get('绩效奖金', 0)),
                deduction=float(row.get('扣减金额', 0)),
                net_salary=float(row.get('实发工资', 0))
            )
            msg.attach(MIMEText(body_html, 'html', 'utf-8'))

            # 处理附件挂载
            attach_file = str(row.get('附件文件名', '')).strip()
            if attach_file and attach_file.lower() != 'nan':
                attach_full_path = os.path.join(CONFIG["attachments_dir"], attach_file)
                if os.path.exists(attach_full_path):
                    with open(attach_full_path, 'rb') as f:
                        part = MIMEBase('application', 'octet-stream')
                        part.set_payload(f.read())
                    encoders.encode_base64(part)
                    part.add_header('Content-Disposition', 'attachment', filename=('utf-8', '', attach_file))
                    msg.attach(part)
                else:
                    print(f"  [提示] 找不到附件 {attach_full_path}，仅发送正文")

            # 尝试投递
            try:
                server.sendmail(CONFIG["sender_email"], [recipient], msg.as_string())
                timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
                df.at[idx, '发送状态'] = 'SUCCESS'
                df.at[idx, '发信时间'] = timestamp
                df.at[idx, '备注日志'] = '已成功投递至网关'
                success_count += 1
                print(f"[成功 {success_count}/{total_tasks}] 员工: {name} ({recipient})")
            except smtplib.SMTPRecipientsRefused:
                df.at[idx, '发送状态'] = 'FAILED'
                df.at[idx, '备注日志'] = '收件人地址不存在或被拒(550)'
                fail_count += 1
                print(f"[失败] 员工: {name} 收件地址被拒")
            except Exception as send_err:
                df.at[idx, '发送状态'] = 'FAILED'
                df.at[idx, '备注日志'] = f"发信异常: {str(send_err)[:40]}"
                fail_count += 1
                print(f"[异常] 员工: {name} 发送失败: {send_err}")

            # 限流休眠 (1.5 ~ 3.0 秒)
            time.sleep(random.uniform(1.5, 3.0))

    finally:
        try:
            server.quit()
            print("\nSMTP 安全连接已正常释放关闭。")
        except Exception:
            pass

    # 3. 结果原子性写回原 Excel 文件
    print("正在将发送状态与回执写回 Excel 工作簿...")
    wb = openpyxl.load_workbook(excel_file)
    ws = wb.active
    
    header_indices = {}
    for c in range(1, ws.max_column + 1):
        val = ws.cell(row=1, column=c).value
        if val:
            header_indices[str(val).strip()] = c

    for col_name in ['发送状态', '发信时间', '备注日志']:
        if col_name not in header_indices:
            new_col = ws.max_column + 1
            ws.cell(row=1, column=new_col, value=col_name)
            header_indices[col_name] = new_col

    for idx, row in df.iterrows():
        target_row = idx + 2
        for col_name in ['发送状态', '发信时间', '备注日志']:
            col_pos = header_indices[col_name]
            cell_val = str(row[col_name]) if pd.notna(row[col_name]) else ''
            ws.cell(row=target_row, column=col_pos, value=cell_val)

    wb.save(excel_file)
    print(f"处理完成！成功: {success_count} 封，失败: {fail_count} 封。状态已持久化更新。")

if __name__ == '__main__':
    main()

六、生产环境落地与常见排错决策

自动化脚本在个人电脑上运行成功与在生产环境长期无人值守运行，面临的稳定性挑战截然不同。总结以下关键防护维度与速查指引：

常见故障速查与定位策略

• 端口被屏蔽（Timeout 连接超时）：许多云服务商（如阿里云、腾讯云、AWS）出于防垃圾邮件治理，在底层默认完全封禁了外网出方向的 25 端口。如果在云服务器或容器中运行脚本，千万不能使用默认的 25 端口，必须使用 smtplib.SMTP_SSL 指定 465 端口，或者使用 587 配合 starttls()。
• 授权码失效（535 Authentication Failed）：某些企业邮箱管理员策略会在 30 天或 90 天后强制重置外部应用授权码；另外修改邮箱账户登录密码后，历史生成的授权码会自动全部作废。若出现 535 错误，请优先进入邮箱网页端重新生成授权码。
• 附件在客户端显示为 winmail.dat 或乱码：这是由于文件名编码不兼容或者客户端解析差异导致的。使用 Header(filename, 'utf-8') 结合 add_header('Content-Disposition', 'attachment', filename=('utf-8', '', filename)) 双重规范，能兼容 99% 以上的现代邮件客户端。
• 群发中途触发反垃圾策略（554 Transaction Failed）：服务商风控系统监测到短时间内高频次发送相似结构的内容。解决策略：一是将休眠时间由固定值调整为 2~5 秒的动态浮动随机数；二是适当增加单批次正文的动态差异（如插入收件人全名、个性化问候语和唯一流水时间戳）。