Как объединить CSV-файлы на Python и убрать дубли

Чтобы объединить CSV-файлы на Python, прочитайте их через модуль csv, проверьте одинаковый набор колонок, нормализуйте ключ записи и сохраните результат в новый файл. Ниже — запускаемый пример: он собирает все CSV из папки, удаляет повторные строки по email и не перезаписывает входные файлы.

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

Подготовьте папки и CSV

Нужен Python 3.10 или новее. Создайте структуру проекта:

csv-report/
  data/
    inbox/
      customers-a.csv
      customers-b.csv
    out/
  merge_csv.py

У входных файлов должны быть одинаковые заголовки, например:

email,name,city
ALICE@example.com,Alice,Minsk

Разделитель здесь — запятая, кодировка — UTF-8. Если поставщик присылает CSV с другим разделителем или кодировкой, сначала укажите это явно, а не пытайтесь угадать по содержимому.

Объедините файлы и уберите повторы

Сохраните код в merge_csv.py:

import csv
from pathlib import Path


ROOT = Path(__file__).parent
INPUT_DIR = ROOT / "data" / "inbox"
OUTPUT_FILE = ROOT / "data" / "out" / "customers-merged.csv"
DEDUPLICATION_KEY = "email"


def merge_csv_files() -> int:
    files = sorted(INPUT_DIR.glob("*.csv"))
    if not files:
        raise FileNotFoundError(f"No CSV files found in {INPUT_DIR}")

    OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
    expected_headers: list[str] | None = None
    seen_keys: set[str] = set()
    written = 0

    with OUTPUT_FILE.open("w", newline="", encoding="utf-8") as output:
        writer: csv.DictWriter | None = None

        for file_path in files:
            with file_path.open(newline="", encoding="utf-8-sig") as source:
                reader = csv.DictReader(source)
                headers = reader.fieldnames
                if not headers or DEDUPLICATION_KEY not in headers:
                    raise ValueError(f"{file_path.name}: missing {DEDUPLICATION_KEY!r} column")
                if expected_headers is None:
                    expected_headers = headers
                    writer = csv.DictWriter(output, fieldnames=headers)
                    writer.writeheader()
                elif headers != expected_headers:
                    raise ValueError(f"{file_path.name}: CSV columns do not match the first file")

                for row in reader:
                    key = (row.get(DEDUPLICATION_KEY) or "").strip().casefold()
                    if not key:
                        continue
                    if key in seen_keys:
                        continue
                    seen_keys.add(key)
                    writer.writerow(row)
                    written += 1

    return written


if __name__ == "__main__":
    count = merge_csv_files()
    print(f"Saved {count} unique rows to {OUTPUT_FILE}")

utf-8-sig читает обычный UTF-8 и убирает BOM, который иногда добавляют табличные редакторы. casefold() сравнивает email без различия регистра, а strip() убирает пробелы по краям. Пустые ключи пропускаются, потому что без ключа надёжно определить повтор нельзя.

Скрипт сохраняет первую встреченную строку для каждого email. Если нужно оставить самую свежую запись, добавьте явное правило сравнения даты и протестируйте его на примерах. Не выбирайте «последнюю» строку случайно: порядок файлов сам по себе не доказывает, какая запись новее.

Запустите и проверьте результат

Из папки проекта выполните:

python merge_csv.py

Ожидаемый вывод при двух уникальных клиентах:

Saved 2 unique rows to .../data/out/customers-merged.csv

Итоговый файл появится в data/out/. Исходные CSV останутся в data/inbox/. Повторный запуск создаёт новый отчёт заново, поэтому он не добавляет те же строки в конец результата.

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

Если исходные CSV большие, набор колонок меняется или результат нужно регулярно отправлять в CRM, вынесите правила нормализации в отдельные функции и добавьте тесты на реальные крайние случаи. В курсе по автоматизации Python такой CSV-процесс входит в практическую траекторию; более сложный сбор данных с сайтов разобран в статье про парсеры на Python, а настройка ежедневного выполнения — в руководстве по запуску Python по расписанию через systemd.