Python3中如何忽略格式错误的UTF-8字节以处理损坏XML文件

一、核心思路 #

问题根源是未完全解码的zip文件导致XML末尾出现截断的UTF-8字符，我们可以从两个层面解决：先修复损坏的UTF-8字节流，再增强lxml解析器的容错性。

二、修改后的完整代码 #

import sys
import lxml.etree as ET
from docx.oxml.ns import qn  # 依赖python-docx的命名空间工具

def fix_truncated_utf8(data):
    """修复截断的UTF-8字节流，返回有效UTF-8字符串"""
    # 读取输入的二进制数据（兼容文件对象和字节串）
    if hasattr(data, 'read'):
        bytes_data = data.read()
    else:
        bytes_data = data.encode('utf-8') if isinstance(data, str) else data
    
    # 尝试解码，遇到错误时截断到有效位置
    try:
        return bytes_data.decode('utf-8')
    except UnicodeDecodeError as e:
        # 只保留错误发生前的有效字节
        valid_bytes = bytes_data[:e.start]
        return valid_bytes.decode('utf-8')

def xml2text(xml_content):
    """提取XML文本内容，兼容损坏的XML文件"""
    text = ''
    # 第一步：修复UTF-8截断问题
    cleaned_xml = fix_truncated_utf8(xml_content)
    
    # 配置更宽松的XML解析器
    parser = ET.XMLParser(
        recover=True,  # 开启恢复模式，忽略解析错误
        encoding='utf-8',
        no_network=True,  # 禁止网络请求，避免DTD加载错误
        resolve_entities=False  # 不解析实体，减少报错可能
    )
    
    # 尝试解析，失败则做兜底处理
    try:
        root = ET.fromstring(cleaned_xml, parser)
    except ET.XMLSyntaxError:
        # 去掉末尾可能的空字节和无效字符后重试
        cleaned_xml = cleaned_xml.rstrip('\x00').rstrip()
        root = ET.fromstring(cleaned_xml, parser)
    
    # 提取文本逻辑保持不变
    for child in root.iter():
        if child.tag == qn('w:t'):
            t_text = child.text
            text += t_text if t_text is not None else ''
        elif child.tag == qn('w:tab'):
            text += '\t'
        elif child.tag in (qn('w:br'), qn('w:cr')):
            text += '\n'
        elif child.tag == qn("w:p"):
            text += '\n\n'
    return text

def process(docx_input):
    text = xml2text(docx_input)
    return text.strip()

if __name__ == '__main__':
    # 参数处理函数（示例）
    def process_args():
        import argparse
        parser = argparse.ArgumentParser(description='提取docx中的文本内容')
        parser.add_argument('--docx', type=argparse.FileType('rb'), help='输入docx文件路径')
        return parser.parse_args()
    
    args = process_args()
    # 从stdin读取时用二进制模式，避免提前解码出错
    if args.docx is None:
        text = process(sys.stdin.buffer)
    else:
        text = process(args.docx)
    sys.stdout.write(text)

三、关键优化说明 #

• 修复UTF-8截断：通过捕获UnicodeDecodeError，精准截断到有效字节位置，确保输入XML是合法UTF-8，从根源解决lxml因编码问题抛出的错误
• 增强解析器容错：在recover=True基础上，添加no_network=True和resolve_entities=False，避免解析时触发额外的网络请求或实体解析错误
• 统一输入处理：不管是文件还是stdin，都先读取二进制数据再修复，避免提前解码导致的编码异常
• 兜底重试机制：如果第一次解析失败，清理末尾的空字节和空白字符后重试，覆盖更多损坏场景

四、简化方案可选 #

如果不需要保留损坏部分的内容，fix_truncated_utf8函数可以简化为直接忽略无效字节：

def fix_truncated_utf8(data):
    if hasattr(data, 'read'):
        bytes_data = data.read()
    else:
        bytes_data = data.encode('utf-8') if isinstance(data, str) else data
    # 直接忽略所有无效UTF-8字节
    return bytes_data.decode('utf-8', errors='ignore')

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