Andrey Epifancev 4a16f193af feat(parser): классификация строк по group_schema (§1.3)
classify.py: collect_body_rows — строки таблицы до «Итого» (пустые, из пробелов,
повторы заголовка и строки правее таблицы пропускаются; подвал после «Итого» не данные),
«Итого» по тексту и корневому оформлению до уровней, без требования сумм (R-13, T3.3),
шесть сумм через to_money с номером строки в предупреждениях, свёрнутое сальдо по знаку
(R-1); classify_rows — ACCOUNT по ACCOUNT_RE, позиция в схеме, OTHER за схемой, TOTAL,
COUNTERPARTY с ИНН (inn_digits), Вал./Кол./код валюты -> data_type, extra без
дубликатов ключей и без колонок подвала (T4.5).

Fixtures-тесты иерархии (R-2) и классификации на всех 66 реальных ОСВ.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
2026-10-05 14:07:33 +04:00

607 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Классификация строк ОСВ (spec «Классификация строк», design D-4, `05-domain-rules.md` §1.2–§1.4).
Сетки собираются `gridbuilder`; конвейер теста — тот же, что у парсера: заголовок таблицы → схема группировок →
`collect_body_rows` → каскад иерархии → `classify_rows`. Позиции колонок в сетках произвольные: классификатор знает их
только из `TableHeader`.
"""
from dataclasses import dataclass
from decimal import Decimal
import pytest
from gridbuilder import C, L, M, sheet_from_rows
from src.contract.common import DataType
from src.parser.classify import ACCOUNT_RE, ClassifiedRow, classify_rows, collect_body_rows
from src.parser.deadline import Deadline
from src.parser.detect import find_table_header
from src.parser.errors import ParseTimeoutError
from src.parser.grid import Sheet
from src.parser.header import build_group_schema
from src.parser.hierarchy import TRIVIAL_STRATEGY, Levels, resolve_levels
from src.parser.rules import Rules, load_rules
TITLE = "Оборотно-сальдовая ведомость по счету 60 за 1 квартал 2025 г."
OPENING = "Сальдо на начало периода"
TURNOVER = "Обороты за период"
CLOSING = "Сальдо на конец периода"
TIMEOUT_SECONDS = 60
EXPIRED_AFTER_SECONDS = 1000
ACCOUNT_ROW = 7 # первая строка данных в сетке `osv(...)`: шапка 3 строки + блок заголовка 3 строки
DUPLICATED_EXTRA_COLUMNS = 2
Sums = tuple[object, object, object, object, object, object]
NO_SUMS: Sums = (None, None, None, None, None, None)
@pytest.fixture(scope="module")
def rules() -> Rules:
return load_rules()
@dataclass(frozen=True)
class Parsed:
rows: list[ClassifiedRow]
warnings: list[str]
levels: Levels
total_row: int | None
def by_text(self, text: str) -> list[ClassifiedRow]:
return [row for row in self.rows if row.body.text == text]
def one(self, text: str) -> ClassifiedRow:
found = self.by_text(text)
assert len(found) == 1, (text, [row.body.text for row in self.rows])
return found[0]
def parse(
sheet: Sheet, rules: Rules, doc_data_type: DataType | None = "BU", deadline: Deadline | None = None
) -> Parsed:
deadline = deadline or Deadline(TIMEOUT_SECONDS)
header = find_table_header(sheet, rules)
schema, _ = build_group_schema(sheet, header, rules)
body, total_row, warnings = collect_body_rows(sheet, header, rules, deadline)
levels = resolve_levels(body, total_row, deadline)
rows = classify_rows(body, levels, schema, header, sheet, doc_data_type, total_row, rules)
return Parsed(rows, warnings, levels, total_row)
# --- Сетка: «Счет | ИНН | КПП | начало Дт/Кт | обороты Дт/Кт | конец Дт/Кт», группировки Счет/Контрагенты/Договоры ---
def _head() -> list[list[object]]:
return [
["ООО Ромашка"],
[TITLE],
[],
["Счет", "ИНН", "КПП", M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)], # 4
["Контрагенты", None, None, "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"], # 5
["Договоры"], # 6
]
def r(level: int | None, text: object, sums: Sums = NO_SUMS, inn: object = None, kpp: object = None) -> list[object]:
"""Строка данных: outline `level` (`None` — без группировки Excel), текст, ИНН, КПП и шесть сумм."""
cells: list[object] = [text, inn, kpp, *sums]
return cells if level is None else [L(level), *cells]
def osv(*rows: list[object]) -> Sheet:
return sheet_from_rows([*_head(), *rows])
S100: Sums = (100, None, 50, 30, 120, None)
def _basic() -> Sheet:
return osv(
r(0, "60", S100), # 7
r(1, "60.01", S100), # 8
r(2, "ООО Поставщик", S100, inn="7701234567", kpp="770101001"), # 9
r(3, "Договор № 1 от 01.02.2025", S100), # 10
r(None, "Итого", S100), # 11
)
# --- Регулярка счёта ---
@pytest.mark.parametrize(
("text", "code"),
[
("60", "60"),
("60.01", "60.01"),
("76.07.1", "76.07.1"),
("001", "001"),
("60.1", "60.1"),
("60.01, Расчеты с поставщиками", "60.01"),
("60,Расчеты", "60"),
],
)
def test_account_re_matches_account_codes(text: str, code: str) -> None:
found = ACCOUNT_RE.match(text)
assert found is not None
assert found.group(1) == code
@pytest.mark.parametrize(
"text", ["6", "6001", "60.001", "60.01.1.1", "ООО 60", "60 лет Октября", "Итого", "12.2025 г."]
)
def test_account_re_rejects_non_accounts(text: str) -> None:
assert ACCOUNT_RE.match(text) is None
# --- Группы по позиции в схеме ---
def test_account_subaccount_counterparty_contract(rules: Rules) -> None:
parsed = parse(_basic(), rules)
kinds = [(row.body.text, row.group_type, row.group_name, row.account_code_raw, row.level) for row in parsed.rows]
assert kinds == [
("60", "ACCOUNT", "Счет", "60", 0),
("60.01", "ACCOUNT", "Счет", "60.01", 1),
("ООО Поставщик", "COUNTERPARTY", "Контрагенты", "60.01", 2), # schema[1]
("Договор № 1 от 01.02.2025", "CONTRACT", "Договоры", "60.01", 3), # schema[2]
("Итого", "TOTAL", None, None, 0),
]
assert parsed.levels.strategy == "outline"
def test_parents_follow_tree(rules: Rules) -> None:
parsed = parse(_basic(), rules)
assert [row.parent for row in parsed.rows] == [None, 7, 8, 9, None]
def test_beyond_schema_is_other(rules: Rules) -> None:
sheet = osv(
r(0, "60", S100),
r(1, "ООО Поставщик", S100, inn="7701234567"),
r(2, "Договор № 1", S100),
r(3, "Спецификация 7", S100),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
beyond = parsed.one("Спецификация 7")
assert (beyond.group_type, beyond.group_name, beyond.account_code_raw) == ("OTHER", None, "60")
assert parsed.one("ООО Поставщик").group_type == "COUNTERPARTY"
assert parsed.one("Договор № 1").group_type == "CONTRACT"
def test_account_with_name_after_comma(rules: Rules) -> None:
sheet = osv(
r(0, "60", S100),
r(1, "60.01, Расчеты с поставщиками", S100),
r(2, "ООО Поставщик", S100),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
account = parsed.one("60.01, Расчеты с поставщиками")
assert (account.group_type, account.account_code_raw) == ("ACCOUNT", "60.01")
assert account.body.text == "60.01, Расчеты с поставщиками" # row_value — исходный текст
assert parsed.one("ООО Поставщик").account_code_raw == "60.01"
def test_same_counterparty_under_different_subaccounts(rules: Rules) -> None:
"""Review Focus 3: одноимённые контрагенты различаются родителем и счётом ветки, а не именем."""
sheet = osv(
r(0, "66", S100),
r(1, "66.03", (None, None, 40, 40, None, None)), # 8
r(2, "Юг-Финанс", (None, None, 40, 40, None, None)), # 9
r(1, "66.04", (100, None, 10, -10, 120, None)), # 10
r(2, "Юг-Финанс", (100, None, 10, -10, 120, None)), # 11
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
first, second = parsed.by_text("Юг-Финанс")
assert (first.parent, first.account_code_raw) == (8, "66.03")
assert (second.parent, second.account_code_raw) == (10, "66.04")
# --- Контрагент и ИНН ---
@pytest.mark.parametrize(
("inn", "expected"),
[
("7701234567", "7701234567"),
("ИНН 770123456789", "770123456789"),
(Decimal("123456789"), "0123456789"), # число с потерянным ведущим нулём
("№ 12 от 01.02.2025", None),
("12345", None),
(None, None),
],
)
def test_counterparty_inn_from_inn_column(rules: Rules, inn: object, expected: str | None) -> None:
sheet = osv(
r(0, "60", S100),
r(1, "ООО Поставщик", S100, inn=inn),
r(1, "ООО Второй", (None, None, None, None, None, None), inn="7702222222"),
r(1, "ООО Третий", (None, None, None, None, None, None), inn="7703333333"),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
counterparty = parsed.one("ООО Поставщик")
assert counterparty.group_type == "COUNTERPARTY"
assert counterparty.counterparty_name == "ООО Поставщик"
assert counterparty.counterparty_inn == expected
def test_counterparty_fields_only_on_counterparty_rows(rules: Rules) -> None:
parsed = parse(_basic(), rules)
assert [(row.counterparty_name, row.counterparty_inn) for row in parsed.rows] == [
(None, None),
(None, None),
("ООО Поставщик", "7701234567"),
(None, None),
(None, None),
]
# --- data_type строк ---
def test_currency_and_quantity_rows(rules: Rules) -> None:
sheet = osv(
r(0, "60", S100),
r(1, "ООО Поставщик", S100, inn="7701234567"),
r(2, "USD", (5, None, 1, None, 6, None)),
r(2, "Кол.", (3, None, None, None, 3, None)),
r(2, "Вал.", (5, None, 1, None, 6, None)),
r(2, "eur", (5, None, 1, None, 6, None)),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules, doc_data_type="NU")
types = {row.body.text: row.data_type for row in parsed.rows}
assert types == {
"60": "NU",
"ООО Поставщик": "NU",
"USD": "CURRENCY",
"Кол.": "QUANTITY",
"Вал.": "CURRENCY",
"eur": "CURRENCY",
"Итого": "NU",
}
usd = parsed.one("USD")
assert usd.body.sums[0] == Decimal("5.00") # суммы остаются в шести полях
# Строка-показатель — не группировка: OTHER, а не следующая группа схемы (CONTRACT)
assert (usd.group_type, usd.group_name, usd.account_code_raw) == ("OTHER", None, "60")
def test_doc_data_type_none_passes_through(rules: Rules) -> None:
parsed = parse(_basic(), rules, doc_data_type=None)
assert {row.data_type for row in parsed.rows} == {None}
# --- extra ---
def test_extra_from_unmatched_columns(rules: Rules) -> None:
parsed = parse(_basic(), rules)
assert parsed.one("ООО Поставщик").extra == {"КПП": "770101001"}
assert parsed.one("60").extra is None # все значения пусты
assert parsed.one("Итого").extra is None
def test_account_row_without_sums_and_duplicate_extra_titles(rules: Rules) -> None:
"""Review Focus 1: строка счёта со всеми null не падает; две колонки с одним заголовком — разные ключи extra."""
sheet = sheet_from_rows(
[
["ООО Ромашка"],
[TITLE],
[],
["Счет", "Кол-во", "Кол-во", M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)],
["Контрагенты", None, None, "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"],
[L(0), "60"], # счёт без сумм
[L(1), "ООО Поставщик", "1", "2", None, None, 10, None, 10, None],
[L(1), "ООО Второй", None, "3"],
["Итого", None, None, None, None, 10, None, 10, None],
]
)
parsed = parse(sheet, rules)
account = parsed.one("60")
assert account.group_type == "ACCOUNT"
assert account.body.sums == (None,) * 6
assert account.extra is None
first = parsed.one("ООО Поставщик").extra
assert first is not None
assert len(first) == len(set(first)) == DUPLICATED_EXTRA_COLUMNS
assert sorted(value or "" for value in first.values()) == ["1", "2"]
second = parsed.one("ООО Второй").extra
assert second is not None
assert set(second) == set(first)
assert sorted(value or "" for value in second.values()) == ["", "3"]
def test_footer_after_total_is_not_data_nor_extra(rules: Rules) -> None:
"""T4.5: подвал после «Итого» (подписи, числа правее таблицы) не попадает ни в строки, ни в extra."""
sheet = osv(
r(0, "60", S100),
r(1, "ООО Поставщик", S100, inn="7701234567"),
r(None, "Итого", S100),
[],
["Главный бухгалтер", None, "Иванова"], # подпись в колонке КПП
[None, None, None, None, None, None, None, None, None, None, 375490642], # правее таблицы
)
parsed = parse(sheet, rules)
assert [row.body.text for row in parsed.rows] == ["60", "ООО Поставщик", "Итого"]
assert all(row.extra is None for row in parsed.rows)
assert parsed.total_row == parsed.one("Итого").body.row
def test_column_filled_only_in_footer_is_not_an_extra_key(rules: Rules) -> None:
"""T4.5: колонка, заполненная только в подвале, не становится ключом `extra` у строк таблицы."""
sheet = sheet_from_rows(
[
["ООО Ромашка"],
[TITLE],
[],
["Счет", "КПП", "Код", M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)],
["Контрагенты", None, None, "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"],
[L(0), "60", None, None, 100, None, None, None, 100, None],
[L(1), "ООО Поставщик", "770101001", None, 100, None, None, None, 100, None],
["Итого", None, None, 100, None, None, None, 100, None],
["Главный бухгалтер", None, "подпись"],
]
)
parsed = parse(sheet, rules)
assert parsed.one("ООО Поставщик").extra == {"КПП": "770101001"}
def test_rows_with_values_only_right_of_table_are_skipped(rules: Rules) -> None:
sheet = osv(
r(0, "60", S100),
[None, None, None, None, None, None, None, None, None, None, 777], # только правее таблицы
r(1, "ООО Поставщик", S100, inn="7701234567"),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
assert [row.body.text for row in parsed.rows] == ["60", "ООО Поставщик", "Итого"]
# --- Пустые строки, пробелы, повтор заголовка ---
def test_blank_and_whitespace_rows_do_not_break_tree(rules: Rules) -> None:
"""Review Focus 5: пустая строка и строка из пробелов между блоками пропускаются, дерево не рвётся."""
sheet = osv(
r(0, "60", S100), # 7
[], # 8
[" ", " ", None, " "], # 9
r(1, "ООО Поставщик", S100, inn="7701234567"), # 10
r(None, "Итого", S100), # 11
)
parsed = parse(sheet, rules)
assert [row.body.row for row in parsed.rows] == [7, 10, 11]
assert parsed.one("ООО Поставщик").parent == ACCOUNT_ROW
def test_header_repeat_is_skipped(rules: Rules) -> None:
"""Review Focus 4: повтор заголовка внутри данных (печать по страницам) — не строка данных и не OTHER."""
sheet = osv(
r(0, "60", S100), # 7
["Счет", "ИНН", "КПП", M(OPENING, 2), M(TURNOVER, 2), M(CLOSING, 2)], # 8
["Контрагенты", None, None, "Дебет", "Кредит", "Дебет", "Кредит", "Дебет", "Кредит"], # 9
r(1, "ООО Поставщик", S100, inn="7701234567"), # 10
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
assert [row.body.text for row in parsed.rows] == ["60", "ООО Поставщик", "Итого"]
assert parsed.one("ООО Поставщик").parent == ACCOUNT_ROW
# --- TOTAL ---
def test_empty_osv_total_without_sums(rules: Rules) -> None:
"""R-13: пустая ОСВ — только «Итого» без сумм: одна строка TOTAL, иерархия trivial."""
parsed = parse(osv(r(None, "Итого")), rules)
assert [(row.group_type, row.level, row.parent) for row in parsed.rows] == [("TOTAL", 0, None)]
assert parsed.levels.strategy == TRIVIAL_STRATEGY
assert parsed.total_row == parsed.rows[0].body.row
def test_total_without_sums_in_account_pattern_sheet_stays_root(rules: Rules) -> None:
"""T6-pointer: «Итого» без сумм выбирается по тексту и положению до уровней — иначе account_pattern сделал бы
его ребёнком последнего счёта.
"""
sheet = osv(
r(None, "60", S100),
r(None, "60.01", S100),
r(None, "ООО Поставщик", S100),
r(None, "Итого"),
)
parsed = parse(sheet, rules)
assert parsed.levels.strategy == "account_pattern"
total = parsed.one("Итого")
assert (total.group_type, total.level, total.parent) == ("TOTAL", 0, None)
def test_counterparty_named_like_total_in_the_middle(rules: Rules) -> None:
"""T3.3: «Итого-Сервис» в середине данных — контрагент; TOTAL — последняя строка «Итого»."""
sheet = osv(
r(0, "60", S100),
r(1, "Итого-Сервис", S100, inn="7701234567"),
r(None, "Итого", S100),
)
parsed = parse(sheet, rules)
assert parsed.one("Итого-Сервис").group_type == "COUNTERPARTY"
assert parsed.one("Итого").group_type == "TOTAL"
assert [row.group_type for row in parsed.rows].count("TOTAL") == 1
def test_last_row_named_like_total_is_rejected_by_outline_level(rules: Rules) -> None:
"""Последняя строка «Итого-Сервис» без настоящего «Итого»: outline глубже минимального — не TOTAL."""
sheet = osv(
r(0, "60", S100),
r(1, "Итого-Сервис", S100, inn="7701234567"),
)
parsed = parse(sheet, rules)
assert parsed.total_row is None
assert parsed.one("Итого-Сервис").group_type == "COUNTERPARTY"
def test_known_risk_last_counterparty_named_like_total_without_outline(rules: Rules) -> None:
"""Известный риск (документирован): без outline и отступов последняя строка «Итого-Сервис» при отсутствии
настоящего «Итого» неотличима от итоговой строки и становится TOTAL; контрольные суммы §1.4 (TOTAL = корень)
это поймают, если суммы не совпадают.
"""
sheet = osv(
r(None, "60", S100),
r(None, "Итого-Сервис", (1, None, None, None, 1, None)),
)
parsed = parse(sheet, rules)
assert parsed.one("Итого-Сервис").group_type == "TOTAL"
# --- Свёрнутое сальдо (R-1) ---
def test_collapsed_closing_split_by_sign(rules: Rules) -> None:
"""Spec «ОСВ без сальдо на начало и со свёрнутым сальдо»: отрицательное свёрнутое — в closing_credit."""
sheet = sheet_from_rows(
[
["ООО Ромашка"],
[TITLE],
[],
["Счет", M(TURNOVER, 2), M(CLOSING, 1, 2)],
["Контрагенты", "Дебет", "Кредит", None],
[L(0), "60", 10, 30, -20],
[L(1), "ООО Первый", 10, None, 10],
[L(1), "ООО Второй", None, 30, -30],
[L(1), "ООО Третий", None, None, 0],
[L(1), "ООО Четвёртый"],
["Итого", 10, 30, -20],
]
)
parsed = parse(sheet, rules)
sums = {row.body.text: row.body.sums for row in parsed.rows}
two = Decimal("0.00")
assert sums["ООО Первый"] == (None, None, Decimal("10.00"), None, Decimal("10.00"), None)
assert sums["ООО Второй"] == (None, None, None, Decimal("30.00"), None, Decimal("30.00"))
assert sums["ООО Третий"] == (None, None, None, None, two, None)
assert sums["ООО Четвёртый"] == (None,) * 6
assert sums["Итого"] == (None, None, Decimal("10.00"), Decimal("30.00"), None, Decimal("20.00"))
collapsed = [warning for warning in parsed.warnings if "свёрнутое сальдо" in warning]
assert len(collapsed) == 1
assert "closing" in collapsed[0]
def test_no_collapsed_warning_for_regular_sheet(rules: Rules) -> None:
parsed = parse(_basic(), rules)
assert parsed.warnings == []
# --- Суммы и предупреждения ---
def test_sums_in_column_map_order_and_money_warnings_with_row(rules: Rules) -> None:
sheet = osv(
r(0, "60", ("1 234,50", "—", "(10,00)", "abc", Decimal("0.005"), None)), # 7
r(None, "Итого"),
)
parsed = parse(sheet, rules)
assert parsed.one("60").body.sums == (
Decimal("1234.50"),
None,
Decimal("-10.00"),
None,
Decimal("0.01"),
None,
)
assert len(parsed.warnings) == 1
assert "строка 7" in parsed.warnings[0]
assert "abc" in parsed.warnings[0]
def test_body_row_style_fields(rules: Rules) -> None:
sheet = osv(
r(None, C("60", bold=True), S100),
r(None, C("ООО Поставщик", indent=2), S100),
r(None, "Итого", S100),
)
header = find_table_header(sheet, rules)
body, _, _ = collect_body_rows(sheet, header, rules, Deadline(TIMEOUT_SECONDS))
account, counterparty, _ = body
assert (account.is_account, account.account_rank, account.bold) == (True, 1, True)
assert (counterparty.is_account, counterparty.account_rank, counterparty.indent) == (False, None, 2)
def test_account_rank_counts_segments(rules: Rules) -> None:
sheet = osv(
r(0, "76", S100),
r(1, "76.07", S100),
r(2, "76.07.1, Прочие", S100),
r(None, "Итого", S100),
)
header = find_table_header(sheet, rules)
body, _, _ = collect_body_rows(sheet, header, rules, Deadline(TIMEOUT_SECONDS))
assert [row.account_rank for row in body] == [1, 2, 3, None]
# --- Дедлайн ---
def test_collect_body_rows_checks_deadline(rules: Rules) -> None:
data = [r(1, f"ООО Поставщик {index}", S100) for index in range(1200)]
sheet = osv(r(0, "60", S100), *data, r(None, "Итого", S100))
header = find_table_header(sheet, rules)
# Часы: 0 при создании дедлайна, дальше — далеко за лимитом; первая же проверка внутри обхода строк бросает
readings = iter([0.0])
deadline = Deadline(EXPIRED_AFTER_SECONDS, clock=lambda: next(readings, EXPIRED_AFTER_SECONDS * 2.0))
with pytest.raises(ParseTimeoutError):
collect_body_rows(sheet, header, rules, deadline)
def test_classified_rows_cover_every_body_row(rules: Rules) -> None:
parsed = parse(_basic(), rules)
assert [row.level for row in parsed.rows] == [parsed.levels.by_row[row.body.row] for row in parsed.rows]