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

Этот skill содержит паттерны Python-кода для работы с CSV/Excel в песочнице. Загруженные пользователем файлы лежат в `/tmp/`. Скрипты навыка смонтированы в `$SKILLS_ROOT/csv_data_processing_ru/scripts/` и доступны только внутри песочницы (REPL / sandbox_bash). Выполняй весь код через песочницу. Python-код с `cw.*`-хелперами и глобалами (`schema`, `llm_query`, `FINAL`) выполняй ячейкой `repl_execute` — они живут в её namespace; `sandbox_bash` используй для shell-команд и запуска готовых скриптов навыка.

---

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

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

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

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

---

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

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

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

```python
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` сравнит лексикографически.

```python
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"])      # {колонка: сколько НЕ распозналось}
```

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

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

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

Числа `coerce(numeric=...)` прогоняет через `to_number_ru` — единственную нормализацию РФ-чисел: разрядные пробелы/nbsp, запятая-десятичная, 1С/бухгалтерские минусы `1 250,00-` и `(1 250,00)`, валютные хвосты (`₽`, `руб.`). Нужна разовая нормализация вне `coerce` — зови `to_number_ru(series)`, не собирай `pd.to_numeric(...)` руками.

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

```python
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"Выручка: {cw.fmt_money(df['amount'].sum(), kopecks=False)}, средний чек: {cw.fmt_money(df['amount'].mean(), kopecks=False)}")
print(df.groupby("category")["amount"].sum().sort_values(ascending=False).to_string())
```

---

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

1. **EXPLORE**: `python $SKILLS_ROOT/csv_data_processing_ru/scripts/explore.py <path>` — структура, типы-подсказки, превью. `status: failed` → СТОП, верни findings.
2. **VALIDATE_INPUT** (обязательно): `python $SKILLS_ROOT/csv_data_processing_ru/scripts/validate.py <path>`. `СТАТУС: FAILED` → СТОП, EXECUTE не запускай.
3. **EXECUTE**: `read_table` → `coerce(объяви колонки)` → обработка (фильтрация, агрегация, трансформации).
4. **VALIDATE_OUTPUT** (обязательно): `python $SKILLS_ROOT/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)

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

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

```python
df.to_csv(cw.output_path("result.csv"), index=False, encoding="utf-8-sig")
df.to_csv(cw.output_path("result.csv"), index=False, sep=";", encoding="utf-8-sig")
df[["id", "name", "amount", "date"]].to_csv(cw.output_path("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 «как назвал → как записал»), плюс закрепляет жирную шапку и подбирает ширину колонок.

```python
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(cw.output_path("report.xlsx"), {
    "Данные": df,
    "По категориям": summary,
    "По месяцам": monthly,
})
```

---

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

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

```python
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 читай чанками по пути:

```python
import pandas as pd
from io_table import to_number_ru
chunks = []
for chunk in pd.read_csv("/tmp/big.csv", chunksize=100_000, dtype=str):
    chunk["amount"] = to_number_ru(chunk["amount"])
    chunks.append(chunk[chunk["amount"] > 0])
df = pd.concat(chunks, ignore_index=True)
print(f"Обработано строк: {len(df)}")
```

---

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

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

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

```python
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")]
```

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

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

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

```python
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)]
```

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

```python
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()

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

---

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

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

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

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

```python
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()
```

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

```python
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-таблица

```python
pivot = pd.pivot_table(df, values="amount", index="client", columns="month",
                       aggfunc="sum", fill_value=0, margins=True, margins_name="Итого")
pivot.to_excel(cw.output_path("pivot.xlsx"))
```

## Паттерн: Crosstab

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

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

```python
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: Трансформации

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

```python
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=["Малый", "Средний", "Крупный", "Очень крупный"])
```

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

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

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

```python
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)

```python
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
```

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

```python
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))
```

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

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

---

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

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

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

```python
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()} пропусков")
```

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

```python
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)

```python
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).

```python
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(cw.output_path("data_quality.xlsx"), index=False)
```

---

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

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

```python
df_clients = df_clients.rename(columns={"id": "client_id"})
result = cw.merge_on(df_orders, df_clients, on="client_id", how="left", suffixes=("", "_client"))
result = cw.merge_on(df1, df2, on="common_key", how="inner")
result = cw.merge_on(df1, df2, on=["key1", "key2"], how="left")
```

Сливай по id-ключам (ИНН, артикул) через `cw.merge_on(...)` — он сам приводит ключ к общему типу на обеих сторонах (`cw.normalize_id`), поэтому `7701234567` (int) и `"7701234567"` (str) сходятся. Сырой `pd.merge(on=...)` по id-колонкам не бери: int64 и object не сольются.

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

```python
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)}")
```

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

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

---

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

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

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

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

```python
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: Статистический анализ

```python
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}")
```

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

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

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

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

---

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

```python
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`).

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

```python
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"Транзакций: {cw.fmt_int(len(df))}, выручка: {cw.fmt_money(df['amount'].sum(), kopecks=False)}, средний чек: {cw.fmt_money(df['amount'].mean(), kopecks=False)}")
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(cw.output_path("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(),
})
```

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

```python
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(cw.output_path("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)
```

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

```python
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 = cw.merge_on(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(cw.output_path("enriched.xlsx"), index=False)
```

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

```python
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"Границы: [{cw.fmt_money(lo, kopecks=False)}; {cw.fmt_money(hi, kopecks=False)}]; выбросов {cw.fmt_int(len(out))} ({len(out)/len(df)*100:.2f}%)")
print(out.nlargest(10, "amount").to_string(index=False))
out.to_excel(cw.output_path("outliers.xlsx"), index=False)
```

---

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