feat: шаг 8 — needs_review по покрытию §7 и конвенции знака, проверки полей шапки
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
parent
e03040d16d
commit
6067893db8
171
arm_statements/review.py
Normal file
171
arm_statements/review.py
Normal file
@ -0,0 +1,171 @@
|
||||
"""Шаг 8: `needs_review` значений и строк, проверки полей шапки, сводная `evaluate`."""
|
||||
|
||||
from collections import Counter
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import date
|
||||
|
||||
from arm_statements.amounts import sign_form_of
|
||||
from arm_statements.checks import RelationOutcome, evaluate_relations, form_variant
|
||||
from arm_statements.columns import DocumentDates
|
||||
from arm_statements.evaluation import EvalLine, EvalValue, is_check_line
|
||||
from arm_statements.issues import WarningCode, make_warning
|
||||
from arm_statements.loader import Rules
|
||||
from arm_statements.model import (
|
||||
DocumentType,
|
||||
Edition,
|
||||
FormVariant,
|
||||
HeaderField,
|
||||
LineStatus,
|
||||
SignForm,
|
||||
StatementHeader,
|
||||
StatementWarning,
|
||||
)
|
||||
from arm_statements.structure import StructureResult, confirm_codes
|
||||
|
||||
_CONFIDENT = 100
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Evaluation:
|
||||
variant: FormVariant | None
|
||||
edition: Edition | None
|
||||
outcomes: tuple[RelationOutcome, ...]
|
||||
line_needs_review: Mapping[str, bool]
|
||||
value_needs_review: Mapping[tuple[str, str], bool]
|
||||
hard_failures: int
|
||||
|
||||
|
||||
def _needs_coverage(value: EvalValue) -> bool:
|
||||
return (value.confidence is not None and value.confidence < _CONFIDENT) or not value.is_recognized
|
||||
|
||||
|
||||
def _is_covered(line: EvalLine, value: EvalValue, outcomes: Sequence[RelationOutcome]) -> bool:
|
||||
if value.amount is None or line.line_code is None:
|
||||
return False
|
||||
for item in outcomes:
|
||||
if item.column != value.column or line.line_code not in item.codes or not item.computed:
|
||||
continue
|
||||
if item.within_tolerance or (item.rule.strict and item.within_rounding):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _sign_confirmed(
|
||||
line: EvalLine, value: EvalValue, lines: Sequence[EvalLine], abs_codes: frozenset[str], rules: Rules
|
||||
) -> bool:
|
||||
"""Форма знака ячейки совпадает с формой строгого большинства прочих членов `abs_terms` колонки."""
|
||||
forms: Counter[SignForm] = Counter()
|
||||
for other in lines:
|
||||
if other is line or not is_check_line(other) or other.line_code not in abs_codes:
|
||||
continue
|
||||
for item in other.values:
|
||||
if item.column != value.column or item.amount is None or not item.is_recognized or item.fixed:
|
||||
continue
|
||||
form = sign_form_of(item.text, rules)
|
||||
if form is not None:
|
||||
forms[form] += 1
|
||||
total = sum(forms.values())
|
||||
own = sign_form_of(value.text, rules)
|
||||
return own is not None and forms[own] * 2 > total
|
||||
|
||||
|
||||
def value_needs_review(
|
||||
lines: Sequence[EvalLine],
|
||||
outcomes: Sequence[RelationOutcome],
|
||||
abs_codes: frozenset[str],
|
||||
rules: Rules,
|
||||
*,
|
||||
accepted: bool = False,
|
||||
) -> dict[tuple[str, str], bool]:
|
||||
hard_codes: set[tuple[str, str]] = set()
|
||||
if not accepted:
|
||||
for item in outcomes:
|
||||
if item.hard_failed:
|
||||
hard_codes |= {(item.column, code) for code in item.codes}
|
||||
result: dict[tuple[str, str], bool] = {}
|
||||
for line in lines:
|
||||
for value in line.values:
|
||||
key = (line.key, value.column)
|
||||
if line.line_status is LineStatus.DETAIL or value.fixed:
|
||||
result[key] = False
|
||||
elif is_check_line(line) and (value.column, line.line_code or "") in hard_codes:
|
||||
result[key] = True
|
||||
elif not _needs_coverage(value):
|
||||
result[key] = False
|
||||
else:
|
||||
covered = _is_covered(line, value, outcomes)
|
||||
signed = line.line_code not in abs_codes or _sign_confirmed(line, value, lines, abs_codes, rules)
|
||||
result[key] = not (covered and signed)
|
||||
return result
|
||||
|
||||
|
||||
def line_needs_review(lines: Sequence[EvalLine], structure: StructureResult) -> dict[str, bool]:
|
||||
return {
|
||||
line.key: line.key in structure.unconfirmed and not line.code_fixed and line.line_status is LineStatus.OK
|
||||
for line in lines
|
||||
}
|
||||
|
||||
|
||||
def evaluate( # noqa: PLR0913 -- сигнатура задана публичным API пакета
|
||||
lines: Sequence[EvalLine],
|
||||
columns: Sequence[str],
|
||||
document_type: DocumentType,
|
||||
effective: date | None,
|
||||
rules: Rules,
|
||||
*,
|
||||
accepted: bool = False,
|
||||
) -> Evaluation:
|
||||
codes = [item.line_code for item in lines if is_check_line(item) and item.line_code is not None]
|
||||
variant = form_variant(document_type, codes)
|
||||
outcomes = evaluate_relations(lines, columns, document_type, variant, rules)
|
||||
structure = confirm_codes(lines, document_type, variant, effective, rules)
|
||||
abs_codes = rules.checks.abs_codes(document_type, variant or FormVariant.FULL)
|
||||
return Evaluation(
|
||||
variant=variant,
|
||||
edition=structure.edition,
|
||||
outcomes=tuple(outcomes),
|
||||
line_needs_review=line_needs_review(lines, structure),
|
||||
value_needs_review=value_needs_review(lines, outcomes, abs_codes, rules, accepted=accepted),
|
||||
hard_failures=sum(1 for item in outcomes if item.hard_failed),
|
||||
)
|
||||
|
||||
|
||||
def _uncertain(field: HeaderField) -> bool:
|
||||
return field.confidence is not None and field.confidence < _CONFIDENT
|
||||
|
||||
|
||||
def review_header(
|
||||
header: StatementHeader, dates: DocumentDates, rules: Rules
|
||||
) -> tuple[StatementHeader, list[StatementWarning]]:
|
||||
"""Поля шапки (спека «Шаг 8 — поля шапки»). Для `DISCLOSURE` не вызывается: предупреждений шапки там нет (P1-6)."""
|
||||
warnings: list[StatementWarning] = []
|
||||
review_fields: list[str] = []
|
||||
org, inn, report = header.organization_name, header.inn, header.report_date
|
||||
|
||||
def warn(code: WarningCode, message: str, field: str | None = None) -> None:
|
||||
warnings.append(make_warning(code, message, rules, field=field))
|
||||
|
||||
if org.value is None and inn.value is None:
|
||||
warn(WarningCode.NO_ORGANIZATION_NO_INN, "В шапке нет ни организации, ни ИНН")
|
||||
elif inn.value is None:
|
||||
warn(WarningCode.INN_NOT_FOUND, "ИНН не найден", "inn")
|
||||
elif org.value is None:
|
||||
warn(WarningCode.ORGANIZATION_NOT_FOUND, "Организация не найдена", "organization_name")
|
||||
if org.value is not None and _uncertain(org):
|
||||
warn(WarningCode.ORGANIZATION_UNCERTAIN, "Название организации распознано неуверенно", "organization_name")
|
||||
if inn.value is not None and _uncertain(inn):
|
||||
warn(WarningCode.INN_UNCERTAIN, "ИНН распознан неуверенно", "inn")
|
||||
|
||||
if report.value is not None:
|
||||
confidence = report.confidence
|
||||
if confidence is not None and confidence < _CONFIDENT and report.value != dates.confident_latest_absolute:
|
||||
review_fields.append("report_date")
|
||||
warn(WarningCode.REPORT_DATE_UNCERTAIN, "Дата отчёта неуверенна и не подтверждена колонками", "report_date")
|
||||
elif report.text is not None and report.text.strip() and dates.latest_absolute is not None:
|
||||
warn(WarningCode.REPORT_DATE_UNPARSED, "Дата отчёта не разобрана", "report_date")
|
||||
if report.value is None and dates.latest_absolute is None:
|
||||
warn(WarningCode.PERIOD_UNKNOWN, "Отчётный период не определён: нет ни даты отчёта, ни дат колонок")
|
||||
if header.units.value is None:
|
||||
warn(WarningCode.UNITS_NOT_FOUND, "Единицы измерения не найдены", "units")
|
||||
return replace(header, needs_review_fields=tuple(review_fields)), warnings
|
||||
139
arm_statements/tests/test_review.py
Normal file
139
arm_statements/tests/test_review.py
Normal file
@ -0,0 +1,139 @@
|
||||
from datetime import date
|
||||
from decimal import Decimal as D
|
||||
|
||||
from arm_statements.columns import DocumentDates
|
||||
from arm_statements.evaluation import EvalLine, EvalValue
|
||||
from arm_statements.header import parse_header
|
||||
from arm_statements.loader import Rules
|
||||
from arm_statements.model import DocumentType, LineStatus
|
||||
from arm_statements.review import evaluate, review_header
|
||||
from arm_statements.tests.builders import hdr
|
||||
|
||||
OK, DETAIL, UNREADABLE = LineStatus.OK, LineStatus.DETAIL, LineStatus.UNREADABLE
|
||||
NO_DATES = DocumentDates(None, None, None, None)
|
||||
|
||||
|
||||
def ln( # noqa: PLR0913, PLR0917 -- сигнатура задана планом (Task 15)
|
||||
code: str | None,
|
||||
text: str,
|
||||
amount: str | None,
|
||||
conf: float = 100.0,
|
||||
recognized: bool = True,
|
||||
status: LineStatus = OK,
|
||||
order: int = 0,
|
||||
) -> EvalLine:
|
||||
value = EvalValue("0", None if amount is None else D(amount), text, conf, recognized)
|
||||
return EvalLine(f"{code}-{order}", order, status, code, 100.0, False, (value,))
|
||||
|
||||
|
||||
def flags(lines: list[EvalLine], doc: DocumentType, rules: Rules) -> dict[str, bool]:
|
||||
result = evaluate(lines, ["0"], doc, date(2025, 9, 30), rules)
|
||||
return {key: flag for (key, _), flag in result.value_needs_review.items()}
|
||||
|
||||
|
||||
def test_guess_confirmed_by_relation(rules: Rules) -> None:
|
||||
lines = [
|
||||
ln("1210", "100", "100", order=0),
|
||||
ln("1230", "42831 1", "428311", recognized=False, order=1),
|
||||
ln("1200", "428411", "428411", order=2),
|
||||
]
|
||||
assert flags(lines, DocumentType.BALANCE, rules)["1230-1"] is False
|
||||
|
||||
|
||||
def test_sign_in_curly_bracket_confirmed(rules: Rules) -> None:
|
||||
lines = [
|
||||
ln("2110", "10000000", "10000000", order=0),
|
||||
ln("2120", "(7744536}", "-7744536", recognized=False, order=1),
|
||||
ln("2100", "2255464", "2255464", order=2),
|
||||
ln("2210", "(100)", "-100", order=3),
|
||||
ln("2220", "(200)", "-200", order=4),
|
||||
ln("2200", "2255164", "2255164", order=5),
|
||||
]
|
||||
assert flags(lines, DocumentType.INCOME, rules)["2120-1"] is False
|
||||
|
||||
|
||||
def test_sign_outside_convention(rules: Rules) -> None:
|
||||
lines = [
|
||||
ln("2110", "100000", "100000", order=0),
|
||||
ln("2120", "38378", "38378", conf=60.0, order=1),
|
||||
ln("2100", "61622", "61622", order=2),
|
||||
ln("2210", "(100)", "-100", order=3),
|
||||
ln("2220", "(200)", "-200", order=4),
|
||||
]
|
||||
assert flags(lines, DocumentType.INCOME, rules)["2120-1"] is True
|
||||
|
||||
|
||||
def test_null_amount_and_unchecked_code(rules: Rules) -> None:
|
||||
lines = [ln("2120", "{38378)", None, recognized=False, order=0), ln("2900", "15", "15", conf=80.0, order=1)]
|
||||
result = flags(lines, DocumentType.INCOME, rules)
|
||||
assert (result["2120-0"], result["2900-1"]) == (True, True)
|
||||
|
||||
|
||||
def test_detail_values_never_flagged(rules: Rules) -> None:
|
||||
lines = [ln("2110", "100", "100", order=0), ln(None, "1 425 130", "1425130", conf=78.0, status=DETAIL, order=1)]
|
||||
assert flags(lines, DocumentType.INCOME, rules)["None-1"] is False
|
||||
|
||||
|
||||
def test_soft_relation_covers_only_exactly(rules: Rules) -> None:
|
||||
exact = [
|
||||
ln("2300", "1000", "1000", order=0),
|
||||
ln("2410", "200", "200", order=1),
|
||||
ln("2400", "800", "800", conf=66.0, order=2),
|
||||
]
|
||||
assert flags(exact, DocumentType.INCOME, rules)["2400-2"] is False
|
||||
rounding = [
|
||||
ln("2300", "17000000000", "17000000000", order=0),
|
||||
ln("2410", "0", "0", order=1),
|
||||
ln("2400", "17000000009", "17000000009", conf=66.0, order=2),
|
||||
]
|
||||
assert flags(rounding, DocumentType.INCOME, rules)["2400-2"] is True
|
||||
|
||||
|
||||
def test_hard_failure_flags_clean_cells(rules: Rules) -> None:
|
||||
lines = [ln("1210", "100", "100", order=0), ln("1230", "50", "50", order=1), ln("1200", "999", "999", order=2)]
|
||||
result = evaluate(lines, ["0"], DocumentType.BALANCE, None, rules)
|
||||
assert result.hard_failures >= 1
|
||||
assert all(result.value_needs_review[(f"{c}-{i}", "0")] for i, c in enumerate(["1210", "1230", "1200"]))
|
||||
accepted = evaluate(lines, ["0"], DocumentType.BALANCE, None, rules, accepted=True)
|
||||
assert not any(accepted.value_needs_review.values())
|
||||
|
||||
|
||||
def test_unreadable_values(rules: Rules) -> None:
|
||||
lines = [ln(None, "7", "7", conf=60.0, status=UNREADABLE), ln(None, "5", "5", status=UNREADABLE, order=1)]
|
||||
result = flags(lines, DocumentType.BALANCE, rules)
|
||||
assert (result["None-0"], result["None-1"]) == (True, False)
|
||||
|
||||
|
||||
def test_report_date_confirmed_or_not(rules: Rules) -> None:
|
||||
head = parse_header(
|
||||
hdr(report_date=("30.09.2025", 60.0), organization_name="ООО «Пример»", inn="7700000000", units="тыс. руб."),
|
||||
rules,
|
||||
)
|
||||
ok_dates = DocumentDates(date(2025, 9, 30), date(2025, 9, 30), date(2025, 9, 30), date(2025, 9, 30))
|
||||
reviewed, warnings = review_header(head, ok_dates, rules)
|
||||
assert (reviewed.needs_review_fields, warnings) == ((), [])
|
||||
relative_only = DocumentDates(date(2025, 9, 30), None, None, None)
|
||||
reviewed, warnings = review_header(head, relative_only, rules)
|
||||
assert reviewed.needs_review_fields == ("report_date",)
|
||||
found = [(w.code, w.severity.value, w.field) for w in warnings]
|
||||
assert found == [("REPORT_DATE_UNCERTAIN", "error", "report_date")]
|
||||
|
||||
|
||||
def test_header_presence_warnings(rules: Rules) -> None:
|
||||
cells = hdr(organization_name="ООО «Пример»", report_date="30.09.2025", units="тыс. руб.")
|
||||
_, no_inn = review_header(parse_header(cells, rules), NO_DATES, rules)
|
||||
assert [w.code for w in no_inn] == ["INN_NOT_FOUND"]
|
||||
_, nothing = review_header(parse_header(hdr(), rules), NO_DATES, rules)
|
||||
assert {w.code for w in nothing} == {"NO_ORGANIZATION_NO_INN", "PERIOD_UNKNOWN", "UNITS_NOT_FOUND"}
|
||||
|
||||
|
||||
def test_uncertain_inn_and_unparsed_date(rules: Rules) -> None:
|
||||
cells = hdr(organization_name=("ООО «Пример»", 70.0), inn=("7700000000", 80.0), report_date="дата?", units="руб.")
|
||||
dates = DocumentDates(None, None, date(2025, 9, 30), None)
|
||||
reviewed, warnings = review_header(parse_header(cells, rules), dates, rules)
|
||||
assert reviewed.needs_review_fields == ()
|
||||
assert [(w.code, w.field) for w in warnings] == [
|
||||
("ORGANIZATION_UNCERTAIN", "organization_name"),
|
||||
("INN_UNCERTAIN", "inn"),
|
||||
("REPORT_DATE_UNPARSED", "report_date"),
|
||||
]
|
||||
Loading…
x
Reference in New Issue
Block a user