在Python 3中以二进制模式打开具有通用换行符的文件

Question

我们终于要将一个应用程序升级到Python 3。

我们需要升级的一件事是使用正常换行符重写CSV文件。

原始（Python 2）代码如下：

import csv

IN_PATH = 'in.csv'
OUT_PATH = 'out.csv'

# Opens the original file in 'text mode' (which has no effect on Python 2)
# and with 'universal newlines',
# meaning \r, \n, and \r\n all get treated as line separators.
with open(IN_PATH, 'rU') as in_csv:
    with open(OUT_PATH, 'w') as out_csv:
        csv_reader = csv.reader(in_csv)
        csv_writer = csv.writer(out_csv)

        for tupl in csv_reader:
            csv_writer.writerow(tupl)

这些CSV文件由用户提供。这意味着:

• 我们无法控制它们使用哪种换行符，因此我们需要处理所有的换行符。
• 在此阶段，我们不知道文件的编码方式。

因为我们不知道编码方式，所以无法将字节流解码为文本。

为了使其在Python 3上工作，首先我们改用io.open()，它与py3的open()大多兼容。现在我们不能再使用“文本模式”了，因为在Python 3上需要对字节串进行解码，而我们不知道编码方式。

但是，使用“二进制模式”意味着我们不能再使用通用换行符，因为那只在文本模式下可用。

# Opens the original file in 'binary mode'
# (because we don't know the encoding, so we can't decode it)
# FIXME: How to get universal newline support?
with io.open(IN_PATH, 'rb') as in_csv:
    with io.open(OUT_PATH, 'wb') as out_csv:

请注意，虽然在Python 3中不再支持U模式字符，但它默认在文本模式下使用通用换行符。它似乎没有任何方法在二进制模式下使用通用换行符。

我们如何在Python 3中使此代码工作？

- craigds

---

请注意，csv模块会处理自己的换行符（以正确管理嵌入式换行符）。您是否尝试过只使用普通的“open”和“csv.reader”？https://docs.python.org/3/library/csv.html#id3 - MisterMiyagi

在 Python 3 中，plain open 相当于 Python 2 中的 io.open，因此使用 io.open 可以实现兼容性，这就是我选择使用它的原因。 - craigds

好的观点。只是再重申一下，您确定CSV文件中的换行符处理不足够吗？ - MisterMiyagi

请注意，在Python3中，您必须以文本模式打开文件 - csv 无法处理字节。 - MisterMiyagi

顺便提一下，在Python 2中，“csv”模块期望以二进制模式打开文件。请参阅https://docs.python.org/2/library/csv.html#csv.reader - PM 2Ring

Answer 1

简述：在Python3中使用带代理转义的ASCII：

def text_open(*args, **kwargs):
    return open(*args, encoding='ascii', errors='surrogateescape', **kwargs)

如果您只知道部分编码（例如ASCII的`\r`和`\n`），则建议的方法是使用替代转义符来处理未知代码点：

``` 如果需要更改文件但不知道文件的编码，该怎么办呢？如果你知道编码是 ASCII 兼容的，并且只想检查或修改 ASCII 部分，那么可以使用 surrogateescape 错误处理程序打开文件： ```

这将使用保留占位符在文本流中嵌入未知字节。例如，字节`b'\x99'`变成了“Unicode”代码点`\udc99`。这适用于读写，允许您保留任意嵌入的数据。

常见的行尾符（`\n`、`\r`、`\r\n`）在ASCII中都有明确定义。因此，使用ASCII编码与替代转义符就足够了。

为了兼容性代码，最好提供Python 2和Python 3版本的分离功能。`open`已经足够相似，大多数情况下，只需要插入替代转义处理即可。

if sys.version_info[0] == 3:
    def text_open(*args, **kwargs):
        return open(*args, encoding='ascii', errors='surrogateescape', **kwargs)
else:
    text_open = open

这允许使用通用的换行符，而不需要知道确切的编码。您可以使用此功能直接读取或转录文件：

with text_open(IN_PATH, 'rU') as in_csv:
    with text_open(OUT_PATH, 'wU') as out_csv:
        for line in in_csv:
            out_csv.write(line)

如果您需要进一步格式化 csv 模块，由 text_open 提供的文本流也足够了。要处理非ASCII分隔符/填充/引号，请将它们从字节串转换为适当的代理。

if sys.version_info[0] == 3:
    def surrogate_escape(symbol):
        return symbol.decode(encoding='ascii', errors='surrogateescape')
else:
    surrogate_escape = lambda x: x

Dezimeter = surrogate_escape(b'\xA9\x87')

- MisterMiyagi

---

谢谢，这看起来就是正确答案，至少对于只支持py3的代码而言。这对于支持_两个_版本来说有点沮丧，因为surrogateescape在py2上不可用。看起来python-future有一个surrogateescape回溯，但它很难插入io.open，这意味着（正如您指出的）我们需要为py2和py3编写完全不同的代码路径。感谢您对此所做的研究。 - craigds

Answer 2

我认为在Python 3中没有内置的方法可以实现你想要的功能。如果不知道编码方式，你只能确定有一堆字节 - 你不确定它们中的哪些是表示字符\r或\n。

你的Python 2代码可能使用了系统默认编码方式，根据sys.getdefaultencoding()来通知内置的通用换行符规范化程序（不要引用我，我还没有看过实现），如果你的系统像我的一样，那么可能是ascii。

幸运的是，我认为大多数编码方式（包括utf-8）只在其高阶字符的映射方面不同（超过了ascii范围）。因此，假设字节10表示\n，字节13表示\r对于所有常见编码方式来说并不是一个糟糕的假设 - 这意味着你可以通过逐字节读取输入（或者更确切地说，使用滑动的两个字节窗口）自己进行替换。

警告：我并没有对以下代码进行全面测试，以确保其在处理重复序列（如\r\r\r）或奇怪的事物（如\n\r）时的行为表现得良好，因此虽然它可能会合理地处理这些情况，但也可能不行。请在您自己的数据上进行测试。

from __future__ import print_function

import io
import six  # optional (but hugely helpful for a 2 to 3 port)


def normalize(prev, curr):
    ''' Given current and previous bytes, get tuple of bytes that should be written

    :param prev: The byte just before the read-head
    :type  prev: six.binary_type
    :param curr: The byte at the read-head
    :type  curr: six.binary_type
    :returns   : A tuple containing 0, 1, or 2 bytes that should be written
    :rtype     : Tuple[six.binary_type]
    '''
    R = six.binary_type(b'\r')
    N = six.binary_type(b'\n')
    if   curr == R:         # if we find R, can't dump N yet because it might be start of RN sequence and we must consume N too
        return ()
    elif curr == N:         # if we find N, doesn't matter what previous byte was - dump N
        return (N,)
    elif prev == R:         # we know current not N or R; if previous byte was R - dump N, then the current byte
        return (N, curr)
    else:                   # we know current not N or R and prev not R - dump the current byte
        return (curr,)


if __name__ == '__main__':

    IN_PATH = 'in.csv'
    OUT_PATH = 'out.csv'

    with io.open(IN_PATH, mode='rb') as in_csv:
        with io.open(OUT_PATH, mode='wb') as out_csv:
            prev = None                                 # at start, there is no previous byte
            curr = six.binary_type(in_csv.read(1))      # at start, the current byte is the input file's first byte
            while curr:                                 # loop over all bytes in the input file
                for byte in normalize(prev, curr):      # loop over all bytes returned from the normalizing function
                    print(repr(byte))                   # debugging
                    out_csv.write(byte)                 # write each byte to the output file
                prev = curr                             # update value of previous byte
                curr = six.binary_type(in_csv.read(1))  # update value of current byte

这对我来说在Python 2.7.16和3.7.3上都有效，使用我创建的输入文件（使用Python 3）如下：

import io

with io.open('in.csv', mode='wb', encoding='latin-1') as fp:
    fp.write('à,b,c\n')
    fp.write('1,2,3\r')
    fp.write('4,5,6\r\n')
    fp.write('7,8,9\r')
    fp.write('10,11,12\n')
    fp.write('13,14,15')

它也可以使用encoding='UTF-8'（应该这样做）。

不必像我一样使用six.binary_type()，但我发现这是一个有用的提醒，特别是在编写跨版本代码时，可以明确数据的语义。

我花了一段时间试图找出比手动检查所有字节更好的方法，但没有成功。如果其他人发现了一种方法，我很想看看！

- Peter Henry

---

谢谢，这看起来很有用。我预计性能会相当惊人，但如果我们添加缓冲读取，也许它会被接受。我会进行一些测试，谢谢。 - craigds

Answer 3

Python 3 中的 open 函数有一个参数 newline。将其设置为 None 可以启用通用换行模式。

import csv

IN_PATH = 'in.csv'
OUT_PATH = 'out.csv'

with open(IN_PATH, 'r', newline=None) as in_csv:
    with open(OUT_PATH, 'w') as out_csv:
        csv_reader = csv.reader(in_csv)
        csv_writer = csv.writer(out_csv)

        for tupl in csv_reader:
            csv_writer.writerow(tupl)

示例：

示例文件：

a,b,c\n
1,2,3\r
4,5,6\r\n
7,8,9

示例代码：

with open('file.csv', 'r', newline=None) as fp:
    reader = csv.reader(fp)
    for line in reader:
        print(line)

# prints:
['a', 'b', 'c']
['1', '2', '3']
['4', '5', '6']
['7', '8', '9']

- James

---

好的，我在我的问题中提到了这一点 - newline=None 实际上是默认值，但在二进制模式下不起作用。 - craigds