邮件解析：从提供的 EML 文件读取邮件头与 MIME 部分

邮件解析将提供的 EML 字节转换为邮件头、MIME 部分和可供审核的元数据。使用理解 MIME 结构的解析器保留重复邮件头，并按声明的编码解码文本；不要用正则表达式拆分邮件。解析 From 或 Reply-To 字段，只是在描述提供的邮件，并没有验证发送者身份。

本工作流从一个获准用于检查的 EML 文件开始，不需要连接邮箱。文档提取指南讨论可读文档中的业务字段；这里的首要任务则是理解邮件头与多部分结构。

示例配置，并非已记录的测试。工作流接口有目录文档；Ai Flowi 尚未针对这项 EML 流程进行生产环境测试。下方辅助函数与模拟测试样例为提供的邮件字节定义了一份拟议的验收契约。

1. 定义字节与输出边界

你需要 AI Flowi Workflow、一个提供的模拟或其他获准使用的 EML 样本，以及一名审核人员。确定性的本地解析不需要电子邮件账户或模型 API 密钥。如果之后增加模型或外部服务，应使用你自己的服务商账户/API 密钥，并另行支付用量费用。

保留原始字节。解析前，若通过任意文本读取器重新编码 EML 文件，可能改变其编码信息。本例只接受字节，将邮件大小限制为 1 MiB，最多访问 64 个 MIME 节点，并将每个解码后的纯文本部分限制为 256 KiB。这些是为演示选定的上限，并非产品限制，也不表示能全面防护恶意输入。

预期输出包含原始的重复邮件头、解码后的邮件头值、纯文本部分、其他部分的元数据及警告。应将它们分开。警告表示审核人员必须检查来源；成功返回 JSON 对象，并不代表邮件内容已获批准。

相关内容：Google Docs 音频转文字。

2. 检查带重复邮件头的模拟邮件

From: sender@example.invalid
Reply-To: replies@example.invalid
Subject: =?utf-8?b?Q2Fmw6k=?=
Message-ID: <synthetic@example.invalid>
X-Trace: first
X-Trace: second
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable

Caf=C3=A9

邮件头中的主题和正文分别以不同方式编码“Café”。按原顺序保留两次 X-Trace。地址与邮件标识符都是编写的占位值。Reply-To 是邮件内的回复地址指令，并不能证明发送者控制该地址。

多部分备选内容可能同时包含相同内容的纯文本版本与 HTML 版本。不要将它们拼接为独立事实。本例不渲染 HTML，也不将文本附件提升为主要纯文本正文。嵌入的邮件保留为单独的元数据条目，其邮件头与正文不并入外层邮件。

3. 使用 Python 标准库并明确解析策略

拟议的辅助函数使用 BytesParser(policy=policy.default)，保留 raw_items() 并记录解析器缺陷。它先严格按声明的字符集解码；遇到未知字符集或无效文本序列时，会记录警告，并生成明确标记为有损的 UTF-8 替换文本视图。

有范围限制的 EML 检查辅助函数
from email import policy
from email.parser import BytesParser


def parse_eml(raw):
    if not isinstance(raw, bytes):
        raise TypeError("Supply EML bytes")
    if len(raw) > 1024 * 1024:
        raise ValueError("EML exceeds example byte limit")
    # ponytail: materializes the capped message before the part check; use a streaming parser for stricter hostile-input resource limits.
    msg = BytesParser(policy=policy.default).parsebytes(raw)
    out = {"headers": list(msg.raw_items()), "decoded_headers": [],
           "plain_text_parts": [], "other_parts": [], "warnings": []}
    out["decoded_headers"] = [(name, str(value)) for name, value in msg.items()]
    for name in ("From", "Subject", "Message-ID"):
        if not msg.get_all(name):
            out["warnings"].append("missing " + name)
    pending, seen = [msg], 0
    while pending:
        part = pending.pop()
        seen += 1
        if seen > 64:
            raise ValueError("EML exceeds example part limit")
        kind, filename = part.get_content_type(), part.get_filename()
        for header in part.values():
            out["warnings"].extend(type(d).__name__ for d in getattr(header, "defects", ()))
        if part.get_content_disposition() == "attachment" or filename is not None or kind == "message/rfc822":
            out["other_parts"].append({"type": kind, "filename": filename})
            if filename and (filename in (".", "..") or any(c in filename for c in "/\\") or any(ord(c) < 32 or ord(c) == 127 for c in filename)):
                out["warnings"].append("unsafe attachment name")
        elif part.is_multipart():
            pending.extend(reversed(list(part.iter_parts())))
        elif kind == "text/plain":
            payload = part.get_payload(decode=True) or b""
            if len(payload) > 256 * 1024:
                raise ValueError("Text part exceeds example byte limit")
            charset = part.get_content_charset() or "ascii"
            try:
                text = payload.decode(charset, errors="strict")
            except (LookupError, UnicodeError):
                out["warnings"].append("charset failure: " + charset)
                text = payload.decode("utf-8", errors="replace")
            out["plain_text_parts"].append(text)
        else:
            out["other_parts"].append({"type": kind, "filename": filename})
        out["warnings"].extend(type(d).__name__ for d in part.defects)
    return out

用已经提供的字节调用 parse_eml(raw)。本例既不打开任意文件路径，也不下载内容。附件名始终只是元数据：即便名称试图跳出文件夹（反复出现两个点后跟斜杠的形式），也绝不会用作目标路径。辅助函数报告可疑的分隔符或控制字符；这项检查不是文件名清理器，也不是恶意软件检测器。

4. 单独配置工作流交接

目录对 Python Interpreter 的描述是运行 Python 代码，可选导入模块。使用 print() 查看输出。其必填字段 global_imports 和 python_code 提供执行接口。导出资料并未说明文件访问、可用导入模块，或打印的 JSON 如何包装在 results 中。

测试样例表和工作流表在移动设备上保持在可滚动区域内。通过键盘焦点，可以检查初始视图以外的列。

仍需检查运行时表现的目录设置
阶段拟议输入验证边界
Python Interpreter为运行时允许的标准库模块设置 global_imports；将审核过的辅助函数放入 python_code。确认导入模块与经授权的字节提供机制。模拟调用后，检查声明为 JSON 的 results。
JSON Operationsdata：检查过的解析器 JSON；选择 "Select Keys" 或审核过的 "JQ Expression" 作为 operations 的值。区分邮件头、各部分与警告。不要假定 stdout 会变成不带包装的已解析对象。
JSON Object Builder（可选）field_names：一个审核过的记录键，例如 parsed_message。检查生成的可连接输入与 json_object。导出资料未包含多键语法及生成输入的类型。

用拟议的模拟检查定义预期输出，再在你的工作区中用无害样本检查导入模块与结果。Python Interpreter 没有导出资料中定义的 EML 上传输入。不要凭空设计 Read File 到 Python Interpreter 的连接，也不要假定解释器中存在文件句柄。尝试工作流交接前，先确认字节传递与打印结果映射。

5. 保留一组小型失败检查

拟议的模拟验收检查
测试样例预期检查结果
重复 X-Trace 与编码后的主题/正文保留两个原始值；解码后的主题与正文保留带重音的字符。
缺少 From、Subject 或 Message-ID报告缺失邮件头警告，而非编造值。
多部分纯文本/HTML 备选内容纯文本保留为文本，HTML 保留为未渲染部分的元数据。
未知字符集、无效 UTF-8 或格式错误的 base64记录解码或解析器缺陷警告；替换文本并非精确证据。
具有路径遍历形式的附件名与嵌入邮件保留元数据，不写入文件，也不将嵌入文本提升到外层正文。
缺少多部分边界暴露解析器缺陷，而非将格式错误的结构当作普通完整文本。
输入类型错误，或超出邮件、文本、部分数量限制接受检查记录前，抛出声明的错误。

采用契约前，应将这些标准应用到你邮件格式中获准使用的样本。将预期警告放在每个测试样例旁，以便审核时发现漏报。解析器可以接受受损输入并暴露缺陷；没有异常，并不能证明邮件格式正确或真实可信。

电子邮件解析有哪些信任限制，何时应停止？

From、Reply-To 和 Message-ID 都是提供的邮件头。传输身份验证需要来自可信接收系统的证据；粘贴的身份验证相关邮件头仍只是提供的文本。本页不保证能检测伪造或验证邮箱，也不提供发送或接收邮件的说明。

如果邮件敏感且无需外部处理，应使用受控的本地解析器。不执行附件，不渲染 HTML，也不抓取链接中的图片。在增加下游操作之前拒绝超限输入；只要解码丢失信息，就必须审核来源。

套餐：Go 年付方案每月 USD8 起（月付 USD10），Plus 年付方案每月 USD16 起（月付 USD20），Pro 100 年付方案每月 USD80 起（月付 USD100），Pro 200 年付方案每月 USD160 起（月付 USD200），Pro 500 年付方案每月 USD400 起（月付 USD500），每月包含 5,000 至 250,000 个 Flowi Tokens，并支持充值；每次节点运行消耗 1 个 token；不包含第三方 API 和模型使用费用。

配置工作流交接之后，才统计解释器与处理操作的执行次数。被拒绝的邮件应保留原因与来源引用，不得变成已接受的下游记录。这项拟议的 EML 检查设计没有提供实测工作流用量。

作者：Ai Flowi · 发布于 。