239 lines
11 KiB
Python
239 lines
11 KiB
Python
"""Шаг 2: колонки значений, подписи и даты колонок."""
|
||
|
||
from collections.abc import Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
from arm_statements.amounts import parse_amount
|
||
from arm_statements.codes import is_code_cell
|
||
from arm_statements.dates import expected_dates, parse_period_label, resolve_relative
|
||
from arm_statements.issues import WarningCode, make_warning
|
||
from arm_statements.layout import Grid, cell_at
|
||
from arm_statements.loader import Rules
|
||
from arm_statements.model import Cell, PeriodKind, Source, StatementColumn, StatementHeader, StatementWarning
|
||
|
||
_FULL_CONFIDENCE = 100.0
|
||
_MAX_REFS = 10
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ColumnDraft:
|
||
"""Колонка сегмента до сопоставления; `date` — только у абсолютной подписи."""
|
||
|
||
grid_col: int
|
||
label: str
|
||
label_confidence: float | None
|
||
kind: PeriodKind
|
||
date: date | None
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class DocumentDates:
|
||
effective: date | None
|
||
confident_effective: date | None
|
||
latest_absolute: date | None
|
||
confident_latest_absolute: date | None
|
||
|
||
|
||
def _text(cell: Cell | None) -> str:
|
||
return "" if cell is None else cell.text.strip()
|
||
|
||
|
||
def _code_rows(rows: Grid, data_rows: Sequence[int], code_col: int, rules: Rules) -> list[int]:
|
||
return [r for r in data_rows if is_code_cell(_text(cell_at(rows, r, code_col)), rules)]
|
||
|
||
|
||
def _is_number(cell: Cell | None, source: Source, rules: Rules) -> bool:
|
||
if cell is None:
|
||
return False
|
||
if isinstance(cell.value, Decimal):
|
||
return True
|
||
return parse_amount(cell.text, source, rules).amount is not None
|
||
|
||
|
||
def has_numbers( # noqa: PLR0913, PLR0917 - сигнатура задана планом (Task 9)
|
||
rows: Grid, data_rows: Sequence[int], col: int, code_col: int, source: Source, rules: Rules
|
||
) -> bool:
|
||
"""В строках с кодом есть ячейка столбца с числом (значение-число или разобранная сумма)."""
|
||
return any(_is_number(cell_at(rows, r, col), source, rules) for r in _code_rows(rows, data_rows, code_col, rules))
|
||
|
||
|
||
def _width(rows: Grid, row_indexes: Sequence[int]) -> int:
|
||
return max((len(rows[r]) for r in row_indexes if 0 <= r < len(rows)), default=0)
|
||
|
||
|
||
def amount_columns(rows: Grid, code_col: int, rules: Rules, *, source: Source = Source.EXCEL) -> tuple[int, ...]:
|
||
"""Столбцы правее кодов, где в строках с кодом есть числа, по возрастанию."""
|
||
every = range(len(rows))
|
||
return tuple(
|
||
col
|
||
for col in range(code_col + 1, _width(rows, every))
|
||
if has_numbers(rows, every, col, code_col, source, rules)
|
||
)
|
||
|
||
|
||
def _label(rows: Grid, header_rows: tuple[int, ...], col: int) -> Cell | None:
|
||
"""Подпись колонки: строки заголовка через пробел, уверенность — минимум; одна ячейка берётся как есть."""
|
||
cells = [c for r in header_rows if (c := cell_at(rows, r, col)) is not None and c.text.strip()]
|
||
if not cells:
|
||
return None
|
||
if len(cells) == 1:
|
||
return cells[0]
|
||
text = " ".join(c.text.strip() for c in cells)
|
||
return Cell(text, None, min(c.confidence for c in cells), cells[0].ref)
|
||
|
||
|
||
def segment_columns( # noqa: PLR0913, PLR0917 - сигнатура задана планом (Task 9)
|
||
rows: Grid,
|
||
header_rows: tuple[int, ...],
|
||
data_rows: Sequence[int],
|
||
code_col: int,
|
||
source: Source,
|
||
rules: Rules,
|
||
) -> tuple[list[ColumnDraft], list[StatementWarning]]:
|
||
"""Колонки значений сегмента и предупреждения о пропущенных/недатированных столбцах."""
|
||
drafts: list[ColumnDraft] = []
|
||
warnings: list[StatementWarning] = []
|
||
columns = range(code_col + 1, _width(rows, [*header_rows, *data_rows]))
|
||
last_dated = _last_dated_column(rows, header_rows, data_rows, columns, code_col, source, rules)
|
||
outside: list[int] = []
|
||
for col in columns:
|
||
numbers = has_numbers(rows, data_rows, col, code_col, source, rules)
|
||
if not header_rows:
|
||
if numbers:
|
||
drafts.append(ColumnDraft(col, "", None, PeriodKind.NONE, None))
|
||
continue
|
||
cell = _label(rows, header_rows, col)
|
||
parsed = parse_period_label(cell, rules)
|
||
if cell is not None and parsed.excluded:
|
||
message = f"столбец {col} («{cell.text}») пропущен: не период"
|
||
warnings.append(make_warning(WarningCode.COLUMN_SKIPPED, message, rules, column_index=col))
|
||
elif cell is not None and parsed.kind is not PeriodKind.NONE:
|
||
drafts.append(ColumnDraft(col, cell.text, cell.confidence, parsed.kind, parsed.date))
|
||
elif numbers and cell is None and last_dated is not None and col > last_dated:
|
||
outside.append(col) # числа правее последней колонки с подписью периода — не колонка сумм
|
||
elif numbers:
|
||
label, conf = (cell.text, cell.confidence) if cell is not None else ("", None)
|
||
message = f"у столбца {col} с числами нет разобранной подписи периода"
|
||
warnings.append(make_warning(WarningCode.COLUMN_WITHOUT_DATE, message, rules, column_index=col))
|
||
drafts.append(ColumnDraft(col, label, conf, PeriodKind.NONE, None))
|
||
if outside:
|
||
warnings.append(_outside_table_warning(rows, data_rows, outside, code_col, source, rules))
|
||
return drafts, warnings
|
||
|
||
|
||
def _last_dated_column( # noqa: PLR0913, PLR0917 - внутренняя сборка границы таблицы
|
||
rows: Grid,
|
||
header_rows: tuple[int, ...],
|
||
data_rows: Sequence[int],
|
||
columns: range,
|
||
code_col: int,
|
||
source: Source,
|
||
rules: Rules,
|
||
) -> int | None:
|
||
"""Правый столбец сегмента с подписью периода и числами в строках с кодами; без шапки — `None`.
|
||
|
||
Подписанный столбец без чисел (потерянная подпись объединённой ячейки) границей таблицы не служит.
|
||
"""
|
||
dated = [
|
||
col
|
||
for col in columns
|
||
if header_rows
|
||
and (cell := _label(rows, header_rows, col)) is not None
|
||
and not (parsed := parse_period_label(cell, rules)).excluded
|
||
and parsed.kind is not PeriodKind.NONE
|
||
and has_numbers(rows, data_rows, col, code_col, source, rules)
|
||
]
|
||
return max(dated, default=None)
|
||
|
||
|
||
def _outside_table_warning( # noqa: PLR0913, PLR0917 - внутренняя сборка предупреждения
|
||
rows: Grid, data_rows: Sequence[int], outside: Sequence[int], code_col: int, source: Source, rules: Rules
|
||
) -> StatementWarning:
|
||
refs = [
|
||
cell.ref
|
||
for r in _code_rows(rows, data_rows, code_col, rules)
|
||
for col in outside
|
||
if (cell := cell_at(rows, r, col)) is not None and _is_number(cell, source, rules)
|
||
]
|
||
return outside_table_warning(refs, rules)
|
||
|
||
|
||
def outside_table_warning(refs: Sequence[str], rules: Rules) -> StatementWarning:
|
||
"""`NUMBERS_OUTSIDE_TABLE` (info): адреса ячеек правее таблицы (`locator` — все, через перевод строки)."""
|
||
shown = ", ".join(refs[:_MAX_REFS]) + (f" и ещё {len(refs) - _MAX_REFS}" if len(refs) > _MAX_REFS else "")
|
||
message = f"числа правее таблицы формы не учитываются: {shown}"
|
||
return make_warning(WarningCode.NUMBERS_OUTSIDE_TABLE, message, rules, locator="\n".join(refs))
|
||
|
||
|
||
def merge_outside_table(warnings: Sequence[StatementWarning], rules: Rules) -> list[StatementWarning]:
|
||
"""Предупреждения `NUMBERS_OUTSIDE_TABLE` сегментов — в одно (адреса без повторов)."""
|
||
refs: dict[str, None] = {}
|
||
rest: list[StatementWarning] = []
|
||
for warning in warnings:
|
||
if warning.code == WarningCode.NUMBERS_OUTSIDE_TABLE:
|
||
refs.update(dict.fromkeys((warning.locator or "").split("\n")))
|
||
else:
|
||
rest.append(warning)
|
||
return [*rest, outside_table_warning(list(refs), rules)] if refs else rest
|
||
|
||
|
||
def trailing_outside_refs( # noqa: PLR0913, PLR0917 - внутренняя сборка
|
||
rows: Grid, data_rows: Sequence[int], cols: Sequence[int], code_col: int, source: Source, rules: Rules
|
||
) -> list[str]:
|
||
"""Адреса чисел в строках с кодами по столбцам `cols` (хвост сегмента правее колонок документа)."""
|
||
return [
|
||
cell.ref
|
||
for r in _code_rows(rows, data_rows, code_col, rules)
|
||
for col in cols
|
||
if (cell := cell_at(rows, r, col)) is not None and _is_number(cell, source, rules)
|
||
]
|
||
|
||
|
||
def _latest(drafts: Sequence[ColumnDraft], *, confident: bool) -> date | None:
|
||
days = [
|
||
d.date
|
||
for d in drafts
|
||
if d.kind is PeriodKind.ABSOLUTE
|
||
and d.date is not None
|
||
and (not confident or d.label_confidence == _FULL_CONFIDENCE)
|
||
]
|
||
return max(days, default=None)
|
||
|
||
|
||
def document_dates(drafts: Sequence[ColumnDraft], header: StatementHeader) -> DocumentDates:
|
||
"""Эффективная дата и её «уверенный» двойник: неуверенное само себя не подтверждает."""
|
||
latest = _latest(drafts, confident=False)
|
||
confident_latest = _latest(drafts, confident=True)
|
||
report = header.report_date
|
||
sure_report = report.value if report.confidence == _FULL_CONFIDENCE else None
|
||
return DocumentDates(
|
||
effective=report.value or latest,
|
||
confident_effective=sure_report or confident_latest,
|
||
latest_absolute=latest,
|
||
confident_latest_absolute=confident_latest,
|
||
)
|
||
|
||
|
||
def finalize_columns(
|
||
drafts: Sequence[ColumnDraft], dates: DocumentDates, rules: Rules
|
||
) -> tuple[tuple[StatementColumn, ...], list[StatementWarning]]:
|
||
"""Колонки документа: даты относительных подписей и проверка неуверенных абсолютных."""
|
||
expected = None if dates.confident_effective is None else expected_dates(dates.confident_effective)
|
||
columns: list[StatementColumn] = []
|
||
warnings: list[StatementWarning] = []
|
||
for draft in drafts:
|
||
if draft.kind is PeriodKind.ABSOLUTE:
|
||
period = draft.date
|
||
unsure = draft.label_confidence is not None and draft.label_confidence < _FULL_CONFIDENCE
|
||
if unsure and (expected is None or period not in expected):
|
||
message = f"дата подписи столбца {draft.grid_col} («{draft.label}») не подтверждена"
|
||
warnings.append(
|
||
make_warning(WarningCode.COLUMN_DATE_UNCERTAIN, message, rules, column_index=draft.grid_col)
|
||
)
|
||
else:
|
||
period = resolve_relative(draft.kind, dates.effective)
|
||
columns.append(StatementColumn(draft.grid_col, draft.label, draft.label_confidence, period, draft.kind))
|
||
return tuple(columns), warnings
|