python 3：使用预读从标准输入管道读取字节

Question

我想读取字节。 sys.stdin 以文本模式打开，但它有一个可用于读取字节的缓冲区：sys.stdin.buffer。

我的问题是，当我将数据通过管道传输到 python 时，如果我想要预读，我似乎只有 2 个选项，否则我会得到一个 io.UnsupportedOperation: File or stream is not seekable.

1. 从sys.stdin 读取缓冲文本，将该文本解码为字节，然后返回

   (sys.stdin.read(1).decode(); sys.stdin.seek(-1, io.SEEK_CUR).

   由于输入流中的字节不可编码而无法接受。

2. 使用peek 从标准输入的缓冲区中获取一些字节，将其分割成适当的数字，然后祈祷，因为peek 并不能保证任何事情：它可能提供的比您要求的要少或多……

   (sys.stdin.buffer.peek(1)[:1])

   peek 的文档确实不足，并为您提供了一堆字节，您必须对这些字节进行性能密集型切片。

顺便说一句。该错误仅在管道时适用：对于./myscript.py <somefile，sys.stdin.buffer 支持查找。然而sys.stdin 始终是相同的对象层次结构：

$ cat testio.py
#!/usr/bin/env python3
from sys import stdin
print(stdin)
print(stdin.buffer)
print(stdin.buffer.raw)"
$ ./testio.py
<_io.TextIOWrapper name='<stdin>' mode='r' encoding='UTF-8'>
<_io.BufferedReader name='<stdin>'>
<_io.FileIO name='<stdin>' mode='rb'>
$ ./testio.py <somefile
[the same as above]
$ echo hi | ./testio.py
[the same as above]

一些最初的想法，例如将字节流包装到随机访问缓冲区中，并出现与上述相同的错误：BufferedRandom(sys.stdin.buffer).seek(0) ⇒ io.UnsupportedOperation…

最后，为了您的方便，我介绍一下：

Python 的 io 类层次结构

IOBase
├RawIOBase
│└FileIO
├BufferedIOBase  (buffers a RawIOBase)
│├BufferedWriter┐ 
│├BufferedReader│
││        └─────┴BufferedRWPair
│├BufferedRandom (implements seeking)
│└BytesIO        (wraps a bytes)
└TextIOBase
 ├TextIOWrapper  (wraps a BufferedIOBase)
 └TextIO         (wraps a str)

如果您忘记了这个问题：如何在不解码/编码任何内容且不前进流的光标的情况下从标准输入获取下一个字节？

Answer 1

异常不是来自 Python，而是来自操作系统，它不允许在管道上搜索。 （如果您从常规管道重定向输出，即使它是标准输入，也可以查找它。）这就是为什么您在一种情况下得到错误而不是在另一种情况下得到错误，即使类是相同的。

用于预读的经典 Python 2 解决方案是将流包装在您自己的实现预读的流实现中：

class Peeker(object):
    def __init__(self, fileobj):
        self.fileobj = fileobj
        self.buf = cStringIO.StringIO()

    def _append_to_buf(self, contents):
        oldpos = self.buf.tell()
        self.buf.seek(0, os.SEEK_END)
        self.buf.write(contents)
        self.buf.seek(oldpos)

    def peek(self, size):
        contents = self.fileobj.read(size)
        self._append_to_buf(contents)
        return contents

    def read(self, size=None):
        if size is None:
            return self.buf.read() + self.fileobj.read()
        contents = self.buf.read(size)
        if len(contents) < size:
            contents += self.fileobj.read(size - len(contents))
        return contents

    def readline(self):
        line = self.buf.readline()
        if not line.endswith('\n'):
            line += self.fileobj.readline()
        return line

sys.stdin = Peeker(sys.stdin)

在 Python 3 中支持完整的sys.stdin，同时窥视未解码的流很复杂——如上所示，将stdin.buffer 包装起来，然后在可窥视的流上实例化一个新的TextIOWrapper，并将TextIOWrapper 安装为@ 987654326@.

但是，由于您只需要查看sys.stdin.buffer，因此将cStringIO.StringIO 更改为io.BytesIO 并将'\n' 更改为b'\n' 后，上述代码将正常工作。

Answer 2

user4815162342 的解决方案虽然非常有用，但似乎存在一个问题，即它与 io.BufferedReader peek 方法的当前行为不同。

内置方法将为连续 peek() 调用返回相同的数据（从当前读取位置开始）。

user4815162342 的解决方案将为每个连续的 peek 调用返回连续的数据块。这意味着如果用户希望多次使用相同的数据，则必须再次包装 peek 以连接输出。

这是返回内置行为的修复：

def _buffered(self):
    oldpos = self.buf.tell()
    data = self.buf.read()
    self.buf.seek(oldpos)
    return data

def peek(self, size):
    buf = self._buffered()[:size]
    if len(buf) < size:
        contents = self.fileobj.read(size - len(buf))
        self._append_to_buf(contents)
        return self._buffered()
    return buf

See the full version here

还有其他可以应用的优化，例如在耗尽缓冲区的读取调用时删除先前缓冲的数据。当前的实现将任何偷看的数据留在缓冲区中，但这些数据是不可访问的。

Answer 3

试试这个：

import sys

ssb = sys.stdin.buffer.read(1)
if ssb == b'h':
    print(ssb+sys.stdin.buffer.read())

回显一个字符串：

a@fuhq:~$ echo 'hi' | python3 buf_test.py 
b'hi\n'

重定向文件：

a@fuhq:~$ cat hi.text
hi
a@fuhq:~$ python3 buf_test.py   <  hi.text
b'hi\n'