Обработка CSV-данных средствами Python

Пример 1: чтение CSV с преобразованием типов и обработкой дат

Предположим, CSV содержит столбцы 'date', 'value' (число) и 'description'. Необходимо преобразовать строку даты в объект datetime и значение в float.

import csv
from datetime import datetime

filename = 'transactions.csv'
# Содержимое файла:
# date,value,description
# 2024-01-15,123.45,Оплата услуг
# 2024-02-20,67.0,Возврат

with open(filename, newline='', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        try:
            row['date'] = datetime.strptime(row['date'], '%Y-%m-%d')
            row['value'] = float(row['value'])
            print(f"{row['date'].strftime('%d.%m.%Y')}: {row['value']} - {row['description']}")
        except ValueError as e:
            print(f"Ошибка в строке: {e}")

15.01.2024: 123.45 - Оплата услуг
20.02.2024: 67.0 - Возврат

Особенность: обработка ошибок важна, так как данные могут быть некорректными. В реальном проекте можно логировать проблемные строки.

Пример 2: запись CSV с динамическими заголовками и разными разделителями

Необходимо создать файл с разделителем точка с запятой, заголовки формируются на основе ключей первого словаря. Используется DictWriter с автоматической записью заголовков.

import csv

data = [
    {'product': 'Молоко', 'price': 89.5, 'quantity': 2},
    {'product': 'Хлеб', 'price': 45.0, 'quantity': 1}
]

with open('order.csv', 'w', newline='', encoding='utf-8') as f:
    fieldnames = data[0].keys()  # заголовки из первого элемента
    writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=';',
                            quoting=csv.QUOTE_NONNUMERIC)
    writer.writeheader()
    for item in data:
        writer.writerow(item)

# Содержимое order.csv:
"product";"price";"quantity"
"Молоко";89.5;2
"Хлеб";45.0;1

Числовые столбцы не заключаются в кавычки благодаря QUOTE_NONNUMERIC.

Пример 3: автоматическое определение формата CSV с помощью Sniffer

Если формат CSV неизвестен (разделитель, кавычки), класс csv.Sniffer анализирует образец файла и определяет диалект.

import csv

with open('unknown.csv', newline='', encoding='utf-8') as f:
    sample = f.read(1024)        # читаем часть файла
    dialect = csv.Sniffer().sniff(sample)
    f.seek(0)                    # возвращаемся в начало
    reader = csv.reader(f, dialect)
    for row in reader:
        print(row)

Sniffer может определить разделитель, символ кавычки, escape-символ. Осторожно: для очень коротких или нестандартных файлов результат может быть неточным.

Пример 4: параллельная обработка большого CSV с использованием multiprocessing

Для ускорения обработки большого файла его можно разбить на части и обрабатывать параллельно. В примере используется пул процессов.

import csv
import multiprocessing
from itertools import islice

def process_chunk(chunk):
    # chunk - список строк, каждая строка - список полей
    result = []
    for row in chunk:
        if row:  # непустая строка
            # например, извлечение первой буквы
            result.append((row[0], len(row)))
    return result

filename = 'large.csv'
chunk_size = 10000

with open(filename, newline='', encoding='utf-8') as f:
    reader = csv.reader(f)
    # Пропускаем заголовок, если он есть
    header = next(reader, None)
    chunks = []
    while True:
        chunk = list(islice(reader, chunk_size))
        if not chunk:
            break
        chunks.append(chunk)

with multiprocessing.Pool() as pool:
    results = pool.map(process_chunk, chunks)

# Объединение результатов
all_rows = [item for sublist in results for item in sublist]
print(f"Обработано {len(all_rows)} строк")

Обработано 123456 строк

Замечание: данный подход эффективен для CPU-интенсивных операций. Для IO-bound задач лучше использовать асинхронное чтение.

Пример 5: преобразование CSV в JSON с кастомными ключами

Иногда требуется сохранить данные CSV в формате JSON, изменив названия полей или добавив вложенные структуры.

import csv
import json

csv_data = """name,age,city
Иван,30,Москва
Анна,25,Санкт-Петербург
"""

reader = csv.DictReader(csv_data.splitlines())
output = []
for row in reader:
    new_record = {
        'full_name': row['name'],
        'years_old': int(row['age']),
        'location': row['city']
    }
    output.append(new_record)

print(json.dumps(output, ensure_ascii=False, indent=2))

[
  {
    "full_name": "Иван",
    "years_old": 30,
    "location": "Москва"
  },
  {
    "full_name": "Анна",
    "years_old": 25,
    "location": "Санкт-Петербург"
  }
]

Метод splitlines() создаёт список строк из текстовой переменной, имитируя файл. Для реального файла используется open.