RSHB/arm-tabular-parser
9
0
Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(statements): адаптер шага 1 — книга → FormCandidate, склейка листов, segment_breaks (#54)

Browse Source 
Тест test_npz_candidates красный: НПЗ — BLOCKED (Р-31), см. отчёт задачи 5.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev 2026-10-18 20:24:47 +04:00
parent 12f4701754
commit c858a85d85
4 changed files with 424 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
api/src/parser/detect.py
View File

@ -31,6 +31,7 @@ __all__ = [
"first_data_row",
"inn_from_cell",
"is_header_repeat",
"is_osv_sheet",
]
CollapsedGroup = Literal["opening", "closing"]
@ -154,11 +155,16 @@ _STATEMENT_KINDS = {
}
def is_osv_sheet(sheet: Sheet, rules: Rules) -> bool:
"""Признак ведомости (`sheet_markers`) в первых `SHEET_MARKER_ROWS` строках листа."""
head = [text for row in range(1, min(sheet.n_rows, SHEET_MARKER_ROWS) + 1) for _, text in _texts(sheet, row)]
return any(_search(rules.headers.sheet_markers, text) for text in head)
def classify_sheet(sheet: Sheet, rules: Rules) -> SheetKind:
"""ОСВ — если в первых строках есть признак ведомости; иначе форма ББ/ОФР, её продолжение или прочее — по
`statements.detect` (признаки формы ищутся по всему листу до таблицы)."""
head = [text for row in range(1, min(sheet.n_rows, SHEET_MARKER_ROWS) + 1) for _, text in _texts(sheet, row)]
if any(_search(rules.headers.sheet_markers, text) for text in head):
if is_osv_sheet(sheet, rules):
return SheetKind.OSV
return _STATEMENT_KINDS[classify_statement_sheet(sheet, rules)]

221
api/src/parser/statements/adapter.py Normal file
View File

@ -0,0 +1,221 @@
"""Адаптер книги -> `FormCandidate` (план 4, задача 5; спека «Шаг 1 — многостраничная форма», §11.2 п. 2–4).
`plan_sheets` раскладывает листы по видам (ОСВ / форма / продолжение / прочее), `build_candidates` собирает из листов
форм кандидатов для пакета: блок формы -> кандидат, лист-продолжение присоединяется к открытой форме (`segment_breaks`
— границы листов в общей сетке) либо становится отдельным кандидатом с предупреждением.
"""
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from arm_statements import Cell as StatementCell
from arm_statements import (
DocumentType,
Evidence,
FormCandidate,
HeaderCells,
Source,
StatementWarning,
WarningCode,
continuation_type,
make_warning,
sheet_joins,
)
from src.parser.deadline import Deadline
from src.parser.detect import SheetKind
from src.parser.grid import Sheet
from src.parser.statements.cells import grid_rows
from src.parser.statements.forms import (
StatementRules,
classify_statement_sheet,
code_rows,
codes_of,
continuation_marker,
find_blocks,
is_weak_continuation,
sheet_amount_columns,
sheet_code_column,
)
from src.parser.statements.header_cells import header_cells
__all__ = ["SheetPlan", "build_candidates", "plan_sheets"]
MAX_FORMAT_LENGTH = 30
_OPEN_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME})
_Row = tuple[StatementCell | None, ...]
@dataclass(frozen=True)
class SheetPlan:
osv: tuple[Sheet, ...]
statements: tuple[tuple[Sheet, SheetKind], ...]
other: tuple[Sheet, ...]
def plan_sheets(sheets: Sequence[Sheet], is_osv: Callable[[Sheet], bool], rules: StatementRules) -> SheetPlan:
"""Листы по порядку книги: ОСВ; форма; продолжение (в том числе слабое — лист из 1–2 кодов сразу после листа
формы или продолжения); прочее. Лист ОСВ или «прочее» прерывает «сразу после»."""
osv: list[Sheet] = []
statements: list[tuple[Sheet, SheetKind]] = []
other: list[Sheet] = []
last_type: DocumentType | None = None # тип последнего блока последнего листа формы; None — формы «сразу до» нет
for sheet in sheets:
if is_osv(sheet):
osv.append(sheet)
last_type = None
continue
kind = classify_statement_sheet(sheet, rules)
if kind == "statement":
statements.append((sheet, SheetKind.STATEMENT))
blocks = find_blocks(sheet, rules)
last_type = blocks[-1].type_hint if blocks else None
elif kind == "continuation":
statements.append((sheet, SheetKind.STATEMENT_CONTINUATION))
elif last_type is not None and is_weak_continuation(sheet, rules, last_type):
statements.append((sheet, SheetKind.STATEMENT_CONTINUATION))
else:
other.append(sheet)
last_type = None
return SheetPlan(tuple(osv), tuple(statements), tuple(other))
@dataclass
class _Draft:
"""Кандидат в сборке."""
type_hint: DocumentType | None
sheet_names: list[str]
evidence: Evidence
header: HeaderCells
rows: list[_Row]
form_amount_columns: int = 0
segment_breaks: list[int] = field(default_factory=list)
formats: set[str] = field(default_factory=set)
warnings: list[StatementWarning] = field(default_factory=list)
@property
def locator(self) -> str:
return "sheet:" + ",".join(self.sheet_names)
def _shorten(fmt: str) -> str:
return fmt if len(fmt) <= MAX_FORMAT_LENGTH else fmt[: MAX_FORMAT_LENGTH - 1] + "…"
def _form_block_drafts(sheet: Sheet, rules: StatementRules) -> list[_Draft]:
drafts: list[_Draft] = []
for block in find_blocks(sheet, rules):
formats: set[str] = set()
col = sheet_code_column(sheet, rules, block.first_row, block.last_row)
rows_with_codes = [] if col is None else code_rows(sheet, rules, col, block.first_row, block.last_row)
header_last = max(block.first_row, rows_with_codes[0] - 1) if rows_with_codes else block.last_row
header, evidence = header_cells(sheet, rules, block, header_last, formats)
amounts = 0 if col is None else sheet_amount_columns(sheet, rules, col, block.first_row, block.last_row)
drafts.append(
_Draft(
type_hint=block.type_hint,
sheet_names=[sheet.name],
evidence=evidence,
header=header,
rows=grid_rows(sheet, block.first_row, block.last_row, sheet.n_cols, formats),
form_amount_columns=amounts,
formats=formats,
)
)
return drafts
def _separate(sheet: Sheet, type_hint: DocumentType | None) -> _Draft:
formats: set[str] = set()
return _Draft(
type_hint=type_hint,
sheet_names=[sheet.name],
evidence=Evidence(),
header=HeaderCells(),
rows=grid_rows(sheet, 1, sheet.n_rows, sheet.n_cols, formats),
formats=formats,
)
def _sheet_joins(sheet: Sheet, form: _Draft, rules: StatementRules) -> bool:
"""Лист без маркера присоединяется по кодам и числу колонок сумм (как сегмент шага 2 пакета)."""
form_type = form.type_hint
assert form_type is not None # noqa: S101 - открытая форма всегда с типом
col = sheet_code_column(sheet, rules, 1, sheet.n_rows)
if col is None:
col = sheet_code_column(sheet, rules, 1, sheet.n_rows, relaxed=form_type)
if col is None:
return False
codes = codes_of(sheet, rules, col, 1, sheet.n_rows)
amounts = sheet_amount_columns(sheet, rules, col, 1, sheet.n_rows)
return sheet_joins(form_type, codes, amounts, form.form_amount_columns, rules)
def _continuation(sheet: Sheet, drafts: list[_Draft], rules: StatementRules) -> None:
open_form = next((draft for draft in reversed(drafts) if draft.type_hint in _OPEN_TYPES), None)
marker = continuation_marker(sheet, rules)
marker_type = None if marker is None else continuation_type(marker.text, rules)
if open_form is None:
drafts.append(_separate(sheet, marker_type))
return
locator = f"sheet:{sheet.name}"
if marker_type is not None and marker_type != open_form.type_hint:
separate = _separate(sheet, marker_type)
message = (
f"лист «{sheet.name}»: тип формы в маркере продолжения ({marker_type}) не совпадает с типом "
f"открытой формы ({open_form.type_hint})"
)
separate.warnings.append(make_warning(WarningCode.CONTINUATION_TYPE_CONFLICT, message, rules, locator=locator))
drafts.append(separate)
return
if marker is None and not _sheet_joins(sheet, open_form, rules):
separate = _separate(sheet, None)
message = (
f"лист «{sheet.name}» не присоединён к форме «{open_form.sheet_names[0]}»: коды или число колонок "
"сумм не подходят"
)
separate.warnings.append(make_warning(WarningCode.SHEET_NOT_JOINED, message, rules, locator=locator))
drafts.append(separate)
return
open_form.segment_breaks.append(len(open_form.rows))
open_form.rows.extend(grid_rows(sheet, 1, sheet.n_rows, sheet.n_cols, open_form.formats))
open_form.sheet_names.append(sheet.name)
def _finish(draft: _Draft, rules: StatementRules, *, xlsb: bool) -> FormCandidate:
width = max((len(row) for row in draft.rows), default=0)
rows = tuple(row + (None,) * (width - len(row)) for row in draft.rows)
warnings = list(draft.warnings)
locator = draft.locator
if xlsb:
message = "формат чисел xlsb не читается: значения как в ячейках"
warnings.append(make_warning(WarningCode.XLSB_NO_FORMATS, message, rules, locator=locator))
if draft.formats:
shown = ", ".join(f"«{_shorten(fmt)}»" for fmt in sorted(draft.formats))
message = f"формат чисел не разобран, значения как в ячейках: {shown}"
warnings.append(make_warning(WarningCode.NUMBER_FORMAT_UNPARSED, message, rules, locator=locator))
return FormCandidate(
source=Source.EXCEL,
locator=locator,
type_hint=draft.type_hint,
evidence=draft.evidence,
header=draft.header,
rows=rows,
segment_breaks=tuple(draft.segment_breaks),
warnings=tuple(warnings),
)
def build_candidates(
sheets: Sequence[tuple[Sheet, SheetKind]], rules: StatementRules, deadline: Deadline, *, xlsb: bool
) -> list[FormCandidate]:
"""Кандидаты форм из листов `STATEMENT` / `STATEMENT_CONTINUATION` в порядке книги."""
drafts: list[_Draft] = []
for sheet, kind in sheets:
deadline.check(f"лист «{sheet.name}»: форма")
if kind is SheetKind.STATEMENT:
drafts.extend(_form_block_drafts(sheet, rules))
elif kind is SheetKind.STATEMENT_CONTINUATION:
_continuation(sheet, drafts, rules)
return [_finish(draft, rules, xlsb=xlsb) for draft in drafts]

42
api/tests/fixtures/test_statements_adapter_real.py vendored Normal file
View File

@ -0,0 +1,42 @@
"""Адаптер книги -> `FormCandidate` на реальных выгрузках (план 4, задача 5)."""
from pathlib import Path
import pytest
from arm_statements import DocumentType, default_rules
from src.parser.deadline import Deadline
from src.parser.detect import is_osv_sheet
from src.parser.readers import read_book
from src.parser.rules import load_rules
from src.parser.statements.adapter import build_candidates, plan_sheets
pytestmark = pytest.mark.fixtures
STATEMENTS = "statements-xlsx"
NPZ = "npz/2025-Q3/Бухгалтерская отчетность за Январь - Сентябрь 2025 г. (НОВОРОССИЙСКИЙ ПРОКАТНЫЙ ЗАВОД).xlsx"
UNAGRANDE_BALANCE = "unagrande-group/2025-Q3/ББ. Лист 1 3кв.25.xls"
def _candidates(path: Path): # noqa: ANN202
sheets, _ = read_book(path.read_bytes(), path.suffix.lstrip(".").upper()) # type: ignore[arg-type]
osv_rules = load_rules()
rules = default_rules()
plan = plan_sheets(sheets, lambda sheet: is_osv_sheet(sheet, osv_rules), rules)
return build_candidates(plan.statements, rules, Deadline(600.0), xlsb=False)
def test_npz_candidates(fixtures_dir: Path) -> None:
candidates = _candidates(fixtures_dir / STATEMENTS / NPZ)
assert [(c.type_hint, c.locator.count(",") + 1, len(c.segment_breaks)) for c in candidates] == [
(DocumentType.BALANCE, 5, 4),
(DocumentType.INCOME, 3, 2),
]
codes = {w.code for c in candidates for w in c.warnings}
assert not codes & {"SHEET_NOT_JOINED", "CONTINUATION_TYPE_CONFLICT"}
def test_unagrande_title_block_header(fixtures_dir: Path) -> None:
candidates = _candidates(fixtures_dir / STATEMENTS / UNAGRANDE_BALANCE)
header = candidates[0].header
assert header.okopf is not None and header.okopf.text == "12267"
assert header.inn is not None and "7728757141" in header.inn.text

153
api/tests/unit/parser/statements/test_adapter.py Normal file
View File

@ -0,0 +1,153 @@
"""Адаптер книги -> `FormCandidate`: план листов, склейка листов-продолжений, предупреждения (план 4, задача 5)."""
import pytest
from gridbuilder import at, sheet_from_rows, with_formats
from arm_statements import DocumentType, Severity, Source, default_rules
from src.parser.deadline import Deadline
from src.parser.detect import SheetKind, is_osv_sheet
from src.parser.grid import Sheet
from src.parser.rules import load_rules
from src.parser.statements.adapter import build_candidates, plan_sheets
from src.parser.statements.forms import StatementRules
NO_DEADLINE = Deadline(10_000.0)
STATEMENT = SheetKind.STATEMENT
CONTINUATION = SheetKind.STATEMENT_CONTINUATION
BALANCE_CODES = (1110, 1150, 1170, 1100)
INCOME_CODES = (2110, 2120, 2400)
@pytest.fixture(scope="module")
def rules() -> StatementRules:
return default_rules()
def _form(title: str, codes: tuple[int, ...], name: str, *, amounts: int = 3, marker: str | None = None) -> Sheet:
head = [[marker]] if marker else [[title]]
caption = at({1: "Показатель", 3: "Код", **{4 + i: f"На 3{i + 1}.12.2024" for i in range(amounts)}})
body = [at({1: "Статья", 3: code, **{4 + i: 10 + i for i in range(amounts)}}) for code in codes]
return sheet_from_rows([*head, caption, *body], name=name)
def test_balance_with_marked_continuation_and_income(rules: StatementRules) -> None:
balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс")
marked = _form("", (1210, 1230, 1600), "Баланс (2)", marker="Форма 0710001 с.2")
ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР")
candidates = build_candidates(
[(balance, STATEMENT), (marked, CONTINUATION), (ofr, STATEMENT)], rules, NO_DEADLINE, xlsb=False
)
assert [(c.type_hint, c.locator) for c in candidates] == [
(DocumentType.BALANCE, "sheet:Баланс,Баланс (2)"),
(DocumentType.INCOME, "sheet:ОФР"),
]
assert candidates[0].segment_breaks == (balance.n_rows,)
assert len({len(row) for row in candidates[0].rows}) == 1
assert candidates[0].source is Source.EXCEL
assert len(candidates[0].rows) == balance.n_rows + marked.n_rows
def test_refs_of_cells(rules: StatementRules) -> None:
rows = [at({1: "Бухгалтерский баланс"})] + [[] for _ in range(24)]
rows.append(at({3: "Код", 4: "На 31.12.2024"}))
rows.append(at({1: "Статья", 3: 1110, 4: 1}))
rows += [at({1: "Статья", 3: code, 4: 1}) for code in (1150, 1170, 1100)]
assert rows[26] == at({1: "Статья", 3: 1110, 4: 1})
for name, ref in (("Баланс", "Баланс!C27"), ("Бухгалтерский баланс. Лист 1", "'Бухгалтерский баланс. Лист 1'!C27")):
[candidate] = build_candidates([(sheet_from_rows(rows, name=name), STATEMENT)], rules, NO_DEADLINE, xlsb=False)
cells = [cell for row in candidate.rows for cell in row if cell is not None and cell.text == "1110"]
assert [cell.ref for cell in cells] == [ref]
def test_marker_type_conflict_is_separate_candidate(rules: StatementRules) -> None:
balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс")
other = _form("", INCOME_CODES, "Лист 2", marker="Форма 0710002 с.2")
first, second = build_candidates([(balance, STATEMENT), (other, CONTINUATION)], rules, NO_DEADLINE, xlsb=False)
assert first.warnings == ()
assert second.type_hint is DocumentType.INCOME
assert [(w.code, w.severity) for w in second.warnings] == [("CONTINUATION_TYPE_CONFLICT", Severity.ERROR)]
assert second.segment_breaks == ()
assert len(second.rows) == other.n_rows
def test_single_line_continuation_joins(rules: StatementRules) -> None:
ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР")
tail = sheet_from_rows([at({1: "Чистая прибыль", 3: 2400, 4: 1, 5: 2, 6: 3})], name="ОФР (2)")
[candidate] = build_candidates([(ofr, STATEMENT), (tail, CONTINUATION)], rules, NO_DEADLINE, xlsb=False)
assert candidate.segment_breaks == (ofr.n_rows,)
assert candidate.warnings == ()
def test_wider_sheet_with_three_codes_is_not_joined(rules: StatementRules) -> None:
balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс", amounts=3)
wide = sheet_from_rows(
[at({1: "Статья", 3: code, 4: 1, 5: 2, 6: 3, 7: 4}) for code in (1210, 1230, 1250)], name="Лист 2"
)
first, second = build_candidates([(balance, STATEMENT), (wide, CONTINUATION)], rules, NO_DEADLINE, xlsb=False)
assert first.segment_breaks == ()
assert second.type_hint is None
assert [(w.code, w.severity) for w in second.warnings] == [("SHEET_NOT_JOINED", Severity.ERROR)]
def test_lonely_continuation_has_no_warning(rules: StatementRules) -> None:
marked = _form("", (1210, 1230, 1600), "Баланс (2)", marker="Форма 0710001 с.2")
[candidate] = build_candidates([(marked, CONTINUATION)], rules, NO_DEADLINE, xlsb=False)
assert candidate.type_hint is DocumentType.BALANCE
assert candidate.warnings == ()
assert candidate.header.inn is None
assert len(candidate.rows) == marked.n_rows
def test_two_forms_on_one_sheet_share_locator(rules: StatementRules) -> None:
balance = [at({1: "Бухгалтерский баланс"}), at({3: "Код", 4: "На 31.12.2024"})]
balance += [at({1: "Статья", 3: code, 4: 1}) for code in BALANCE_CODES]
income = [at({1: "Отчет о финансовых результатах"}), at({3: "Код", 4: "За 2024"})]
income += [at({1: "Статья", 3: code, 4: 1}) for code in INCOME_CODES]
sheet = sheet_from_rows([*balance, *income], name="Формы")
candidates = build_candidates([(sheet, STATEMENT)], rules, NO_DEADLINE, xlsb=False)
assert [(c.type_hint, c.locator) for c in candidates] == [
(DocumentType.BALANCE, "sheet:Формы"),
(DocumentType.INCOME, "sheet:Формы"),
]
assert len(candidates[0].rows) == len(balance)
assert len(candidates[1].rows) == len(income)
def test_xlsb_and_unparsed_formats_warnings(rules: StatementRules) -> None:
balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс", amounts=1)
tail = _form("", (1210, 1230, 1250), "Баланс (2)", amounts=1, marker="Форма 0710001 с.2")
formats = {(row, 4): "0.0%" for row in range(3, balance.n_rows + 1)}
balance = with_formats(balance, formats)
tail = with_formats(tail, {(row, 4): "0.0%" for row in range(3, tail.n_rows + 1)})
ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР", amounts=1)
candidates = build_candidates(
[(balance, STATEMENT), (tail, CONTINUATION), (ofr, STATEMENT)], rules, NO_DEADLINE, xlsb=True
)
codes = [[w.code for w in c.warnings] for c in candidates]
assert codes == [["XLSB_NO_FORMATS", "NUMBER_FORMAT_UNPARSED"], ["XLSB_NO_FORMATS"]]
assert "«0.0%»" in candidates[0].warnings[1].message
assert candidates[0].warnings[1].message.count("0.0%") == 1
def test_plan_sheets_weak_continuation_only_after_form(rules: StatementRules) -> None:
ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР")
weak = sheet_from_rows([["Чистая прибыль", "2400", 10, 8]], name="2400")
notes = sheet_from_rows([["Примечания"], ["Текст пояснения к отчетности"]], name="Примечания")
osv = sheet_from_rows([["Оборотно-сальдовая ведомость"]], name="ОСВ")
def is_osv(sheet: Sheet) -> bool:
return sheet.name == "ОСВ"
plan = plan_sheets([ofr, weak], is_osv, rules)
assert [(s.name, k) for s, k in plan.statements] == [("ОФР", STATEMENT), ("2400", CONTINUATION)]
plan = plan_sheets([osv, weak], is_osv, rules)
assert (plan.osv, plan.statements, plan.other) == ((osv,), (), (weak,))
plan = plan_sheets([ofr, notes, weak], is_osv, rules)
assert [s.name for s in plan.other] == ["Примечания", "2400"]
def test_is_osv_sheet_by_marker() -> None:
rules = load_rules()
osv = sheet_from_rows([["Оборотно-сальдовая ведомость по счету 60"]])
assert is_osv_sheet(osv, rules)
assert not is_osv_sheet(sheet_from_rows([["Бухгалтерский баланс"]]), rules)