Python邮件收取教程:使用POP3协议

1. 邮件收取协议概述

在电子邮件的生态里,发送靠SMTP,收取则有两大主流方案:

  1. POP3 (Post Office Protocol version 3) - 专注「下载+本地处理」的轻量级协议
  2. IMAP (Internet Message Access Protocol) - 支持「云端同步+多设备管理」的进阶方案

本教程聚焦于更简单易用的POP3,用Python内置模块快速实现从邮箱抓取邮件的全流程。


2. POP3极简工作流

不用太纠结细节,记住这几步就能理解代码逻辑:

  1. 客户端连服务器:110端口(明文,强烈不推荐)/995端口(SSL加密,生产必用)
  2. 验证身份(登录)
  3. 拉取所有邮件的ID和大小列表
  4. 按需下载指定邮件
  5. (可选)标记/删除服务器上的已下载邮件
  6. 主动断开连接

3. Python全流程实现

不需要安装第三方库,Python标准库poplib+email全家桶就够了。

3.1 模块引入

先把要用的工具全部备好:

import poplib
from email.parser import BytesParser
from email.header import decode_header
from email.utils import parseaddr

3.2 安全连接函数

现代邮箱基本禁用明文POP3,封装一个默认走SSL的通用连接函数更方便:

def connect_to_pop3_server(email, password, pop3_server, use_ssl=True, port=None):
    """
    建立POP3安全连接
    :param email: 完整邮箱地址
    :param password: 主密码或应用专用密码(双重验证邮箱必须用后者)
    :param pop3_server: 服务商提供的POP3地址(如pop.qq.com)
    :param use_ssl: 是否启用SSL/TLS加密
    :param port: 自定义端口(留空则自动匹配加密/明文默认端口)
    :return: 已登录的POP3连接实例
    """
    # 自动匹配端口
    if not port:
        port = 995 if use_ssl else 110
    
    # 初始化连接
    try:
        server = poplib.POP3_SSL(pop3_server, port=port) if use_ssl else poplib.POP3(pop3_server, port=port)
    except Exception as e:
        raise ConnectionError(f"连接服务器失败: {str(e)}") from e
    
    # 调试开关(可选,开发时开,生产关)
    # server.set_debuglevel(1)
    
    # 打印欢迎信息(验证连接成功)
    print("✅ 服务器响应:", server.getwelcome().decode('utf-8'))
    
    # 身份验证
    server.user(email)
    server.pass_(password)
    print("✅ 登录成功")
    
    return server

3.3 获取邮件元数据列表

先不要下载大邮件,拉取ID和大小可以先判断下载范围:

def get_email_metadata(server):
    """
    获取邮件元数据(ID+大小)
    :param server: 已登录的POP3连接
    :return: [(邮件ID(字符串), 邮件大小(字节)), ...] 按时间从旧到新排序
    """
    _, raw_mails, _ = server.list()
    email_list = []
    for raw in raw_mails:
        mail_id, mail_size = raw.decode('utf-8').split()
        email_list.append((mail_id, int(mail_size)))
    return email_list

3.4 下载单封原始邮件

拿到ID后直接下载邮件的字节流原始内容(方便后续用email模块解析):

def download_single_email(server, mail_id):
    """
    下载指定ID的邮件
    :param server: 已登录的POP3连接
    :param mail_id: 邮件元数据中的ID
    :return: 邮件原始字节流
    """
    _, raw_lines, _ = server.retr(mail_id)
    return b'\r\n'.join(raw_lines)

3.5 核心解析工具

POP3只传原始字节,邮件头/正文/附件的解析全靠email模块,需要解决两个关键问题:

  • 编码混乱:中文邮件头/正文经常用GBK/GB2312/UTF-8
  • 多部分邮件:带附件、纯文本+HTML双格式的邮件是嵌套结构

3.5.1 通用字符串解码函数

def safe_decode_str(s):
    """
    安全解码邮件头中的编码字符串(支持多编码拼接)
    :param s: 待解码的邮件头字符串
    :return: 纯文本结果
    """
    if not s:
        return ""
    
    decoded_parts = []
    for part, charset in decode_header(s):
        if isinstance(part, bytes):
            charset = charset or "utf-8"
            try:
                part = part.decode(charset)
            except UnicodeDecodeError:
                # 尝试国内常用编码兜底
                part = part.decode("gbk", errors="replace")
        decoded_parts.append(part)
    return "".join(decoded_parts)

3.5.2 猜测邮件部分的编码

def guess_part_charset(msg_part):
    """
    猜测邮件单部分的字符编码
    :param msg_part: email模块解析的邮件子对象
    :return: 编码名称
    """
    charset = msg_part.get_charset()
    if charset:
        return charset
    
    content_type = msg_part.get("Content-Type", "").lower()
    for token in content_type.split(";"):
        token = token.strip()
        if token.startswith("charset="):
            return token[8:].strip('"').strip("'")
    
    return "utf-8"  # 最终兜底编码

3.5.3 递归打印邮件信息

处理嵌套的多部分邮件,只打印关键内容(正文前200字、附件名称):

def print_email_summary(msg, indent=0):
    """
    递归打印邮件摘要(避免控制台太长)
    :param msg: email模块解析的完整邮件对象
    :param indent: 缩进层级(内部递归用)
    """
    prefix = " " * indent
    
    # 最外层打印基本邮件头
    if indent == 0:
        print("\n" + "="*50)
        headers_to_show = ["From", "To", "Subject", "Date"]
        for header in headers_to_show:
            value = msg.get(header, "")
            if value:
                if header == "Subject" or header == "Date":
                    value = safe_decode_str(value)
                else:
                    # 格式化发件人/收件人
                    name, addr = parseaddr(value)
                    name = safe_decode_str(name)
                    value = f"{name} <{addr}>" if name else addr
                print(f"{prefix}{header}: {value}")
        print("="*50 + "\n")
    
    # 处理多部分/单部分
    if msg.is_multipart():
        # 递归遍历子部分
        for i, part in enumerate(msg.get_payload()):
            print(f"{prefix}--- Part {i+1} ---")
            print_email_summary(part, indent + 2)
    else:
        content_type = msg.get_content_type()
        charset = guess_part_charset(msg)
        
        # 纯文本/HTML正文
        if content_type.startswith("text/"):
            content_bytes = msg.get_payload(decode=True)
            try:
                content = content_bytes.decode(charset)
            except UnicodeDecodeError:
                content = content_bytes.decode("gbk", errors="replace")
            # 只打印前200个字符(防刷屏)
            print(f"{prefix}{content_type}, {charset}】")
            print(f"{prefix}{content[:200].strip()}{'...' if len(content) > 200 else ''}")
        # 附件
        else:
            filename = msg.get_filename()
            filename = safe_decode_str(filename) if filename else "未命名附件"
            print(f"{prefix}📎 附件: {filename} ({content_type})")

4. 完整可运行的示例

把上面的函数串起来,封装成一个主入口:

def main(email, password, pop3_server, max_fetch=3):
    """
    POP3收取邮件主流程
    :param max_fetch: 最多收取最新的N封邮件
    """
    server = None
    try:
        # 1. 连接并登录
        server = connect_to_pop3_server(email, password, pop3_server)
        
        # 2. 获取邮件总数和元数据
        total_count, _ = server.stat()
        print(f"📬 邮箱总共有 {total_count} 封邮件")
        if total_count == 0:
            print("没有邮件可收取")
            return
        
        email_metadata = get_email_metadata(server)
        # 取最新的max_fetch封(列表是旧→新,直接切片末尾)
        target_mails = email_metadata[-max_fetch:]
        print(f"📥 准备收取最新的 {len(target_mails)} 封邮件\n")
        
        # 3. 逐个下载+解析
        for i, (mail_id, size) in enumerate(target_mails, 1):
            print(f"🔍 正在处理第 {i}/{len(target_mails)} 封 (ID: {mail_id}, 大小: {size/1024:.2f}KB)")
            raw_content = download_single_email(server, mail_id)
            msg = BytesParser().parsebytes(raw_content)
            print_email_summary(msg)
            
            # 可选:标记为已删除(断开连接后才生效,可通过server.rset()取消)
            # server.dele(mail_id)
            # print(f"🗑️ 已标记ID为{mail_id}的邮件删除")
            
    except Exception as e:
        print(f"❌ 发生错误: {str(e)}")
    finally:
        # 无论成功失败都要关闭连接
        if server:
            server.quit()
            print("\n🔌 已断开服务器连接")


if __name__ == "__main__":
    # --------------------------
    # 请替换成你自己的邮箱配置
    # --------------------------
    MY_EMAIL = "your_email@example.com"
    MY_PWD = "your_app_password_or_main_password"
    MY_POP3 = "pop.example.com"  # 比如QQ是pop.qq.com,网易是pop.163.com
    
    # 收取最新2封
    main(MY_EMAIL, MY_PWD, MY_POP3, max_fetch=2)

5. 避坑指南(安全+实用)

5.1 安全第一

  1. 永远不要硬编码密码:用环境变量(os.getenv("EMAIL_PWD"))、加密配置文件(如python-dotenv+本地Git忽略的.env
  2. 双重验证邮箱必须用「应用专用密码」:QQ/163/Gmail等都要单独生成,不能用主密码
  3. 必须开SSL:默认走995端口,明文110端口现在几乎都被封了

5.2 实用注意事项

  1. 避免频繁调用:短时间内重复连接会被邮箱服务商限流甚至拉黑
  2. 邮件ID是会话级的:每次重新登录,邮件ID可能会变,不要跨会话保存
  3. 删除邮件要谨慎server.dele(mail_id)只是标记,必须等server.quit()才会真删,反悔的话用server.rset()
  4. 编码问题别慌:国内邮箱常用GBK,国外常用UTF-8,我们的代码已经加了两层兜底

6. 简单进阶方向

如果想拓展功能,可以试试:

  1. 附件下载:遍历多部分邮件,找到Content-Disposition: attachment的部分,保存到本地
  2. 邮件过滤:先拉取所有邮件头(用server.top(mail_id, 0)),根据主题/发件人/日期筛选后再下载正文
  3. 定时抓取:结合schedule库做定时任务
  4. 数据持久化:把解析后的邮件存入SQLite/MySQL/Elasticsearch

7. 总结

这篇教程用Python内置模块实现了POP3收取邮件的核心功能:从建立安全连接、拉取元数据、下载解析,到避坑注意事项都覆盖到了。整个流程代码量不大,逻辑清晰,很适合作为邮件自动化的入门实践。