读取大型XML文件时如何处理ParseError无效字符问题？

问题场景 #

使用ElementTree的iterparse读取大型XML文件并转换为pandas DataFrame时，触发ParserError，排查后确认是XML中存在


&#xA这类重复的回车换行转义字符导致的。原代码如下：

import numpy as np
import pandas as pd
import xml.etree.ElementTree as et

def read_xml(filepath,tag,*args):
    atts=[arg for arg in args]
    dict_list = []
    
    for _, elem in et.iterparse(filepath, events=("end",)):

        tempdict={}
        
        if elem.tag == tag:
            for i in range(len(atts)):
                tempdict[atts[i]]=elem.attrib[atts[i]]
        
            dict_list.append(tempdict)
            elem.clear()

    return pd.DataFrame(dict_list)

解决方案 #

方法一：预处理XML文件，清理重复转义字符 #

直接读取XML文件内容，用正则替换掉重复的回车换行转义序列，再用StringIO模拟文件对象供iterparse读取。同时优化属性获取逻辑，避免属性不存在时报错：

import re
import xml.etree.ElementTree as et
import pandas as pd
from io import StringIO

def clean_xml(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        content = f.read()
    # 将重复的回车换行转义序列替换为单个
    cleaned_content = re.sub(r'(
)+', '
', content)
    # 若不需要保留转义格式，也可直接替换为普通换行符：
    # cleaned_content = re.sub(r'(
)+', '\n', content)
    return cleaned_content

def read_xml(filepath, tag, *args):
    atts = list(args)
    dict_list = []
    cleaned_content = clean_xml(filepath)
    
    for _, elem in et.iterparse(StringIO(cleaned_content), events=("end",)):
        tempdict = {}
        if elem.tag == tag:
            for att in atts:
                # 用get方法避免属性不存在时抛出KeyError
                tempdict[att] = elem.attrib.get(att, None)
            dict_list.append(tempdict)
            elem.clear()
    
    return pd.DataFrame(dict_list)

方法二：使用宽容解析器处理非法字符 #

如果XML中还存在其他类型的无效字符，可以基于HTMLParser实现自定义宽容解析器，对字符引用做过滤处理后再交给ElementTree解析：

import xml.etree.ElementTree as et
import pandas as pd
from io import StringIO
from html.parser import HTMLParser

class TolerantXMLParser(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=False)
        self.xml_content = []
    
    def handle_starttag(self, tag, attrs):
        attr_str = ' '.join([f'{k}="{v}"' for k, v in attrs])
        self.xml_content.append(f'<{tag} {attr_str}>' if attr_str else f'<{tag}>')
    
    def handle_endtag(self, tag):
        self.xml_content.append(f'</{tag}>')
    
    def handle_data(self, data):
        self.xml_content.append(data)
    
    def handle_charref(self, name):
        # 过滤重复的回车换行转义字符
        if name in ['D', 'A']:
            # 避免连续添加相同的转义字符
            if not (self.xml_content and self.xml_content[-1] == f'&#{name};'):
                self.xml_content.append(f'&#{name};')
        else:
            self.xml_content.append(f'&#{name};')

def read_xml(filepath, tag, *args):
    atts = list(args)
    dict_list = []
    
    # 用自定义解析器处理XML内容
    parser = TolerantXMLParser()
    with open(filepath, 'r', encoding='utf-8') as f:
        parser.feed(f.read())
    cleaned_xml = ''.join(parser.xml_content)
    
    for _, elem in et.iterparse(StringIO(cleaned_xml), events=("end",)):
        tempdict = {}
        if elem.tag == tag:
            for att in atts:
                tempdict[att] = elem.attrib.get(att, None)
            dict_list.append(tempdict)
            elem.clear()
    
    return pd.DataFrame(dict_list)

内容的提问来源于stack exchange，提问作者jmparejaz