arm-tabular-parser/api/tests/unit/parser/test_book_parser.py

"""Оркестратор `BookParser` (design D-6, D-8; spec «Листы ОСВ находятся по содержимому», «Контрольные суммы определяют
статус», «Форматы, лимиты, время», «Словари — данные, не код»).

Книги собираются openpyxl в памяти: лист описывается строками сверху вниз, как в `gridbuilder` (`M` — объединённая
ячейка, `L` — outline level строки первым элементом). Позиции колонок произвольные — парсер находит их по тексту.
"""

import datetime
import io
import itertools
import shutil
import time
import traceback
from collections.abc import Callable, Iterator, Sequence
from decimal import Decimal
from pathlib import Path
from typing import Any

import openpyxl
import pytest
from gridbuilder import L, M, sheet_from_rows
from loguru import logger
from result_contract import assert_valid_result

from src.contract import ErrorCode, OsvDocument, ResultStatus, StatementDocument
from src.contract.result import StatementHeader
from src.parser import ParseOutcome, hierarchy
from src.parser.book import BookParser, StatementSheetParser
from src.parser.classify import SUM_FIELDS
from src.parser.deadline import Deadline
from src.parser.detect import SheetKind
from src.parser.errors import ParseTimeoutError
from src.parser.grid import Sheet
from src.parser.hierarchy import BodyRow
from src.parser.osv import OsvParser
from src.parser.rules import DEFAULT_RULES_DIR, HEADERS_FILE, Rules, load_rules
from src.parser.statements.statement import StatementParser

OPENING = "Сальдо на начало периода"
TURNOVER = "Обороты за период"
CLOSING = "Сальдо на конец периода"
TITLE = "Оборотно-сальдовая ведомость по счету 60 за 1 квартал 2025 г."
MAX_BYTES = 50 * 1024 * 1024
TIMEOUT_SECONDS = 120
# Строки мини-ОСВ: шапка 6 строк, блок заголовка 2 строки → данные с 9-й
ACCOUNT_ROW, SUPPLIER_ROW, CONTRACTOR_ROW, TOTAL_ROW = 9, 10, 11, 12
DATA_ROWS = 3
# Часы-заглушка: каждый вызов +100 с; лимит 150 с истекает на второй проверке дедлайна
CLOCK_STEP = 100
SHORT_TIMEOUT = 150

Cells = list[object]


@pytest.fixture(scope="module")
def rules() -> Rules:
    return load_rules()


def xlsx(sheets: dict[str, Sequence[Cells]]) -> bytes:
    """Книга XLSX: имя листа → строки (значение, `M` — merged-диапазон, `L` — outline level первым элементом)."""
    workbook = openpyxl.Workbook()
    default = workbook.active
    assert default is not None
    workbook.remove(default)
    for name, rows in sheets.items():
        sheet = workbook.create_sheet(name)
        for row_index, cells in enumerate(rows, start=1):
            col = 1
            for value in cells:
                if isinstance(value, L):
                    sheet.row_dimensions[row_index].outline_level = value.level
                    continue
                if isinstance(value, M):
                    sheet.cell(row_index, col, value.value)
                    sheet.merge_cells(
                        start_row=row_index,
                        start_column=col,
                        end_row=row_index + value.height - 1,
                        end_column=col + value.width - 1,
                    )
                    col += value.width
                    continue
                if value is not None:
                    sheet.cell(row_index, col, value)
                col += 1
    buffer = io.BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()


def head(opening: str = OPENING, title: str = TITLE) -> list[Cells]:
    """Шапка варианта 1 и блок заголовка «Счет | начало | обороты | конец», группировки Счет / Контрагенты (с ИНН)."""
    return [
        ['АО "Ромашка"'],  # 1
        [title],  # 2
        [],  # 3
        ["Выводимые данные: БУ (данные бухгалтерского учета)"],  # 4
        ["Параметры:", None, "Валюта БУ, НУ: RUB"],  # 5
        [],  # 6
        ["Счет", None, M(opening, 2), M(TURNOVER, 2), M(CLOSING, 2)],  # 7
        ["Контрагенты", "ИНН", "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"],  # 8
    ]


def data(level: int, text: str, sums: Sequence[object], inn: object = None) -> Cells:
    return [L(level), text, inn, *sums]


ROOT = (100, None, 50, 30, 120, None)
SUPPLIER = (60, None, 20, 10, 70, None)
CONTRACTOR = (40, None, 30, 20, 50, None)


def mini_osv(supplier: Sequence[object] = SUPPLIER, *, total: bool = True, opening: str = OPENING) -> list[Cells]:
    rows = [
        *head(opening),
        data(0, "60", ROOT),  # 9
        data(1, "ООО Поставщик", supplier, inn="7701234567"),  # 10
        data(1, "ООО Подрядчик", CONTRACTOR, inn="7707654321"),  # 11
    ]
    if total:
        rows.append(data(0, "Итого", ROOT))  # 12
    return rows


def statement() -> list[Cells]:
    return [["Бухгалтерский баланс"], ["на 31 марта 2025 г."], ["Актив", "Код", "На 31 марта 2025 г."]]


def book(
    rules: Rules,
    *,
    statements: StatementSheetParser | None = None,
    max_bytes: int = MAX_BYTES,
    timeout_seconds: float = TIMEOUT_SECONDS,
    clock: Callable[[], float] = time.monotonic,
) -> BookParser:
    return BookParser(
        rules,
        osv=OsvParser(rules),
        statements=statements or StatementParser(rules),
        max_bytes=max_bytes,
        timeout_seconds=timeout_seconds,
        clock=clock,
    )


def only_document(outcome: ParseOutcome) -> OsvDocument:
    (document,) = outcome.documents
    assert isinstance(document, OsvDocument)
    return document


def warnings_of(outcome: ParseOutcome) -> list[str]:
    assert outcome.metrics is not None
    assert outcome.metrics.warnings is not None
    return outcome.metrics.warnings


# --- Штатный разбор ---


def test_mini_osv_is_done_with_full_fields(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")

    assert outcome.status is ResultStatus.DONE
    assert outcome.error is None
    document = only_document(outcome)
    assert document.document_type == "OSV"
    assert document.locator == "sheet:ОСВ 60"
    header = document.header
    assert (header.organization_name, header.account_code_raw, header.period_text) == (
        'АО "Ромашка"',
        "60",
        "1 квартал 2025 г.",
    )
    assert (header.period_year, header.period_quarter, header.is_cumulative) == (2025, "Q1", True)
    assert (header.data_type, header.currency, header.header_variant) == ("BU", "RUB", 1)
    assert [(group.type, group.name, group.has_inn) for group in document.group_schema] == [
        ("ACCOUNT", "Счет", None),
        ("COUNTERPARTY", "Контрагенты", True),
    ]
    assert document.columns == {
        "opening_debit": "C",
        "opening_credit": "D",
        "turnover_debit": "E",
        "turnover_credit": "F",
        "closing_debit": "G",
        "closing_credit": "H",
        "inn": "B",
    }
    rows = document.rows
    assert [(row.row_number, row.level, row.parent_row_number, row.group_type) for row in rows] == [
        (ACCOUNT_ROW, 0, None, "ACCOUNT"),
        (SUPPLIER_ROW, 1, ACCOUNT_ROW, "COUNTERPARTY"),
        (CONTRACTOR_ROW, 1, ACCOUNT_ROW, "COUNTERPARTY"),
        (TOTAL_ROW, 0, None, "TOTAL"),
    ]
    supplier = rows[1]
    assert (supplier.row_value, supplier.counterparty_name, supplier.counterparty_inn) == (
        "ООО Поставщик",
        "ООО Поставщик",
        "7701234567",
    )
    assert (supplier.group_name, supplier.account_code_raw, supplier.data_type) == ("Контрагенты", "60", "BU")
    sums = [getattr(supplier, name) for name in SUM_FIELDS]
    assert sums == [Decimal(60), None, Decimal(20), Decimal(10), Decimal(70), None]
    assert document.checks is not None
    assert document.checks.total_row_matches_root is True
    assert document.checks.balance_identity_violations == []
    assert document.checks.parent_sum_violations == []
    assert outcome.metrics is not None
    assert outcome.metrics.rows == DATA_ROWS
    assert outcome.metrics.duration_ms is not None
    assert outcome.metrics.duration_ms >= 0
    assert warnings_of(outcome) == ["sheet=ОСВ 60 hierarchy=outline"]
    payload = assert_valid_result(outcome)
    assert payload["schema_version"] == "1.2"


def test_spoiled_sum_needs_check_with_row_numbers(rules: Rules) -> None:
    """Spec «Испорченная сумма»: `closing_debit` контрагента +1 → тождество в его строке и сумма родителя."""
    spoiled = (60, None, 20, 10, 71, None)

    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(spoiled)}), "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK
    checks = only_document(outcome).checks
    assert checks is not None
    assert checks.balance_identity_violations == [SUPPLIER_ROW]
    assert checks.parent_sum_violations == [ACCOUNT_ROW]
    assert checks.total_row_matches_root is True
    warnings = warnings_of(outcome)
    assert warnings[0] == "sheet=ОСВ 60 hierarchy=outline"
    assert any(f"строки {SUPPLIER_ROW}" in warning for warning in warnings), warnings
    assert any(f"строки {ACCOUNT_ROW}" in warning for warning in warnings), warnings
    assert_valid_result(outcome)


def test_period_bounds_reach_osv_header(rules: Rules) -> None:
    """Контракт v1.2: границы периода и признак «с начала года» заполнены парсером ОСВ."""
    header = only_document(book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")).header

    assert (header.period_start, header.period_end, header.is_from_year_start) == (
        datetime.date(2025, 1, 1),
        datetime.date(2025, 3, 31),
        True,
    )


def test_period_bounds_reach_osv_header_not_from_year_start(rules: Rules) -> None:
    title = TITLE.replace("1 квартал", "3 квартал")
    rows = [*head(title=title), *mini_osv()[len(head()) :]]

    header = only_document(book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")).header

    assert (header.period_start, header.period_end, header.is_from_year_start) == (
        datetime.date(2025, 7, 1),
        datetime.date(2025, 9, 30),
        False,
    )


def test_missing_total_needs_check(rules: Rules) -> None:
    """R-20: нет «Итого» — сверку с корнем выполнить нельзя: `false`, предупреждение, `needs_check`."""
    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(total=False)}), "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK
    checks = only_document(outcome).checks
    assert checks is not None
    assert checks.total_row_matches_root is False
    assert "строка «Итого» не найдена" in warnings_of(outcome)


def test_empty_osv_is_done_with_total_only(rules: Rules) -> None:
    """Spec «Пустая ОСВ» (R-13): заголовок и сразу «Итого» без сумм."""
    outcome = book(rules).parse(xlsx({"ОСВ 66": [*head(), [None, "Итого"]]}), "XLSX")

    assert outcome.status is ResultStatus.DONE
    document = only_document(outcome)
    assert [(row.group_type, row.level, row.parent_row_number) for row in document.rows] == [("TOTAL", 0, None)]
    assert document.checks is not None
    assert document.checks.total_row_matches_root is True
    assert outcome.metrics is not None
    assert outcome.metrics.rows == 0
    assert warnings_of(outcome) == ["sheet=ОСВ 66 hierarchy=trivial", "нет строк данных"]
    assert_valid_result(outcome)


def test_number_warnings_carry_row_numbers(rules: Rules) -> None:
    rows = mini_osv()
    rows[CONTRACTOR_ROW - 1] = data(1, "ООО Подрядчик", (40, None, 30, 20, 50, "н/д"), inn="7707654321")

    outcome = book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")

    assert outcome.status is ResultStatus.DONE
    assert any(warning.startswith(f"строка {CONTRACTOR_ROW}:") for warning in warnings_of(outcome))


# --- Шапка без меток и код/наименование счёта в двух колонках (issue #11 п. 4, OQ-134) ---

NO_LABEL_ORG_SUMS = (None, 600, 100, None, None, 500)
NO_LABEL_SUPPLIER = (None, 400, None, None, None, 400)
NO_LABEL_CONTRACTOR = (None, 200, 100, None, None, 100)


def code_name_osv(*, total_label: str | None = None) -> list[Cells]:
    """ОСВ 63 Новоросметалл: «Бухгалтерский | период» отдельной строкой, организация — в названии ведомости, колонки
    «Счет: Код / Наименование», контрагенты — в колонке правее кода (уровень — сдвиг колонки), «Итого» без подписи.
    """
    blank = [None] * 3
    sums_head = ["Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"]

    def line(label_cells: Sequence[object], sums: Sequence[object]) -> Cells:
        return [*label_cells, *sums]

    return [
        ["Оборотно-сальдовая ведомость НОВОРОСМЕТАЛЛ ООО"],  # 1
        [],  # 2
        ["Бухгалтерский", *([None] * 10), "9 месяцев 2025 г."],  # 3
        ["Счет", *([None] * 5), M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)],  # 4
        ["Код", *blank, "Наименование", None, *sums_head],  # 5
        line([63, *blank, "Резервы по сомнительным долгам", None], NO_LABEL_ORG_SUMS),  # 6
        line([None, "ООО Поставщик", None, None, None, None], NO_LABEL_SUPPLIER),  # 7
        line([None, "ООО Подрядчик", None, None, None, None], NO_LABEL_CONTRACTOR),  # 8
        line([total_label, None, None, None, None, None], NO_LABEL_ORG_SUMS),  # 9
    ]


def test_header_without_labels_and_code_name_columns_are_found(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Лист1": code_name_osv()}), "XLSX")

    document = only_document(outcome)
    header = document.header
    assert header.organization_name == "НОВОРОСМЕТАЛЛ ООО"
    assert header.account_code_raw == "63"
    assert (header.period_text, header.period_year, header.period_quarter, header.is_cumulative) == (
        "9 месяцев 2025 г.",
        2025,
        "Q3",
        True,
    )
    assert header.data_type == "BU"
    warnings = warnings_of(outcome)
    assert not any("период" in warning and "не найден" in warning for warning in warnings), warnings
    assert not any("тип выводимых данных" in warning for warning in warnings), warnings


def test_code_name_columns_rows_are_account_and_children_by_label_column(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Лист1": code_name_osv(total_label="Итого")}), "XLSX")

    document = only_document(outcome)
    assert [(row.row_value, row.level, row.group_type) for row in document.rows] == [
        ("63", 0, "ACCOUNT"),
        ("ООО Поставщик", 1, "OTHER"),
        ("ООО Подрядчик", 1, "OTHER"),
        ("Итого", 0, "TOTAL"),
    ]
    assert document.rows[0].extra == {"Наименование": "Резервы по сомнительным долгам"}
    assert outcome.status is ResultStatus.DONE


def test_total_row_without_caption_is_the_last_row_of_sums(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Лист1": code_name_osv()}), "XLSX")

    document = only_document(outcome)
    assert document.rows[-1].group_type == "TOTAL"
    assert document.checks is not None
    assert document.checks.total_row_matches_root is True
    assert outcome.status is ResultStatus.DONE
    assert any("без подписи" in warning for warning in warnings_of(outcome)), warnings_of(outcome)


# --- Несколько листов ---


def test_statement_sheet_goes_to_real_pipeline_next_to_osv(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Баланс": statement(), "ОСВ 60": mini_osv()}), "XLSX")

    assert [type(document).__name__ for document in outcome.documents] == ["OsvDocument", "StatementDocument"]
    assert outcome.documents[0].locator == "sheet:ОСВ 60" and outcome.documents[1].locator == "sheet:Баланс"
    assert "sheet=ОСВ 60 hierarchy=outline" in warnings_of(outcome)
    assert_valid_result(outcome)


def test_two_osv_sheets_and_other_sheet(rules: Rules) -> None:
    """Spec «Два листа ОСВ и посторонний лист»: два документа в порядке листов, пропуск «Справки» в warnings."""
    other: list[Cells] = [["Справка о дебиторской задолженности"], ["Контрагент", "Сумма"], ["ООО Лютик", 5]]
    content = xlsx({"ОСВ 60": mini_osv(), "Справка": other, "ОСВ 62": mini_osv()})

    outcome = book(rules).parse(content, "XLSX")

    assert outcome.status is ResultStatus.DONE
    assert [document.locator for document in outcome.documents] == ["sheet:ОСВ 60", "sheet:ОСВ 62"]
    assert outcome.metrics is not None
    assert outcome.metrics.rows == 2 * DATA_ROWS
    warnings = warnings_of(outcome)
    assert "sheet=ОСВ 60 hierarchy=outline" in warnings
    assert "sheet=ОСВ 62 hierarchy=outline" in warnings
    assert any(warning.startswith("sheet=Справка skipped") for warning in warnings), warnings
    assert_valid_result(outcome)


def test_one_spoiled_sheet_makes_whole_result_needs_check(rules: Rules) -> None:
    spoiled = (60, None, 20, 10, 71, None)
    content = xlsx({"ОСВ 60": mini_osv(), "ОСВ 62": mini_osv(spoiled)})

    outcome = book(rules).parse(content, "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK
    assert len(outcome.documents) == 2  # noqa: PLR2004 - два листа ОСВ


# --- Ошибки (карта D-6) ---


def test_book_without_osv_is_invalid_document(rules: Rules) -> None:
    """Spec «Нет ОСВ»: сообщение называет число просмотренных листов."""
    outcome = book(rules).parse(
        xlsx({"Справка": [["Наименование", "Сумма"], ["Аренда", 100]], "Данные": [["a", "b"], [1, 2]]}), "XLSX"
    )

    assert outcome.status is ResultStatus.ERROR
    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
    assert "листов просмотрено 2" in outcome.error.message
    assert outcome.documents == ()


def test_osv_sheet_without_table_header_is_invalid_format(rules: Rules) -> None:
    rows: list[Cells] = [['АО "Ромашка"'], [TITLE], [], ["Счет", "Сумма"], ["60", 100]]

    outcome = book(rules).parse(xlsx({"ОСВ 60": rows}), "XLSX")

    assert outcome.status is ResultStatus.ERROR
    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.INVALID_FORMAT
    assert "«ОСВ 60»" in outcome.error.message
    assert "заголовок таблицы" in outcome.error.message
    assert outcome.error.message.count("ОСВ 60") == 1


def test_undetermined_hierarchy_is_invalid_format_with_sheet_and_step(
    rules: Rules, monkeypatch: pytest.MonkeyPatch
) -> None:
    def nothing(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> None:
        return None

    monkeypatch.setattr(hierarchy, "STRATEGIES", (("nothing", nothing),))

    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.INVALID_FORMAT
    assert "«ОСВ 60»" in outcome.error.message
    assert "иерархия" in outcome.error.message


def test_size_over_limit_is_unsupported_format_before_reading(rules: Rules) -> None:
    """Spec «Превышен лимит»: на байт больше лимита — `UNSUPPORTED_FORMAT`, ровно лимит — разбор."""
    content = xlsx({"ОСВ 60": mini_osv()})

    over = book(rules, max_bytes=len(content) - 1).parse(content, "XLSX")
    exact = book(rules, max_bytes=len(content)).parse(content, "XLSX")

    assert over.status is ResultStatus.ERROR
    assert over.error is not None
    assert over.error.code is ErrorCode.UNSUPPORTED_FORMAT
    assert "лимит" in over.error.message
    assert exact.status is ResultStatus.DONE


def test_tiny_limit_rejects_any_file(rules: Rules) -> None:
    outcome = book(rules, max_bytes=10).parse(b"PK\x03\x04" + b"x" * 7, "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT


def test_unknown_signature_is_unsupported_format(rules: Rules) -> None:
    outcome = book(rules).parse(b"%PDF-1.7 not a workbook", "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT


def test_broken_zip_is_unreadable(rules: Rules) -> None:
    outcome = book(rules).parse(b"PK\x03\x04 broken archive", "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.UNREADABLE


def test_deadline_exceeded_is_timeout(rules: Rules) -> None:
    clock = itertools.count(step=CLOCK_STEP)

    outcome = book(rules, timeout_seconds=SHORT_TIMEOUT, clock=lambda: float(next(clock))).parse(
        xlsx({"ОСВ 60": mini_osv()}), "XLSX"
    )

    assert outcome.status is ResultStatus.ERROR
    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.TIMEOUT
    assert "превышен лимит" in outcome.error.message


def test_unexpected_exception_is_recognition_failed_in_one_line(rules: Rules, monkeypatch: pytest.MonkeyPatch) -> None:
    def broken(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
        raise ValueError("сбой стратегии\nвторая строка")

    monkeypatch.setattr(hierarchy, "STRATEGIES", (("broken", broken),))

    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")

    assert outcome.status is ResultStatus.ERROR
    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.RECOGNITION_FAILED
    assert outcome.error.message == "ValueError: сбой стратегии вторая строка"


@pytest.fixture
def log_records() -> Iterator[list[Any]]:
    """Записи loguru, выпущенные за время теста."""
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
    yield records
    logger.remove(handler_id)


def test_unexpected_exception_is_logged_with_traceback(
    rules: Rules, monkeypatch: pytest.MonkeyPatch, log_records: list[Any]
) -> None:
    """I-2: ответ прежний (RECOGNITION_FAILED одной строкой), но стек непредвиденной ошибки уходит в лог."""

    def broken(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
        raise ValueError("сбой стратегии\nвторая строка")

    monkeypatch.setattr(hierarchy, "STRATEGIES", (("broken", broken),))

    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv()}), "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.RECOGNITION_FAILED
    assert outcome.error.message == "ValueError: сбой стратегии вторая строка"
    (record,) = [record for record in log_records if record["level"].name == "ERROR"]
    assert "Непредвиденная ошибка разбора" in record["message"]
    assert "ValueError: сбой стратегии вторая строка" in record["message"]
    exception = record["exception"]
    assert exception is not None
    assert exception.type is ValueError
    assert exception.traceback is not None
    assert "broken" in "".join(traceback.format_tb(exception.traceback))


def test_unreadable_book_is_logged_as_warning_without_traceback(rules: Rules, log_records: list[Any]) -> None:
    outcome = book(rules).parse(b"PK\x03\x04 broken archive", "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.UNREADABLE
    (record,) = [record for record in log_records if record["level"].name == "WARNING"]
    assert outcome.error.message in record["message"]
    assert record["exception"] is None
    assert not [record for record in log_records if record["level"].name == "ERROR"]


def test_expected_errors_other_than_unreadable_are_not_logged(rules: Rules, log_records: list[Any]) -> None:
    outcome = book(rules).parse(b"%PDF-1.7", "XLSX")

    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT
    assert not [record for record in log_records if record["level"].no >= logger.level("WARNING").no]


def test_header_search_is_interrupted_by_deadline(rules: Rules) -> None:
    """R-23 (в): дедлайн из `parse_sheet` доходит до поиска заголовка таблицы — цикла по всему листу."""
    sheet = sheet_from_rows([[TITLE], *([["строка"]] * 1000)])
    deadline = Deadline(1, clock=itertools.chain([0.0], itertools.repeat(1e9)).__next__)

    with pytest.raises(ParseTimeoutError, match="поиск заголовка таблицы"):
        OsvParser(rules).parse_sheet(sheet, deadline)


# --- Словари — данные (R-8, spec «Новый синоним без релиза кода») ---

NEW_OPENING = "Нач. остаток"
NEW_SYNONYM_LINE = "    - 'нач\\.\\s*остаток'\n"


def rules_with_new_opening_synonym(tmp_path: Path) -> Path:
    """Копия штатного каталога словарей, в `osv_headers.yaml` синоним «Нач. остаток» добавлен к группе opening."""
    copy = tmp_path / "rules"
    shutil.copytree(DEFAULT_RULES_DIR, copy)
    headers = copy / HEADERS_FILE
    text = headers.read_text(encoding="utf-8")
    anchor = "  opening:\n"
    assert text.count(anchor) == 1
    headers.write_text(text.replace(anchor, anchor + NEW_SYNONYM_LINE), encoding="utf-8")
    return copy


def test_new_synonym_parses_without_code_changes(tmp_path: Path, rules: Rules) -> None:
    content = xlsx({"ОСВ 60": mini_osv(opening=NEW_OPENING)})

    outcome = book(load_rules(rules_with_new_opening_synonym(tmp_path))).parse(content, "XLSX")

    assert outcome.status is ResultStatus.DONE
    document = only_document(outcome)
    assert document.columns is not None
    assert (document.columns["opening_debit"], document.columns["opening_credit"]) == ("C", "D")
    assert document.rows[0].opening_debit == Decimal(100)
    assert_valid_result(outcome)


def test_default_dictionary_does_not_know_the_new_synonym(rules: Rules) -> None:
    """До правки словаря «Нач. остаток» не распознаётся: заголовок находится по «оборотам» и «концу» (начало
    необязательно, R-12), колонки начала уходят в `extra`, `opening_*` = null, и тождество сальдо не сходится в
    строках с ненулевым началом → `needs_check` (фактическое поведение, зафиксировано тестом).
    """
    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(opening=NEW_OPENING)}), "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK
    document = only_document(outcome)
    assert document.columns is not None
    assert "opening_debit" not in document.columns
    assert all(row.opening_debit is None and row.opening_credit is None for row in document.rows)
    assert document.rows[0].extra is not None
    assert "100" in document.rows[0].extra.values()
    checks = document.checks
    assert checks is not None
    assert checks.balance_identity_violations == [ACCOUNT_ROW, SUPPLIER_ROW, CONTRACTOR_ROW, TOTAL_ROW]


# --- Формы (задача 7, оркестрация книги) ---

BALANCE = [["Бухгалтерский баланс"], ["Код", "На 31.12.2024"], ["1600", 1]]
CONTINUATION = [["Форма 0710001 с.2"], ["Код", "На 31.12.2024"], ["1700", 1]]


class _FakeStatements:
    """Возвращает по одному документу DISCLOSURE на лист формы и запоминает, что получил."""

    def __init__(self) -> None:
        self.received: list[tuple[str, SheetKind]] = []

    def parse_sheets(
        self, sheets: Sequence[tuple[Sheet, SheetKind]], deadline: Deadline
    ) -> tuple[list[StatementDocument], list[str], bool]:
        self.received = [(sheet.name, kind) for sheet, kind in sheets]
        documents = [
            StatementDocument(
                document_type="DISCLOSURE",
                locator=f"sheet:{sheet.name}",
                header=StatementHeader(),
                columns=[],
                lines=[],
            )
            for sheet, _ in sheets
        ]
        return documents, [f"sheet={sheet.name} statements-fake" for sheet, _ in sheets], True


def test_statement_sheets_go_to_statement_parser_in_sheet_order(rules: Rules) -> None:
    fake = _FakeStatements()
    content = xlsx({"ББ": BALANCE, "Лист_1": mini_osv(), "с.2": CONTINUATION})

    outcome = book(rules, statements=fake).parse(content, "XLSX")

    assert fake.received == [("ББ", SheetKind.STATEMENT), ("с.2", SheetKind.STATEMENT_CONTINUATION)]
    assert [type(document).__name__ for document in outcome.documents] == [
        "OsvDocument",
        "StatementDocument",
        "StatementDocument",
    ]
    assert outcome.status is ResultStatus.DONE
    assert "sheet=ББ statements-fake" in (outcome.metrics.warnings if outcome.metrics else [])


def test_balance_sheet_gives_statement_document_without_stub(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"ББ": BALANCE, "Лист_1": mini_osv()}), "XLSX")

    assert [type(document).__name__ for document in outcome.documents] == ["OsvDocument", "StatementDocument"]
    assert not any("skipped: statement" in warning for warning in warnings_of(outcome))


def test_statement_continuation_is_not_parsed_as_osv(rules: Rules) -> None:
    """R-5: лист-продолжение формы идёт в конвейер форм (здесь заглушка), а не в разбор ОСВ."""
    outcome = book(rules).parse(xlsx({"ОСВ 60": mini_osv(), "с.2": CONTINUATION}), "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK  # лист-продолжение потерян: в нём меньше трёх кодов
    assert [document.locator for document in outcome.documents] == ["sheet:ОСВ 60"]
    assert "sheet=с.2 колонка кодов не найдена" in warnings_of(outcome)


def test_book_without_osv_and_forms_is_invalid_document(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Справка": [["Наименование", "Сумма"], ["Аренда", 100]]}), "XLSX")

    assert outcome.status is ResultStatus.ERROR
    assert outcome.error is not None
    assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
    assert "нет листов ОСВ или форм" in outcome.error.message


def test_statement_not_ok_makes_book_needs_check(rules: Rules) -> None:
    class _NotOk(_FakeStatements):
        def parse_sheets(
            self, sheets: Sequence[tuple[Sheet, SheetKind]], deadline: Deadline
        ) -> tuple[list[StatementDocument], list[str], bool]:
            documents, warnings, _ = super().parse_sheets(sheets, deadline)
            return documents, warnings, False

    outcome = book(rules, statements=_NotOk()).parse(xlsx({"ББ": BALANCE}), "XLSX")

    assert outcome.status is ResultStatus.NEEDS_CHECK
    assert len(outcome.documents) == 1


# --- N-2/N-3: лист с одной кодовой строкой — продолжение только сразу после листа формы ---

BALANCE_ROWS: list[Cells] = [
    ["Бухгалтерский баланс"],
    ["на 31 марта 2025 г."],
    ["Организация", "ООО Ромашка"],
    ["Наименование показателя", "Код", "На 31 марта 2025 г."],
    ["Основные средства", "1150", 100],
    ["Итого по разделу I", "1100", 100],
    ["Запасы", "1210", 50],
    ["Итого по разделу II", "1200", 50],
    ["БАЛАНС", "1600", 150],
]


def test_single_code_sheet_after_osv_sheet_is_not_a_continuation(rules: Rules) -> None:
    """Баланс, лист ОСВ, затем лист с одной строкой 1700: между формой и листом стоит ОСВ — лист пропущен."""
    data_ = xlsx({"ББ": BALANCE_ROWS, "Лист_1": mini_osv(), "Хвост": [["БАЛАНС", "1700", 150]]})

    outcome = book(rules).parse(data_, "XLSX")

    assert "sheet=Хвост skipped: нет признаков ОСВ или формы" in warnings_of(outcome)
    statements = [document for document in outcome.documents if not isinstance(document, OsvDocument)]
    assert len(statements) == 1 and "1700" not in [line.line_code for line in statements[0].lines]


def test_lone_single_code_sheet_is_not_a_form(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"Справка": [["Основные средства", "1150", 100]]}), "XLSX")

    assert outcome.status is ResultStatus.ERROR and outcome.error is not None
    assert outcome.error.code is ErrorCode.INVALID_DOCUMENT


def test_single_code_sheet_right_after_form_joins(rules: Rules) -> None:
    outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "ББ_2": [["БАЛАНС", "1700", 150]]}), "XLSX")

    (document,) = outcome.documents
    assert not isinstance(document, OsvDocument)
    assert document.locator == "sheet:ББ,ББ_2" and [line.line_code for line in document.lines][-1] == "1700"


def test_statement_sheets_are_reported_when_osv_sheet_fails(rules: Rules) -> None:
    """T7.1: лист ОСВ сломан — ответ с ошибкой всё равно называет листы форм, которые остались неразобранными."""
    broken: list[Cells] = [['АО "Ромашка"'], [TITLE], [], ["Счет", "Сумма"], ["60", 100]]

    outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS, "ОСВ 60": broken}), "XLSX")

    assert outcome.status is ResultStatus.ERROR
    assert "sheet=ББ форма не разобрана: книга завершилась ошибкой листа ОСВ «ОСВ 60»" in warnings_of(outcome)