Как читать записи, завершенные пользовательским разделителем из файла в Python?

Question

Я хотел бы сделать способ for line in file в Python, где конец строки переопределен, чтобы быть любой строкой, которую я хочу. Другой способ сказать, что я хочу читать записи из файла, а не строк; Я хочу, чтобы это было так же быстро и удобно, как чтение строк.

Это эквивалент Python для установки Perl $/ разделитель входной записи или использование Scanner в Яве. Это не обязательно использовать for line in file (в частности, итератор не может быть файловым объектом). Просто что-то эквивалентное, что позволяет избежать чтения слишком большого количества данных в память.

См. Также: добавлена ​​поддержка чтения записей с произвольными разделителями в стандартный стек ввода-вывода.

Answer 1

В Python 2.x ничего нет file объект, или Python 3.3 io классы, которые позволяют указать пользовательский разделитель для readline, (The for line in file в конечном итоге использует тот же код, что и readline.)

Но это довольно легко построить самостоятельно. Например:

def delimited(file, delimiter='\n', bufsize=4096):
    buf = ''
    while True:
        newbuf = file.read(bufsize)
        if not newbuf:
            yield buf
            return
        buf += newbuf
        lines = buf.split(delimiter)
        for line in lines[:-1]:
            yield line
        buf = lines[-1]

---

Вот глупый пример этого в действии:

>>> s = io.StringIO('abcZZZdefZZZghiZZZjklZZZmnoZZZpqr')
>>> d = delimited(s, 'ZZZ', bufsize=2)
>>> list(d)
['abc', 'def', 'ghi', 'jkl', 'mno', 'pqr']

---

Если вы хотите сделать это правильно как для двоичных, так и для текстовых файлов, особенно в 3.x, это немного сложнее. Но если он должен работать только для одного или другого (и одного языка или другого), вы можете игнорировать это.

Аналогично, если вы используете Python 3.x (или используете io объекты в Python 2.x), и хотят использовать буферы, которые уже поддерживаются в BufferedIOBase вместо того, чтобы просто помещать буфер поверх буфера, это хитрее. io Документы действительно объясняют, как делать все... но я не знаю ни одного простого примера, так что вам действительно придется прочитать хотя бы половину этой страницы и просмотреть остальные. (Конечно, вы можете просто использовать необработанные файлы напрямую... но не в том случае, если хотите найти разделители Юникода...)

Answer 2

Связанный OP для обсуждения проблемы предлагает еще одно решение для чтения строк данных, заканчивающихся настраиваемым разделителем, из файла, опубликованного Аланом Барнетом. Он работает как с текстовыми, так и с двоичными файлами и является большим улучшением по сравнению сfileLineIter рецепт Дугласа Алана.

Вот моя полированная версия Алана Барнета. resplit. Я заменил добавление строки+=с якобы более быстрым "".joinконкатенация строк, и я добавил подсказки по типу для еще большей производительности. Моя версия настроена на работу с бинарными файлами. Я должен использовать шаблон регулярного выражения для разделения, потому что мой разделитель в его простой форме также встречается внутри строк данных в функции без разделителей, поэтому мне нужно учитывать его контекст. Однако вы можете перенастроить его для текстовых файлов и заменить шаблон регулярного выражения общимstr если у вас есть простой и уникальный разделитель, нигде больше не используемый.

import pathlib
import functools
import re
from typing import Iterator, Iterable, ByteString
import logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger().setLevel(logging.DEBUG)
logger = logging.getLogger(__name__)


def resplit(chunks_of_a_file: Iterator, split_pattern: re.Pattern) -> Iterable[ByteString]:
    """
    Reads chunks of a file one chunk at a time, 
    splits them into data rows by `split_pattern` 
    and joins partial data rows across chunk boundaries.
    borrowed from https://bugs.python.org/issue1152248#msg223491
    """
    partial_line = None
    for chunk in chunks_of_a_file:
        if partial_line:
            partial_line = b"".join((partial_line, chunk))
        else:
            partial_line = chunk
        if not chunk:
            break
        lines = split_pattern.split(partial_line)
        partial_line = lines.pop()
        yield from lines
    if partial_line:
        yield partial_line


if __name__ == "__main__":
    path_to_source_file = pathlib.Path("source.bin")
    with open(path_to_source_file, mode="rb") as file_descriptor:
        buffer_size = 8192
        sentinel = b""
        chunks = iter(functools.partial(file_descriptor.read, buffer_size), sentinel)
        data_rows_delimiter = re.compile(b"ABC")
        lines = resplit(chunks, data_rows_delimiter)
        for line in lines:
            logger.debug(line)