arm-tabular-parser/api/src/parser/osv.py

"""Конвейер листа ОСВ: лист → документ `OsvDocument`, предупреждения, признак «контрольные суммы сошлись».

Книгу (лимиты, чтение, дедлайн, вид листа, статус и карта ошибок) разбирает `book.BookParser`. Здесь: заголовок
таблицы → шапка и схема группировок → строки → каскад иерархии → классификация → контрольные суммы §1.4. Имя листа
только подставляется в `locator` и сообщения, поиск по нему не ведётся.
"""

import datetime
from collections.abc import Iterable, Sequence
from dataclasses import replace

from src.contract import OsvDocument
from src.contract.result import OsvChecks, OsvGroup, OsvHeader, OsvRow
from src.parser.checks import run_checks
from src.parser.classify import SUM_FIELDS, ClassifiedRow, classify_rows, collect_body_rows
from src.parser.deadline import Deadline
from src.parser.detect import TableHeader, find_table_header
from src.parser.errors import InvalidFormatError
from src.parser.grid import Sheet, column_letter
from src.parser.header import ACCOUNT_NOT_FOUND_WARNING, DocHeader, build_group_levels, parse_doc_header
from src.parser.hierarchy import resolve_levels
from src.parser.inn_rows import glue_inn_rows
from src.parser.rules import Rules

__all__ = ["OsvParser", "data_rows", "sheet_line", "unique"]

LOCATOR_PREFIX = "sheet:"
INN_COLUMN_KEY = "inn"
NO_DATA_ROWS_WARNING = "нет строк данных"


def sheet_line(sheet: Sheet, text: str) -> str:
    return f"sheet={sheet.name} {text}"


def unique(items: Iterable[str]) -> list[str]:
    """Без повторов, в порядке первого появления."""
    return list(dict.fromkeys(items))


def _columns(header: TableHeader) -> dict[str, str]:
    """Диагностика `columns`: шесть колонок сумм и, если найдена, колонка ИНН (буквы — только здесь)."""
    columns = header.columns.as_diagnostics()
    if header.inn_col is not None:
        columns[INN_COLUMN_KEY] = column_letter(header.inn_col)
    return columns


def _from_year_start(start: datetime.date | None) -> bool | None:
    return None if start is None else (start.month, start.day) == (1, 1)


def _header(doc: DocHeader) -> OsvHeader:
    return OsvHeader(
        organization_name=doc.organization_name,
        organization_inn=None,
        account_code_raw=doc.account_code_raw,
        period_text=doc.period_text,
        period_year=doc.period_year,
        period_quarter=doc.period_quarter,
        is_cumulative=doc.is_cumulative,
        period_start=doc.period_start,
        period_end=doc.period_end,
        is_from_year_start=_from_year_start(doc.period_start),
        data_type=doc.data_type,
        currency=doc.currency,
        header_variant=doc.header_variant,
    )


def _row(item: ClassifiedRow) -> OsvRow:
    sums = dict(zip(SUM_FIELDS, item.body.sums, strict=True))
    return OsvRow(
        row_number=item.body.row,
        level=item.level,
        parent_row_number=item.parent,
        row_value=item.body.text,
        group_type=item.group_type,
        group_name=item.group_name,
        account_code_raw=item.account_code_raw,
        counterparty_name=item.counterparty_name,
        counterparty_inn=item.counterparty_inn,
        data_type=item.data_type,
        opening_debit=sums["opening_debit"],
        opening_credit=sums["opening_credit"],
        turnover_debit=sums["turnover_debit"],
        turnover_credit=sums["turnover_credit"],
        closing_debit=sums["closing_debit"],
        closing_credit=sums["closing_credit"],
        extra=item.extra,
    )


def _with_table_account(doc: DocHeader, rows: Sequence[ClassifiedRow]) -> DocHeader:
    """Код счёта шапки, которой нет («по счету …» в названии нет), — код первой строки-счёта таблицы.

    «Счет: Код / Наименование»: код стоит в строке таблицы, а не в названии ведомости. Предупреждение «код счёта в
    шапке не найден» заменяется пометкой, откуда код взят.
    """
    if doc.account_code_raw:
        return doc
    accounts = (item.account_code_raw for item in rows if item.group_type == "ACCOUNT")
    code = next((code for code in accounts if code), None)
    if code is None:
        return doc
    warnings = [warning for warning in doc.warnings if warning != ACCOUNT_NOT_FOUND_WARNING]
    warnings.append(f"код счёта «{code}» взят из первой строки-счёта таблицы: в шапке «по счету …» нет")
    return replace(doc, account_code_raw=code, warnings=tuple(warnings))


def _is_ok(checks: OsvChecks) -> bool:
    return (
        bool(checks.total_row_matches_root)
        and not checks.balance_identity_violations
        and not (checks.parent_sum_violations)
    )


def data_rows(document: OsvDocument) -> int:
    return sum(1 for row in document.rows if row.group_type != "TOTAL")


def _invalid_format(sheet: Sheet, step: str, exc: InvalidFormatError) -> InvalidFormatError:
    """Сообщение `INVALID_FORMAT` с листом и шагом; префикс «лист «…»:» детектора не дублируется."""
    reason = exc.message.removeprefix(f"лист «{sheet.name}»: ")
    return InvalidFormatError(f"лист «{sheet.name}», шаг «{step}»: {reason}")


class OsvParser:
    """Конвейер листа ОСВ по словарям `rules`; книгу разбирает `BookParser`."""

    def __init__(self, rules: Rules) -> None:
        self._rules = rules

    def parse_sheet(self, sheet: Sheet, deadline: Deadline) -> tuple[OsvDocument, list[str], bool]:
        """Документ листа ОСВ, его предупреждения (первое — `sheet=<имя> hierarchy=<стратегия>`) и признак
        «контрольные суммы сошлись»; нет заголовка/колонок/иерархии — `InvalidFormatError` с листом и шагом.
        """
        rules = self._rules
        step = "заголовок таблицы"
        try:
            header = find_table_header(sheet, rules, deadline)
            step = "шапка документа"
            doc = parse_doc_header(sheet, header, rules)
            schema = build_group_levels(sheet, header, rules)
            deadline.check(f"лист «{sheet.name}»: строки таблицы")
            step = "строки таблицы"
            body, total_row, body_warnings = collect_body_rows(sheet, header, rules, deadline)
            step = "иерархия строк"
            levels = resolve_levels(body, total_row, deadline)
            step = "строки ИНН"
            glue = glue_inn_rows(
                body,
                levels,
                total_row,
                sheet,
                inn_col=header.inn_col,
                schema=schema.groups,
                hint=schema.inn_hint,
                rules=rules,
            )
            step = "классификация строк"
            classified = classify_rows(
                glue.rows,
                glue.levels,
                schema.groups,
                header,
                sheet,
                doc.data_type,
                total_row,
                rules,
                inns=glue.inns,
                unpaired=glue.unpaired,
            )
        except InvalidFormatError as exc:
            raise _invalid_format(sheet, step, exc) from exc
        doc = _with_table_account(doc, classified)
        deadline.check(f"лист «{sheet.name}»: контрольные суммы")
        checks, check_warnings = run_checks(classified, rules.checks)
        document = self._document(
            sheet=sheet, header=header, doc=doc, schema=schema.groups, rows=classified, checks=checks
        )
        warnings = [sheet_line(sheet, f"hierarchy={levels.strategy}"), *doc.warnings, *body_warnings, *glue.warnings]
        if not any(item.group_type != "TOTAL" for item in classified):
            warnings.append(NO_DATA_ROWS_WARNING)
        warnings.extend(check_warnings)
        # Строка ИНН без пары, похожая на ИНН строка или расхождение с колонкой (#35) — needs_check
        return document, unique(warnings), _is_ok(checks) and glue.ok

    # --- внутреннее ---

    @staticmethod
    def _document(  # noqa: PLR0913 - части документа собираются в parse_sheet и передаются явно
        *,
        sheet: Sheet,
        header: TableHeader,
        doc: DocHeader,
        schema: list[OsvGroup],
        rows: list[ClassifiedRow],
        checks: OsvChecks,
    ) -> OsvDocument:
        return OsvDocument(
            document_type="OSV",
            locator=f"{LOCATOR_PREFIX}{sheet.name}",
            header=_header(doc),
            group_schema=schema,
            columns=_columns(header),
            rows=[_row(item) for item in rows],
            checks=checks,
        )