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

Question

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

这是来自服务器的响应示例：

服务器响应如下：

• 'success' 是了解是否使用私有数据访问的关键 正确。
• 'message' 是成功为 False 的关键（在这种情况下为 成功 == 真，消息未显示
• '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 块返回：

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

根据您在评论中提到的要求，我在下面指出当前的问题和可能的未来问题：

在您的 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给出的第一个块：

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

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

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:。

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

2. 不包含文件的块可以包含}}，因此它也可以破坏您尝试实现的目标。

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

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

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

如果有任何进一步的困惑，请随时询问。

Answer 2

我试图分析您的问题，但找不到比@devReddir 提供的更好的解决方案。

原因是 - 在完全下载之前解析数据是不可能的（或非常困难的）。

解决方法可能是将数据按原样保存在一个大文件中，并由单独的工作人员对其进行解析。这将允许在下载文件时减少服务器内存使用并避免丢失数据。

1. 按原样保存文件

...
while True:
    chunk = next(iter_content, None)
    if not chunk:
        break
    with open(file, "ab") as file_object:
        file_object.write(chunk)
...

1. 在分离的工作人员中读取文件

import json
import base64

with open("saved_as_is.json") as json_file:
    json_object = json.load(json_file)

encoded_base64 = json_object['data']['file']
decoded = base64.b64decode(encoded_base64)
...

为什么动态解析数据如此困难？

1. file 分隔符可以分成两块：

b'... ... ... .., "fi'
b'le": "AAAB... ... .'

1. 其实\\是一个转义符号，你必须手动处理（别忘了\\可以被分块→b'...\', b'\...'）：

b'dTAAAOpgAAA6mAAAF3CculE8AAAABmJLR0QA\\/wD\\/AP+gvaeTAACAAElEQVR42u'

1. 如果文件超小，块行可能如下所示：

b'"file":"SUPERTINY_BASE64_DECODED", "fileName":"Python_lo'

而chunk.split('file:')[1] 将不起作用

4. base64 块必须是 4 的倍数，因此如果您的第一个块（"file": 之后的字符）将是 3 个字符长度，您将需要读取下一个块并将一个第一个字符添加到前一个块的末尾以进行所有后续操作迭代

因此，如果您尝试手动解析数据，这里有一些细微差别。

但是，如果你想选择这种硬方式，here is 如何解码 base64 块。

和here is 允许的base64 字符列表

如果您想使用@devReddir 的解决方案并将整个数据存储在内存中，不确定这里是否有任何使用stream 的好处。

Answer 3

好的，这是完整的工作解决方案：

服务器端（main.py）：

我添加了这段代码，以便能够运行测试服务器，该服务器使用 base64 编码文件响应 json 数据。
我还添加了一些随机性以响应能够检查字符串解析是否独立于字符位置

import base64 as b
import json as j
from fastapi import FastAPI as f
import requests as r
import random as rr
import string as s
import uvicorn as u

banana_url = 'https://upload.wikimedia.org/wikipedia/commons/c/ce/PNG_demo_Banana.png'
banana_b64 = b.encodebytes(
    r.get(banana_url, stream=True).raw.read())
banana_b64 = banana_b64.decode('ascii').replace('\n', '').encode('ascii')

def get_response(banana_file, banana_file_name):
    random_status = ''
    for i in range(rr.randint(3, 30)): random_status += rr.choice(s.ascii_letters)

    banana_response = {
        'status': random_status,
        'data': {
            'fileName': banana_file_name.split('/')[-1],
            'file': banana_file,
        }
    }

    if len(random_status) % 2 == 0:
        banana_response['data']['random_payload'] = 'hello_world'
        banana_response['random_payload'] = '%hello_world_again%'

    return banana_response

app = f()

@app.get("/")
async def read_root():
    resp = get_response(banana_b64, banana_url.split('/')[-1])
    print('file length:', len(resp['data']['file']))
    return resp

if __name__ == "__main__":
    u.run('main:app', host="0.0.0.0", port=8000, reload=True, workers=1)

客户端（文件下载器decoder.py）：

import requests
import base64

# must be larger than len('"file":')
CHUNK_SIZE = 64

# iterable response
r = requests.get('http://127.0.0.1:8000', stream=True).iter_content(chunk_size=CHUNK_SIZE)

class ChunkParser:

    file = None
    total_length = 0

    def close(self):
        if self.file:
            self.file.close()

    def __init__(self, file_name) -> None:
        self.file = open(file_name, 'ab')

    def add_chunk(self, chunk):

        # remove all escape symbols if existing
        chunk = chunk.decode('ascii').replace('\\', '').encode('ascii')

        # if chunk size is not multiple of 4, return modulo to be able add it in next chunk
        modulo = b''
        if not (l := len(chunk)) % 4 == 0:
            modulo = chunk[l-(l%4):]
            chunk = chunk[:l-(l%4)]

        self.file.write(base64.b64decode(chunk))
        self.total_length += len(chunk)

        return modulo



prev_chunk = None
cur_chunk = None
writing_started = False
last_chunk = False
parser = ChunkParser('temp_file.png')
file_found = False
while True:
    
    # set previous chunk on first iterations before modulo may be returned
    if cur_chunk is not None and not writing_started:
        prev_chunk = cur_chunk
    
    # get current chunk
    cur_chunk = next(r, None)
    
    # skip first iteration
    if prev_chunk is None:
        continue
    
    # break loop if no data
    if not cur_chunk:
        break
    
    # concatenate two chunks to avoid b' ... "fil', b'e": ... ' patern
    two_chunks = prev_chunk + cur_chunk

    # if file key found get real base64 encoded data
    if not file_found and '"file":' in two_chunks.decode('ascii'):
        file_found = True

        # get part after "file" key
        two_chunks = two_chunks.decode('ascii').split('"file":')[1].encode('ascii')
        
    if file_found and not writing_started:
        # data should be started after first "-quote
        # so cut all data before "
        if '"' in (t := two_chunks.decode('ascii')):
            two_chunks = t[t.find('"')+1:].encode('ascii')
            writing_started = True
        # handle b' ... "file":', b'"... ' patern
        else:
            cur_chunk = b''
            continue

    # check for last data chunk
    # "-quote means end of value
    if writing_started and '"' in (t := two_chunks.decode('ascii')):
        two_chunks = t[:t.find('"')].encode('ascii')
        last_chunk = True

    if writing_started:

        # decode and write data in file
        prev_chunk = parser.add_chunk(two_chunks)

        # end operation
        if last_chunk:
            if (l := len(prev_chunk)) > 0:
                # if last modulo length is larget than 0, that meaning the data total length is not multiple of 4
                # probably data loss appear? 
                raise ValueError(f'Bad end of data. length is {str(l)} and last characters are {prev_chunk.decode("ascii")}')
            break

parser.close()
print(parser.total_length)

测试此脚本时不要忘记下载后比较文件：

# get md5 of downloaded by chunks file
$ md5 temp_file.png
MD5 (temp_file.png) = 806165d96d5f9a25cebd2778ae4a3da2
# get md5 of downloaded file using browser
$ md5 PNG_demo_Banana.png
MD5 (PNG_demo_Banana.png) = 806165d96d5f9a25cebd2778ae4a3da2

Answer 4

您可以将其流式传输到这样的文件中（pip install base64io）：

class decoder():
    def __init__(self, fh):
        self.fileh = open(fh, 'rb')
        self.closed = False
        search = ''
        start_tag = '"file": "'
        for i in range(1024):
            search += self.fileh.read(1).decode('UTF8')
            if len(start_tag) > len(search)+1:
                continue
            if search[-len(start_tag):] == start_tag:
                break

    def read(self, chunk=1200):
        data = self.fileh.read(chunk)
        if not data:
            self.close()
            return b''
        return data if not data.decode('UTF8').endswith('"}}') else data[:-3]

    def close(self):
        self.fileh.close()
        self.closed = True

    def closed(self):
        return self.closed

    def flush(self):
        pass

    def write(self):
        pass

    def readable(self):
        return True

然后像这样使用类：

from base64io import Base64IO
encoded_source = decoder(fh)
with open("target_file.jpg", "wb") as target, Base64IO(encoded_source) as source:
    for line in source:
        target.write(line)

当然，您需要将本地文件的流式传输更改为 requests.raw 对象的流式传输。