SAX（Simple API for XML）采用事件驱动模式解析 XML。与 DOM 和 ElementTree 不同，SAX 不会将整个文档载入内存，而是边读边触发事件，适合处理超大 XML 文件。
注意：SAX 只能单向遍历，无法随意回读、修改、随机访问节点，修改 XML 一般不用 SAX。
SAX 解析器像扫描仪一样逐字节读取 XML，遇到特定结构时触发回调：
XML 文档 事件序列
<root> → start_element("root")
<item> → start_element("item")
文本 → characters("文本")
</item> → end_element("item")
</root> → end_element("root")上述示例中，SAX 会从 <root> 开始扫描，每当扫描到一个节点的还是标签或者结束标签都会触发对应的事件，我们处理事件即可。
下面将通过实例简单介绍如何使用 SAX 解析 XML 文档。
Python 的 xml.sax.handler.ContentHandler 是 SAX 流式解析 XML 的事件回调基类，SAX 不加载完整文档到内存，边读边触发事件。
当解析器读到标签、文本、注释等节点时，自动调用对应方法，用户继承该类重写方法实现自定义处理。
ContentHandler 的常用关键方法：
startDocument()：文档解析开始
endDocument()：文档解析结束
startElement(name, attrs)：遇到起始标签（带属性）
endElement(name)：遇到闭合标签
characters(content)：读取标签内文本内容
示例代码：
import xml.sax
from io import StringIO
from xml.sax.handler import ContentHandler
# 继承ContentHandler回调基类
class MyHandler(ContentHandler):
"""自定义内容处理器"""
def startDocument(self):
"""文档开始"""
print("开始解析文档")
def endDocument(self):
"""文档结束"""
print("文档解析完成")
def startElement(self, name, attrs):
"""元素开始标签"""
print(f"开始元素: <{name}>")
if attrs:
for attr_name, value in attrs.items():
print(f" 属性: {attr_name}={value}")
def endElement(self, name):
"""元素结束标签"""
print(f"结束元素: </{name}>")
def characters(self, content):
"""文本内容"""
text = content.strip()
if text:
print(f" 文本: '{text}'")
# 使用示例
xml_data = '''<?xml version="1.0"?>
<message id="1">
<to>张三</to>
<body>你好</body>
</message>'''
parser = xml.sax.make_parser()
parser.setContentHandler(MyHandler()) # 创建自定义的事件处理器，设置给 parser
parser.parse(StringIO(xml_data))当面示例中，自定义的 MyHandler 继承了 ContentHandler，并且重写了回调方法。在 parser.parse() 之前通过 parser.setContentHandler() 将自定义的 ContentHandler 设置给 parser。当 parser 解析 XML 文档时，遇到开始标签、结束标签、文本时会自动调用 ContentHandler 中的方法。
运行示例，输出结果如下：
开始解析文档
开始元素: <message>
属性: id=1
开始元素: <to>
文本: '张三'
结束元素: </to>
开始元素: <body>
文本: '你好'
结束元素: </body>
结束元素: </message>
文档解析完成前面通过一个示例演示了如何使用 ContentHandler 实现 SAX 解析 XML 文档。
下面通过解析一个关于图书馆 XML，统计书籍数量和书籍基础信息，巩固 ContentHandler 的用法，代码如下：
import xml.sax
from io import StringIO
from xml.sax.handler import ContentHandler
class BookCollector(ContentHandler):
"""收集书籍信息"""
def __init__(self):
# 存放所有书籍
self.books = []
# 当前书籍信息
self.current_book = None
self.current_field = None
self.text_buffer = []
def startElement(self, name, attrs):
if name == 'book':
# 创建当前书籍信息，通过属性获取 id 和 genre 信息
self.current_book = {
'id': attrs.get('id'),
'genre': attrs.get('genre', 'unknown')
}
elif name in ('title', 'author', 'price'):
# 解析子标签
self.current_field = name
self.text_buffer = []
def characters(self, content):
# 提取标签的文本内容
if self.current_field:
self.text_buffer.append(content)
def endElement(self, name):
if name == 'book':
# book 标签结束时，当前书籍信息存入所有书籍中，并且将当前书籍节点置空
self.books.append(self.current_book)
self.current_book = None
elif name in ('title', 'author', 'price'):
# 子标签结束时，将取的值设置到指定字段的值
text = ''.join(self.text_buffer).strip()
self.current_book[name] = text
self.current_field = None
# 使用
xml_data = '''
<library>
<book id="b1" genre="sci-fi">
<title>三体</title>
<author>刘慈欣</author>
<price>59.00</price>
</book>
<book id="b2" genre="novel">
<title>活着</title>
<author>余华</author>
<price>45.00</price>
</book>
</library
'''
handler = BookCollector() # 自定义的解析器
parser = xml.sax.make_parser()
parser.setContentHandler(handler)
parser.parse(StringIO(xml_data))
print(f"共收集 {len(handler.books)} 本书:")
for book in handler.books:
print(f" {book['title']} ({book['author']}) - ¥{book['price']}")运行示例，输出如下：
共收集 2 本书:
三体 (刘慈欣) - ¥59.00
活着 (余华) - ¥45.00SAX 的核心优势是低内存占用，假设有一个 10GB 的日志文件 huge_log.xml 内容如下（为了演示，只列举了部分内容）：
<?xml version="1.0" encoding="utf-8"?>
<logs>
<!-- info 普通日志 -->
<log level="info">系统启动成功，服务端口8080已监听</log>
<log level="info">用户id=1001发起登录请求</log>
<log>数据库连接池初始化完成（无level属性，默认info）</log>
<!-- warning 警告日志 -->
<log level="warning">内存占用超过80%，建议扩容服务器</log>
<log level="warning">接口请求耗时大于2000ms，存在性能隐患</log>
<!-- error 错误日志 -->
<log level="error">数据库查询异常：Table 'user' not exists</log>
<log level="error">文件上传失败，存储空间不足</log>
<log level="error">第三方支付接口超时，重试3次仍失败</log>
<!-- 可无限复制下方log节点模拟超大文件 -->
<log level="info">定时任务001执行完成</log>
<log level="warning">缓存命中率低于60%</log>
<log level="error">Redis连接断开</log>
</logs>编写代码解析上面的日志文件，统计各级别日志的总记录数：
import xml.sax
import sys
from xml.sax.handler import ContentHandler
class LogAnalyzer(ContentHandler):
"""分析大型日志文件"""
def __init__(self):
# 错误总数
self.error_count = 0
# 警告总数
self.warning_count = 0
# 总数
self.total_count = 0
# 当前级别
self.current_level = None
def startElement(self, name, attrs):
if name == 'log':
# log 开始标签，总数加一，同时获取级别信息
self.total_count += 1
self.current_level = attrs.get('level', 'info')
def endElement(self, name):
if name == 'log':
# log 结束标签，根据级别进行+1操作
if self.current_level == 'error':
self.error_count += 1
elif self.current_level == 'warning':
self.warning_count += 1
self.current_level = None
def report(self):
print(f"日志统计:")
print(f" 总数: {self.total_count}")
print(f" 错误: {self.error_count}")
print(f" 警告: {self.warning_count}")
# 分析文件（内存占用极小）
analyzer = LogAnalyzer()
parser = xml.sax.make_parser()
parser.setContentHandler(analyzer)
parser.parse('huge_log.xml') # 处理大文件
analyzer.report()运行示例，输出如下：
日志统计:
总数: 11
错误: 4
警告: 3使用 ErrorHandler 错误处理器，可以统一捕获、处理程序运行异常，解耦业务逻辑与报错处理，避免大量重复 try-catch。
ErrorHandler 可以在解析 XML 时拦截标签闭合、格式错乱等问题，避免程序直接崩溃。如果不设置 setErrorHandler()，SAX 默认直接打印错误到控制台，不会收集。
SAX 把 XML 错误分为 3 类，对应三个回调方法：
warning()：警告，不影响解析，可继续
error()：标准错误（标签不匹配、非法字符等），解析可继续
fatalError()：致命错误（XML 格式彻底损坏），解析终止
示例代码：
import xml.sax
from xml.sax.handler import ContentHandler
from xml.sax.handler import ErrorHandler
class LogAnalyzer(ContentHandler):
"""分析大型日志文件"""
def __init__(self):
self.error_count = 0
self.warning_count = 0
self.total_count = 0
self.current_level = None
def startElement(self, name, attrs):
if name == 'log':
self.total_count += 1
self.current_level = attrs.get('level', 'info')
def endElement(self, name):
if name == 'log':
if self.current_level == 'error':
self.error_count += 1
elif self.current_level == 'warning':
self.warning_count += 1
self.current_level = None
def report(self):
print(f"日志统计:")
print(f" 总数: {self.total_count}")
print(f" 错误: {self.error_count}")
print(f" 警告: {self.warning_count}")
# 看这里，继承了 ErrorHandler 错误处理器
class StrictErrorHandler(ErrorHandler):
"""严格错误处理"""
def warning(self, exception):
"""警告（可忽略）"""
print(f"警告: {exception}")
def error(self, exception):
"""可恢复错误"""
print(f"错误: {exception}")
raise exception # 继续抛出
def fatalError(self, exception):
"""致命错误"""
print(f"致命错误: {exception}")
raise exception
analyzer = LogAnalyzer()
parser = xml.sax.make_parser()
parser.setContentHandler(analyzer)
parser.setErrorHandler(StrictErrorHandler()) # 设置错误处理器
parser.parse('huge_log.xml') # 处理大文件
analyzer.report()XML 允许自定义标签，不同业务 / 厂商可能出现同名标签，造成冲突。命名空间（Namespace）通过唯一 URI区分不同标签，解决重名冲突。
命名空间语法如下：
声明命名空间
默认命名空间：xmlns="http://xxx.com/ns"（无前缀，当前元素及子元素默认使用）
带前缀命名空间：xmlns:f="http://furniture.com/ns"（前缀 f 绑定 URI）
带命名空间标签写法
前缀形式：<f:table>
完整限定名（Clark notation）：{http://furniture.com/ns}table
属性也支持命名空间：<f:table h:width="100">
Python 的 xml.sax 提供两套元素回调，它们互斥，只能实现一套：
如果是普通无命名空间，则使用 startElement / endElement。如果只能拿到带前缀原始字符串（f:table），无法拆分 URI 和本地名，不推荐处理带 ns XML。
如果存在命名空间，则使用 startElementNS / endElementNS，它会自动拆分 URI、本地标签名、前缀，是标准处理方案。
注意：要触发 *NS 系列方法，解析器必须开启命名空间支持。
示例：
import xml.sax
from io import StringIO
from xml.sax.handler import ContentHandler
class NamespaceHandler(ContentHandler):
"""处理带命名空间的 XML"""
def startPrefixMapping(self, prefix, uri):
"""命名空间声明"""
print(f"命名空间声明: xmlns:{prefix}='{uri}'")
def endPrefixMapping(self, prefix):
"""命名空间结束"""
print(f"命名空间结束: {prefix}")
def startElementNS(self, name, qname, attrs):
"""带命名空间的元素开始"""
# name 是 (uri, localname) 元组
uri, localname = name
print(f"元素开始: {{{uri}}}{localname}")
# 遍历带命名空间的属性
for attr_name, value in attrs.items():
attr_uri, attr_local = attr_name
print(f" 属性: {{{attr_uri}}}{attr_local}={value}")
def endElementNS(self, name, qname):
"""带命名空间的元素结束"""
uri, localname = name
print(f"元素结束: {{{uri}}}{localname}")
# 使用
xml_data = '''
<root xmlns:dc="http://purl.org/dc/elements/1.1/"
xmlns:media="http://search.yahoo.com/mrss/">
<dc:title>文档标题</dc:title>
</root>
'''
handler = NamespaceHandler()
parser = xml.sax.make_parser()
# 开启NS解析
parser.setFeature(xml.sax.handler.feature_namespaces, True)
parser.setContentHandler(handler)
parser.parse(StringIO(xml_data))运行实例，输出如下：
命名空间声明: xmlns:dc='http://purl.org/dc/elements/1.1/'
命名空间声明: xmlns:media='http://search.yahoo.com/mrss/'
元素开始: {None}root
元素开始: {http://purl.org/dc/elements/1.1/}title
元素结束: {http://purl.org/dc/elements/1.1/}title
元素结束: {None}root
命名空间结束: media
命名空间结束: dc通过输出可以看出，命名空间声明和结束、元素开始和结束都会触发回调。
编写一个 XMLToCSVConverter 类，用来将 XML 转 CSV 格式的数据并打印到控制条，你可以根据需要将 CSV 保存到文件。代码如下：
import xml.sax
import csv
from io import StringIO
from xml.sax.handler import ContentHandler
class XMLToCSVConverter(ContentHandler):
"""
无命名空间XML转CSV通用SAX处理器
"""
def __init__(self, output_file, fieldnames):
"""
初始化转换器
"""
self.output_file = output_file # CSV输出流对象
self.fieldnames = fieldnames # CSV表头字段集合
self.writer = None # csv.DictWriter写入器实例
self.current_row = {} # 缓存当前单条record行数据
self.current_field = None # 当前正在解析的子字段名称
self.text_buffer = [] # 文本缓冲区，解决characters分段读取问题
self.in_record = False # 标记解析器是否进入<record>标签内部
def startDocument(self):
""" 整个XML文档开始解析时触发，写入CSV表头 """
self.writer = csv.DictWriter(
self.output_file,
fieldnames=self.fieldnames,
extrasaction='ignore' # 过滤XML多余属性，不存在表头时不抛异常
)
# 写入CSV第一行表头
self.writer.writeheader()
def startElement(self, name, attrs):
"""
解析到任意开始标签时触发
:param name: 标签名称字符串
:param attrs: 标签属性字典 key=属性名 value=属性值
"""
# 命中记录根标签，初始化一行空数据
if name == "record":
# 预先填充所有表头字段为空字符串，保证CSV列对齐不缺失
self.current_row = {field: "" for field in self.fieldnames}
self.in_record = True
# 遍历当前 record 所有属性，匹配表头字段后赋值
for attr_k, attr_v in attrs.items():
if attr_k in self.current_row:
self.current_row[attr_k] = attr_v
# 处于record内部，且当前标签是需要采集的字段
elif self.in_record and name in self.fieldnames:
self.current_field = name
self.text_buffer.clear() # 清空缓冲区，准备接收标签内文本
def characters(self, content):
"""
SAX回调：捕获标签中间的文本内容
注意：长文本、换行、空格会被拆分成多次调用，必须用缓冲区拼接
:param content: 单次读取到的文本片段
"""
# 仅当正在采集某个字段时，才缓存文本
if self.current_field is not None:
self.text_buffer.append(content)
def endElement(self, name):
"""
SAX回调：解析到闭合标签时触发
:param name: 闭合标签名称
"""
# record标签闭合，将当前行写入CSV
if name == "record":
self.writer.writerow(self.current_row)
self.current_row.clear()
self.in_record = False
# 字段子标签闭合，拼接缓冲区文本存入行数据
elif self.in_record and name == self.current_field:
# 拼接所有文本片段并去除首尾空白换行空格
full_text = "".join(self.text_buffer).strip()
self.current_row[name] = full_text
self.current_field = None # 退出字段采集状态
# 程序入口，仅直接运行脚本时执行
if __name__ == '__main__':
xml_data = '''<?xml version="1.0"?>
<customers>
<record id="1" country="CN">
<name>张三</name>
<email>zhangsan@example.com</email>
<city>北京</city>
</record>
<record id="2" country="US">
<name>John Doe</name>
<email>john@example.com</email>
<city>New York</city>
</record>
<!-- 边界测试：缺少country、city，name为空标签 -->
<record id="3">
<name></name>
<email>test@demo.com</email>
</record>
</customers>
'''
# 内存IO流，替代本地文件用于演示输出
output = StringIO()
handler = XMLToCSVConverter(
output,
fieldnames=["id", "name", "email", "city", "country"]
)
# 创建SAX解析器并绑定自定义处理器
parser = xml.sax.make_parser()
parser.setContentHandler(handler)
parser.parse(StringIO(xml_data))
# 打印转换完成后的CSV内容
print("===== 生成CSV结果 =====")
print(output.getvalue())运行实例，输出如下：
===== 生成CSV结果 =====
id,name,email,city,country
1,张三,zhangsan@example.com,北京,CN
2,John Doe,john@example.com,New York,US
3,,test@demo.com,,注意，SAX 是 Python 标准库中处理超大 XML 的唯一可靠方案。虽然回调机制让代码稍显复杂，但它保证了无论文件多大，内存占用都保持稳定。下一章将介绍 xmltodict 等第三方库，提供更简洁的 XML 处理方式。
川公网安备51010802032098