如何处理含.msg附件的.msg文件并生成结构化数据表用于NLP分析

我完全懂你现在的痛点——处理带嵌套.msg附件的邮件时，现有代码没法把所有层级的内容都整理成你想要的扁平结构化表格，对吧？咱们一步步来修改代码，解决这个问题，最终生成你需要的那种每行对应一个文档的结构化数据表。

问题根源分析 #

你当前的process_msg函数返回的是嵌套的字典结构，没法直接转换成扁平的表格；而且缺少两个关键逻辑：一是统一的文档ID生成机制，二是追踪每个文档（包括附件）对应的父文档ID，这两个正好是你要的表格里的核心字段。

解决方案：重构代码实现扁平结构化收集 #

咱们重写代码，核心思路是用一个全局列表来收集所有层级的文档条目（主邮件、附件里的.msg、Word/PDF等其他附件），同时记录每个条目的父ID，最后直接转换成DataFrame。

第一步：准备辅助函数

先搞定一些基础的辅助功能，比如清理文件名、生成唯一文件名避免重复：

import os
import extract_msg
import pandas as pd
import re
import glob
from docx import Document
import pdfplumber

# 辅助函数：清理文件名，移除系统不允许的非法字符
def sanitize_filename(filename):
    return re.sub(r'[\\/*?:"<>|]', '_', filename)

# 辅助函数：生成唯一文件名，避免同文件夹下重复
def get_unique_filename(filename, folder):
    base, ext = os.path.splitext(filename)
    counter = 1
    unique_name = filename
    while os.path.exists(os.path.join(folder, unique_name)):
        unique_name = f"{base}_{counter}{ext}"
        counter += 1
    return unique_name

第二步：递归收集所有文档条目

重写处理函数，让它递归遍历所有层级的.msg附件，同时把每个文档（主邮件、附件邮件、其他类型附件）都作为单独条目加入列表：

def process_msg_with_attachments(file_path, input_folder, entries, parent_id=None):
    """
    递归处理.msg文件及其所有附件，收集扁平结构化条目
    :param file_path: 当前要处理的文件路径
    :param input_folder: 输入文件夹根路径
    :param entries: 用于存储所有条目的列表（可变对象，递归中修改会自动保留）
    :param parent_id: 父文档的Doc ID（当前条目是某个文档的附件时传入）
    """
    try:
        msg = extract_msg.openMsg(file_path)
        # 生成当前文档的唯一Doc ID（用列表长度+1，自动递增）
        current_doc_id = len(entries) + 1
        
        # 把主邮件加入条目列表
        entries.append({
            "Doc ID": current_doc_id,
            "Data": f"Email: {msg.subject[:20]}..." if msg.subject else "Email (无主题)",
            "File path": file_path,
            "Attachment": "N" if parent_id is None else "Y",
            "ID of Attachment": parent_id if parent_id is not None else ""
            # 如果你需要NLP用的邮件内容，可以在这里加字段：
            # "Sender": msg.sender,
            # "Body": msg.body or msg.htmlBody or "无内容",
            # "Received Time": msg.receivedTime
        })
        
        # 创建附件存储文件夹
        attachments_folder = os.path.join(input_folder, "attachments")
        os.makedirs(attachments_folder, exist_ok=True)
        
        for att in msg.attachments:
            att_name = get_unique_filename(sanitize_filename(att.name), attachments_folder)
            att_ext = re.search(r"\.(\w+)$", att_name)
            att_ext = att_ext.group(1).lower() if att_ext else None
            
            saved_path = os.path.join(attachments_folder, att_name)
            att.save(customPath=attachments_folder, customFilename=att_name)
            
            if att_ext == "msg":
                # 递归处理.msg附件，父ID就是当前主邮件的Doc ID
                process_msg_with_attachments(saved_path, input_folder, entries, parent_id=current_doc_id)
            else:
                # 处理非.msg类型的附件（Word/PDF/TXT等）
                att_doc_id = len(entries) + 1
                # 生成附件的描述文本
                att_data_desc = f"{att_ext.upper()} 文件: {att_name[:20]}..."
                entries.append({
                    "Doc ID": att_doc_id,
                    "Data": att_data_desc,
                    "File path": saved_path,
                    "Attachment": "Y",
                    "ID of Attachment": current_doc_id
                    # 同样可以加附件内容字段用于NLP：
                    # "Content": 这里可以调用对应的提取函数（比如docx、pdf的文本提取）
                })
                
                # 【可选】提取非.msg附件的文本内容，用于后续NLP
                if att_ext == "docx":
                    doc = Document(saved_path)
                    att_content = "\n".join([para.text for para in doc.paragraphs])
                    entries[-1]["Content"] = att_content
                elif att_ext == "pdf":
                    with pdfplumber.open(saved_path) as pdf:
                        att_content = "\n".join([page.extract_text() for page in pdf.pages if page.extract_text()])
                        entries[-1]["Content"] = att_content
                elif att_ext == "txt":
                    with open(saved_path, "r", encoding="utf-8", errors="ignore") as f:
                        entries[-1]["Content"] = f.read()
                else:
                    entries[-1]["Content"] = "暂不支持提取该类型文件内容"
                
    except Exception as e:
        print(f"处理文件 {file_path} 时出错: {str(e)}")
        # 把错误条目也加入列表，方便排查
        error_doc_id = len(entries) + 1
        entries.append({
            "Doc ID": error_doc_id,
            "Data": f"处理失败: {os.path.basename(file_path)}",
            "File path": file_path,
            "Attachment": "N" if parent_id is None else "Y",
            "ID of Attachment": parent_id if parent_id is not None else "",
            "Error": str(e)
        })

第三步：主函数生成结构化数据表

最后，遍历所有.msg文件，收集所有条目并转换成你需要的DataFrame：

def generate_document_table(input_folder):
    # 初始化条目列表，用于存储所有扁平化的文档记录
    all_doc_entries = []
    # 遍历文件夹下所有.msg文件
    for msg_file in glob.glob(os.path.join(input_folder, "*.msg")):
        process_msg_with_attachments(msg_file, input_folder, all_doc_entries)
    
    # 转换成DataFrame，调整列顺序和你要的表格一致
    doc_df = pd.DataFrame(all_doc_entries)
    doc_df = doc_df[["Doc ID", "Data", "File path", "Attachment", "ID of Attachment"]]
    return doc_df

# 示例调用
if __name__ == "__main__":
    # 替换成你的实际输入文件夹路径
    your_input_folder = "./your_email_folder"
    result_table = generate_document_table(your_input_folder)
    
    # 打印结果表格
    print(result_table)
    # 保存为Excel文件，方便后续分析
    result_table.to_excel(os.path.join(your_input_folder, "邮件结构数据表.xlsx"), index=False)

关键改动说明 #

1. 扁平结构化收集：用可变列表all_doc_entries存储所有层级的文档，递归时直接追加条目，最终所有内容都是扁平的，完美匹配你要的表格格式。
2. 父ID追踪：通过parent_id参数记录当前文档是哪个文档的附件，对应表格里的ID of Attachment字段。
3. 自动ID生成：利用列表长度自动生成唯一的Doc ID，不用额外维护计数器。
4. NLP友好扩展：预留了提取邮件内容、附件文本内容的字段，直接可以用于后续的NLP处理。

注意事项 #

• 如果嵌套的.msg附件层级极深（比如超过100层），可能会触发Python的递归深度限制，这时候可以把递归改成迭代方式，但一般邮件嵌套不会这么深，递归足够用。
• 处理大量文件前建议先备份原始数据，避免意外覆盖。
• 你可以根据自己的NLP需求，在条目里加入更多字段（比如邮件发件人、正文内容、附件文本等）。

备注：内容来源于stack exchange，提问作者ds_1234