Работа с данными в формате CSV средствами Python

Расширенные примеры работы с CSV

1. Чтение CSV из строки (StringIO)

Иногда данные CSV приходят в виде текстовой строки (например, из HTTP-запроса). Для имитации файлового объекта используется io.StringIO.

import csv
from io import StringIO

csv_text = "Имя,Возраст\nИван,25\nМария,30"
f = StringIO(csv_text)
reader = csv.DictReader(f)
for row in reader:
    print(row)

{'Имя': 'Иван', 'Возраст': '25'}
{'Имя': 'Мария', 'Возраст': '30'}

2. Запись CSV с указанием порядка столбцов через DictWriter

Порядок столбцов определяется списком fieldnames. Если словарь содержит лишние ключи, они будут проигнорированы (если не задан extrasaction).

import csv

data = [{'Имя': 'Анна', 'Возраст': 28, 'Город': 'Москва'}]
with open('ordered.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=['Возраст', 'Имя'], extrasaction='ignore')
    writer.writeheader()
    writer.writerows(data)

Содержимое файла ordered.csv:
Возраст,Имя
28,Анна

3. Фильтрация строк с помощью генератора

Можно построчно отфильтровать данные, не загружая весь файл.

import csv

with open('data.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    filtered = (row for row in reader if int(row['Возраст']) > 25)
    for person in filtered:
        print(person['Имя'])

Мария

4. Пропуск строк с комментариями

Если CSV содержит строки, начинающиеся с символа комментария (например, '#'), их можно проигнорировать.

import csv

with open('comments.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(line for line in f if not line.startswith('#'))
    for row in reader:
        print(row)

Файл comments.csv:

# Это комментарий
Имя,Возраст
Иван,25

5. Чтение CSV с разными разделителями и автоматическим определением

Пример с использованием csv.Sniffer для файла с разделителем табуляции.

import csv

with open('tab_sep.csv', newline='', encoding='utf-8') as f:
    sample = f.read(2048)
    dialect = csv.Sniffer().sniff(sample)
    f.seek(0)
    reader = csv.reader(f, dialect)
    for row in reader:
        print(row)

['Имя', 'Возраст', 'Зарплата']
['Иван', '25', '50000']

6. Обработка больших файлов по кускам (chunks) с pandas

Pandas позволяет читать CSV частями и агрегировать результат.

import pandas as pd

chunksize = 1000
results = []
for chunk in pd.read_csv('huge.csv', chunksize=chunksize):
    # пример: средний возраст по каждому куску
    mean_age = chunk['Возраст'].mean()
    results.append(mean_age)
overall_mean = sum(results) / len(results)
print(f"Средний возраст: {overall_mean:.2f}")

Средний возраст: 34.78

7. Обработка ошибок кодировки с errors='replace'

Когда файл содержит символы не в UTF-8, можно заменить их на знак замены.

import csv

with open('broken_encoding.csv', 'r', encoding='utf-8', errors='replace') as f:
    reader = csv.reader(f)
    for row in reader:
        print(row)

8. Преобразование строк в даты при чтении

Использование параметра parse_dates в pandas или ручной конвертации с datetime.strptime.

import csv
from datetime import datetime

with open('dates.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        date = datetime.strptime(row['Дата'], '%Y-%m-%d')
        print(date.year)

9. Объединение нескольких CSV-файлов в один

Построчная запись всех данных из нескольких файлов в один результирующий.

import csv

input_files = ['part1.csv', 'part2.csv']
with open('merged.csv', 'w', newline='', encoding='utf-8') as out:
    writer = csv.writer(out)
    for i, fname in enumerate(input_files):
        with open(fname, 'r', encoding='utf-8') as f:
            reader = csv.reader(f)
            if i == 0:
                writer.writerow(next(reader))  # заголовок только из первого
            else:
                next(reader)  # пропустить заголовки остальных
            for row in reader:
                writer.writerow(row)

10. Сортировка CSV по столбцу без pandas

Загрузка всех строк, сортировка и запись обратно. Подходит для файлов, помещающихся в память.

import csv

with open('data.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    rows = list(reader)

rows.sort(key=lambda x: int(x['Возраст']))

with open('sorted.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=reader.fieldnames)
    writer.writeheader()
    writer.writerows(rows)

11. Агрегация данных (сумма, среднее) по группам

Группировка по одному столбцу и вычисление агрегатов для другого с помощью словарей.

import csv
from collections import defaultdict

with open('sales.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    totals = defaultdict(float)
    counts = defaultdict(int)
    for row in reader:
        product = row['Продукт']
        price = float(row['Цена'])
        totals[product] += price
        counts[product] += 1

for product, total in totals.items():
    avg = total / counts[product]
    print(f"{product}: средняя цена {avg:.2f}")

12. Создание собственного диалекта с помощью csv.Dialect

Можно определить кастомный формат, если файл использует нестандартные параметры (например, двойные кавычки и обратную косую черту для экранирования).

import csv

class CustomDialect(csv.Dialect):
    delimiter = '|'
    quotechar = '"'
    doublequote = True
    skipinitialspace = True
    lineterminator = '\n'
    quoting = csv.QUOTE_ALL

with open('custom.csv', 'r', newline='', encoding='utf-8') as f:
    reader = csv.reader(f, dialect=CustomDialect)
    for row in reader:
        print(row)

13. Работа с вложенными кавычками и экранированием

Модуль csv корректно обрабатывает поля, содержащие символы-разделители внутри кавычек, а также удвоенные кавычки для экранирования.

import csv

csv_text = '"Иван","25","город Москва, Россия"'
f = StringIO(csv_text)
reader = csv.reader(f)
for row in reader:
    print(row)

['Иван', '25', 'город Москва, Россия']

14. Потоковая обработка с записью результата в другой CSV

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

import csv

with open('source.csv', 'r', encoding='utf-8') as src, \
     open('result.csv', 'w', newline='', encoding='utf-8') as dst:
    reader = csv.DictReader(src)
    writer = csv.DictWriter(dst, fieldnames=reader.fieldnames)
    writer.writeheader()
    for row in reader:
        row['Возраст'] = int(row['Возраст']) + 1  # увеличим возраст на 1
        writer.writerow(row)