"""Чистые функции маппинга формы (spec osv-result-mapping «Документ формы ББ/ОФР»; design D-2, D-3, D-6, D-14).""" import datetime from decimal import Decimal import pytest from src.contract.result import StatementDocument from src.domain.statement_mapping import ( build_statement, edition_for, quarter_of, statement_period, statement_status, value_clean, values_ok, ) from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType d = datetime.date @pytest.mark.parametrize( ("day", "quarter"), [ (d(2025, 3, 31), PeriodQuarter.Q1), (d(2025, 6, 30), PeriodQuarter.Q2), (d(2025, 9, 30), PeriodQuarter.Q3), (d(2024, 12, 31), PeriodQuarter.Q4), (d(2025, 10, 31), None), (d(2025, 6, 29), None), ], ) def test_quarter_of(day: datetime.date, quarter: PeriodQuarter | None) -> None: assert quarter_of(day) is quarter def test_period_and_edition() -> None: assert statement_period(d(2025, 9, 30), [d(2024, 12, 31)]) == d(2025, 9, 30) assert statement_period(None, [d(2024, 12, 31), d(2025, 6, 30), None]) == d(2025, 6, 30) assert statement_period(None, [None]) is None assert edition_for(d(2024, 12, 31)) == "66n" assert edition_for(d(2025, 3, 31)) == "fsbu4" assert edition_for(None) is None def _doc(**overrides: object) -> StatementDocument: data: dict[str, object] = { "document_type": "BALANCE", "locator": "sheet:ББ", "header": { "form_name": "Бухгалтерский баланс", "inn": "7700000000", "inn_confidence": 100, "report_date": "2025-09-30", "report_date_text": "на 30 сентября 2025 г.", "units": "тыс. руб.", "organization_name": "ООО Пример", "okud": "0710001", }, "columns": [ {"column_index": 0, "label": "На 30.09.2025", "period_date": "2025-09-30"}, {"column_index": 1, "label": "На 31.12.2024", "period_date": "2024-12-31"}, ], "lines": [ { "line_code": "1150", "line_code_confidence": 100, "line_name": "ОС", "line_name_confidence": 100, "values": [{"column_index": 0, "amount": "10", "confidence": 100, "is_recognized": True}], }, {"line_code": "1150", "line_name": "дубль", "values": []}, { "line_code": "1600", "line_name": "БАЛАНС", "values": [{"column_index": 5, "amount": "1", "confidence": 100, "is_recognized": True}], }, ], } data.update(overrides) return StatementDocument.model_validate(data) def test_build_statement_cells_and_header() -> None: draft = build_statement(_doc()) expected_cells = 4 # 2 строки × 2 колонки assert draft.document_type is DocumentType.BALANCE and draft.statement_type is StatementType.balance assert (draft.period_year, draft.period_quarter, draft.period_date) == (2025, PeriodQuarter.Q3, d(2025, 9, 30)) assert draft.header["inn"] == "7700000000" and draft.header["inn_original_value"] == "7700000000" assert draft.header["inn_verified"] is True assert draft.header["report_date_original_value"] == "на 30 сентября 2025 г." assert draft.header["units"] == draft.header["units_original_value"] == "тыс. руб." assert [(c.column_index, c.label, c.period_date, c.is_current) for c in draft.columns] == [ (0, "На 30.09.2025", d(2025, 9, 30), True), (1, "На 31.12.2024", d(2024, 12, 31), False), ] assert [line.line_code for line in draft.lines] == ["1150", "1600"] # дубль 1150 отброшен cells = {(v.line_code, v.column_index): v for v in draft.values} assert len(cells) == expected_cells assert cells[("1150", 0)].amount == Decimal("10") and cells[("1150", 1)].amount is None assert cells[("1150", 1)].is_recognized is None and cells[("1600", 0)].confidence is None assert any("1150" in w and "повтор" in w for w in draft.warnings) assert any("колонк" in w and "5" in w for w in draft.warnings) def test_value_outside_columns_skipped() -> None: """Review Focus 3: значение с column_index вне колонок — пропущено с предупреждением, остальное смаплено.""" draft = build_statement(_doc()) assert all(v.column_index in (0, 1) for v in draft.values) def test_period_from_columns_and_non_quarter() -> None: no_date = _doc(header={"form_name": "ББ", "report_date": None}) draft = build_statement(no_date) assert (draft.period_year, draft.period_quarter) == (2025, PeriodQuarter.Q3) assert draft.header["report_date"] is None october = _doc(header={"form_name": "ББ", "report_date": "2025-10-31"}) draft = build_statement(october) assert (draft.period_year, draft.period_quarter) == (2025, None) assert any("конец квартала" in w for w in draft.warnings) def test_no_dates_at_all() -> None: draft = build_statement(_doc(header={"form_name": "ББ"}, columns=[{"column_index": 0, "label": ""}])) assert (draft.period_year, draft.period_quarter, draft.period_date) == (None, None, None) assert [(c.column_index, c.label, c.period_date, c.is_current) for c in draft.columns] == [(0, "", None, None)] def test_status_rules() -> None: ok = dict( job_done=True, checks_passed=True, counterparty=DocumentCheckResult.MATCH, period_known=True, values_ok=True, has_lines=True, warnings=[], ) assert statement_status(**ok) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] for key, bad in [ ("job_done", False), ("checks_passed", False), ("counterparty", DocumentCheckResult.MISMATCH), ("period_known", False), ("values_ok", False), ("has_lines", False), ("warnings", ["x"]), ]: assert statement_status(**{**ok, key: bad}) is DocumentStatus.NEEDS_CHECK # type: ignore[arg-type] assert statement_status(**{**ok, "counterparty": DocumentCheckResult.UNKNOWN}) is DocumentStatus.RECOGNIZED # type: ignore[arg-type] def test_values_ok() -> None: hundred = Decimal("100") assert values_ok([(False, True, hundred), (False, None, None)]) assert not values_ok([(False, False, hundred)]) assert not values_ok([(False, True, Decimal("99"))]) assert values_ok([(True, False, Decimal("10"))]) # правленое аналитиком значение не тянет статус вниз def test_equal_column_dates_current_is_lowest_index_only() -> None: columns = [ {"column_index": 2, "label": "б", "period_date": "2025-09-30"}, {"column_index": 1, "label": "а", "period_date": "2025-09-30"}, {"column_index": 3, "label": "в", "period_date": "2024-12-31"}, ] draft = build_statement(_doc(columns=columns, lines=[])) assert {c.column_index: c.is_current for c in draft.columns} == {1: True, 2: False, 3: False} def test_column_without_date_next_to_dated_is_not_current() -> None: columns = [ {"column_index": 0, "label": "примечание"}, {"column_index": 1, "label": "На 30.09.2025", "period_date": "2025-09-30"}, ] draft = build_statement(_doc(columns=columns, lines=[])) assert {c.column_index: c.is_current for c in draft.columns} == {0: False, 1: True} def test_duplicate_code_values_belong_to_first_line() -> None: lines = [ { "line_code": "1150", "line_name": "первая", "values": [{"column_index": 0, "amount": "10", "confidence": 100, "is_recognized": True}], }, { "line_code": "1150", "line_name": "дубль", "values": [{"column_index": 0, "amount": "999", "confidence": 100, "is_recognized": True}], }, ] draft = build_statement(_doc(lines=lines)) assert [line.line_name for line in draft.lines] == ["первая"] assert [(v.line_code, v.column_index, v.amount) for v in draft.values] == [ ("1150", 0, Decimal("10")), ("1150", 1, None), ] def test_fractional_confidence_kept_exactly() -> None: lines = [ { "line_code": "1150", "line_code_confidence": 99.5, "line_name": "ОС", "line_name_confidence": 98.25, "values": [{"column_index": 0, "amount": "10", "confidence": 99.5, "is_recognized": True}], } ] draft = build_statement(_doc(lines=lines)) assert draft.lines[0].line_code_confidence == Decimal("99.5") assert draft.lines[0].line_name_confidence == Decimal("98.25") assert draft.values[0].confidence == Decimal("99.5") def test_duplicate_column_index_keeps_first_with_warning() -> None: """I-1: повтор `column_index` в колонках — берётся первая, остальные отброшены с предупреждением; текущая колонка и значения считаются по оставшимся.""" columns = [ {"column_index": 0, "label": "На 30.09.2025", "period_date": "2025-09-30"}, {"column_index": 0, "label": "дубль", "period_date": "2026-01-31"}, {"column_index": 1, "label": "На 31.12.2024", "period_date": "2024-12-31"}, ] draft = build_statement(_doc(columns=columns)) assert [(c.column_index, c.label, c.period_date, c.is_current) for c in draft.columns] == [ (0, "На 30.09.2025", d(2025, 9, 30), True), (1, "На 31.12.2024", d(2024, 12, 31), False), ] assert "sheet:ББ: колонка 0 повторяется — оставлена первая" in draft.warnings assert (draft.period_date, draft.period_quarter) == (d(2025, 9, 30), PeriodQuarter.Q3) assert [(v.line_code, v.column_index) for v in draft.values] == [("1150", 0), ("1150", 1), ("1600", 0), ("1600", 1)] def test_duplicate_cell_keeps_first_value_with_warning() -> None: """T2.3: повтор пары (строка, колонка) среди значений — первое значение и предупреждение.""" lines = [ { "line_code": "1150", "line_name": "ОС", "values": [ {"column_index": 0, "amount": "10", "confidence": 100, "is_recognized": True}, {"column_index": 0, "amount": "77", "confidence": 100, "is_recognized": True}, ], } ] draft = build_statement(_doc(lines=lines)) assert {(v.column_index): v.amount for v in draft.values} == {0: Decimal("10"), 1: None} assert "sheet:ББ: строка 1150: значение колонки 0 повторяется — оставлено первое" in draft.warnings def test_1x_line_fields_domain() -> None: """Строки 1.x несут `row_index` — индекс строки в `doc.lines` (пропущенный повтор индекс не занимает).""" lines = [ {"line_code": "1150", "line_name": "а", "values": []}, {"line_code": "1150", "line_name": "повтор", "values": []}, {"line_code": "1600", "line_name": "б", "values": []}, ] draft = build_statement(_doc(lines=lines)) assert [(line.row_index, line.line_code) for line in draft.lines] == [(0, "1150"), (2, "1600")] assert all( line.line_status == "ok" and line.source_ref is None and line.line_code_text is None for line in draft.lines ) assert all(line.line_code_original_value == line.line_code and line.needs_review is False for line in draft.lines) assert {v.row_index for v in draft.values} == {0, 2} def test_value_clean_and_value_flags() -> None: assert value_clean(Decimal("100"), True) is True assert value_clean(None, None) is True assert value_clean(Decimal("80"), True) is False assert value_clean(Decimal("100"), False) is False lines = [ { "line_code": "1150", "line_name": "а", "values": [ {"column_index": 0, "amount": "10", "string_value": "10,0", "confidence": 100, "is_recognized": True}, {"column_index": 1, "amount": "5", "string_value": "5", "confidence": 80, "is_recognized": True}, ], } ] draft = build_statement(_doc(lines=lines)) by_column = {v.column_index: v for v in draft.values} assert (by_column[0].source_text, by_column[0].needs_review) == ("10,0", False) assert (by_column[1].source_text, by_column[1].needs_review) == ("5", True) def test_1x_columns_period_kind() -> None: draft = build_statement(_doc(columns=[{"column_index": 0, "label": "x"}])) assert [(c.period_date, c.period_kind, c.label_confidence) for c in draft.columns] == [(None, "none", None)] full = build_statement(_doc()) assert [c.period_kind for c in full.columns] == ["absolute", "absolute"] def test_1x_header_source() -> None: draft = build_statement(_doc()) source = draft.header["header_source"] assert source["inn"] == {"text": "7700000000", "confidence": 100} assert source["report_date"]["text"] == "на 30 сентября 2025 г." assert source["organization_name"] == {"text": "ООО Пример", "confidence": None} assert source["okei"] == {"text": None, "confidence": None} assert set(source) == { "organization_name", "inn", "report_date", "units", "okud", "okpo", "okopf", "okfs", "okei", "address", } assert draft.header["organization_name_original_value"] == "ООО Пример" assert draft.header["organization_name_modified"] is False assert draft.header["needs_review_fields"] == [] and draft.header["form_variant"] is None