Python headerregistry 的 defects 怎样保留邮件头解析问题

处理来源不稳定的邮件时，最容易丢掉的不是正文，而是“这个头部为什么不符合规范”的线索。email.headerregistry 会把已经发现的邮件头问题放在具体 Header 对象的 defects 属性中；它是一个缺陷记录元组，不等同于异常。更稳妥的做法是同时保留可继续处理的邮件对象、原始邮件 bytes，以及一份可 JSON 化的诊断副本。

官方文档：https://docs.python.org/3/library/email/headerregistry.html

先区分 defects 与异常

在默认的 EmailPolicy 下，邮件头会由 HeaderRegistry 创建定制的 Header 对象。对象的字符串值负责继续使用，defects 则是解析器发现的 RFC 合规问题集合。官方文档明确把它定义为只读的 HeaderDefect 元组，所以应用代码应该读取并复制诊断信息，而不是尝试修改这个属性。

这和 HeaderParseError 不是一回事：前者表示“解析结果带有问题说明”，后者表示某些解析场景直接抛出错误。把两者混在一个 except 分支里，会让下游无法区分“可以人工复核的邮件”和“没有得到可用解析结果的输入”。

从具体 Header 读取缺陷

读取时不要只检查 Message.defects。邮件头的解析问题通常挂在具体的 Header 对象上，例如 Subject、From 或 Content-Type。下面的函数把字段名、缺陷类型和文本整理成普通字典，后续就可以交给日志系统、队列或数据库。

from email import policy
from email.parser import BytesParser


def defect_record(field_name, defect):
    # 只保留稳定、可序列化的诊断字段，不把 HeaderDefect 对象直接写入 JSON。
    return {
        "field": field_name,
        "type": type(defect).__name__,
        "message": str(defect),
    }


def collect_header_defects(message):
    records = []
    for field_name, header in message.items():
        # headerregistry 的 Header.defects 是当前字段自己的问题列表。
        for defect in getattr(header, "defects", ()):
            records.append(defect_record(field_name, defect))
    return records


raw = b"Subject: =?utf-8?Q?demo?=\r\n\r\nbody\r\n"
message = BytesParser(policy=policy.default).parsebytes(raw)
diagnostics = collect_header_defects(message)

这段代码的关键是 getattr(header, "defects", ())：未知头部仍然会得到一个普通的 Header 对象，没有缺陷时返回空元组；有缺陷时只把诊断信息取出来。示例中的输入仅用于说明数据形状，文章不把它描述成某次本地运行结果。

建立可落盘的诊断副本

如果只把 str(message["Subject"]) 存下来，问题类型和解析上下文就可能丢失。推荐保留三部分：原始邮件 bytes 的存储引用、正常解析后的消息对象，以及独立的诊断列表。这样业务仍能使用解析结果，同时可以按缺陷类型统计和复核。

from email.errors import HeaderParseError
from email.parser import BytesParser
from email import policy


def collect_message_diagnostics(message, path="root"):
    records = []
    # 消息级 defect 记录在发现问题的 Message 节点，不一定在最外层。
    for defect in getattr(message, "defects", ()):
        records.append({
            "path": path,
            "scope": "message",
            "type": type(defect).__name__,
            "message": str(defect),
        })

    for field_name, header in message.items():
        # 字段级 defect 归属于当前 Header，而不是自动提升到外层 Message。
        for defect in getattr(header, "defects", ()):
            records.append({
                "path": f"{path}.{field_name}",
                "scope": "header",
                "type": type(defect).__name__,
                "message": str(defect),
            })

    if message.is_multipart():
        # 递归检查嵌套邮件，避免只看根节点而漏掉附件或子消息的问题。
        for index, part in enumerate(message.iter_parts()):
            records.extend(collect_message_diagnostics(part, f"{path}.parts[{index}]"))
    return records


def parse_with_diagnostics(raw_bytes):
    try:
        # policy.default 才会按 headerregistry 生成专用 Header 对象。
        message = BytesParser(policy=policy.default).parsebytes(raw_bytes)
    except HeaderParseError as exc:
        # 真正的头部解析异常单独返回，不能伪装成普通 defect。
        return {"message": None, "defects": [], "error": type(exc).__name__}

    return {
        "message": message,
        "defects": collect_message_diagnostics(message),
        "error": None,
    }

实际落盘时，message 不直接进入 JSON；可以保存原始输入的对象存储键、解析时间和上面返回的 defects 列表。诊断记录的形状可以是下面这样，严格 JSON 不写注释，字段含义放在代码块外解释：

{
  "path": "root.Subject",
  "scope": "header",
  "type": "HeaderDefect",
  "message": "解析器提供的缺陷描述"
}

path 用于定位字段或嵌套消息，scope 区分字段级与消息级问题，type 便于聚合统计，message 保留人类可读的原始说明。不要把 str(header) 当作原始邮件备份；需要复核原始字节时，仍应保存输入 bytes。

按风险决定后续动作

发现 defect 后不必一律丢弃邮件。可以按业务风险分层：主题中的编码瑕疵通常进入低优先级复核；地址头、日期头或内容类型头的异常可能影响路由、时间判断或附件处理，应提高告警等级；解析异常则进入隔离队列，并保留异常类型和原始输入引用。

常见问题

defects 为空是不是说明邮件完全合法？

不能这样推断。它只表示当前解析路径没有报告这些缺陷；业务仍需结合字段是否存在、类型是否符合自己的要求，以及是否发生了异常。

为什么只检查 message.defects 会漏问题？

字段级缺陷挂在具体 Header 对象上，嵌套消息的缺陷也归属于发现问题的节点。应遍历头部并递归检查子消息。

能不能直接修改 Header.defects？

不建议也没有这个必要。它是只读诊断属性；要追加业务标签，请在旁边建立自己的诊断记录。