diff --git a/api/src/domain/statement_status.py b/api/src/domain/statement_status.py new file mode 100644 index 0000000..0181b30 --- /dev/null +++ b/api/src/domain/statement_status.py @@ -0,0 +1,107 @@ +"""Условие `RECOGNIZED` при приёме контракта v2.0 — чистые функции (05-domain-rules §9.3). + +«Документ формы при приёме контракта v2.0» (Р-16, Р-18, Р-19) и «Документ ОСВ при приёме контракта v2.0» (Р-28). +Подключаются в приёме результата и общем пересчёте статуса; здесь нет ни БД, ни разбора файла. +""" + +from collections.abc import Hashable, Iterable +from dataclasses import dataclass +from decimal import Decimal +from typing import Any + +from src.domain.statement_lines import LineCatalog +from src.domain.statement_names import names_match +from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementLineStatus + +_CONFIDENT = Decimal(100) +_PERIOD_FORMS = (DocumentType.BALANCE, DocumentType.INCOME) + + +def has_error_warning(warnings: Any) -> bool: + """Среди предупреждений файла есть объект с `severity == "error"`. + Старые строковые предупреждения (1.x), мусор и не-список — не считаются.""" + if not isinstance(warnings, list): + return False + return any(isinstance(item, dict) and item.get("severity") == "error" for item in warnings) + + +def duplicate_period_forms( + forms: Iterable[tuple[Hashable, DocumentType, int | None, PeriodQuarter | None]], +) -> set[Hashable]: + """Ключи форм ББ/ОФР, у которых в файле есть другая форма того же типа и той же пары (год, квартал) (Р-18). + Формы без квартала (и без года) не считаются.""" + groups: dict[tuple[DocumentType, int, PeriodQuarter], list[Hashable]] = {} + for key, doc_type, year, quarter in forms: + if doc_type in _PERIOD_FORMS and year is not None and quarter is not None: + groups.setdefault((doc_type, year, quarter), []).append(key) + return {key for keys in groups.values() if len(keys) > 1 for key in keys} + + +@dataclass(frozen=True) +class NameCheckLine: + """Вход проверки наименования: строка формы с кодом и уверенностью.""" + + key: Hashable + line_status: str + line_code: str | None + line_code_confidence: Decimal | None + line_name: str | None + code_fixed: bool + + +def name_mismatches(lines: Iterable[NameCheckLine], catalog: LineCatalog, form: str) -> list[Hashable]: + """Ключи строк `ok` с неуверенным кодом (`line_code_confidence < 100`), без закреплённого кода (`code_fixed`), + чьё наименование не совпало ни с одним справочным именем кода (Р-16). Кода нет в словаре — несовпадение.""" + result: list[Hashable] = [] + for line in lines: + if line.line_status != StatementLineStatus.ok or line.code_fixed: + continue + confidence = line.line_code_confidence + if confidence is None or confidence >= _CONFIDENT: + continue + if not line.line_code or not names_match(line.line_name, catalog.names(form, line.line_code)): + result.append(line.key) + return result + + +def _recognized(ok: bool) -> DocumentStatus: + return DocumentStatus.RECOGNIZED if ok else DocumentStatus.NEEDS_CHECK + + +def form_status_v2( # noqa: PLR0913 + *, + service_done: bool, + file_error_warning: bool, + counterparty: DocumentCheckResult | None, + period_known: bool, + names_ok: bool, + duplicate_in_file: bool, +) -> DocumentStatus: + """Статус формы ББ/ОФР: `RECOGNIZED` только при выполнении всех условий (Р-16, Р-18, Р-19), иначе `NEEDS_CHECK`.""" + return _recognized( + service_done + and not file_error_warning + and counterparty is not DocumentCheckResult.MISMATCH + and period_known + and names_ok + and not duplicate_in_file + ) + + +def osv_status_v2( + *, + osv_done: bool, + checks_ok: bool, + file_error_warning: bool, + counterparty: DocumentCheckResult | None, + sheet_mapping_warnings: bool, +) -> DocumentStatus: + """Статус ОСВ: `RECOGNIZED` только если лист `done`, проверки без нарушений, нет `error` у файла, контрагент не + `MISMATCH` и нет предупреждений маппинга по листу (Р-28); статус конверта не учитывается.""" + return _recognized( + osv_done + and checks_ok + and not file_error_warning + and counterparty is not DocumentCheckResult.MISMATCH + and not sheet_mapping_warnings + ) diff --git a/api/tests/test_domain_statement_status.py b/api/tests/test_domain_statement_status.py new file mode 100644 index 0000000..4c7d758 --- /dev/null +++ b/api/tests/test_domain_statement_status.py @@ -0,0 +1,122 @@ +"""Условие RECOGNIZED при приёме 2.0 (05-domain-rules §9.3; Р-16, Р-18, Р-19, Р-28).""" + +from decimal import Decimal + +import pytest + +from src.domain.statement_lines import load_line_catalog +from src.domain.statement_status import ( + NameCheckLine, + duplicate_period_forms, + form_status_v2, + has_error_warning, + name_mismatches, + osv_status_v2, +) +from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter + +BAL, INC = DocumentType.BALANCE, DocumentType.INCOME +Q3 = PeriodQuarter.Q3 +Q4 = PeriodQuarter.Q4 +CATALOG = load_line_catalog() + +OK = dict( + service_done=True, + file_error_warning=False, + counterparty=DocumentCheckResult.MATCH, + period_known=True, + names_ok=True, + duplicate_in_file=False, +) + + +def test_form_recognized() -> None: + assert form_status_v2(**OK) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "change", + [ + {"service_done": False}, + {"file_error_warning": True}, + {"counterparty": DocumentCheckResult.MISMATCH}, + {"period_known": False}, + {"names_ok": False}, + {"duplicate_in_file": True}, + ], +) +def test_form_needs_check(change: dict) -> None: + assert form_status_v2(**OK | change) is DocumentStatus.NEEDS_CHECK # type: ignore[arg-type] + + +@pytest.mark.parametrize("cp", [None, DocumentCheckResult.UNKNOWN]) +def test_unknown_counterparty_allowed(cp: DocumentCheckResult | None) -> None: + assert form_status_v2(**OK | {"counterparty": cp}) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] + + +def test_duplicate_period_forms() -> None: + forms = [ + (1, BAL, 2025, Q3), + (2, BAL, 2025, Q3), + (3, BAL, 2024, Q4), + (4, INC, 2025, Q3), + (5, BAL, 2025, None), + (6, BAL, 2025, None), + ] + assert duplicate_period_forms(forms) == {1, 2} + + +def test_has_error_warning() -> None: + assert has_error_warning([{"code": "JSON_FAILED", "severity": "error", "message": "m"}]) + assert not has_error_warning([{"code": "LEGACY_WARNING", "severity": "info", "message": "m"}, "строка"]) + assert not has_error_warning(["error", 5, None, {"severity": "ERROR"}]) + assert not has_error_warning(None) + assert not has_error_warning({"severity": "error"}) + + +OSV_OK = dict( + osv_done=True, + checks_ok=True, + file_error_warning=False, + counterparty=DocumentCheckResult.MATCH, + sheet_mapping_warnings=False, +) + + +def test_osv_recognized() -> None: + assert osv_status_v2(**OSV_OK) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] + assert osv_status_v2(**OSV_OK | {"counterparty": None}) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "change", + [ + {"osv_done": False}, + {"checks_ok": False}, + {"file_error_warning": True}, + {"counterparty": DocumentCheckResult.MISMATCH}, + {"sheet_mapping_warnings": True}, + ], +) +def test_osv_needs_check(change: dict) -> None: + assert osv_status_v2(**OSV_OK | change) is DocumentStatus.NEEDS_CHECK # type: ignore[arg-type] + + +def _line(key: int, code: str | None, conf: str | None, name: str | None, *, fixed: bool = False, status: str = "ok"): # noqa: PLR0913 + return NameCheckLine(key, status, code, None if conf is None else Decimal(conf), name, fixed) + + +def test_name_mismatches() -> None: + lines = [ + _line(1, "1120", "50", "Нематериальные активы"), + _line(2, "1220", "50", "НДС, начисленный по отгрузке"), + _line(3, "1200", "50", "Итого по разделу I"), + _line(4, "1500", "56", "Итого по разделу V1"), + _line(5, "1105", "67", "АКТИВ I. Внеоборотные активы Гудвил"), + _line(6, "1120", "50", "Нематериальные активы", fixed=True), + _line(7, "1110", "100", "Чужое имя"), + _line(8, "9999", "50", "Что угодно"), + _line(9, "1120", "50", ""), + _line(10, None, None, "Без кода", status="unreadable"), + ] + assert name_mismatches(lines, CATALOG, "BALANCE") == [1, 3, 8, 9]