pylib_ai_arm_statements/arm_statements/pipeline.py

"""Шаг 6 и конвейер кандидата: шаги 2–8 одного кандидата и нормализация набора кандидатов."""

from collections.abc import Sequence
from dataclasses import replace

from arm_statements.checks import checks_from
from arm_statements.codes import document_type_of, majority_section
from arm_statements.columns import DocumentDates
from arm_statements.evaluation import EvalLine, EvalValue
from arm_statements.header import parse_header
from arm_statements.issues import WarningCode, make_warning
from arm_statements.lines import build_lines, units_warnings
from arm_statements.loader import Rules, default_rules
from arm_statements.model import (
    CandidateResult,
    DocStatus,
    DocumentType,
    FormCandidate,
    FormVariant,
    LineStatus,
    NormalizationResult,
    Source,
    StatementDocument,
    StatementHeader,
    StatementLine,
    StatementWarning,
)
from arm_statements.review import evaluate, review_header
from arm_statements.rows import RawLine, collect_raw_lines
from arm_statements.segments import Layout, find_layout
from arm_statements.status import document_status, status_input
from arm_statements.text import has_digit, is_disguised_digit

_SHEET_PREFIX = "sheet:"


def _eval_lines(lines: Sequence[StatementLine]) -> list[EvalLine]:
    """Строки сервиса в общий вид шагов 6–8: ключ — `ref`, колонка — `str(column_index)`."""
    return [
        EvalLine(
            key=line.ref,
            order=line.row_index,
            line_status=line.line_status,
            line_code=line.line_code,
            code_confidence=line.line_code_confidence,
            code_fixed=False,
            values=tuple(
                EvalValue(str(v.column_index), v.amount, v.text, v.confidence, v.is_recognized, fixed=False)
                for v in line.values
            ),
        )
        for line in lines
    ]


def _form_name(candidate: FormCandidate) -> str | None:
    title = candidate.evidence.title
    if title is None or not title.text.strip():
        return None
    return title.text


def _has_content(candidate: FormCandidate, rules: Rules) -> bool:
    """Содержимое таблицы: ячейка с цифрой или «цифрой под видом буквы»."""
    return any(
        cell is not None and (has_digit(cell.text) or is_disguised_digit(cell.text, rules.lookalikes))
        for row in candidate.rows
        for cell in row
    )


def _lost_table(candidate: FormCandidate, rules: Rules) -> StatementWarning:
    """Документа нет (нет кодов и `type_hint`): предупреждение файла по источнику и содержимому."""
    if candidate.source is Source.EXCEL:
        sheet = candidate.locator.removeprefix(_SHEET_PREFIX).split(",")[0]
        message = f"лист «{sheet}» не содержит формы ББ/ОФР"
        return make_warning(WarningCode.SHEET_OTHER, message, rules, locator=f"{_SHEET_PREFIX}{sheet}")
    if _has_content(candidate, rules):
        message = "таблица с данными не распознана как форма: нет столбца кодов и типа формы"
        return make_warning(WarningCode.TABLE_LOST, message, rules, locator=candidate.locator)
    return make_warning(WarningCode.TABLE_SKIPPED, "таблица без данных пропущена", rules, locator=candidate.locator)


def _without_lines(  # noqa: PLR0913 -- сборка документа без строк: поля заданы схемой
    candidate: FormCandidate,
    document_type: DocumentType,
    header: StatementHeader,
    warnings: Sequence[StatementWarning],
    *,
    dates: DocumentDates | None,
    rules: Rules,
) -> StatementDocument:
    """Документ без строк: `DISCLOSURE` (`dates=None` — шапку не проверяем, P1-6) или без столбца кодов (P1-19)."""
    collected = list(warnings)
    if dates is not None:
        header, header_warnings = review_header(header, dates, rules)
        collected.extend(header_warnings)
    document = StatementDocument(
        document_type=document_type,
        form_variant=None if document_type is DocumentType.DISCLOSURE else FormVariant.FULL,
        locator=candidate.locator,
        header=header,
        columns=(),
        lines=(),
        checks=None,
        status=DocStatus.DONE,
        warnings=tuple(collected),
    )
    return replace(document, status=document_status(status_input(document)))


def _lines_document(  # noqa: PLR0913 -- сборка документа со строками: поля заданы схемой
    candidate: FormCandidate,
    document_type: DocumentType,
    header: StatementHeader,
    layout: Layout,
    *,
    raw: Sequence[RawLine],
    row_warnings: Sequence[StatementWarning],
    conflict: Sequence[StatementWarning],
    rules: Rules,
) -> StatementDocument:
    lines, range_warnings = build_lines(raw, candidate.source, rules)
    units = units_warnings(lines, header, candidate.warnings, rules)
    column_keys = [str(column.column_index) for column in layout.columns]
    evaluation = evaluate(_eval_lines(lines), column_keys, document_type, layout.dates.effective, rules, accepted=False)
    checks = checks_from(evaluation.outcomes, {key: int(key) for key in column_keys})
    header, header_warnings = review_header(header, dates=layout.dates, rules=rules)
    reviewed = tuple(
        replace(
            line,
            needs_review=evaluation.line_needs_review.get(line.ref, False),
            values=tuple(
                replace(v, needs_review=evaluation.value_needs_review.get((line.ref, str(v.column_index)), False))
                for v in line.values
            ),
        )
        for line in lines
    )
    document = StatementDocument(
        document_type=document_type,
        form_variant=evaluation.variant,
        locator=candidate.locator,
        header=header,
        columns=layout.columns,
        lines=reviewed,
        checks=checks,
        status=DocStatus.DONE,
        warnings=(
            *candidate.warnings,
            *layout.warnings,
            *row_warnings,
            *range_warnings,
            *units,
            *conflict,
            *header_warnings,
        ),
    )
    return replace(document, status=document_status(status_input(document)))


def normalize_candidate(candidate: FormCandidate, rules: Rules | None = None) -> CandidateResult:
    """Шаги 2–8 кандидата; документа нет — предупреждение файла (`TABLE_LOST`, `TABLE_SKIPPED`, `SHEET_OTHER`).

    Когда документа нет, `candidate.warnings` (предупреждения адаптера) никуда не попадают: хост обрабатывает их сам.
    """
    rules = rules if rules is not None else default_rules()
    header = replace(parse_header(candidate.header, rules), form_name=_form_name(candidate))
    layout = find_layout(candidate, header, rules)
    hint = candidate.type_hint

    if not layout.code_found:
        if hint is None:
            return CandidateResult(None, (_lost_table(candidate, rules),))
        base = [*candidate.warnings, *layout.warnings]
        if hint is DocumentType.DISCLOSURE:
            return CandidateResult(
                _without_lines(candidate, hint, header, candidate.warnings, dates=None, rules=rules), ()
            )
        message = "столбец кодов строк не найден"
        base.append(make_warning(WarningCode.CODE_COLUMN_NOT_FOUND, message, rules))
        return CandidateResult(_without_lines(candidate, hint, header, base, dates=layout.dates, rules=rules), ())

    raw, row_warnings = collect_raw_lines(candidate, layout, rules)
    ok_codes = [item.code for item in raw if item.line_status is LineStatus.OK and item.code is not None]
    section = majority_section(ok_codes, rules)
    if section is not None:
        document_type = document_type_of(section)
    elif hint is not None:
        document_type = hint
    else:
        return CandidateResult(None, (_lost_table(candidate, rules),))

    conflict: list[StatementWarning] = []
    if hint is not None and hint is not document_type:
        message = f"тип по кодам строк ({document_type}) противоречит типу по заголовку ({hint})"
        conflict.append(make_warning(WarningCode.TYPE_CONFLICT, message, rules))

    if document_type is DocumentType.DISCLOSURE:
        warnings = [*candidate.warnings, *conflict]
        return CandidateResult(_without_lines(candidate, document_type, header, warnings, dates=None, rules=rules), ())
    document = _lines_document(
        candidate, document_type, header, layout, raw=raw, row_warnings=row_warnings, conflict=conflict, rules=rules
    )
    return CandidateResult(document, ())


def normalize(candidates: Sequence[FormCandidate], rules: Rules | None = None) -> NormalizationResult:
    """Кандидаты по порядку; исключения не перехватываются (изоляцию кандидатов делает хост)."""
    rules = rules if rules is not None else default_rules()
    documents: list[StatementDocument] = []
    file_warnings: list[StatementWarning] = []
    for candidate in candidates:
        result = normalize_candidate(candidate, rules)
        if result.document is not None:
            documents.append(result.document)
        file_warnings.extend(result.file_warnings)
    return NormalizationResult(tuple(documents), tuple(file_warnings))