Перейти к содержанию

Табличные данные

Шаг 1

В предыдущем разделе мы извлекали из документов последовательный текст. Для TXT, Markdown, PDF и DOCX этого было достаточно, ведь приложение получало полный текст и отдельные страницы или разделы.

С таблицами такой подход быстро разрушит полезную структуру. Если просто соединить все ячейки в одну строку, станет непонятно, какое значение относится к какой колонке, где заканчивается одна запись и начинается другая, а в Excel дополнительно потеряются названия листов.

Например число 12 само по себе ничего не объясняет. В колонке resolution_hours оно означает двенадцать часов обработки обращения, в колонке satisfaction такое значение было бы ошибкой, а в колонке order_id являлось бы идентификатором, а не числом для вычислений.

Поэтому для CSV и Excel создадим отдельный табличный контур. Он по-прежнему будет получать только ValidatedFile, но вернёт не ExtractedDocument, а новый объект ExtractedTable.

Для работы с CSV и Excel мы будем использовать библиотеку pandas. Она умеет читать табличные файлы, представлять их содержимое в виде DataFrame, выбирать строки и колонки, обрабатывать пропущенные значения, группировать данные и выполнять точные вычисления. Благодаря этому нам не придётся самостоятельно разбирать структуру каждого табличного формата.

Ранее мы не устанавливали pandas, поэтому добавим её в проект сейчас. А для чтения файлов .xlsx нам также понадобится библиотека openpyxl, которую pandas использует как специальный движок для работы с Excel. Выполните в терминале с активированным виртуальным окружением:

pip install pandas openpyxl
pip freeze > requirements.txt

После этого мы сможем импортировать pandas и использовать её для обработки CSV и Excel.

Что такое DataFrame?

Для работы с таблицами мы будем использовать библиотеку pandas и её основной объект называется DataFrame. Упрощённо его можно представить как таблицу в памяти Python, где у неё есть названия колонок, индекс строк и значения в ячейках.

DataFrame умеет выбирать строки и колонки, находить пропуски, группировать записи и выполнять вычисления. Эти операции детерминированы, то есть одинаковые входные данные и одинаковый код дают один и тот же результат. Именно поэтому сумму, среднее значение и количество обращений должен считать pandas, а не языковая модель.

Перейдем к практике, в существующем пакете app/files создайте файл table_extraction.py и в начало нового файла добавьте:

from dataclasses import dataclass
from typing import Protocol

import pandas as pd

from app.files.validation import FileKind, ValidatedFile

Обратите внимание, что мы используем псевдоним pd, он является общепринятым способом импортировать pandas. Поэтому объект таблицы будет записываться как pd.DataFrame, а функции библиотеки как pd.read_csv() и pd.read_excel().

Далее добавьте прикладные ошибки:

class TableExtractionError(RuntimeError):
    pass

class UnsupportedTableFormatError(TableExtractionError):
    pass

class TableReadError(TableExtractionError):
    pass

class EmptyTableError(TableExtractionError):
    pass

class TableLimitError(TableExtractionError):
    pass

class TableSheetError(TableExtractionError):
    pass

Тут TableExtractionError станет общей границей ожидаемых ошибок, UnsupportedTableFormatError означает отсутствие обработчика для формата, TableReadError проблему при чтении, а EmptyTableError отсутствие полезных строк или колонок.

А ошибка TableLimitError будет использоваться, если таблица превышает прикладные ограничения. И ошибка TableSheetError относится уже к Excel, например если пользователь попросил лист, которого в книге нет.

Ни одна из этих ошибок не заменяет FileValidationError. Файловый валидатор отвечает на вопрос, разрешено ли передавать файл парсеру. А новые ошибки появятся позже, когда допустимый CSV или XLSX уже начали разбирать как таблицу.

После ошибок добавьте политику табличного извлечения:

@dataclass(frozen=True)
class TableExtractionPolicy:
    max_rows: int = 5_000
    max_columns: int = 100

    def __post_init__(self) -> None:
        if self.max_rows <= 0:
            raise ValueError("max_rows должен быть больше нуля.")
        if self.max_columns <= 0:
            raise ValueError(
                "max_columns должен быть больше нуля."
            )

Лимит размера файла из раздела 4.1 остаётся в силе, но для таблицы нужны дополнительные границы. Файл небольшого размера способен содержать тысячи коротких строк или очень много колонок. Загружать всё только потому, что файл поместился в лимит байтов, не следует.

Теперь давайте реализуем то, что будет возвращать табличный экстрактор. После политики добавьте метаданные:

@dataclass(frozen=True)
class TableMetadata:
    original_name: str
    kind: FileKind
    size_bytes: int
    sheet_name: str | None
    available_sheets: tuple[str, ...]
    loaded_row_count: int
    column_count: int
    truncated: bool

Тут значения полей original_name, kind и size_bytes приходят из ValidatedFile. Они позволяют связать результат с исходным файлом, не сохраняя сам временный путь.

А уже поле sheet_name содержит имя выбранного листа Excel. Для CSV оно будет равно None, потому что CSV не поддерживает листы. available_sheets сохраняет все найденные названия листов книги и для CSV будет пустым кортежем.

Поля loaded_row_count и column_count описывают уже подготовленный DataFrame. Поле truncated показывает, пришлось ли отбросить строки из-за max_rows. Это важно, ведь приложение не должно выдавать ограниченную выборку за полный набор данных.

Следом добавьте общий результат:

@dataclass
class ExtractedTable:
    frame: pd.DataFrame
    metadata: TableMetadata
    warnings: tuple[str, ...]

Тут frame содержит строки и колонки, metadata — общие сведения, а warnings — проблемы, которые не помешали получить полезную таблицу. Например предупреждение сообщит о сокращении количества строк или автоматическом имени пустой колонки.

В отличие от предыдущих результатов, ExtractedTable не объявлен как frozen=True. Причина в самом DataFrame, это уже изменяемый объект.

Теперь опишем единый интерфейс обработчиков:

class TableFormatExtractor(Protocol):
    @property
    def supported_kinds(self) -> frozenset[FileKind]:
        ...

    def extract(
        self,
        file: ValidatedFile,
        sheet_name: str | None = None,
    ) -> ExtractedTable:
        ...

supported_kinds сообщает поддерживаемые форматы. Параметр sheet_name понадобится XLSX, но присутствует в общем контракте, чтобы маршрутизатор всегда вызывал один и тот же метод.

Перед созданием результата таблицу нужно привести к понятному виду. После протокола добавьте функцию:

def normalize_table_frame(
    frame: pd.DataFrame,
    policy: TableExtractionPolicy,
) -> tuple[pd.DataFrame, bool, tuple[str, ...]]:
    normalized = frame.copy()
    normalized = normalized.dropna(how="all")
    normalized = normalized.dropna(
        axis="columns",
        how="all",
    )

    if normalized.empty:
        raise EmptyTableError(
            "В таблице не найдено строк с данными."
        )

    warnings: list[str] = []
    column_names: list[str] = []

    for index, value in enumerate(normalized.columns):
        name = str(value).strip()
        if not name or name.startswith("Unnamed:"):
            name = f"column_{index + 1}"
            warnings.append(
                "Пустым заголовкам колонок присвоены имена."
            )
        column_names.append(name)

    if pd.Index(column_names).duplicated().any():
        raise TableReadError(
            "После нормализации найдены одинаковые колонки."
        )

    normalized.columns = column_names

    if len(normalized.columns) > policy.max_columns:
        raise TableLimitError(
            "Таблица содержит слишком много колонок."
        )

    truncated = len(normalized) > policy.max_rows
    if truncated:
        normalized = normalized.head(policy.max_rows).copy()
        warnings.append(
            "Количество строк ограничено политикой извлечения."
        )

    normalized = normalized.reset_index(drop=True)
    return normalized, truncated, tuple(warnings)

Давайте разберем данный код. Эта функция получает DataFrame, прочитанный конкретным парсером, и сразу создаёт копию. Поэтому последующие операции не меняют объект, принадлежащий библиотеке чтения.

Два вызова dropna() удаляют полностью пустые строки и полностью пустые колонки. Значение how="all" принципиально: строка удаляется только тогда, когда пусты все её ячейки. Строки с одним пропуском сохраняются.

После этого normalized.empty проверяет, остались ли данные. Пустой заголовок без записей не является полезной таблицей, поэтому функция выбрасывает EmptyTableError.

Цикл приводит названия колонок к строкам и удаляет пробелы по краям. Если pandas создал имя вида Unnamed: 2 для пустого заголовка, мы заменяем его на понятное column_3 и добавляем предупреждение.

После нормализации имена должны быть уникальными. Иначе выражение вроде frame["status"] могло бы неожиданно вернуть сразу несколько колонок. Такая неоднозначность останавливается через TableReadError.

Далее проверяется ширина. Строки сверх лимита не приводят к полному отказу: мы оставляем первые max_rows, устанавливаем truncated=True и явно предупреждаем о сокращении. Колонки сокращать автоматически опаснее, потому что мы могли бы молча отбросить как раз нужное поле, поэтому превышение max_columns считается ошибкой.

Последний reset_index(drop=True) создаёт простой индекс 0, 1, 2.... Исходные номера строк файла не используются как бизнес-данные и не должны превращаться в дополнительную колонку.

Осталось собрать общий объект. В конце файла добавьте:

def build_extracted_table(
    file: ValidatedFile,
    frame: pd.DataFrame,
    policy: TableExtractionPolicy,
    sheet_name: str | None = None,
    available_sheets: tuple[str, ...] = (),
    warnings: tuple[str, ...] = (),
) -> ExtractedTable:
    normalized, truncated, normalization_warnings = (
        normalize_table_frame(frame, policy)
    )
    all_warnings = warnings + normalization_warnings

    metadata = TableMetadata(
        original_name=file.original_name,
        kind=file.kind,
        size_bytes=file.size_bytes,
        sheet_name=sheet_name,
        available_sheets=available_sheets,
        loaded_row_count=len(normalized),
        column_count=len(normalized.columns),
        truncated=truncated,
    )

    return ExtractedTable(
        frame=normalized,
        metadata=metadata,
        warnings=tuple(dict.fromkeys(all_warnings)),
    )

Конкретный обработчик передаст сюда прочитанный DataFrame и сведения, относящиеся к формату. Общая функция нормализует таблицу, создаёт метаданные и объединяет предупреждения.

Выражение warnings + normalization_warnings соединяет замечания парсера и общей очистки. dict.fromkeys() удаляет повторения, сохраняя порядок. В результате вызывающий код получает один объект и не собирает его вручную для CSV и Excel по-разному.

В этом шаге мы ещё не открывали таблицы, но мы определили границы, общий результат и правила нормализации. В следующем шаге мы создадим обработчики CSV и XLSX, научимся выбирать лист Excel и соберём их в единый TableExtractor.

Шаг 2

Общий контракт у нас готов и теперь мы научимся превращать проверенные CSV и XLSX в одинаковый ExtractedTable. То есть внешний результат будет общим, но способы чтения останутся разными. Давайте только обсудим разницу в файлах.

CSV — это текстовый файл. Обычно первая строка содержит названия колонок, остальные строки — значения, а поля разделяются запятой, точкой с запятой или другим символом. В одном CSV нет нескольких листов.

XLSX — это уже ZIP-контейнер со структурой книги Excel. В нём может быть несколько листов, типизированные числа и даты, формулы, стили и служебные элементы. В этом разделе нас интересуют только значения ячеек и названия листов, а не визуальное оформление и мы реализуем следующее взаимодействие:

Начнём с CSV. Создадим отдельный обработчик, который прочитает проверенный файл через pandas, ограничит количество строк и передаст полученный DataFrame в общую функцию build_extracted_table().

В пакете app/files создайте файл csv_table_extraction.py и добавьте:

import pandas as pd

from app.files.table_extraction import (
    ExtractedTable,
    TableExtractionPolicy,
    TableReadError,
    TableSheetError,
    UnsupportedTableFormatError,
    build_extracted_table,
)
from app.files.validation import FileKind, ValidatedFile

class CsvTableExtractor:
    supported_kinds = frozenset({FileKind.CSV})

    def __init__(self, policy: TableExtractionPolicy) -> None:
        self._policy = policy

    def extract(
        self,
        file: ValidatedFile,
        sheet_name: str | None = None,
    ) -> ExtractedTable:
        if file.kind not in self.supported_kinds:
            raise UnsupportedTableFormatError(
                f"CsvTableExtractor не поддерживает {file.kind}."
            )

        if sheet_name is not None:
            raise TableSheetError("CSV не содержит листов.")

        try:
            frame = pd.read_csv(
                file.path,
                encoding="utf-8",
                sep=None,
                engine="python",
                nrows=self._policy.max_rows + 1,
                index_col=False,
                keep_default_na=True,
                on_bad_lines="error",
            )
        except (
            OSError,
            UnicodeError,
            ValueError,
            pd.errors.ParserError,
            pd.errors.EmptyDataError,
        ) as error:
            raise TableReadError(
                "Не удалось прочитать CSV."
            ) from error

        return build_extracted_table(
            file=file,
            frame=frame,
            policy=self._policy,
        )

Класс получает общую политику через конструктор и только сохраняет её. Файл открывается позднее, во время extract(). Сначала метод проверяет FileKind, а затем убеждается, что вызывающий код не передал имя листа.

pd.read_csv() получает путь уже проверенной временной копии. Кодировка UTF-8 зафиксирована явно и совпадает с проверкой из FileValidator.

sep=None вместе с engine="python" просит pandas определить разделитель по содержимому. Это позволит прочитать как файл с запятыми, так и распространённый CSV с точками с запятой. Автоматическое определение не является безошибочным, поэтому в реальном интерфейсе для нестандартного файла можно дополнительно дать пользователю выбор разделителя.

nrows ограничивает количество прочитанных строк данных и не включает строку заголовков. Мы запрашиваем на одну строку больше политики, чтобы общая нормализация увидела превышение.

index_col=False говорит не превращать первую колонку в индекс. Все исходные колонки останутся данными. keep_default_na=True включает стандартное распознавание пропусков, а on_bad_lines="error" запрещает молча пропускать строки с неправильным количеством полей.

Ожидаемые ошибки pandas и файловой системы преобразуются в TableReadError. При успехе обработчик не собирает метаданные самостоятельно, а передаёт DataFrame в уже написанную build_extracted_table().

Как выбирается лист Excel?

CSV-обработчику достаточно одного вызова чтения, но для XLSX сначала нужно открыть книгу, получить список листов и решить, какой из них разбирать.

Если вызывающий код передал sheet_name, используем именно его. Если название отсутствует, выберем первый лист и добавим предупреждение, по сути это удобное значение по умолчанию, но все равно пользователь должен знать, какой лист фактически попал в результат.

В app/files создайте файл excel_table_extraction.py:

from zipfile import BadZipFile

import pandas as pd

from app.files.table_extraction import (
    ExtractedTable,
    TableExtractionError,
    TableExtractionPolicy,
    TableReadError,
    TableSheetError,
    UnsupportedTableFormatError,
    build_extracted_table,
)
from app.files.validation import FileKind, ValidatedFile

class ExcelTableExtractor:
    supported_kinds = frozenset({FileKind.XLSX})

    def __init__(self, policy: TableExtractionPolicy) -> None:
        self._policy = policy

    def extract(
        self,
        file: ValidatedFile,
        sheet_name: str | None = None,
    ) -> ExtractedTable:
        if file.kind not in self.supported_kinds:
            raise UnsupportedTableFormatError(
                f"ExcelTableExtractor не поддерживает {file.kind}."
            )

        try:
            with pd.ExcelFile(
                file.path,
                engine="openpyxl",
            ) as workbook:
                available_sheets = tuple(workbook.sheet_names)
                if not available_sheets:
                    raise TableSheetError(
                        "В книге Excel не найдено листов."
                    )

                selected_sheet = sheet_name or available_sheets[0]
                if selected_sheet not in available_sheets:
                    raise TableSheetError(
                        f"Лист {selected_sheet!r} не найден."
                    )

                frame = workbook.parse(
                    sheet_name=selected_sheet,
                    nrows=self._policy.max_rows + 1,
                )
        except TableExtractionError:
            raise
        except (
            OSError,
            BadZipFile,
            ImportError,
            KeyError,
            ValueError,
        ) as error:
            raise TableReadError(
                "Не удалось прочитать XLSX."
            ) from error

        warnings = ()
        if sheet_name is None:
            warnings = (
                f"Автоматически выбран первый лист: "
                f"{selected_sheet}.",
            )

        return build_extracted_table(
            file=file,
            frame=frame,
            policy=self._policy,
            sheet_name=selected_sheet,
            available_sheets=available_sheets,
            warnings=warnings,
        )

В этом коде pd.ExcelFile открывает книгу и позволяет несколько раз обращаться к её структуре без повторного открытия исходного файла. Контекстный менеджер with гарантирует закрытие ресурса после чтения или ошибки.

workbook.sheet_names возвращает названия листов в их исходном порядке. Мы превращаем список в кортеж и сохраняем его в метаданных. Если листов нет, обработка завершается понятным TableSheetError.

Выражение sheet_name or available_sheets[0] выбирает запрошенное имя либо первый лист. После этого отдельная проверка не позволяет передать несуществующее название в pandas и получить менее понятную ошибку библиотеки.

workbook.parse() читает только выбранный лист и возвращает DataFrame. Как и в CSV, количество строк ограничено значением max_rows + 1.

TableExtractionError перехватывается отдельно и сразу передаётся дальше. Благодаря этому собственный TableSheetError не превращается в общий TableReadError. Низкоуровневые ошибки XLSX и отсутствие openpyxl, наоборот, скрываются за общей прикладной границей чтения.

После выхода из with книга закрыта, но frame уже находится в памяти. Если лист был выбран автоматически, создаётся предупреждение. Затем общая функция нормализует таблицу и добавляет метаданные.

Теперь оба обработчика имеют одинаковый интерфейс, поэтому откройте app/files/table_extraction.py и в конец файла добавьте единый маршрутизатор:

class TableExtractor:
    def __init__(
        self,
        extractors: tuple[TableFormatExtractor, ...],
    ) -> None:
        if not extractors:
            raise ValueError(
                "Нужен хотя бы один табличный экстрактор."
            )

        registry: dict[FileKind, TableFormatExtractor] = {}

        for extractor in extractors:
            for kind in extractor.supported_kinds:
                if kind in registry:
                    raise ValueError(
                        f"Для формата {kind} уже есть экстрактор."
                    )
                registry[kind] = extractor

        self._registry = registry

    def extract(
        self,
        file: ValidatedFile,
        sheet_name: str | None = None,
    ) -> ExtractedTable:
        extractor = self._registry.get(file.kind)
        if extractor is None:
            raise UnsupportedTableFormatError(
                f"Табличное извлечение для {file.kind} "
                "не поддерживается."
            )

        return extractor.extract(file, sheet_name=sheet_name)

Устройство повторяет знакомый DocumentExtractor и при создании объекта один раз строится словарь форматов. При вызове extract() маршрутизатор находит обработчик по file.kind и передаёт ему файл вместе с необязательным именем листа. То есть маршрутизатор не знает, как устроены CSV и Excel, и не вызывает pandas самостоятельно. Его единственная задача, это только направить запрос правильному объекту.

Чтобы не собирать стандартный набор вручную в каждом демонстрационном файле, создайте app/files/default_table_extractor.py:

from app.files.csv_table_extraction import CsvTableExtractor
from app.files.excel_table_extraction import ExcelTableExtractor
from app.files.table_extraction import (
    TableExtractionPolicy,
    TableExtractor,
)

def build_default_table_extractor(
    policy: TableExtractionPolicy | None = None,
) -> TableExtractor:
    extraction_policy = policy or TableExtractionPolicy()

    return TableExtractor(
        extractors=(
            CsvTableExtractor(extraction_policy),
            ExcelTableExtractor(extraction_policy),
        )
    )

Функция создаёт одну неизменяемую политику и передаёт её обоим обработчикам. Поэтому CSV и XLSX используют одинаковые ограничения строк и колонок. Если вызывающий код не передал настройки, создаются стандартные.

Теперь приложение умеет получить одинаковый ExtractedTable из двух форматов, но ведь читать таблицу ещё не значит правильно использовать её в AI-функции. Поэтому в следующем шаге мы создадим отдельный TableAnalyzer, он покажет пропуски и типы колонок, выполнит точную агрегацию обычным кодом и подготовит ограниченное представление для будущей передачи модели.

Шаг 3

TableExtractor уже умеет читать CSV и Excel и возвращать их содержимое в виде DataFrame. Однако полученную таблицу нельзя без подготовки целиком передавать языковой модели.

Во-первых, модель не получает объект DataFrame напрямую. Перед отправкой таблицу пришлось бы преобразовать в текст, например в CSV или JSON. Если файл содержит тысячи строк и десятки колонок, получится очень большой запрос. Он займёт значительную часть контекстного окна, увеличит количество входных токенов, время ожидания и стоимость обращения к модели.

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

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

Поэтому работу лучше разделить. Обычный Python-код и pandas должны выбирать нужные строки и колонки, проверять пропуски и выполнять точные вычисления. Модель будет получать уже подготовленный компактный результат и использовать его для того, в чём она действительно полезна: объяснить показатели простым языком, сформулировать вывод или ответить на смысловой вопрос пользователя.

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

В пакете app/files создайте файл table_analysis.py и добавьте:

from dataclasses import dataclass
from typing import Literal

import pandas as pd

from app.files.table_extraction import ExtractedTable

AggregationOperation = Literal[
    "count",
    "sum",
    "mean",
    "min",
    "max",
]

class TableAnalysisError(RuntimeError):
    pass

@dataclass(frozen=True)
class ColumnProfile:
    name: str
    data_type: str
    missing_count: int
    unique_count: int

@dataclass(frozen=True)
class TableProfile:
    row_count: int
    column_count: int
    columns: tuple[ColumnProfile, ...]

@dataclass(frozen=True)
class PreparedTableContext:
    text: str
    columns: tuple[str, ...]
    included_rows: int
    omitted_rows: int
    source_truncated: bool

@dataclass(frozen=True)
class TableContextPolicy:
    max_preview_rows: int = 20
    max_characters: int = 12_000

    def __post_init__(self) -> None:
        if self.max_preview_rows <= 0:
            raise ValueError(
                "max_preview_rows должен быть больше нуля."
            )
        if self.max_characters <= 0:
            raise ValueError(
                "max_characters должен быть больше нуля."
            )

Давайте разберем этот код. И начнем с AggregationOperation, эта переменная ограничивает поддерживаемые операции, а именно количество, сумма, среднее, минимум и максимум. Мы не разрешаем передать произвольное имя метода pandas, тем самым это сохранит контролируемую границу между будущим AI-сценарием и Python-кодом.

ColumnProfile описывает одну колонку: её имя, тип pandas, количество пропусков и количество уникальных непустых значений. TableProfile объединяет профили с размерами таблицы.

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

TableContextPolicy относится уже не к чтению файла, а к подготовке данных для модели. max_preview_rows ограничивает число строк в представлении, а max_characters не позволяет нескольким очень длинным ячейкам создать огромную строку.

После моделей добавьте начало анализатора:

class TableAnalyzer:
    def __init__(
        self,
        context_policy: TableContextPolicy | None = None,
    ) -> None:
        self._context_policy = (
            context_policy or TableContextPolicy()
        )

    @staticmethod
    def profile(table: ExtractedTable) -> TableProfile:
        column_profiles: list[ColumnProfile] = []

        for name in table.frame.columns:
            series = table.frame[name]
            column_profiles.append(
                ColumnProfile(
                    name=name,
                    data_type=str(series.dtype),
                    missing_count=int(series.isna().sum()),
                    unique_count=int(
                        series.nunique(dropna=True)
                    ),
                )
            )

        return TableProfile(
            row_count=len(table.frame),
            column_count=len(table.frame.columns),
            columns=tuple(column_profiles),
        )

Конструктор сохраняет политику контекста. Если настройки не переданы, создаются стандартные, при этом сам вызов profile() от этой политики не зависит, поэтому метод объявлен статическим. Внутри цикл идёт по названиям колонок. Выражение table.frame[name] возвращает series — одну колонку pandas вместе со всеми её значениями.

series.dtype содержит определённый pandas тип, например int64, float64, datetime64 или строковый тип. Мы сохраняем его текстовое представление, потому что профиль предназначен для диагностики, а не для восстановления объекта dtype.

series.isna().sum() считает пропуски. series.nunique(dropna=True) считает разные непустые значения. Оба результата могут иметь числовой тип NumPy, поэтому int() превращает их в обычные целые числа Python.

После завершения цикла список профилей преобразуется в кортеж. Исходный DataFrame при этом не изменяется.

Теперь внутри TableAnalyzer, после profile(), добавьте подготовку ограниченного представления:

    def prepare_context(
        self,
        table: ExtractedTable,
        columns: tuple[str, ...] | None = None,
    ) -> PreparedTableContext:
        selected_columns = (
            tuple(table.frame.columns)
            if columns is None
            else columns
        )

        if not selected_columns:
            raise TableAnalysisError(
                "Нужно выбрать хотя бы одну колонку."
            )

        self._require_columns(table.frame, selected_columns)

        preview = table.frame.loc[
            :, list(selected_columns)
        ].head(self._context_policy.max_preview_rows).copy()
        text = preview.to_csv(index=False)

        if len(text) > self._context_policy.max_characters:
            raise TableAnalysisError(
                "Подготовленное представление слишком большое. "
                "Выберите меньше колонок."
            )

        return PreparedTableContext(
            text=text,
            columns=selected_columns,
            included_rows=len(preview),
            omitted_rows=max(
                len(table.frame) - len(preview),
                0,
            ),
            source_truncated=table.metadata.truncated,
        )

Если columns равно None, используются все колонки. Пустой кортеж, наоборот, считается ошибкой: вызывающий код явно не выбрал ни одного поля.

_require_columns(), который добавим в конце класса, заранее проверяет имена. Благодаря этому опечатка не превращается в длинный KeyError pandas.

frame.loc[:, list(selected_columns)] означает: взять все строки и только перечисленные колонки. Затем head() оставляет первые строки согласно политике, а copy() создаёт независимый DataFrame. Исходная таблица не меняется.

preview.to_csv(index=False) превращает небольшую таблицу в компактный текст с заголовками. Мы используем CSV как внутреннее текстовое представление, потому что оно сохраняет границы строк и колонок и занимает меньше места, чем подробный JSON.

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

included_rows сообщает, сколько строк вошло в представление. omitted_rows считает строки, оставшиеся внутри загруженного DataFrame, а source_truncated отдельно показывает, был ли сокращён сам файл. Это две разные границы.

Далее добавьте точную агрегацию:

    def aggregate(
        self,
        table: ExtractedTable,
        group_by: str,
        operation: AggregationOperation,
        value_column: str | None = None,
    ) -> pd.DataFrame:
        allowed_operations = {
            "count",
            "sum",
            "mean",
            "min",
            "max",
        }
        if operation not in allowed_operations:
            raise TableAnalysisError(
                f"Операция {operation!r} не поддерживается."
            )

        self._require_columns(table.frame, (group_by,))
        frame = table.frame.copy()

        if operation == "count":
            return (
                frame.groupby(
                    group_by,
                    dropna=False,
                    sort=True,
                )
                .size()
                .reset_index(name="count")
            )

        if value_column is None:
            raise TableAnalysisError(
                "Для числовой операции нужна колонка значений."
            )

        self._require_columns(table.frame, (value_column,))
        frame[value_column] = pd.to_numeric(
            frame[value_column],
            errors="coerce",
        )
        frame = frame.dropna(subset=[value_column])

        if frame.empty:
            raise TableAnalysisError(
                f"В колонке {value_column!r} нет чисел."
            )

        result = (
            frame.groupby(
                group_by,
                dropna=False,
                sort=True,
            )[value_column]
            .agg(operation)
            .reset_index()
        )
        result_name = f"{operation}_{value_column}"
        return result.rename(
            columns={value_column: result_name}
        )

Метод поддерживает два сценария. Для операции count достаточно указать колонку группировки. Например groupby("category") объединит обращения с одинаковой категорией, size() посчитает строки в каждой группе, а reset_index() вернёт результат в виде обычного DataFrame. Параметр sort=True обеспечивает предсказуемый порядок групп.

Для суммы, среднего, минимума и максимума дополнительно указывается value_column — колонка со значениями для вычисления. Например можно найти среднее время обработки resolution_hours для каждой категории.

Перед вычислением pd.to_numeric(errors="coerce") пытается преобразовать значения в числа. Неподходящие значения становятся пропусками и удаляются через dropna(). Если чисел не осталось, метод возвращает понятную ошибку.

Обработка выполняется в копии DataFrame, поэтому исходная таблица не изменяется. Метод .agg(operation) запускает одну из разрешённых операций, а итоговая колонка получает понятное имя, например mean_resolution_hours.

Чтобы закончить класс, добавьте проверку колонок:

    @staticmethod
    def _require_columns(
        frame: pd.DataFrame,
        columns: tuple[str, ...],
    ) -> None:
        missing = [
            name
            for name in columns
            if name not in frame.columns
        ]
        if missing:
            raise TableAnalysisError(
                "В таблице отсутствуют колонки: "
                + ", ".join(missing)
            )

Метод строит список запрошенных имён, которых нет в frame.columns. Если список не пуст, наружу выходит одно понятное сообщение сразу со всеми отсутствующими колонками.

Теперь точные операции находятся в обычном Python-коде. TableAnalyzer умеет описать таблицу, выбрать релевантные поля, ограничить объём и выполнить безопасную агрегацию. В следующем шаге мы подготовим CSV и Excel и запустим весь сценарий и сравним полную таблицу с компактным результатом, который позднее получит модель.

Шаг 4

В этом разделе появилось несколько новых объектов, поэтому перед запуском восстановим общую последовательность. FileValidator подтверждает файл, TemporaryFileStorage создаёт временную копию, TableExtractor выбирает CSV- или XLSX-обработчик, а TableAnalyzer работает уже с готовым ExtractedTable.

На этом этапе мы работаем только с pandas и пока не обращаемся к ProxyAPI. Ведь сначала важно проверить, правильно ли прочитались строки и колонки, какие типы получили данные и верные ли результаты возвращают вычисления. И когда мы это проверим, тогда уже можно безопасно передавать модели результат для дальнейшего объяснения.

Подготавливаем CSV и Excel

Продолжим использовать каталог sample_files. Это каталог с учебными данными, и теперь создайте в нём файл support_requests.csv и сохраните следующий текст в UTF-8:

request_id,category,status,resolution_hours,satisfaction
1001,Доставка,Закрыто,4.5,5
1002,Возврат,Закрыто,12,4
1003,Доставка,Открыто,,
1004,Оплата,Закрыто,2.5,5
1005,Возврат,Закрыто,8,3
1006,Доставка,Закрыто,6,4

У обращения 1003 отсутствуют время решения и оценка. Эти пустые ячейки нужны, чтобы увидеть работу isna() и убедиться, что среднее значение не рассчитывается по выдуманному нулю.

Теперь создайте книгу support_requests.xlsx в Excel или Numbers. Первый лист назовите Requests и перенесите в него ту же таблицу.

Добавьте второй лист Dictionary:

field              | description
request_id         | Идентификатор обращения
category           | Категория проблемы
status             | Текущий статус
resolution_hours   | Время решения в часах
satisfaction       | Оценка клиента от 1 до 5

Сохраните книгу именно в формате XLSX. Если у вас возникли трудности, то можете скачать оба файла по этой ссылке. В результате набор учебных файлов должен содержать:

sample_files/
├── support_requests.csv
└── support_requests.xlsx
    ├── Requests
    └── Dictionary

То есть на диске существует один файл support_requests.xlsx, а Requests и Dictionary находятся внутри него.

Для проверки создадим отдельную точку входа table_extraction_demo.py в корне проекта. Этот временный демонстрационный файл позволит проверить новый контур отдельно и позднее будет удалён после подключения Document AI Service.

Сначала добавьте импорты и пути:

from dataclasses import replace
from pathlib import Path

from app.files.default_table_extractor import (
    build_default_table_extractor,
)
from app.files.table_analysis import (
    TableAnalysisError,
    TableAnalyzer,
)
from app.files.table_extraction import (
    ExtractedTable,
    TableExtractionError,
    TableExtractionPolicy,
    TableExtractor,
)
from app.files.temporary_storage import (
    TemporaryFileStorage,
    TemporaryFileStorageError,
)
from app.files.validation import (
    FileValidationError,
    FileValidationPolicy,
    FileValidator,
)

PROJECT_ROOT = Path(__file__).resolve().parent
SAMPLE_FILES_ROOT = PROJECT_ROOT / "sample_files"

Импорты отражают всю цепочку. Из файлового контура нужны валидатор и временное хранилище, из нового табличного контура — экстрактор, анализатор и их прикладные ошибки.

PROJECT_ROOT строится от расположения самого демонстрационного файла. Поэтому запуск из IDE и запуск командой из корня проекта будут использовать один и тот же каталог sample_files.

Ниже добавьте функцию вывода:

def print_table_details(
    table: ExtractedTable,
    analyzer: TableAnalyzer,
    context_columns: tuple[str, ...],
) -> None:
    metadata = table.metadata
    print(f"Формат: {metadata.kind}")
    print(f"Выбранный лист: {metadata.sheet_name}")
    print(f"Доступные листы: {metadata.available_sheets}")
    print(f"Загружено строк: {metadata.loaded_row_count}")
    print(f"Колонок: {metadata.column_count}")
    print(f"Таблица сокращена: {metadata.truncated}")

    for warning in table.warnings:
        print(f"Предупреждение: {warning}")

    profile = analyzer.profile(table)
    print("Профиль колонок:")
    for column in profile.columns:
        print(
            f"- {column.name}: тип={column.data_type}, "
            f"пропусков={column.missing_count}, "
            f"уникальных={column.unique_count}"
        )

    context = analyzer.prepare_context(
        table,
        columns=context_columns,
    )
    print("Компактное представление:")
    print(context.text)
    print(f"Строк включено: {context.included_rows}")
    print(f"Строк не включено: {context.omitted_rows}")

Функция получает уже извлечённую таблицу и ничего не читает с диска. Сначала она выводит метаданные и предупреждения. Для CSV имя листа будет None, а список листов — пустым.

Затем analyzer.profile() создаёт описание колонок. Вложенный цикл печатает тип, число пропусков и число уникальных значений. По этому выводу можно заметить неожиданное распознавание типа до вычислений.

context_columns передаются явно для каждого сценария. Это показывает важный принцип: релевантные поля выбирает прикладной сценарий, а не случайный порядок колонок в файле.

prepare_context() возвращает CSV-строку, но функция пока только печатает её. Никакого скрытого вызова модели здесь нет.

Теперь добавьте функцию обработки одного файла:

def run_case(
    source_path: Path,
    file_policy: FileValidationPolicy,
    validator: FileValidator,
    storage: TemporaryFileStorage,
    extractor: TableExtractor,
    analyzer: TableAnalyzer,
    context_columns: tuple[str, ...],
    sheet_name: str | None = None,
) -> None:
    print(f"\nТаблица: {source_path.name}")
    if sheet_name is not None:
        print(f"Запрошенный лист: {sheet_name}")

    try:
        validated = validator.validate(source_path)

        with storage.stage(validated) as temporary_path:
            staged_policy = replace(
                file_policy,
                allowed_root=temporary_path.parent,
            )
            staged_validator = FileValidator(staged_policy)
            staged_file = staged_validator.validate(
                temporary_path,
                declared_media_type=validated.media_type,
            )
            staged_file = replace(
                staged_file,
                original_name=validated.original_name,
            )
            table = extractor.extract(
                staged_file,
                sheet_name=sheet_name,
            )

        print_table_details(
            table,
            analyzer,
            context_columns,
        )

        if {
            "category",
            "resolution_hours",
        }.issubset(table.frame.columns):
            counts = analyzer.aggregate(
                table,
                group_by="category",
                operation="count",
            )
            averages = analyzer.aggregate(
                table,
                group_by="category",
                value_column="resolution_hours",
                operation="mean",
            )
            print("Количество обращений по категориям:")
            print(counts.to_string(index=False))
            print("Среднее время решения по категориям:")
            print(averages.to_string(index=False))
    except (
        FileValidationError,
        TemporaryFileStorageError,
        TableExtractionError,
        TableAnalysisError,
    ) as error:
        print(f"Обработка не выполнена: {error}")

Функция получает готовые зависимости, поэтому не создаёт валидатор и анализатор заново для каждого файла. Исходная file_policy также передаётся явно: она понадобится, чтобы создать её копию с временным разрешённым каталогом. Параметры source_path, sheet_name и context_columns описывают различия конкретного сценария.

После первой проверки создаётся временная копия. Как и в демонстрации документов, для неё собирается политика с новым allowed_root и выполняется повторная проверка.

После повторной проверки через replace() восстанавливается исходное имя. Затем таблица извлекается внутри with, пока временный файл существует. Готовый DataFrame уже загружен в память, поэтому печать и вычисления выполняются после автоматического удаления копии.

Условие с issubset() проверяет наличие двух колонок, необходимых для учебной агрегации. На листе Dictionary их нет, поэтому профиль и компактное представление будут показаны, а расчёты категорий пропустятся.

Для подходящей таблицы анализатор выполняет две точные операции. Первая считает количество обращений в каждой категории. Вторая преобразует resolution_hours в числа и рассчитывает среднее. Пропуск обращения 1003 не превращается в ноль и не искажает среднее значение.

В конце файла добавьте основную функцию:

def main() -> None:
    file_policy = FileValidationPolicy(
        allowed_root=SAMPLE_FILES_ROOT,
        max_size_bytes=10 * 1_024 * 1_024,
    )
    validator = FileValidator(file_policy)
    storage = TemporaryFileStorage(
        max_size_bytes=file_policy.max_size_bytes
    )

    table_policy = TableExtractionPolicy(
        max_rows=100,
        max_columns=20,
    )
    extractor = build_default_table_extractor(table_policy)
    analyzer = TableAnalyzer()

    request_columns = (
        "request_id",
        "category",
        "status",
        "resolution_hours",
    )

    run_case(
        SAMPLE_FILES_ROOT / "support_requests.csv",
        file_policy,
        validator,
        storage,
        extractor,
        analyzer,
        context_columns=request_columns,
    )
    run_case(
        SAMPLE_FILES_ROOT / "support_requests.xlsx",
        file_policy,
        validator,
        storage,
        extractor,
        analyzer,
        context_columns=request_columns,
        sheet_name="Requests",
    )
    run_case(
        SAMPLE_FILES_ROOT / "support_requests.xlsx",
        file_policy,
        validator,
        storage,
        extractor,
        analyzer,
        context_columns=("field", "description"),
        sheet_name="Dictionary",
    )

if __name__ == "__main__":
    main()

В main() отдельно создаются две политики, где FileValidationPolicy ограничивает файл на диске, а TableExtractionPolicy задает логические размеры таблицы. Далее один TableExtractor и один TableAnalyzer используются во всех трёх вызовах, где сначала читается CSV, затем лист Requests, а после него Dictionary из той же книги.

Запускаем и разбираем результат

Давайте запустим данный файл и для CSV ожидаем результат примерно следующего вида:

Таблица: support_requests.csv
Формат: csv
Выбранный лист: None
Доступные листы: ()
Загружено строк: 6
Колонок: 5
Таблица сокращена: False
Профиль колонок:
- request_id: тип=int64, пропусков=0, уникальных=6
- category: тип=str, пропусков=0, уникальных=3
- status: тип=str, пропусков=0, уникальных=2
- resolution_hours: тип=float64, пропусков=1, уникальных=5
- satisfaction: тип=float64, пропусков=1, уникальных=3

Конкретные названия строковых типов могут отличаться между версиями pandas, например вместо object может появиться строковый dtype. Для нас важнее содержимое, количество пропусков и успешное выполнение вычислений.

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

Количество обращений по категориям:
category  count
 Возврат      2
Доставка      3
  Оплата      1
Среднее время решения по категориям:
category  mean_resolution_hours
 Возврат                  10.00
Доставка                   5.25
  Оплата                   2.50

Для доставки есть три обращения, но среднее рассчитано по двум известным значениям 4.5 и 6. Получается 5.25. Пустая ячейка не считается нулём.

Для XLSX метаданные дополнительно покажут:

Выбранный лист: Requests
Доступные листы: ('Requests', 'Dictionary')

Во втором запуске той же книги будет выбран Dictionary. Это подтверждает, что XLSX не превращается автоматически в одну смешанную таблицу: каждый лист читается как отдельный набор данных.

Чтобы увидеть ограничение на практике, временно установите max_rows=3 в TableExtractionPolicy и запустите демонстрацию повторно. Метаданные покажут truncated=True, а предупреждение сообщит о сокращении. После проверки верните значение 100.

Если указать sheet_name="Unknown", Excel-обработчик вернёт контролируемую ошибку со списком фактически доступных листов в метаданных успешных запусков. Если в context_columns допустить опечатку, TableAnalyzer остановится до подготовки контекста и перечислит отсутствующую колонку.

Подведём итоги раздела. Мы научились читать CSV и выбранный лист XLSX через единый интерфейс, сохранять типы и пропуски pandas, ограничивать строки и колонки, строить профиль и выбирать релевантные поля.

Точные вычисления выполняются до модели: pandas считает группы и средние значения, а будущая AI-функция получит компактную таблицу или уже рассчитанный результат. Модель не должна повторять арифметику — её задача будет заключаться в понимании вопроса и понятном объяснении проверяемых данных.

Пока оставьте table_extraction_demo.py рядом с document_extraction_demo.py. Эти файлы позволяют отдельно проверять извлечение документов и таблиц без обращения к модели. Такой подход упрощает поиск ошибок: если pandas неправильно прочитал колонку или экстрактор не нашёл текст в PDF, мы увидим проблему ещё до подключения AI-анализа.

В следующем разделе мы объединим подготовленные компоненты в первую версию DocumentAIService. Перед этим проведём небольшой рефакторинг каталога app/files, разделив обработчики документов и таблиц по отдельным пакетам. Затем сервис свяжет проверку файла, временное хранение, извлечение данных, подготовку контекста и обращение к модели. После успешной проверки общего сценария отдельные файлы document_extraction_demo.py и table_extraction_demo.py можно будет удалить, а для дальнейших запусков мы оставим единый document_ai_demo.py.

Шаг 5

Почему таблицу нельзя сразу превратить в одну длинную строку?

Правильный ответ
  • ✅ Потеряется связь значений со строками, колонками и типами данных

Шаг 6

Какие действия входят в нормализацию DataFrame из лекции?

Правильный ответ
  • ✅ Удаление полностью пустых строк и колонок
  • ✅ Ограничение числа строк и колонок политикой
  • ✅ Назначение стабильных имён пустым заголовкам
  • ✅ Отклонение одинаковых имён колонок после нормализации

Шаг 7

Как различается выбор листа для CSV и XLSX?

Правильный ответ
  • ✅ Для CSV выбор листа запрещён, а для XLSX можно выбрать существующий лист

Шаг 8

Какие данные безопаснее передать модели для объяснения результата табличного анализа?

Правильный ответ
  • ✅ Ограниченный preview нужных колонок и точный результат вычисления pandas

Шаг 9

CSV-экстрактор ограничивает результат:

rows = list(islice(reader, max_rows + 1))
truncated = len(rows) > max_rows
rows = rows[:max_rows]

В файле четыре строки данных, а max_rows равен 3. Что будет после выполнения этого кода?

Правильный ответ
  • ✅ truncated будет равен True, а в rows останутся три строки

Шаг 10

Приведение заголовков таблицы к стабильным именам

После чтения CSV или Excel названия колонок могут содержать пробелы, числа, пустые значения или автоматически созданные имена вида Unnamed: 1. Перед анализом заголовки нужно нормализовать и убедиться, что они остаются уникальными.

Задача

Реализуйте функцию normalize_column_names(columns). Она должна вернуть нормализованные имена и предупреждение, если хотя бы одному пустому заголовку было присвоено новое имя.

Требования

  • Не изменяйте имя функции. Она принимает один аргумент columns.
  • columns должен быть списком или кортежем. Для другого типа выбросьте TypeError.
  • Обрабатывайте элементы последовательно, сохраняя исходный порядок.
  • Каждый элемент преобразуйте в строку через str() и удалите внешние пробелы через strip().
  • Если получено пустое имя или имя начинается с "Unnamed:", замените его на "column_N", где N — позиция колонки, начиная с единицы.
  • Если была выполнена хотя бы одна замена, добавьте предупреждение "Пустым заголовкам колонок присвоены имена.".
  • Это предупреждение должно присутствовать в результате не более одного раза, даже если заменено несколько заголовков.
  • После нормализации проверьте уникальность всех имён.
  • Если два или более имени совпали, выбросьте ValueError.
  • Верните пару (tuple(normalized_names), tuple(warnings)).

Пример результата

names, warnings = normalize_column_names([
    " category ",
    "Unnamed: 1",
    "",
])
print(names)
print(warnings)

Результат:

('category', 'column_2', 'column_3')
('Пустым заголовкам колонок присвоены имена.',)

Для входа ["a", " a "] после очистки возникнет дубликат, поэтому ожидается ValueError.

Создавать DataFrame и импортировать pandas не нужно.

Правильный ответ 
def normalize_column_names(columns) -> tuple[tuple[str, ...], tuple[str, ...]]:
    if not isinstance(columns, (list, tuple)):
        raise TypeError

    normalized_names = []
    replaced = False

    for index, column in enumerate(columns, start=1):
        name = str(column).strip()

        if not name or name.startswith("Unnamed:"):
            name = f"column_{index}"
            replaced = True

        normalized_names.append(name)

    if len(normalized_names) != len(set(normalized_names)):
        raise ValueError

    warnings = ()

    if replaced:
        warnings = ("Пустым заголовкам колонок присвоены имена.",)

    return tuple(normalized_names), warnings

Шаг 11

Подготовка ограниченного CSV-preview таблицы

Передавать модели всю таблицу часто дорого и небезопасно. Вместо этого приложение выбирает только нужные колонки и первые строки, формирует CSV-текст и сообщает, сколько строк осталось за пределами preview. Исходный DataFrame при этом изменять нельзя.

Задача

Реализуйте функцию prepare_table_preview(frame, columns, max_rows). Она должна создать ограниченное CSV-представление выбранных колонок и вернуть его вместе с метаданными.

Требования

  • Не изменяйте имя функции и её аргументы.
  • frame должен быть экземпляром pandas.DataFrame. Для другого типа выбросьте TypeError.
  • columns должен быть списком или кортежем. Пустая последовательность должна приводить к ValueError.
  • max_rows должен быть положительным целым числом. Значение типа bool, ноль, отрицательное число или другой тип должны приводить к ValueError.
  • Проверьте наличие каждого имени из columns в frame.columns. Для отсутствующей колонки выбросьте KeyError.
  • Выберите колонки в том порядке, в котором они переданы в columns.
  • Возьмите первые max_rows строк выбранной части и создайте отдельную копию.
  • Преобразуйте копию в CSV через to_csv(index=False). В text должны попасть строка заголовков и выбранные строки без индекса.
  • included_rows должен содержать фактическое число строк в preview. Если таблица короче лимита, это число будет меньше max_rows.
  • omitted_rows должен быть равен числу строк исходной таблицы, не вошедших в preview, и не может быть отрицательным.
  • Верните словарь с ключами text, columns, included_rows и omitted_rows. В columns сохраните кортеж выбранных имён.
  • Не изменяйте исходный frame.

Пример результата

В таблице три строки, а preview ограничен двумя:

preview = prepare_table_preview(
    frame, ("category", "hours"), 2
)
print(preview)

Результат:

{
    'text': 'category,hours\na,1\nb,2\n',
    'columns': ('category', 'hours'),
    'included_rows': 2,
    'omitted_rows': 1,
}

Библиотека pandas и импорт pd уже доступны в заготовке.

Правильный ответ 
import pandas as pd

def prepare_table_preview(frame, columns, max_rows: int) -> dict:
    if not isinstance(frame, pd.DataFrame):
        raise TypeError

    if not isinstance(columns, (list, tuple)):
        raise TypeError

    if not columns:
        raise ValueError

    if (
        isinstance(max_rows, bool)
        or not isinstance(max_rows, int)
        or max_rows <= 0
    ):
        raise ValueError

    for column in columns:
        if column not in frame.columns:
            raise KeyError(column)

    preview_frame = frame.loc[:, list(columns)].head(max_rows).copy()

    included_rows = len(preview_frame)
    omitted_rows = max(len(frame) - included_rows, 0)

    return {
        "text": preview_frame.to_csv(index=False),
        "columns": tuple(columns),
        "included_rows": included_rows,
        "omitted_rows": omitted_rows,
    }

Шаг 12

Группировка и точная агрегация таблицы

Языковая модель не должна самостоятельно вычислять суммы и средние значения по таблице. Такие операции выполняются кодом pandas, а модели передаётся уже готовый результат. Функция должна поддерживать подсчёт строк и четыре числовые агрегации.

Задача

Реализуйте функцию aggregate_table(frame, group_by, operation, value_column=None). Она должна сгруппировать строки по колонке group_by и вернуть новый DataFrame с результатом выбранной операции.

Требования

  • Не изменяйте имя функции, её аргументы и значение value_column по умолчанию.
  • frame должен быть экземпляром pandas.DataFrame. Для другого типа выбросьте TypeError.
  • Не изменяйте исходный frame. Для промежуточных преобразований используйте копию нужных колонок.
  • Разрешите только операции "count", "sum", "mean", "min" и "max". Для другого значения выбросьте ValueError.
  • Колонка group_by должна существовать. Иначе выбросьте KeyError.
  • Во всех операциях группируйте с параметрами dropna=False и sort=True.
  • Для операции "count" параметр value_column не используется. Подсчитайте число строк в каждой группе и назовите итоговую колонку count.
  • Для "sum", "mean", "min" и "max" необходимо передать value_column. Если значение не задано, выбросьте ValueError.
  • Если value_column отсутствует в таблице, выбросьте KeyError.
  • Преобразуйте значения выбранной колонки через pd.to_numeric(errors="coerce").
  • Удалите строки, в которых после преобразования нет числового значения.
  • Если после очистки не осталось ни одного числа, выбросьте ValueError.
  • Выполните выбранную агрегацию и назовите итоговую колонку по шаблону operation_value_column, например mean_hours.
  • Верните новый DataFrame с колонкой группировки и колонкой результата.

Пример результата

Для строк b:4, a:2 и a:bad нечисловое значение исключается из вычисления:

result = aggregate_table(
    frame, "category", "mean", "hours"
)
print(result.to_dict("records"))

Результат:

[{'category': 'a', 'mean_hours': 2.0}, {'category': 'b', 'mean_hours': 4.0}]

Импорт pandas как pd уже находится в заготовке. Проверка сравнивает содержимое возвращённого DataFrame.

Правильный ответ 
import pandas as pd

def aggregate_table(frame, group_by: str, operation: str, value_column: str | None = None):
    if not isinstance(frame, pd.DataFrame):
        raise TypeError

    allowed_operations = {"count", "sum", "mean", "min", "max"}

    if operation not in allowed_operations:
        raise ValueError

    if group_by not in frame.columns:
        raise KeyError(group_by)

    if operation == "count":
        data = frame[[group_by]].copy()

        return (
            data.groupby(group_by, dropna=False, sort=True)
            .size()
            .reset_index(name="count")
        )

    if value_column is None:
        raise ValueError

    if value_column not in frame.columns:
        raise KeyError(value_column)

    data = frame[[group_by, value_column]].copy()

    data[value_column] = pd.to_numeric(
        data[value_column],
        errors="coerce",
    )

    data = data.dropna(subset=[value_column])

    if data.empty:
        raise ValueError

    result_column = f"{operation}_{value_column}"

    result = (
        data.groupby(group_by, dropna=False, sort=True)[value_column]
        .agg(operation)
        .reset_index(name=result_column)
    )

    return result