Обработка CSV и табличных данных

Этот skill содержит паттерны Python-кода для работы с CSV/Excel в песочнице. Загруженные пользователем файлы лежат в /tmp/. Скрипты навыка смонтированы в /mnt/skills/csv_data_processing_ru/scripts/ и доступны только внутри песочницы (REPL / sandbox_bash). Выполняй весь код через песочницу.

Доступные библиотеки

В песочнице библиотеки не предустановлены. В начале первого скрипта запусти один раз:

import subprocess
subprocess.run(["pip", "install", "pandas", "openpyxl", "chardet"], capture_output=True)

Дальше во всех скриптах сессии — сразу import pandas as pd без повторного pip install. Состав: pandas (всегда), openpyxl (xlsx), numpy (числовые операции).

ЕДИНЫЙ ВВОД И ТИПИЗАЦИЯ (ОБЯЗАТЕЛЬНО)

Весь ввод идёт через ОДИН модуль io_table — тот же, что используют explore.py и validate.py. Это гарантирует: что зеленит валидатор, то и получит обработка. Никакого ручного pd.read_csv с подбором sep/encoding и никакого base64 — это устаревшие пути, которые расходятся с валидатором.

Шаг 1 — прочитать (всё строкой)

from io_table import read_table, coerce

df, meta = read_table("/tmp/sales.csv")   # .csv и .xlsx по расширению
print(meta)   # format, encoding, delimiter, rows, cols, bad_lines

read_table сам определяет кодировку (utf-8/cp1251/…), разделитель (, ; \t |) и читает ВСЁ строкой (dtype=str) — лидирующие нули ИНН/артикулов сохранены. meta["bad_lines"] > 0 означает потерянные строки → не обрабатывай, верни findings.

Шаг 2 — типизировать (ЕДИНСТВЕННАЯ точка приведения типов)

После read_table колонки — строки. Перед ЛЮБОЙ арифметикой, сравнением, агрегацией, сортировкой по числам/датам — объяви колонки в coerce. Иначе df["amount"].sum() молча склеит строки, а df["amount"] > 100000 сравнит лексикографически.

df = coerce(
    df,
    numeric=("amount", "quantity"),   # '1 234,56' / nbsp -> float
    dates=("date",),                  # ISO, dd.mm.yyyy, dd/mm/yyyy (dayfirst) -> datetime
    ids=("inn", "phone", "артикул"),  # остаются строкой (нули/точность целы)
)
print(df.attrs["coerce_report"])      # {колонка: сколько НЕ распозналось}

Проверь потери явно, если они критичны:

rep = df.attrs["coerce_report"]
assert rep.get("amount", 0) == 0, f"{rep['amount']} нераспознанных сумм"

coerce — ровно то место, где раньше были разрозненные сниппеты «Числа (русский формат)», «Даты», «ID-колонки». Не дублируй их вручную — объяви колонки и получи типизированный df.

Сквозной пример

from io_table import read_table, coerce

df, meta = read_table("/tmp/sales.csv")
df = coerce(df, numeric=("amount", "quantity"), dates=("date",))

print(f"Период: {df['date'].min():%d.%m.%Y} — {df['date'].max():%d.%m.%Y}")
print(f"Выручка: {df['amount'].sum():,.0f} руб., средний чек: {df['amount'].mean():,.0f} руб.")
print(df.groupby("category")["amount"].sum().sort_values(ascending=False).to_string())

Рабочий процесс

EXPLORE: python /mnt/skills/csv_data_processing_ru/scripts/explore.py <path> — структура, типы-подсказки, превью. status: failed → СТОП, верни findings.
VALIDATE_INPUT (обязательно): python /mnt/skills/csv_data_processing_ru/scripts/validate.py <path>. СТАТУС: FAILED → СТОП, EXECUTE не запускай.
EXECUTE: read_table → coerce(объяви колонки) → обработка (фильтрация, агрегация, трансформации).
VALIDATE_OUTPUT (обязательно): python /mnt/skills/csv_data_processing_ru/scripts/validate.py --output <path_to_output> — формат-aware (xlsx читается как Excel, csv как csv).

Выбор паттерна по задаче

Задача пользователя	Блок	Метод
Сохранить в Excel (.xlsx)	БЛОК 1	`io_table.write_xlsx()`
Сохранить в CSV	БЛОК 1	`df.to_csv()`
Показать данные в чате	БЛОК 8	`print(df.to_markdown())`
Фильтрация / агрегация	БЛОК 3-4	pandas operations

БЛОК 1: Сохранение в файл (.xlsx, .csv)

Результаты сохраняй в /home/user/output/. После сохранения прогони VALIDATE_OUTPUT.

Паттерн: Сохранение в CSV

df.to_csv("/home/user/output/result.csv", index=False, encoding="utf-8-sig")
df.to_csv("/home/user/output/result.csv", index=False, sep=";", encoding="utf-8-sig")
df[["id", "name", "amount", "date"]].to_csv("/home/user/output/result.csv", index=False, encoding="utf-8-sig")

encoding="utf-8-sig" — Excel на Windows откроет кириллицу без «кракозябр».

Паттерн: Сохранение в Excel (один или несколько листов)

Только через io_table.write_xlsx — НЕ пиши свой код на openpyxl/ExcelWriter и свои helper-функции форматирования (set_header и т.п.). Excel запрещает в имени листа символы [ ] : * ? / \ и длину >31 — write_xlsx санирует имена сам (вернёт map «как назвал → как записал»), плюс закрепляет жирную шапку и подбирает ширину колонок.

from io_table import write_xlsx
summary = df.groupby("category")["amount"].sum().reset_index()
monthly = df.groupby(df["date"].dt.to_period("M"))["amount"].sum().reset_index()
write_xlsx("/home/user/output/report.xlsx", {
    "Данные": df,
    "По категориям": summary,
    "По месяцам": monthly,
})

БЛОК 2: Чтение и парсинг

Всегда через read_table (см. «Единый ввод и типизация»). Он закрывает кодировку, разделитель и формат сам — ручной chardet/sep=None не нужен.

from io_table import read_table, coerce
df, meta = read_table("/tmp/file.csv")     # или .xlsx
df = coerce(df, numeric=("amount",), dates=("date",), ids=("inn",))
print(f"Загружено строк: {len(df)} (кодировка {meta['encoding']}, разделитель {meta['delimiter']!r})")

Паттерн: Большой файл — обработка чанками

read_table грузит файл целиком (подходит до сотен тысяч строк). Для очень больших CSV читай чанками по пути:

import pandas as pd
chunks = []
for chunk in pd.read_csv("/tmp/big.csv", chunksize=100_000, dtype=str):
    chunk["amount"] = pd.to_numeric(
        chunk["amount"].str.replace(r"[\s ]", "", regex=True).str.replace(",", ".", regex=False),
        errors="coerce")
    chunks.append(chunk[chunk["amount"] > 0])
df = pd.concat(chunks, ignore_index=True)
print(f"Обработано строк: {len(df)}")

БЛОК 3: Фильтрация

Колонки уже типизированы через coerce (числа — float, даты — datetime).

Паттерн: Простая фильтрация по условию

df_active = df[df["status"] == "active"]
df_large = df[df["amount"] > 100000]
df_complex = df[(df["amount"] > 50000) & (df["status"] == "active")]
df_any = df[(df["status"] == "active") | (df["status"] == "pending")]

Паттерн: Фильтрация по списку значений

df_in = df[df["status"].isin(["active", "pending", "review"])]
df_out = df[~df["status"].isin(["cancelled", "rejected"])]

Паттерн: Фильтрация по текстовым паттернам

df[df["name"].str.contains("ООО", na=False)]
df[df["inn"].str.startswith("77", na=False)]
df[df["email"].str.match(r"^[a-z]+@.*\.ru$", na=False)]
df[df["name"].str.contains("альфа", case=False, na=False)]

Паттерн: Фильтрация по датам

from datetime import datetime, timedelta
df_period = df[(df["date"] >= datetime(2024, 1, 1)) & (df["date"] <= datetime(2024, 3, 31))]
df_2024 = df[df["date"].dt.year == 2024]
df_recent = df[df["date"] >= datetime.now() - timedelta(days=30)]

Паттерн: Фильтрация с query()

min_amount, status = 50000, "active"
df.query("amount > @min_amount and status == @status")

БЛОК 4: Агрегация

Паттерн: Базовая группировка

df.groupby("category")["amount"].sum()
df.groupby("category").agg({"amount": ["sum", "mean", "count"], "quantity": ["sum", "mean"]})

Паттерн: Именованные агрегации

df.groupby("category").agg(
    total_amount=("amount", "sum"),
    avg_amount=("amount", "mean"),
    order_count=("id", "count"),
    unique_clients=("client_id", "nunique"),
    min_date=("date", "min"),
    max_date=("date", "max"),
).reset_index()

Паттерн: Группировка по нескольким полям с итогом

sub = df.groupby(["region", "category"], as_index=False).agg({"amount": "sum"})
total = pd.DataFrame([{"region": "ИТОГО", "category": "", "amount": df["amount"].sum()}])
result = pd.concat([sub, total], ignore_index=True)

Паттерн: Pivot-таблица

pivot = pd.pivot_table(df, values="amount", index="client", columns="month",
                       aggfunc="sum", fill_value=0, margins=True, margins_name="Итого")
pivot.to_excel("/home/user/output/pivot.xlsx")

Паттерн: Crosstab

cross = pd.crosstab(df["region"], df["status"], margins=True, margins_name="Всего")
cross_pct = (pd.crosstab(df["region"], df["status"], normalize="index").round(2) * 100)

Паттерн: Скользящие агрегаты

df = df.sort_values("date")
df["rolling_7d"] = df["amount"].rolling(window=7).sum()
df["cumsum"] = df["amount"].cumsum()
df["cumsum_by_client"] = df.groupby("client_id")["amount"].cumsum()

БЛОК 5: Трансформации

Паттерн: Вычисляемые колонки

df["total"] = df["quantity"] * df["price"]
df["vat"] = df["total"] * 0.2
df["discount"] = df["total"].apply(lambda x: x * 0.1 if x > 100000 else 0)
df["size"] = pd.cut(df["amount"], bins=[0, 10000, 50000, 100000, float("inf")],
                    labels=["Малый", "Средний", "Крупный", "Очень крупный"])

Паттерн: Переименование / переупорядочивание / удаление колонок

df = df.rename(columns={"client_name": "Клиент", "amount": "Сумма"})
df = df[["id", "date", "client", "amount", "status"]]
df = df.drop(columns=["col1", "col2"])

Паттерн: Работа с текстом

df["name_upper"] = df["name"].str.upper()
df["inn"] = df["inn"].str.strip()
df["phone"] = df["phone"].str.replace("+7", "8", regex=False)
df["region_code"] = df["inn"].str.zfill(10).str[:2]
df[["last_name", "first_name"]] = df["full_name"].str.split(" ", n=1, expand=True)

Паттерн: Компоненты даты (после coerce даты уже datetime)

df["year"] = df["date"].dt.year
df["month"] = df["date"].dt.month
df["quarter"] = df["date"].dt.quarter
df["week_number"] = df["date"].dt.isocalendar().week
df["month_name"] = df["date"].dt.strftime("%B %Y")
df["days_since"] = (pd.Timestamp.now() - df["date"]).dt.days

Паттерн: Заполнение пропусков

df["status"] = df["status"].fillna("unknown")
df["amount"] = df["amount"].fillna(df["amount"].median())
df["category"] = df["category"].ffill()
df["amount"] = df.groupby("category")["amount"].transform(
    lambda x: x.fillna(x.mean() if x.notna().any() else 0))

Паттерн: Удаление дубликатов

dups = df[df.duplicated(subset=["inn", "date"], keep=False)]
df_unique = df.drop_duplicates(subset=["inn", "date"], keep="first")

БЛОК 6: Валидация

Структуру файла (кодировка, разделитель, дубли заголовков, рваные строки) проверяет validate.py — не дублируй это в коде. Здесь — доменные проверки ПОСЛЕ coerce.

Паттерн: Обязательные поля

required = ["id", "name", "inn", "amount", "date"]
missing = [c for c in required if c not in df.columns]
if missing:
    print(f"Отсутствуют колонки: {missing}")
for c in required:
    if c in df.columns and df[c].isna().any():
        print(f"Колонка '{c}': {df[c].isna().sum()} пропусков")

Паттерн: Валидация форматов

import re
df["inn_valid"] = df["inn"].fillna("").str.match(r"^(\d{10}|\d{12})$")
df["email_valid"] = df["email"].fillna("").str.match(r"^[\w\.-]+@[\w\.-]+\.\w+$")
print(f"Невалидных ИНН: {(~df['inn_valid']).sum()}")

Паттерн: Диапазоны значений (после coerce)

report = []
report.append({"Проверка": "Положительные суммы", "Ошибок": int((df["amount"] < 0).sum())})
report.append({"Проверка": "Даты не в будущем", "Ошибок": int((df["date"] > pd.Timestamp.now()).sum())})
report.append({"Проверка": "Допустимые статусы",
               "Ошибок": int((~df["status"].isin(["active", "pending", "completed", "cancelled"])).sum())})
print(pd.DataFrame(report).to_string(index=False))

Паттерн: Полный отчёт о качестве данных

Числовую статистику считаем по колонкам, которые объявили числовыми в coerce (после read_table всё было строкой — нельзя полагаться на dtype).

def data_quality_report(df, numeric_cols=()):
    rows = []
    for col in df.columns:
        r = {"Колонка": col, "Тип": str(df[col].dtype), "Всего": len(df),
             "Заполнено": int(df[col].notna().sum()), "Пропуски": int(df[col].isna().sum()),
             "% заполнения": round(df[col].notna().mean() * 100, 1),
             "Уникальных": int(df[col].nunique())}
        if col in numeric_cols:
            r["Мин"] = df[col].min(); r["Макс"] = df[col].max()
            r["Среднее"] = round(df[col].mean(), 2)
        rows.append(r)
    return pd.DataFrame(rows)

rep = data_quality_report(df, numeric_cols=("amount",))
print(rep.to_string(index=False))
rep.to_excel("/home/user/output/data_quality.xlsx", index=False)

БЛОК 7: Объединение

Паттерн: Merge (JOIN)

result = pd.merge(df_orders, df_clients, left_on="client_id", right_on="id",
                  how="left", suffixes=("", "_client"), validate="many_to_one")
result = pd.merge(df1, df2, on="common_key", how="inner", validate="one_to_one")
result = pd.merge(df1, df2, on=["key1", "key2"], how="left", validate="one_to_many")

Ключи слияния (ИНН, артикул) держи строкой через coerce(ids=...) в ОБОИХ датафреймах — иначе 7701234567 (int) и "7701234567" (str) не сольются.

Паттерн: Concat нескольких файлов (по пути, без base64)

from io_table import read_table
frames = []
for path in ["/tmp/jan.csv", "/tmp/feb.csv", "/tmp/mar.csv"]:
    d, _ = read_table(path)
    d["source"] = path.split("/")[-1]
    frames.append(d)
combined = pd.concat(frames, ignore_index=True)
print(f"Объединено файлов: {len(frames)}, всего строк: {len(combined)}")

Паттерн: Обогащение из справочника

regions = {"77": "Москва", "78": "Санкт-Петербург", "50": "Московская область"}
df["region_code"] = df["inn"].str[:2]
df["region_name"] = df["region_code"].map(regions).fillna("Другой регион")

БЛОК 8: Вывод в чат (stdout)

Когда пользователь просит просто показать/прочитать файл без сохранения.

from io_table import read_table
df, _ = read_table("/tmp/file.csv")    # или .xlsx
print(df.to_markdown(index=False))

Excel со всеми листами

import pandas as pd
xls = pd.ExcelFile("/tmp/file.xlsx", engine="openpyxl")
print(f"Листы: {xls.sheet_names}")
for sheet in xls.sheet_names:
    d = xls.parse(sheet)
    if d.empty:
        continue
    print(f"\n## {sheet}\n")
    print(d.to_markdown(index=False))

БЛОК 9: Статистический анализ

print(df.describe())
for col in ("amount", "quantity"):          # колонки, объявленные числовыми в coerce
    print(f"\n{col}: среднее {df[col].mean():,.2f}, медиана {df[col].median():,.2f}, "
          f"ст.откл {df[col].std():,.2f}, мин {df[col].min():,.2f}, макс {df[col].max():,.2f}")

Распределение по категориям

summary = pd.DataFrame({
    "Количество": df["category"].value_counts(),
    "Процент": (df["category"].value_counts(normalize=True) * 100).round(1)})
print(summary)

Корреляция (по числовым колонкам)

corr = df[["amount", "quantity", "price"]].corr().round(2)
print(corr)

БЛОК 10: Выборка данных

df.sample(n=1000)
df.sample(frac=0.1)
df.head(1000); df.tail(1000)
df.iloc[::10]

БЛОК 11: Типовые задачи

Все читают через read_table + coerce (никакого base64/file_content).

Задача: Анализ продаж

from io_table import read_table, coerce, write_xlsx

df, _ = read_table("/tmp/sales.csv")
df = coerce(df, numeric=("amount",), dates=("date",))

print("=== АНАЛИЗ ПРОДАЖ ===")
print(f"Период: {df['date'].min():%d.%m.%Y} — {df['date'].max():%d.%m.%Y}")
print(f"Транзакций: {len(df):,}, выручка: {df['amount'].sum():,.0f} руб., средний чек: {df['amount'].mean():,.0f} руб.")
print("\n=== ТОП-10 КЛИЕНТОВ ===")
print(df.groupby("client")["amount"].sum().nlargest(10).to_string())
print("\n=== ПО МЕСЯЦАМ ===")
print(df.groupby(df["date"].dt.to_period("M"))["amount"].agg(["sum", "count", "mean"]).round(2).to_string())

write_xlsx("/home/user/output/sales_analysis.xlsx", {
    "Данные": df,
    "Топ клиенты": df.groupby("client")["amount"].sum().nlargest(10).reset_index(),
    "По месяцам": df.groupby(df["date"].dt.to_period("M"))["amount"].sum().reset_index(),
})

Задача: Сверка двух списков

from io_table import read_table, coerce

a, _ = read_table("/tmp/list_a.xlsx"); a = coerce(a, ids=("inn",))
b, _ = read_table("/tmp/list_b.xlsx"); b = coerce(b, ids=("inn",))
key = "inn"
only_a = a[~a[key].isin(b[key])]
only_b = b[~b[key].isin(a[key])]
common = a[a[key].isin(b[key])]
print(f"Только в A: {len(only_a)} | Только в B: {len(only_b)} | Общие: {len(common)}")

with pd.ExcelWriter("/home/user/output/comparison.xlsx", engine="openpyxl") as w:
    only_a.to_excel(w, sheet_name="Только A", index=False)
    only_b.to_excel(w, sheet_name="Только B", index=False)
    common.to_excel(w, sheet_name="Общие", index=False)

Задача: Обогащение из справочника

from io_table import read_table, coerce

main, _ = read_table("/tmp/orders.xlsx"); main = coerce(main, ids=("inn",))
ref, _ = read_table("/tmp/ref.xlsx"); ref = coerce(ref, ids=("inn",))
res = pd.merge(main, ref[["inn", "company_name", "region", "industry"]], on="inn", how="left")
matched = res["company_name"].notna().sum()
print(f"Обогащено: {matched}/{len(res)} ({matched/len(res)*100:.1f}%)")
print(f"Не найдено в справочнике: {sorted(set(res[res['company_name'].isna()]['inn']))}")
res.to_excel("/home/user/output/enriched.xlsx", index=False)

Задача: Поиск аномалий (IQR)

from io_table import read_table, coerce

df, _ = read_table("/tmp/transactions.csv")
df = coerce(df, numeric=("amount",))
Q1, Q3 = df["amount"].quantile(0.25), df["amount"].quantile(0.75)
IQR = Q3 - Q1
lo, hi = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
out = df[(df["amount"] < lo) | (df["amount"] > hi)]
print(f"Границы: [{lo:,.0f}; {hi:,.0f}]; выбросов {len(out)} ({len(out)/len(df)*100:.2f}%)")
print(out.nlargest(10, "amount").to_string(index=False))
out.to_excel("/home/user/output/outliers.xlsx", index=False)

Skill-файл содержит паттерны кода для CSV/Excel. Весь ввод — через io_table.read_table + coerce; выполняй код через песочницу; результаты — в /home/user/output/ с обязательным VALIDATE_OUTPUT.

Обработка CSV и табличных данных

Обработка CSV и табличных данных

Доступные библиотеки

ЕДИНЫЙ ВВОД И ТИПИЗАЦИЯ (ОБЯЗАТЕЛЬНО)

Шаг 1 — прочитать (всё строкой)

Шаг 2 — типизировать (ЕДИНСТВЕННАЯ точка приведения типов)

Сквозной пример

Рабочий процесс

Выбор паттерна по задаче

БЛОК 1: Сохранение в файл (.xlsx, .csv)

Паттерн: Сохранение в CSV

Паттерн: Сохранение в Excel (один или несколько листов)

БЛОК 2: Чтение и парсинг

Паттерн: Большой файл — обработка чанками

БЛОК 3: Фильтрация

Паттерн: Простая фильтрация по условию

Паттерн: Фильтрация по списку значений

Паттерн: Фильтрация по текстовым паттернам

Паттерн: Фильтрация по датам

Паттерн: Фильтрация с query()

БЛОК 4: Агрегация

Паттерн: Базовая группировка

Паттерн: Именованные агрегации

Паттерн: Группировка по нескольким полям с итогом

Паттерн: Pivot-таблица

Паттерн: Crosstab

Паттерн: Скользящие агрегаты

БЛОК 5: Трансформации

Паттерн: Вычисляемые колонки

Паттерн: Переименование / переупорядочивание / удаление колонок

Паттерн: Работа с текстом

Паттерн: Компоненты даты (после coerce даты уже datetime)

Паттерн: Заполнение пропусков

Паттерн: Удаление дубликатов

БЛОК 6: Валидация

Паттерн: Обязательные поля

Паттерн: Валидация форматов

Паттерн: Диапазоны значений (после coerce)

Паттерн: Полный отчёт о качестве данных

БЛОК 7: Объединение

Паттерн: Merge (JOIN)

Паттерн: Concat нескольких файлов (по пути, без base64)

Паттерн: Обогащение из справочника

БЛОК 8: Вывод в чат (stdout)

Excel со всеми листами

БЛОК 9: Статистический анализ

Распределение по категориям

Корреляция (по числовым колонкам)

БЛОК 10: Выборка данных

БЛОК 11: Типовые задачи

Задача: Анализ продаж

Задача: Сверка двух списков

Задача: Обогащение из справочника

Задача: Поиск аномалий (IQR)

Попробуйте этот навык