Разбор CSV в Python: BOM, потеря столбца в DictReader и проверка схемы

От байтов файла до проверенных записей

CSV может прочитаться без исключения и всё равно потерять столбец: достаточно двух одинаковых заголовков перед DictReader. На воспроизводимых примерах разберём этот сбой, невидимый BOM, границы strict=True и проверку схемы до передачи записей в импорт.

Материал подготовлен с использованием ИИ: текст, учебный код и иллюстрации. Обложка создана ИИ и не является снимком эксперимента. Результаты действительно получены запуском кода на синтетических данных; опыт работы с реальным магазином не заявляется.

Где пригодится этот разбор

CSV — текстовый способ хранения таблицы: поле соответствует значению в столбце, а запись объединяет значения одной строки таблицы. Например, у товара это артикул, название и примечание. CSV удобно передавать между сервисами: файл текстовый, его можно просмотреть без исходной программы. Однако наличие запятых ещё не означает, что каждую из них нужно считать границей столбца. При импорте каталога или обработке выгрузки задача состоит в восстановлении полей, а не в механическом разрезании текста.

Расширенный опыт выполнен 7 ноября 2026 года в Linux с CPython 3.12.14. Полный код csv_lab.py приведён в конце статьи. Он создаёт собственные файлы во временной папке, не подключается к сети и не меняет реальные каталоги. Excel, Windows и кабинеты маркетплейсов не тестировались.

Три поля и две записи

В тестовом наборе три столбца: id, title, note. Первая запись содержит идентификатор 001, название Desk, small и примечание plain. Во второй записи идентификатор 002, название Lamp «North», а примечание состоит из двух строк: line 1 и line 2.

Набор намеренно небольшой. Запятая проверяет границу поля, кавычки — сохранность самого названия, перенос — границу записи. Начальные нули в идентификаторах позволяют заметить нежелательное превращение текста в число.

Файл создаётся стандартным csv.writer. Указаны UTF-8, запятая в качестве разделителя и CRLF между записями. CRLF — последовательность из двух служебных символов, LF — один символ перевода строки. Внутри примечания второй записи находится один LF. Так можно отдельно увидеть перенос внутри значения и завершение записи. Получившийся текст выглядит так:

id,title,note
001,"Desk, small",plain
002,"Lamp ""North""","line 1
line 2"

В этой текстовой иллюстрации окончания строк не различимы визуально; точные байты сохраняются в файле, созданном скриптом. Кавычки вокруг поля защищают содержащуюся в нём запятую или перевод строки от роли разделителя. Две подряд кавычки внутри обрамлённого поля представляют одну кавычку в значении. Такие правила описаны в RFC 4180 [1]. Документ имеет статус Informational: это описание распространённого формата, а не гарантия одинакового поведения всех программ.

Что получилось при простом разделении

Для ошибочного варианта использована последовательность двух действий: text. splitlines(), затем line. split(',') для каждого фрагмента. Первое действие режет текст по переводам строки, второе — по каждой запятой. Такая последовательность не понимает правил CSV.

Результат — четыре фрагмента с количеством полей 3, 4, 3 и 1. Вторая физическая строка разбилась по запятой внутри Desk, small. Последняя строка line 2″ стала самостоятельным фрагментом, хотя относится к примечанию предыдущей записи.

У reader есть line_num: количество прочитанных физических строк, а не номер записи. В нашем файле после трёх next() получаются пары «логическая запись, физическая строка»: [1,1], [2,2], [3,4]. Это полезно для диагностики: ошибка в третьей записи может относиться к четвёртой строке файла. Счётчик enumerate и reader. line_num отвечают на разные вопросы.

Что восстановил CSV-парсер

Тот же файл прочитан через csv.reader. Это парсер: программа, которая разбирает текст по правилам формата и возвращает значения полей. Открытие выполнено с encoding='utf-8' и newline=''. Эти параметры относятся к известному тестовому файлу, а не к автоматическому определению любой выгрузки. Документация Python рекомендует newline='' при работе модуля csv с файловым объектом [2].

Парсер вернул три записи по три поля. Название Desk, small осталось одним значением. Lamp «North» вернулось с одной парой кавычек внутри строки. Примечание сохранило внутренний перевод строки. Полученный список был целиком сопоставлен с заранее заданным исходным списком и совпал с ним.

Идентификаторы остались строками 001 и 002. В использованном обычном режиме csv. reader возвращает строки, а не угадывает тип каждого столбца [2]. Этот результат ничего не доказывает о поведении табличного редактора: он в опыте не запускался.

Почему правильного парсера ещё недостаточно

Следующая проверка использует те же значения, но при записи разделителем становится точка с запятой. Чтение с настройками по умолчанию, ожидающими запятую, не восстановило исходный набор. Оно даже не обязано закончиться явной ошибкой: некоторые символы можно разобрать в соответствии с неверно выбранными правилами.

При повторном чтении с delimiter=';' набор восстановился полностью. Вывод для практического импорта простой: расширение .csv не сообщает все параметры. Разделитель, кодировку, кавычки и правила экранирования нужно брать из документации конкретной выгрузки или из согласованного формата обмена. Не стоит исправлять файл глобальной заменой запятых на точки с запятой: запятая может быть частью значения, как в этом опыте.

Strict не проверяет вашу схему

strict=True отклонил незакрытую кавычку с csv. Error: unexpected end of data, но принял запись из двух полей при трёх именах в заголовке. Более того, вход 001, a"b, x был прочитан как ['001', 'a"b', 'x']: кавычка внутри необрамлённого поля не вызвала исключения. Поэтому strict нельзя считать полной проверкой соответствия RFC 4180. Он усиливает обработку определённых ошибок выбранного диалекта, а не подтверждает любую правильность CSV.

Валидатор ниже сначала требует существующий заголовок, уникальные имена и точный порядок ['id', 'title', 'note']. Затем он проверяет ширину каждой записи, непустой id и отсутствие повторного id. Эти требования заданы для учебной таблицы. Они не обязательны для любого CSV. Только после них запись превращается в словарь.

Проверки выполняются последовательно и останавливаются на первом нарушении. Для короткой строки сначала проверяется ширина, поэтому случайного IndexError при обращении к полю не возникает. Пробельный id здесь считается заполненным: правило «после strip() не пусто» при необходимости надо добавить явно. Автоматическое удаление пробелов, смена регистра и переименование колонок могут изменить смысл исходных данных.

BOM меняет имя первого столбца

К исходным байтам добавим UTF-8-сигнатуру EF BB BF. При decode('utf-8') она становится символом U+FEFF в начале первого поля. Заголовок выглядит как id, но его представление repr содержит '\ufeffid'. Сравнение с ожидаемым 'id' не проходит. Это не ошибка кавычек или разделителя: отличия появились на этапе декодирования.

decode('utf-8-sig') снимает начальную сигнатуру. В опыте после этого весь набор совпал с исходным, а validated_rows(…, encoding='utf-8-sig') вернул две записи данных. Этот кодек подходит для известного UTF-8 с возможным BOM. Он не превращает Windows-1251 или UTF-16 в UTF-8 и не заменяет договорённость о кодировке. Глобально удалять U+FEFF по всему документу также не нужно.

Как одинаковые заголовки уничтожают отдельное значение

Вход состоит из двух строк: id, title, title и 001, old, new. csv. reader сохраняет три значения. DictReader возвращает {'id': '001', 'title': 'new'} без исключения. Значение old больше не представлено отдельным ключом, хотя число полей было правильным. Ни исправление разделителя, ни strict=True не возвращают потерянное значение после такого преобразования.

Причина в DictReader.__next__: строка сопоставляется заголовку через dict(zip(self.fieldnames, row)). Словарь не хранит два значения под одним ключом: более позднее присваивание title замещает предыдущее. Поэтому проверять уникальность заголовков нужно до DictReader или до собственного dict(zip(…)). Проверка длины уже полученного словаря лишь обнаружит последствия.

Ошибка должна указывать слой и место

В нашем валидаторе повтор имени даёт duplicate_header ещё до чтения данных. Короткая запись даёт record₂/physical_line_2:width_2, лишнее поле — width₄. Повтор идентификатора найден в record₃/physical_line₃. Неожиданный заголовок печатается через repr, чтобы было видно '\ufeffid'. Подстановка None, молчаливое обрезание колонок или автоматическое переименование здесь не выполняются.

При проверке чужой выгрузки различайте UnicodeDecodeError (не удалось декодировать байты), csv. Error (ошибка разбора) и ValueError нашего валидатора (не выполнена схема). Код не ловит всё общим except Exception и не продолжает импорт после ошибки. Пустое note допустимо; пустой id — нет. Такие правила задаёт смысл таблицы, а не расширение файла.

Как повторить эксперимент

Сохраните код из конца статьи в файл csv_lab.py и запустите в подходящей среде Python:

python3 csv_lab.py

Команду выполняют в терминале из папки с кодом. Для понимания опыта запуск необязателен: входы и наблюдаемые результаты описаны здесь. Если Python не установлен, его установка выходит за рамки этого разбора.

Скрипт использует только стандартную библиотеку и создаёт новую папку ixbt-csv-deep-…. Путь печатается в root. Для большого рабочего файла не копируйте list(…) из тестов: в лаборатории маленькие списки используются для сравнения с известным эталоном.

В папке останутся sample. csv, semicolon. csv, bom. csv, файлы восьми отрицательных сценариев и results.json. Они не удаляются автоматически. Пример незакрытой кавычки проверяется в памяти. Перед ручной очисткой убедитесь, что выбрали каталог своего тестового запуска.

В журнале можно проверить:

• naive_counts: [3, 4, 3, 1], а правильный разбор совпадает с исходными тремя записями по три поля;

• record_and_physical_line: [[1,1], [2,2], [3,4]];

• dictreader_duplicate_header: [{'id':'001', 'title':'new'}];

• schema_errors: восемь отклонённых входов, включая короткую и длинную запись, повтор заголовка и id, пустой id, пустой файл, иной порядок колонок и BOM, прочитанный как UTF-8;

• unclosed_quote_error: unexpected end of data;

• first_row_before_late_error: первая запись уже получена; late_error: record₃/physical_line_3:duplicate_id.

В журнале указаны фактические python, system и utc. Скрипт сравнивает восстановленные значения с эталоном и проверяет неизменность sample.csv. Проверки реализованы через RuntimeError, не через assert. Выполнены обычный запуск и python3 -O; в обоих все условия прошли.

Как использовать результат при импорте

В потоковом импорте есть отдельная граница: генератор validated_rows отдаёт первую проверенную запись ещё до чтения конца файла. На второй записи может обнаружиться дубликат. В опыте именно так и произошло. Если сразу записывать каждую строку в боевую таблицу, ошибка оставит частичный импорт. Используйте транзакцию или промежуточный результат с подтверждением только после полной проверки, либо отдельно спроектированное возобновление.

Генератор не хранит все строки, но множество seen растёт с числом уникальных идентификаторов, а csv. reader должен разместить в памяти текущую запись. Это не обещание постоянной памяти для сколь угодно большого CSV. Для крупных наборов проверку уникальности можно вынести в промежуточную БД с уникальным индексом. Здесь не измерены производительность, предел поля, неизвестные кодировки или безопасность открытия формул в табличных редакторах.

Рабочая последовательность: сохранить исходную выгрузку, зафиксировать кодировку и диалект, проверить заголовок до преобразования в словари, проверить каждую запись и завершить весь проход до подтверждения импорта. Полезно различать три результата: текст разобран, схема выполнена, данные целиком приняты целевой системой. Ни один не следует автоматически из предыдущего.

Источники

[1] RFC 4180, раздел 2, описание распространённого формата CSV; статус Informational.

https://www.rfc-editor.org/rfc/rfc4180

[2] Python 3.12, csv. Проверены также codecs. utf₈_sig и реализация DictReader.__next__ в установленном CPython 3.12.14. Текущая документация ветки помечена 3.12.16.

https://docs.python.org/3.12/library/csv.html

https://github.com/python/cpython/blob/v3.12.14/Lib/csv.py

https://docs.python.org/3.12/library/codecs.html#module-encodings.utf_8_sig

Полный код csv_lab.py

"""Synthetic CSV lab; no network, arguments or user files.
Creates its own temporary folder and leaves it for inspection.
"""
import codecs, csv, io, json, platform, tempfile
from datetime import datetime, timezone
from pathlib import Path

EXPECTED = ['id', 'title', 'note']
LF = chr(10)
CRLF = chr(13) + LF
BOM_CHAR = chr(0xfeff)

def require(condition, message):
    if not condition:
        raise RuntimeError(message)

def parse(text, **options):
    return list(csv.reader(io.StringIO(text, newline=''), strict=True, **options))

def validated_rows(path, encoding='utf-8', delimiter=','):
    # Yields validated records, not an all-file transaction.
    with path.open('r', encoding=encoding, newline='') as f:
        reader = csv.reader(f, delimiter=delimiter, strict=True)
        header = next(reader, None)
        if header is None:
            raise ValueError('missing_header')
        if len(set(header)) != len(header):
            raise ValueError('duplicate_header')
        if header != EXPECTED:
            raise ValueError('unexpected_header:' + repr(header))
        seen = set()
        for record, row in enumerate(reader, 2):
            where = f'record_{record}/physical_line_{reader.line_num}'
            if len(row) != len(header):
                raise ValueError(where + f':width_{len(row)}')
            identifier = row[0]
            if not identifier:
                raise ValueError(where + ':empty_id')
            if identifier in seen:
                raise ValueError(where + ':duplicate_id')
            seen.add(identifier)
            yield dict(zip(header, row))

root = Path(tempfile.mkdtemp(prefix='ixbt-csv-deep-'))
sample = [EXPECTED, ['001', 'Desk, small', 'plain'],
          ['002', 'Lamp "North"', 'line 1' + LF + 'line 2']]
for name, delimiter in [('sample.csv', ','), ('semicolon.csv', ';')]:
    with (root/name).open('w', encoding='utf-8', newline='') as f:
        csv.writer(f, delimiter=delimiter, lineterminator=CRLF).writerows(sample)
payload = (root/'sample.csv').read_bytes()
text = payload.decode('utf-8')
naive = [len(row.split(',')) for row in text.splitlines()]
require(naive == [3,4,3,1], 'naive counts')
require(parse(text) == sample, 'roundtrip')
reader = csv.reader(io.StringIO(text, newline=''), strict=True)
positions = []
for record, row in enumerate(reader, 1):
    positions.append([record, reader.line_num])
require(positions == [[1,1],[2,2],[3,4]], 'physical line positions')
semicolon = (root/'semicolon.csv').read_bytes().decode('utf-8')
# The incorrect delimiter may return wrong fields without raising an error.
wrong_delimiter = list(csv.reader(io.StringIO(semicolon, newline='')))
require(wrong_delimiter != sample, 'wrong delimiter matched')
require(parse(semicolon, delimiter=';') == sample, 'semicolon roundtrip')
bom = codecs.BOM_UTF8 + payload
(root/'bom.csv').write_bytes(bom)
require(parse(bom.decode('utf-8'))[0][0] == BOM_CHAR + 'id', 'BOM disappeared')
require(parse(bom.decode('utf-8-sig')) == sample, 'BOM decoding')
require(len(list(validated_rows(root/'bom.csv', encoding='utf-8-sig'))) == 2, 'valid BOM file')

duplicate = LF.join(['id,title,title', '001,old,new', ''])
collapsed = list(csv.DictReader(io.StringIO(duplicate, newline='')))
require(collapsed == [{'id':'001','title':'new'}], 'duplicate mapping behavior')
cases = {
    'short': LF.join(['id,title,note', '001,Desk', '']),
    'long': LF.join(['id,title,note', '001,Desk,note,extra', '']),
    'duplicate_header': duplicate,
    'duplicate_id': LF.join(['id,title,note', '001,Desk,first', '001,Lamp,second', '']),
    'empty_id': LF.join(['id,title,note', ',Desk,note', '']),
    'no_header': '',
    'reordered': LF.join(['title,id,note', 'Desk,001,note', '']),
    'bom_as_utf8': bom.decode('utf-8'),
}
errors = {}
for name, raw in cases.items():
    path = root/(name+'.csv')
    path.write_bytes(raw.encode('utf-8'))
    parse(raw)  # All cases are accepted by csv.reader(strict=True).
    try:
        list(validated_rows(path))
    except ValueError as exc:
        errors[name] = str(exc)
    else:
        raise RuntimeError('schema accepted: '+name)
try:
    parse('id,title,note' + LF + '001,"unterminated')
except csv.Error as exc:
    quote_error = str(exc)
else:
    raise RuntimeError('unclosed quote accepted')
# strict=True is not a complete RFC 4180 conformity checker.
quote_in_unquoted = parse(LF.join(['id,title,note', '001,a"b,x', '']))
require(quote_in_unquoted[1][1] == 'a"b', 'unquoted quote behavior')
valid = list(validated_rows(root/'sample.csv'))
require(valid == [dict(zip(EXPECTED,row)) for row in sample[1:]], 'validated values')
late = validated_rows(root/'duplicate_id.csv')
first_before_failure = next(late)
try:
    next(late)
except ValueError as exc:
    late_error = str(exc)
else:
    raise RuntimeError('late duplicate not rejected')
require((root/'sample.csv').read_bytes() == payload, 'source changed')
report = {'utc': datetime.now(timezone.utc).isoformat(), 'python': platform.python_version(),
          'system': platform.system(), 'root': str(root), 'naive_counts': naive,
          'record_and_physical_line': positions, 'bom_utf8_first_header': parse(bom.decode('utf-8'))[0][0],
          'dictreader_duplicate_header': collapsed, 'schema_errors': errors,
          'unclosed_quote_error': quote_error, 'strict_unquoted_quote': quote_in_unquoted[1],
          'first_row_before_late_error': first_before_failure, 'late_error': late_error,
          'all_checks_passed': True}
(root/'results.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps(report, ensure_ascii=False, indent=2))