python - 如何将带有base64图像文件的流请求管理为json数据响应？

Question

我向服务器发出 requests.post() 调用，服务器用 json 回复我，在这个 json 中有一些键以及 base64 文件。

这是服务器响应的示例:

服务器响应如下:

• 'success' 是了解私有(private)数据访问是否有效的关键 正确。
• 'message' 是成功为 False 时的关键(在本例中为 success == True，消息不显示
• 'data' 是包含文件名和文件名的字典键 Base64格式文件

所以:

{'success': True,
 'message': '',
 'data': {'fileName': 'Python_logo_and_wordmark.svg.png',
          'file': 'iVBORw0KGgoAAAANSUhEUgAABLAAAA....'}} #To limit the space, I cut the very long bytes example

因此 json 中的 respose 也包含该文件，我需要使用 base64.b64decode(r.json()['data']['file']) 进行解码

一切正常，我可以获取我的文件并正确解密。

问题是，对于大文件，我想使用如下的流方法:

file = "G:\Python_logo_and_wordmark.svg.png"
if os.path.isfile(file):
    os.remove(file)

def get_chunk(chunk):

    # Try to decode the base64 file (Chunked)
    # is this a wrong approach?
    chunk = chunk.decode("ascii")
    chunk = chunk.replace('"', '')
    if "file" in chunk:
        chunk = chunk.split('file:')[1]
    elif "}}" in chunk:
        chunk = chunk.split('}}')[0]
    else:
        chunk = chunk
    
    chunk += "=" * ((4 - len(chunk) % 4) % 4)
    chunk_decoded = base64.b64decode(chunk)
    return chunk_decoded

r = requests.post(url=my_url, json=my_data, stream=True)

iter_content = r.iter_content(chunk_size=64)
    
while True:
    chunk = next(iter_content, None)
    if not chunk:
        break
    chunk_decoded = get_chunk(chunk)

    with open(file, "ab") as file_object:
        file_object.write(chunk_decoded)

iter_content block 返回此:

b'{"success":true,"message":"","data":{"fileName":"Python_logo_and'
b'_wordmark.svg.png","file":"iVBORw0KGgoAAAANSUhEUgAABLAAAAFkCAYAA'
b'AAwtsJRAAAABGdBTUEAALGPC\\/xhBQAAACBjSFJNAAB6JgAAgIQAAPoAAACA6AAA'
b'dTAAAOpgAAA6mAAAF3CculE8AAAABmJLR0QA\\/wD\\/AP+gvaeTAACAAElEQVR42u'
b'zdeZwbdf0\\/8Nf7k2Ovdttyt7QIggoth1qUW1AQ5PLeAiK13UwWiqLiBZ4Eb+T6+'

有时在解码过程中填充会存在固有的错误，但经过 1 周的尝试后，我更愿意在这里问这个问题，因为我担心对这种情况采取错误的方法。 我想知道如何以正确的方式处理这种情况

Answer 1

根据您在评论中提到的要求，我指出以下当前问题和 future 可能出现的问题:

在您的 get_chunck 函数中，您正在执行以下操作:

chunk = chunk.decode("ascii")
chunk = chunk.replace('"', '')
if "file" in chunk:
    chunk = chunk.split('file:')[1]
elif "}}" in chunk:
    chunk = chunk.split('}}')[0]
else:
    chunk = chunk

现在查看由 iter_line 给出的第一个 block :

b'{"success":true,"message":"","data":{"fileName":"Python_logo_and'

1. 因此，它将属于 if "file"in chunk: 条件，因为它在 fileName 中包含此 file 字符串。因此，当它尝试根据 file: 拆分此文件时，它将返回一个包含一个元素的列表，因为 file 位于 fileName 中，不是文件:。因此程序将出现以下错误:

Traceback (most recent call last):
  File "main.py", line 7, in <module>
    chunk = chunk.split('file:')[1]
IndexError: list index out of range

尝试使用 if "file:"in chunk: 来代替。

如果 fileName 包含类似“prod_file:someName”的内容，您的程序也可能会失败。您也必须检查这一点。

不包含文件的 block 可以包含 }}，因此它也可能会破坏您想要实现的目标。

您可以修改响应服务器并使用唯一标识符包装文件base64编码字符串的开头和结尾，以便您可以收到如下响应，从而可以在此流方法中保证识别文件的开头和结尾。例如:

{'success': True,
 'message': '',
 'data': {'fileName': 'Python_logo_and_wordmark.svg.png',
          'file': '0000101100iVBORw0KGgoAAAANSUhEUgAABLAAAA....0000101101'}}

我已附加 0000101100 作为起始标识符，并附加 0000101101 作为结束标识符。您可以在写入 block /文件时修剪它们。您可以使用任何其他唯一标识符格式作为您自己的唯一标识符格式，而不与 base64 编码冲突。

如果还有任何困惑，请随时询问。