Python大型XML文件高效流式解析教程

import re
from html.parser import HTMLParser
import pandas as pd

class MyHTMLParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.data = {}  # 用于存储按managedObject class分类的解析结果
        self.current = None  # 存储当前正在处理的managedObject数据
        self.list_name = None  # 标记当前是否在处理一个<list>标签
        self.p_name = None  # 存储当前<p>标签的name属性值

    def handle_starttag(self, tag, attrs):
        """
        处理HTML/XML的开始标签，如 <tag attr="value">
        """
        attrs = dict(attrs)  # 将属性列表转换为字典便于查找

        if tag == "managedobject":
            # 当遇到 <managedObject> 标签时，初始化当前对象的数据字典
            # 从 distName 属性中解析出 MRBTS, NRBTS, NRCELL, NRREL
            # re.findall(r"([^/]+?)-([^/]+)", attrs["distname"])[1:]
            # 例如 "PLMN-PLMN/MRBTS-277215/NRBTS-277215/NRCELL-0/NRREL-1"
            # 会匹配到 [('PLMN', 'PLMN'), ('MRBTS', '277215'), ('NRBTS', '277215'), ('NRCELL', '0'), ('NRREL', '1')]
            # [1:] 表示从第二个元组开始，即跳过 'PLMN-PLMN'
            self.current = dict(re.findall(r"([^/]+?)-([^/]+)", attrs["distname"])[1:])
            # 添加 id 属性
            self.current['id'] = attrs.get('id') # 确保id存在
            # 将当前对象数据添加到对应 class 的列表中
            self.data.setdefault(attrs["class"], []).append(self.current)
        elif tag == "list":
            # 当遇到 <list> 标签时，记录其 name 属性，用于后续 <p> 标签的命名
            self.list_name = attrs["name"]
        elif tag == "p":
            # 当遇到 <p> 标签时，根据是否在 <list> 内部生成不同的键名
            if self.list_name:
                self.p_name = f'Item-{self.list_name}-{attrs["name"]}'
            else:
                self.p_name = attrs["name"]

    def handle_endtag(self, tag):
        """
        处理HTML/XML的结束标签，如 </tag>
        """
        if tag == "managedobject":
            # 当 <managedObject> 结束时，清空当前对象数据，表示一个对象的解析完成
            self.current = None
        elif tag == "list":
            # 当 <list> 结束时，清空 list_name 标记
            self.list_name = None
        elif tag == "p":
            # 当 <p> 结束时，清空 p_name 标记
            self.p_name = None

    def handle_data(self, data):
        """
        处理标签内部的数据内容
        """
        if not self.current:
            # 如果当前没有正在处理的 managedObject，则忽略数据
            return

        if self.p_name is not None:
            # 如果当前正在处理 <p> 标签，将其数据内容存储到 current 字典中
            self.current[self.p_name] = data

# 实例化解析器
parser = MyHTMLParser()

# 逐行读取XML文件并进行解析
# 假设XML文件名为 "data.xml"
try:
    with open("data.xml", "r", encoding="utf-8") as f_in: # 指定编码以避免解析错误
        for line in f_in:
            parser.feed(line)
except FileNotFoundError:
    print("错误：data.xml 文件未找到。请确保文件存在且路径正确。")
except Exception as e:
    print(f"解析文件时发生错误: {e}")
finally:
    parser.close() # 关闭解析器，释放资源

# 将解析结果转换为Pandas DataFrame并写入Excel
output_excel_path = "output_streaming.xlsx"
try:
    with pd.ExcelWriter(output_excel_path) as writer:
        for k, v in parser.data.items():
            if v: # 确保列表不为空
                df = pd.DataFrame(v)
                # 尝试将所有列转换为数值类型，如果失败则忽略（errors="ignore"）
                df = df.apply(pd.to_numeric, errors="ignore")
                df.to_excel(writer, sheet_name=k, index=False)
                print(f"成功将数据写入 Excel 表格 '{k}'。")
            else:
                print(f"'{k}' 类型没有数据，跳过写入。")
    print(f"所有数据已成功导出到 '{output_excel_path}'")
except Exception as e:
    print(f"写入Excel文件时发生错误: {e}")

# 示例：打印其中一个DataFrame
# for k, v in parser.data.items():
#     print(f"\nSheet name: {k}")
#     print("-" * 80)
#     df = pd.DataFrame(v)
#     print(df)
#     break # 只打印第一个