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

"""Блоки субсчетов с кодами контрагентов (`05-domain-rules.md` §1.3, AURORA-1714, решение владельца 27.10.2026).

Выгрузка 1С «ОСВ по счету 60 по контрагентам»: строка-субсчёт без сумм (`60.111`), внутри — код контрагента с суммами,
строка названия без сумм и договоры с суммами, в конце блока — «ИТОГ по 60.111». Шаг работает над строками тела между
сбором строк (`collect_body_rows`) и иерархией:

1. «ИТОГ по <субсчёт>» (`subaccount_sums`, словарь) отдаёт шесть сумм строке-счёту с тем же кодом, у которой сумм нет, и
   сама из тела исчезает (в `rows` контракта отдельной строкой не выдаётся);
2. в блоке такого субсчёта строка с суммами, за которой идёт строка без сумм, — контрагент (`role = counterparty`);
   следующая за ней строка без сумм — его название: склеивается с ним (`names`) и из тела исчезает; прочие строки с
   суммами после контрагента — его договоры (`role = contract`).

Без «ИТОГ по …», сопоставленного субсчёту, шаг ничего не меняет: раскладка узнаётся по ним. Координат и имён нет —
только суммы, порядок строк и словарь.
"""

from collections.abc import Sequence
from dataclasses import dataclass, field, replace

from src.parser.classify import match_account
from src.parser.hierarchy import BodyRow, Role
from src.parser.rules import Rules, normalize_text

__all__ = ["CodeBlocks", "apply_code_blocks"]


@dataclass(frozen=True)
class CodeBlocks:
    """Тело таблицы после шага: строки, названия контрагентов по номеру строки с кодом, признак «раскладка найдена»."""

    rows: list[BodyRow]
    names: dict[int, str] = field(default_factory=dict)
    applied: bool = False


def _has_sums(item: BodyRow) -> bool:
    return any(value is not None for value in item.sums)


def _subtotal_code(item: BodyRow, rules: Rules) -> str | None:
    """Код субсчёта из «ИТОГ по 60.111» (в нормализованном виде) или `None`, если строка — не итог субсчёта."""
    for pattern in rules.headers.subaccount_sums:
        match = pattern.search(item.text)
        if match is not None:
            return match.group("account")
    return None


def _mark_wide_subaccounts(body: Sequence[BodyRow], header_account: str) -> list[BodyRow]:
    """Субсчёт с тремя цифрами в сегменте («60.111» при шапке «60») — строка-счёт (§1.3, правило 1)."""
    result: list[BodyRow] = []
    for item in body:
        under_header = bool(header_account) and item.text.startswith(f"{header_account}.")
        found = match_account(item.text, header_account) if under_header and not item.is_account else None
        if found is None:
            result.append(item)
        else:
            result.append(replace(item, is_account=True, account_rank=found.group(1).count(".") + 1))
    return result


def _block_roles(block: Sequence[BodyRow]) -> tuple[dict[int, Role], dict[int, str], set[int]]:
    """Роли строк блока: `{строка: роль}`, названия `{строка кода: текст}` и номера склеенных (убираемых) строк."""
    roles: dict[int, Role] = {}
    names: dict[int, str] = {}
    glued: set[int] = set()
    has_counterparty = False
    index = 0
    while index < len(block):
        item = block[index]
        if item.indicator or not _has_sums(item):
            index += 1
            continue
        following = block[index + 1] if index + 1 < len(block) else None
        if following is not None and not following.indicator and not _has_sums(following) and following.text.strip():
            roles[item.row] = "counterparty"
            names[item.row] = following.text
            glued.add(following.row)
            has_counterparty = True
            index += 2
            continue
        if has_counterparty:
            roles[item.row] = "contract"
        index += 1
    return roles, names, glued


def apply_code_blocks(body: Sequence[BodyRow], total_row: int | None, rules: Rules, header_account: str) -> CodeBlocks:
    """Итоги субсчетов -> суммы строк-счетов; роли и названия в блоках (см. модуль). Без счёта шапки и без пар
    «субсчёт — ИТОГ по субсчёту» тело возвращается как есть."""
    rows = _mark_wide_subaccounts(body, header_account)
    if not header_account or not rules.headers.subaccount_sums:
        return CodeBlocks(rows)
    data = [item for item in rows if total_row is None or item.row < total_row]
    accounts = {normalize_text(item.text): item for item in data if item.is_account and not _has_sums(item)}
    sums_of: dict[int, tuple] = {}
    dropped: set[int] = set()
    for item in data:
        code = _subtotal_code(item, rules)
        account = None if code is None else accounts.get(code)
        if account is not None and account.row < item.row and account.row not in sums_of:
            sums_of[account.row] = item.sums
            dropped.add(item.row)
    if not sums_of:
        return CodeBlocks(rows)
    roles: dict[int, Role] = {}
    names: dict[int, str] = {}
    block: list[BodyRow] = []
    block_open = False
    for current in [*(item for item in data if item.row not in dropped), None]:
        if current is None or current.is_account:
            if block_open:
                block_roles, block_names, glued = _block_roles(block)
                roles.update(block_roles)
                names.update(block_names)
                dropped |= glued
            block, block_open = [], current is not None and current.row in sums_of
            continue
        block.append(current)
    result: list[BodyRow] = []
    for source in rows:
        if source.row in dropped:
            continue
        sums = sums_of.get(source.row, source.sums)
        role = roles.get(source.row)
        result.append(replace(source, sums=sums, role=role) if (sums is not source.sums or role) else source)
    return CodeBlocks(result, names, applied=True)