我正在尝试使用 Exchange GetAttachment网络服务使用requests , lxml和 base64io .此服务在 SOAP XML HTTP 响应中返回一个 base64 编码的文件。文件内容包含在单个 XML 元素的一行中。 GetAttachment只是一个例子，但问题更普遍。

我想将解码后的文件内容直接流式传输到磁盘，而不会将附件的全部内容随时存储在内存中，因为一个附件可能有几百 MB。

我试过这样的:

r = requests.post('https://example.com/EWS/Exchange.asmx', data=..., stream=True)
with open('foo.txt', 'wb') as f:
    for action, elem in lxml.etree.iterparse(GzipFile(fileobj=r.raw)):
    if elem.tag == 't:Content':
        b64_encoder = Base64IO(BytesIO(elem.text))
        f.write(b64_encoder.read())

但 lxml 仍将附件的副本存储为 elem.text。有什么方法可以创建一个完全流式传输的 XML 解析器，它还可以直接从输入流中流式传输元素的内容吗？

最佳答案

不要使用 iterparse在这种情况下。 iterparse()方法只能发出元素开始和结束事件，因此当找到结束 XML 标记时，元素中的任何文本都会提供给您。

相反，使用 SAX parser interface .这是 XML 解析库的通用标准，用于将解析后的数据传递给内容处理程序。 ContentHandler.characters() callback以 block 的形式传递字符数据(假设实现 XML 库实际上利用了这种可能性)。这是来自 ElementTree API 的较低级别 API，并且 Python 标准库已经捆绑了 Expat 解析器来驱动它。

那么流程就变成了:

• 将传入的请求流包装在 GzipFile 中为了方便减压。或者，更好的是，设置 response.raw.decode_content = True并根据服务器设置的内容编码将解压留给请求库。
• 传递 GzipFile实例或原始流到 .parse() method使用 xml.sax.make_parser() 创建的解析器.然后解析器继续以 block 的形式从流中读取。通过使用 make_parser()您首先可以启用命名空间处理等功能(如果 Exchange 决定更改用于每个命名空间的短前缀，这将确保您的代码不会中断)。
• 内容处理器characters()使用 XML 数据 block 调用方法；检查正确的元素启动事件，以便您知道何时需要 base64 数据。您可以在 chunks of (a multiple of) 4 characters 中解码该 base64 数据一次，并将其写入文件。我不会使用 base64io在这里，只做你自己的分块。

一个简单的内容处理器可以是:

from xml.sax import handler
from base64 import b64decode

class AttachmentContentHandler(handler.ContentHandler):
    types_ns = 'http://schemas.microsoft.com/exchange/services/2006/types'

    def __init__(self, filename):
        self.filename = filename

    def startDocument(self):
        self._buffer = None
        self._file = None

    def startElementNS(self, name, *args):
        if name == (self.types_ns, 'Content'):
            # we can expect base64 data next
            self._file = open(self.filename, 'wb')
            self._buffer = []

    def endElementNS(self, name, *args):
        if name == (self.types_ns, 'Content'):
            # all attachment data received, close the file
            try:
                if self._buffer:
                    raise ValueError("Incomplete Base64 data")
            finally:
                self._file.close()
                self._file = self._buffer = None

    def characters(self, data):
        if self._buffer is None:
            return
        self._buffer.append(data)
        self._decode_buffer()

    def _decode_buffer(self):
        remainder = ''
        for data in self._buffer:
            available = len(remainder) + len(data)
            overflow = available % 4
            if remainder:
                data = (remainder + data)
                remainder = ''
            if overflow:
                remainder, data = data[-overflow:], data[:-overflow]
            if data:
                self._file.write(b64decode(data))
        self._buffer = [remainder] if remainder else []

你会像这样使用它:

import requests
from xml.sax import make_parser, handler

parser = make_parser()
parser.setFeature(handler.feature_namespaces, True)
parser.setContentHandler(AttachmentContentHandler('foo.txt'))

r = requests.post('https://example.com/EWS/Exchange.asmx', data=..., stream=True)
r.raw.decode_content = True  # if content-encoding is used, decompress as we read
parser.parse(r.raw)

这将以最大 64KB 的 block (默认 IncrementalParser buffer size )解析输入 XML，因此附件数据最多解码为 48KB 的原始数据 block 。

我可能会扩展内容处理程序以获取目标目录，然后查找 <t:Name>元素提取文件名，然后使用它为找到的每个附件将数据提取到正确的文件名。您还想验证您实际上正在处理 GetAttachmentResponse记录并处理错误响应。

关于python - 全流式 XML 解析器，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/52989143/