315 lines
18 KiB
Python
315 lines
18 KiB
Python
"""Иерархия строк ОСВ — каскад стратегий (spec «Иерархия строк — каскад стратегий», design D-3, §1.0 п. 5, §1.3).
|
||
|
||
Стратегии — чистые функции над `BodyRow` (строки данных без листа): `outline` → `indent` → `account_pattern` →
|
||
`sums`. Каждая возвращает уровни `{row: level}` или `None`, если неприменима; первая, давшая непротиворечивое дерево
|
||
(`is_consistent`), побеждает, её имя попадает в `Levels.strategy` (`OSV_HIERARCHY_BY_<ИМЯ>` в `warnings[]` документа).
|
||
|
||
Уровни и родители считаются по порядку строк и уровням, а не по номерам и текстам: два одноимённых контрагента под
|
||
разными субсчетами получают разных родителей (Review Focus 3). Строка «Итого» в стратегии не передаётся: она —
|
||
контрольная сумма документа, всегда уровень 0 без родителя, и не может стать ребёнком последнего счёта.
|
||
"""
|
||
|
||
from collections.abc import Iterator, Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
from typing import Protocol
|
||
|
||
from src.contract.common import DataType
|
||
from src.parser.deadline import Deadline
|
||
from src.parser.errors import InvalidFormatError
|
||
|
||
__all__ = [
|
||
"STRATEGIES",
|
||
"TRIVIAL_STRATEGY",
|
||
"BodyRow",
|
||
"Levels",
|
||
"Strategy",
|
||
"is_consistent",
|
||
"parents",
|
||
"resolve_levels",
|
||
"strategy_account_pattern",
|
||
"strategy_indent",
|
||
"strategy_outline",
|
||
"strategy_sums",
|
||
]
|
||
|
||
ROOT_LEVEL = 0
|
||
# Стратегия по оформлению (outline, отступ) применима, только если различает хотя бы два уровня
|
||
MIN_DISTINCT_LEVELS = 2
|
||
# Допуск сравнения «сумма детей = родитель» в стратегии `sums` — тот же, что у контрольных сумм §1.4
|
||
SUM_TOLERANCE = Decimal("0.01")
|
||
# Каскад нужен, только если строк (кроме «Итого») хотя бы две; одна строка или ни одной — тривиальное дерево
|
||
# из корней (пустая ОСВ: только «Итого», в том числе «Итого» без сумм, не распознанное как TOTAL)
|
||
MIN_ROWS_FOR_CASCADE = 2
|
||
TRIVIAL_STRATEGY = "trivial"
|
||
# Кооперативный дедлайн внутри стратегии (design D-8): проверка каждые N строк или N шагов перебора поддеревьев
|
||
DEADLINE_CHECK_EVERY = 500
|
||
_ZERO = Decimal(0)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BodyRow:
|
||
"""Строка данных до классификации: текст, признак счёта, оформление и шесть сумм (по `ColumnMap`)."""
|
||
|
||
row: int
|
||
text: str
|
||
is_account: bool
|
||
account_rank: int | None # число сегментов кода счёта: 60 → 1, 60.01 → 2, 60.01.1 → 3
|
||
bold: bool
|
||
indent: int # alignment.indent или число ведущих пробелов (единицы разные — сравниваются только ранги)
|
||
outline: int
|
||
sums: tuple[Decimal | None, ...]
|
||
label_col: int | None = None # колонка, в которой стоит текст строки (она не всегда колонка текста таблицы)
|
||
indicator: bool = False # строка-показатель (§1.3): вид сумм предыдущей строки данных, не уровень аналитики
|
||
label_data_type: DataType | None = None # вид данных по метке строки (колонка «Показатели» или «Вал.»/«Кол.»)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Levels:
|
||
"""Уровни строк, их родители (`None` — корень) и имя выбравшей их стратегии."""
|
||
|
||
by_row: dict[int, int]
|
||
parent: dict[int, int | None]
|
||
strategy: str
|
||
|
||
|
||
class Strategy(Protocol):
|
||
"""Стратегия каскада: уровни `{row: level}` или `None`, если неприменима; `deadline` проверяется изнутри
|
||
(каждые `DEADLINE_CHECK_EVERY` строк/шагов), без него — не проверяется (чистая функция для тестов).
|
||
"""
|
||
|
||
def __call__(self, rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None: ...
|
||
|
||
|
||
class _Pacer:
|
||
"""Счётчик шагов: каждые `DEADLINE_CHECK_EVERY` вызовов `tick()` — `deadline.check(where)`."""
|
||
|
||
def __init__(self, deadline: Deadline | None, where: str) -> None:
|
||
self._deadline = deadline
|
||
self._where = where
|
||
self._steps = 0
|
||
|
||
def tick(self) -> None:
|
||
self._steps += 1
|
||
if self._deadline is not None and self._steps % DEADLINE_CHECK_EVERY == 0:
|
||
self._deadline.check(self._where)
|
||
|
||
|
||
def _paced(rows: Sequence[BodyRow], deadline: Deadline | None, where: str) -> Iterator[BodyRow]:
|
||
"""Строки по порядку с проверкой дедлайна каждые `DEADLINE_CHECK_EVERY` строк."""
|
||
pacer = _Pacer(deadline, where)
|
||
for item in rows:
|
||
pacer.tick()
|
||
yield item
|
||
|
||
|
||
def _ranks(values: Sequence[int]) -> dict[int, int] | None:
|
||
"""Значение → его ранг среди различных значений (0, 1, 2…); `None`, если различных меньше двух."""
|
||
distinct = sorted(set(values))
|
||
if len(distinct) < MIN_DISTINCT_LEVELS:
|
||
return None
|
||
return {value: rank for rank, value in enumerate(distinct)}
|
||
|
||
|
||
def strategy_outline(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
"""Outline level Excel (группировки строк), сдвинутый так, чтобы минимальный уровень стал 0.
|
||
|
||
Сдвиг нужен, когда вся таблица целиком вложена в группировку (уровни 1, 2, … без 0); пропуски уровней
|
||
сохраняются как в файле. `None`, если различных уровней меньше двух.
|
||
"""
|
||
outlines = [item.outline for item in _paced(rows, deadline, "иерархия (outline)")]
|
||
if len(set(outlines)) < MIN_DISTINCT_LEVELS:
|
||
return None
|
||
base = min(outlines)
|
||
return {item.row: item.outline - base for item in rows}
|
||
|
||
|
||
def strategy_indent(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
"""Отступ ячейки текста: уровень — ранг отступа среди различных отступов листа.
|
||
|
||
`indent` приходит либо как `alignment.indent`, либо как число ведущих пробелов (T2.4), поэтому шаг отступа
|
||
не предполагается: (0, 2, 4), (0, 1, 2) и (0, 3, 7) дают одинаковые уровни 0, 1, 2. `None`, если различных
|
||
отступов меньше двух.
|
||
"""
|
||
ranks = _ranks([item.indent for item in _paced(rows, deadline, "иерархия (indent)")])
|
||
if ranks is None:
|
||
return None
|
||
return {item.row: ranks[item.indent] for item in rows}
|
||
|
||
|
||
def strategy_account_pattern(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
"""Шаблон кода счёта и жирность.
|
||
|
||
Строка-счёт: уровень = ранг кода (число сегментов) − минимальный ранг среди счетов листа (60 → 0, 60.01 → 1,
|
||
если 60 есть; если таблица начинается с субсчетов, они — корни). Не-счёт: уровень последнего счёта + 1.
|
||
|
||
ASSUMPTION (правило брифа Задачи 6, §1.0 п. 5 «по жирности»): жирная не-счётная строка — заголовок группировки
|
||
(например, контрагент); жирные не-счётные строки внутри одного счёта — соседи (не глубже друг друга), а
|
||
следующие за жирной обычные строки — её дети (+1). Если жирных не-счётных строк под счётом нет, все не-счётные
|
||
строки — дети счёта одного уровня; глубже одной группировки по жирности стратегия не различает (CURRENCY/QUANTITY
|
||
и договоры под обычным контрагентом окажутся его соседями — это поймают контрольные суммы §1.4).
|
||
Не-счётные строки до первого счёта считаются корнями. `None`, если на листе нет ни одного счёта.
|
||
"""
|
||
ranks = [item.account_rank for item in rows if item.is_account and item.account_rank is not None]
|
||
if not ranks:
|
||
return None
|
||
base = min(ranks)
|
||
by_row: dict[int, int] = {}
|
||
account_level: int | None = None
|
||
group_level: int | None = None # уровень последней жирной не-счётной строки под текущим счётом
|
||
for item in _paced(rows, deadline, "иерархия (account_pattern)"):
|
||
if item.is_account and item.account_rank is not None:
|
||
level = item.account_rank - base
|
||
account_level, group_level = level, None
|
||
else:
|
||
child_level = ROOT_LEVEL if account_level is None else account_level + 1
|
||
if item.bold:
|
||
level = child_level
|
||
group_level = level
|
||
elif group_level is not None:
|
||
level = group_level + 1
|
||
else:
|
||
level = child_level
|
||
by_row[item.row] = level
|
||
return by_row
|
||
|
||
|
||
def _vector(item: BodyRow) -> tuple[Decimal, ...]:
|
||
return tuple(_ZERO if value is None else value for value in item.sums)
|
||
|
||
|
||
def _add(left: tuple[Decimal, ...], right: tuple[Decimal, ...]) -> tuple[Decimal, ...]:
|
||
return tuple(a + b for a, b in zip(left, right, strict=True))
|
||
|
||
|
||
def _matches(left: tuple[Decimal, ...], right: tuple[Decimal, ...]) -> bool:
|
||
return all(abs(a - b) <= SUM_TOLERANCE for a, b in zip(left, right, strict=True))
|
||
|
||
|
||
def strategy_header_account(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
"""Плоский список под счётом шапки (#59): все строки — уровень 0.
|
||
|
||
Применима, только если ни одна строка тела не является счётом (`is_account`): счёт назван в шапке, а строки таблицы
|
||
— его аналитика без оформления. Условие «в шапке есть код счёта» проверяет `resolve_levels`.
|
||
"""
|
||
if any(item.is_account for item in _paced(rows, deadline, "иерархия (header_account)")):
|
||
return None
|
||
return {item.row: ROOT_LEVEL for item in rows}
|
||
|
||
|
||
def strategy_sums(rows: Sequence[BodyRow], deadline: Deadline | None = None) -> dict[int, int] | None:
|
||
"""Последнее средство: строка — родитель следующих поддеревьев, пока их сумма не совпадёт с её суммами.
|
||
|
||
Жадно и без рекурсии: поддеревья строятся с конца листа (`end[i]` — граница поддерева строки `i`), строка `i`
|
||
перебирает следующие поддеревья подряд и становится родителем на первом совпадении по всем шести колонкам
|
||
(± `SUM_TOLERANCE`; пусто = 0). Строка без сумм (все пусто/0) родителем не бывает. Соседи с равными суммами
|
||
могут быть приняты за родителя и ребёнка — это ограничение жадного правила; результат всё равно проходит
|
||
контрольные суммы §1.4. `None`, если ни одна строка не стала родителем.
|
||
|
||
Худший случай (ни одна строка не сходится) — O(n²) шагов перебора: каждая строка проходит все следующие
|
||
поддеревья до конца листа. Поэтому дедлайн проверяется во внутреннем цикле каждые `DEADLINE_CHECK_EVERY` шагов,
|
||
и такой файл завершается `TIMEOUT`, а не висит.
|
||
"""
|
||
count = len(rows)
|
||
vectors = [_vector(item) for item in rows]
|
||
zero = tuple(_ZERO for _ in vectors[0]) if vectors else ()
|
||
end = [index + 1 for index in range(count)] # граница (исключительно) поддерева строки
|
||
has_children = False
|
||
pacer = _Pacer(deadline, "иерархия (sums): перебор поддеревьев")
|
||
for index in reversed(range(count)):
|
||
pacer.tick()
|
||
target = vectors[index]
|
||
if all(value == _ZERO for value in target):
|
||
continue
|
||
accumulated = zero
|
||
position = index + 1
|
||
while position < count:
|
||
pacer.tick()
|
||
accumulated = _add(accumulated, vectors[position])
|
||
position = end[position]
|
||
if _matches(accumulated, target):
|
||
end[index] = position
|
||
has_children = True
|
||
break
|
||
if not has_children:
|
||
return None
|
||
by_row: dict[int, int] = {}
|
||
open_ends: list[int] = [] # границы поддеревьев открытых предков
|
||
for index, item in enumerate(rows):
|
||
while open_ends and open_ends[-1] <= index:
|
||
open_ends.pop()
|
||
by_row[item.row] = len(open_ends)
|
||
if end[index] > index + 1:
|
||
open_ends.append(end[index])
|
||
return by_row
|
||
|
||
|
||
STRATEGIES: tuple[tuple[str, Strategy], ...] = (
|
||
("outline", strategy_outline),
|
||
("indent", strategy_indent),
|
||
("account_pattern", strategy_account_pattern),
|
||
("sums", strategy_sums),
|
||
)
|
||
|
||
|
||
def is_consistent(rows: Sequence[BodyRow], levels: Mapping[int, int], total_row: int | None) -> bool:
|
||
"""Уровни образуют дерево: у каждой строки есть уровень ≥ 0, первая строка — корень, у каждой строки уровня > 0
|
||
есть предыдущая строка с меньшим уровнем, «Итого» (если есть) — уровень 0.
|
||
"""
|
||
lowest_seen: int | None = None
|
||
for item in rows:
|
||
level = levels.get(item.row)
|
||
if level is None or level < ROOT_LEVEL:
|
||
return False
|
||
if level > ROOT_LEVEL and (lowest_seen is None or lowest_seen >= level):
|
||
return False
|
||
lowest_seen = level if lowest_seen is None else min(lowest_seen, level)
|
||
return total_row is None or levels.get(total_row) == ROOT_LEVEL
|
||
|
||
|
||
def parents(rows: Sequence[BodyRow], by_row: Mapping[int, int]) -> dict[int, int | None]:
|
||
"""Родитель — ближайшая предыдущая строка с меньшим уровнем (`None` для корней); по позициям, не по текстам."""
|
||
result: dict[int, int | None] = {}
|
||
stack: list[tuple[int, int]] = [] # (уровень, номер строки) открытых предков
|
||
for item in rows:
|
||
level = by_row[item.row]
|
||
while stack and stack[-1][0] >= level:
|
||
stack.pop()
|
||
result[item.row] = stack[-1][1] if stack else None
|
||
stack.append((level, item.row))
|
||
return result
|
||
|
||
|
||
def resolve_levels(
|
||
rows: Sequence[BodyRow], total_row: int | None, deadline: Deadline, *, header_account: bool = False
|
||
) -> Levels:
|
||
"""Первая стратегия каскада, давшая непротиворечивое дерево; иначе `InvalidFormatError`.
|
||
|
||
`header_account` — в шапке есть код счёта: тогда после `account_pattern` и перед `sums` пробуется `header_account`
|
||
(плоский список под счётом шапки, #59).
|
||
|
||
«Итого» (`total_row`) и строки после неё (вторые итоговые) в стратегии не передаются и получают уровень 0: это
|
||
контрольная сумма документа, а не ребёнок последнего счёта. Меньше двух строк, кроме «Итого», — тривиальное
|
||
дерево корней `TRIVIAL_STRATEGY`.
|
||
"""
|
||
# «Итого» и вторые итоговые строки за ней («Итого развернутое») — не данные: уровень 0 без участия стратегий
|
||
totals = [] if total_row is None else [item for item in rows if item.row >= total_row]
|
||
body = [item for item in rows if total_row is None or item.row < total_row]
|
||
pinned = {item.row: ROOT_LEVEL for item in totals}
|
||
if len(body) < MIN_ROWS_FOR_CASCADE:
|
||
by_row = {item.row: ROOT_LEVEL for item in rows}
|
||
return Levels(by_row=by_row, parent=parents(rows, by_row), strategy=TRIVIAL_STRATEGY)
|
||
cascade = list(STRATEGIES)
|
||
if header_account:
|
||
before_sums = next((index for index, (name, _) in enumerate(cascade) if name == "sums"), len(cascade))
|
||
cascade.insert(before_sums, ("header_account", strategy_header_account))
|
||
for name, strategy in cascade:
|
||
deadline.check(f"иерархия ({name})")
|
||
found = strategy(body, deadline)
|
||
if found is None:
|
||
continue
|
||
by_row = {**found, **pinned}
|
||
if is_consistent(rows, by_row, total_row):
|
||
return Levels(by_row=by_row, parent=parents(rows, by_row), strategy=name)
|
||
raise InvalidFormatError("иерархия строк не определена")
|