2025, Dec 01 12:06
Как избежать повторов при чтении файла в Python: курсор, newline и EOFError
Наглядно разбираем ошибку с возвратом курсора в начало при построчном чтении файла в Python. read_until и jump_to_line, настройка newline и обработка EOFError.
Создать небольшой вспомогательный класс поверх файлового ввода-вывода в Python — распространённый способ разобраться в курсорах, чтении и границах строк. Но есть тонкая ловушка: если при каждой операции снова переходить к началу файла, вы будете перечитывать один и тот же фрагмент и искусственно завышать количество строк. Ниже — минимальный пример, показывающий, как это происходит, и как исправить ситуацию без внешних модулей и без изменения базовой идеи.
Воспроизводим проблему
Пример ниже записывает в файл две строки, а затем пытается проходить его построчно, пока не возникнет EOFError. Класс держит два файловых дескриптора — для записи и для чтения — и ведёт общий курсор.
def run():
f = QuickIO("numbers.txt")
f.put("1, -2, 5, 0, 19, -7, end\n5, 5, -1, -10, 9, end")
rows = 0
while(True):
try:
print("Line #"+str(rows+1))
f.jump_to_line(rows)
except EOFError:
break
rows += 1
print("The amount of lines are: "+str(rows + 1))
class QuickIO:
# поля: path, out, inp, pos
def __init__(self, path):
self.path = path
self.pos = 0
self.out = open(path, "w")
self.inp = open(path, "r")
self.out.seek(self.pos)
self.inp.seek(self.pos)
def jump_to(self, char_idx):
self.pos = char_idx
self.out.seek(char_idx)
self.inp.seek(char_idx)
tmp = open(self.path, "r")
tmp.seek(char_idx)
if tmp.read(1) == "":
tmp.seek(char_idx - 1)
if tmp.read(1) == "":
tmp.close()
raise EOFError
tmp.close()
def jump_to_line(self, n):
self.jump_to(0)
for i in range(0, n):
print(repr(self.read_until()))
def put(self, s):
self.out.write(s)
self.out.flush()
self.jump_to(self.pos + len(s))
def read_until(self, token="\n"):
data = self.inp.read()
self.jump_to(self.pos)
end = 0
while len(data) > end:
if data[end:end+len(token)] == token:
break
if len(data[end:end+len(token)]) != len(token):
self.jump_to(self.pos + len(data))
return data
end += 1
self.jump_to(self.pos + end + len(token))
return data[0:end]
run()
Что на самом деле идёт не так
Повторы строк и неверные итоги возникают из‑за сброса позиции в файле. Метод, который должен переходить к нужной строке, делает жёсткий сброс: jump_to(0). То есть каждая итерация начинается с начала и снова «прожёвывает» один и тот же контент. В итоге цикл не продвигается по файлу так, как вы ожидаете: вывод зацикливается на первой строке, а счётчик уходит вразнобой.
Есть и платформенная тонкость: при записи текста с переводами строк в Windows последовательность может быть \r\n. Если ваш код рассчитывает на разделитель \n, а в файле фактически лежит \r\n, возникнут «пропуски» строк и несостыковки. Откройте и на чтение, и на запись с параметром newline="\n" — так содержимое будет согласовано с тем разделителем, который вы ищете.
Исправляем логику
Во‑первых, перестаньте каждый раз перематывать файл к началу при переходе по строкам. Читайте относительно текущего положения курсора. Во‑вторых, измените read_until: пусть она возбуждает EOFError, когда больше нечего читать; смещает курсор при обнаружении разделителя; а иначе доходит до конца. И, наконец, нормализуйте переводы строк, передавая newline="\n" и в писатель, и в читатель.
class QuickIO:
# поля: path, out, inp, pos
def __init__(self, path):
self.path = path
self.pos = 0
self.out = open(path, "w", newline="\n")
self.inp = open(path, "r", newline="\n")
self.out.seek(self.pos)
self.inp.seek(self.pos)
def jump_to(self, char_idx):
self.pos = char_idx
self.out.seek(char_idx)
self.inp.seek(char_idx)
tmp = open(self.path, "r")
tmp.seek(char_idx)
if tmp.read(1) == "":
tmp.seek(char_idx - 1)
if tmp.read(1) == "":
tmp.close()
raise EOFError
tmp.close()
def jump_to_line(self, n):
for _ in range(n):
self.read_until('\n')
return self.read_until('\n')
def put(self, s):
self.out.write(s)
self.out.flush()
self.jump_to(self.pos + len(s))
def read_until(self, token="\n"):
data = self.inp.read()
if data == "":
raise EOFError
self.jump_to(self.pos)
end = 0
while len(data) > end:
if data[end:end+len(token)] == token:
self.jump_to(self.pos + end + len(token))
return data[0:end]
end += 1
self.jump_to(self.pos + len(data))
return data[0:end]
Запускайте так, передавая 0, чтобы каждый раз запрашивать «следующую» строку:
def run():
f = QuickIO("numbers.txt")
f.put("1, -2, 5, 0, 19, -7, end\n5, 5, -1, -10, 9, end")
line_count = 0
while True:
try:
print(f"Line #{line_count + 1}")
print(repr(f.jump_to_line(0)))
except EOFError:
break
line_count += 1
print("The amount of lines are:", line_count)
run()
В результате вы получите ожидаемый, без повторов, вывод и корректное количество строк.
Почему это важно
Корректность обхода файла держится на одном инварианте: позиция чтения должна предсказуемо двигаться вперёд. Любой безусловный переход к началу в операции «по строкам» нарушает это правило и приводит к перечитыванию и раздутым счётчикам. Не менее важна единообразная обработка переводов строк: если искать \n, а в файле лежит \r\n, появляются смещения на один символ и «пропавшие» разделители, которые сложно заметить.
Итоги
Продвигайтесь относительно текущего курсора, а не возвращайтесь к нулю; нормализуйте переводы строк, открывая оба потока с newline="\n"; и сделайте так, чтобы ваша read‑until либо возвращала фрагмент и перескакивала за разделитель, либо возбуждала EOFError, если больше данных нет. Если позже захотите упростить ещё, можно открыть файл один раз в режиме r+ и использовать readline(), но описанных правок достаточно, чтобы текущий подход работал корректно и предсказуемо.