904392ddcb
Фикс-волна по финальному ревью: I-1, M-1, M-4, M-5, M-6 (T4.1, T4.2); тесты на ;;;, 5 секций, [$-419], \(…\), [=5]0;[=5]0,. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
789 lines
36 KiB
Python
789 lines
36 KiB
Python
"""Оркестратор `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
|
||
from tests.synthetic.xlsb_writer import write_xlsb
|
||
|
||
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]] = []
|
||
self.formats_unreadable = False
|
||
|
||
def parse_sheets(
|
||
self, sheets: Sequence[tuple[Sheet, SheetKind]], deadline: Deadline, *, formats_unreadable: bool = False
|
||
) -> tuple[list[StatementDocument], list[str], bool]:
|
||
self.formats_unreadable = formats_unreadable
|
||
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, *, formats_unreadable: bool = False
|
||
) -> tuple[list[StatementDocument], list[str], bool]:
|
||
documents, warnings, _ = super().parse_sheets(sheets, deadline, formats_unreadable=formats_unreadable)
|
||
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)
|
||
|
||
|
||
def test_xlsb_book_marks_formats_unreadable_for_statements_only(rules: Rules) -> None:
|
||
"""#42, R-7: xlsb определяется по содержимому книги (без подмены предупреждений); ОСВ предупреждения не получает."""
|
||
cells = {
|
||
name: {position: cell.value for position, cell in sheet_from_rows(rows, name).cells.items()}
|
||
for name, rows in (("ББ", BALANCE_ROWS), ("ОСВ 60", mini_osv()))
|
||
}
|
||
outcome = book(rules).parse(write_xlsb(cells), "XLSB")
|
||
|
||
xlsb_warnings = [w for w in warnings_of(outcome) if "формат чисел xlsb" in w]
|
||
assert xlsb_warnings == ["sheet=ББ формат чисел xlsb не читается: значения как в файле"]
|
||
assert [type(document).__name__ for document in outcome.documents] == ["OsvDocument", "StatementDocument"]
|
||
|
||
|
||
def test_xlsx_book_has_no_xlsb_format_warning(rules: Rules) -> None:
|
||
outcome = book(rules).parse(xlsx({"ББ": BALANCE_ROWS}), "XLSX")
|
||
|
||
assert not any("формат чисел" in w for w in warnings_of(outcome))
|