如何在 Python 3 中将文本流编码为字节流？答案

【问题标题】：How to encode a text stream into a byte stream in Python 3?如何在 Python 3 中将文本流编码为字节流？
【发布时间】：2019-01-06 04:21:09
【问题描述】：

将字节流解码为文本流很容易：

import io
f = io.TextIOWrapper(io.BytesIO(b'Test\nTest\n'), 'utf-8')
f.readline()

在本例中，io.BytesIO(b'Test\nTest\n') 是字节流，f 是文本流。

我想做完全相反的事情。给定一个文本流或类似文件的对象，我想将其编码为字节流或类似文件的对象而不处理整个流。

这是我迄今为止尝试过的：

import io, codecs

f = codecs.getreader('utf-8')(io.StringIO('Test\nTest\n'))
f.readline()
# TypeError: can't concat str to bytes

f = codecs.EncodedFile(io.StringIO('Test\nTest\n'), 'utf-8')
f.readline()
# TypeError: can't concat str to bytes

f = codecs.StreamRecoder(io.StringIO('Test\nTest\n'), None, None,
                         codecs.getreader('utf-8'), codecs.getwriter('utf-8'))
# TypeError: can't concat str to bytes

f = codecs.encode(io.StringIO('Test\nTest\n'), 'utf-8')
# TypeError: utf_8_encode() argument 1 must be str, not _io.StringIO

f = io.TextIOWrapper(io.StringIO('Test\nTest\n'), 'utf-8')
f.readline()
# TypeError: underlying read() should have returned a bytes-like object, not 'str'

f = codecs.iterencode(io.StringIO('Test\nTest\n'), 'utf-8')
next(f)
# This works, but it's an iterator instead of a file-like object or stream.

f = io.BytesIO(io.StringIO('Test\nTest\n').getvalue().encode('utf-8'))
f.readline()
# This works, but I'm reading the whole stream before converting it.

我正在使用 Python 3.7

【问题讨论】：

标签： python python-3.x io character-encoding stream

【解决方案1】：

你可以很容易地自己写这个；您只需要决定如何进行缓冲。

例如：

class BytesIOWrapper(io.RawIOBase):
    def __init__(self, file, encoding='utf-8', errors='strict'):
        self.file, self.encoding, self.errors = file, encoding, errors
        self.buf = b''
    def readinto(self, buf):
        if not self.buf:
            self.buf = self.file.read(4096).encode(self.encoding, self.errors)
            if not self.buf:
                return 0
        length = min(len(buf), len(self.buf))
        buf[:length] = self.buf[:length]
        self.buf = self.buf[length:]
        return length
    def readable():
        return True

我认为这正是您所要求的。

>>> f = BytesIOWrapper(io.StringIO("Test\nTest\n"))
>>> f.readline()
b'Test\n'
>>> f.readline()
b'Test\n'
>>> f.readline()
b''

如果您想变得更聪明，您可能想要包装 codecs.iterencode 而不是一次缓冲 4K。或者，由于我们使用的是缓冲区，您可能希望创建一个BufferedIOBase 而不是RawIOBase。此外，一个名为BytesIOWrapper 的类可能应该处理write，但这是容易的部分。困难的部分是实现seek/tell，因为你不能在TextIOBase 内任意寻找；寻找开始和结束非常容易；另一方面，寻找已知的先前位置是困难的（除非您依赖 TextIOBase.tell 返回一个字节位置——它不能保证这样做，而 TextIOWrapper 确实如此，StringIO 不......） .

无论如何，我认为这是如何编写最复杂的io 类的最简单演示。

【讨论】：