"""Фабрики результата распознавания контракта `2.0` для тестов маппера форм (B4): базовый документ ББ/ОФР, строки, значения, колонки; `statement_result_v2` собирает `RecognitionResultV2` с переопределениями. Реальный пакет `arm_statements` считает статус и редакцию, поэтому базовая форма «чистая»: уверенности 100, `needs_review = false`. """ import copy import uuid from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any from src.contract import RecognitionResultV2 __all__ = [ "BALANCE_DATE", "balance_lines", "column_v2", "header_v2", "line_v2", "result_builder", "statement_doc_v2", "statement_result_v2", "value_v2", ] BALANCE_DATE = "2025-09-30" OWNER_INN = "7700000000" _NO_CHECKS: dict[str, Any] = {"relations_failed": [], "soft_relations_failed": []} def header_v2(**overrides: Any) -> dict[str, Any]: """Шапка с ИНН владельца по умолчанию (`counterparty_check = MATCH`), датой 30.09.2025 и тыс. руб.""" header: dict[str, Any] = { "form_name": "Бухгалтерский баланс", "organization_name": "ООО Ромашка", "organization_name_text": "ООО Ромашка", "organization_name_confidence": 100, "inn": OWNER_INN, "inn_text": OWNER_INN, "inn_confidence": 100, "report_date": BALANCE_DATE, "report_date_text": "30 сентября 2025 г.", "report_date_confidence": 100, "units": "тыс. руб.", "units_text": "тыс. руб.", "units_confidence": 100, "needs_review_fields": [], } return {**header, **overrides} def column_v2( index: int = 2, *, date: str | None = BALANCE_DATE, kind: str = "absolute", label: str | None = None ) -> dict[str, Any]: return { "column_index": index, "label": label or f"На {date}", "label_confidence": 100, "period_date": date, "period_kind": kind, } def value_v2( # noqa: PLR0913 - все поля значения контракта column_index: int = 2, amount: str | None = "100", *, text: str | None = None, confidence: float = 100, is_recognized: bool = True, needs_review: bool = False, ) -> dict[str, Any]: return { "column_index": column_index, "amount": amount, "text": text if text is not None else amount, "confidence": confidence, "is_recognized": is_recognized, "needs_review": needs_review, } def line_v2( # noqa: PLR0913 - все поля строки контракта row_index: int, code: str | None, name: str | None = "Строка", *, amount: str | None = "100", status: str = "ok", ref: str | None = None, code_confidence: float | None = 100, name_confidence: float | None = 100, code_text: str | None = None, parent_ref: str | None = None, needs_review: bool = False, values: Sequence[Mapping[str, Any]] | None = None, ) -> dict[str, Any]: """Строка результата; без `values` — одно значение `amount` в колонке 2 (`amount=None` — значений нет).""" if values is None: values = [] if amount is None else [value_v2(2, amount)] line: dict[str, Any] = { "row_index": row_index, "ref": ref or f"row:{row_index}", "line_status": status, "line_code": code, "line_code_text": code_text if code_text is not None else code, "line_code_confidence": code_confidence, "line_name": name, "line_name_confidence": name_confidence, "needs_review": needs_review, "values": [dict(value) for value in values], } if parent_ref is not None: line["parent_ref"] = parent_ref return line def balance_lines(amount: str = "100") -> list[dict[str, Any]]: """Минимальный чистый баланс (ключевые коды, имена как в словаре `fsbu4`).""" names = [ ("1210", "Запасы"), ("1200", "Итого по разделу II"), ("1600", "БАЛАНС"), ("1300", "Итого по разделу III"), ("1700", "БАЛАНС"), ] return [line_v2(index, code, name, amount=amount) for index, (code, name) in enumerate(names)] def statement_doc_v2( # noqa: PLR0913 - все поля документа контракта document_type: str = "BALANCE", *, lines: Sequence[Mapping[str, Any]] | None = None, columns: Sequence[Mapping[str, Any]] | None = None, header: Mapping[str, Any] | None = None, status: str = "done", form_variant: str | None = "full", warnings: Sequence[Mapping[str, Any]] = (), checks: Mapping[str, Any] | None = None, locator: str | None = None, ) -> dict[str, Any]: """Документ формы `2.0`; `checks` по умолчанию пустые (форма со строками, иначе поле не допускается).""" doc_lines = [dict(line) for line in (balance_lines() if lines is None else lines)] doc: dict[str, Any] = { "document_type": document_type, "locator": locator or f"file:test.json;table:{document_type}", "status": status, "form_variant": form_variant, "warnings": [dict(warning) for warning in warnings], "header": dict(header_v2() if header is None else header), "columns": [dict(column) for column in ([column_v2()] if columns is None else columns)], "lines": doc_lines, } if doc_lines and document_type != "DISCLOSURE": doc["checks"] = copy.deepcopy(dict(_NO_CHECKS if checks is None else checks)) elif checks is not None: doc["checks"] = copy.deepcopy(dict(checks)) return doc def statement_result_v2( # noqa: PLR0913 - все поля результата documents: Sequence[Mapping[str, Any]] | None = None, *, job_id: uuid.UUID | None = None, file_id: uuid.UUID | None = None, status: str = "done", statements_version: str | None = "abc1234", file_warnings: Sequence[Mapping[str, Any]] = (), overrides: Mapping[str, Any] | None = None, ) -> RecognitionResultV2: """Результат `2.0`: базовый словарь + переопределения верхнего уровня (`overrides`).""" payload: dict[str, Any] = { "schema_version": "2.0", "job_id": str(job_id or uuid.uuid4()), "file_id": str(file_id or uuid.uuid4()), "service": {"name": "tabular-parser", "version": "test"}, "status": status, "metrics": {"warnings": [dict(warning) for warning in file_warnings]}, "documents": [dict(doc) for doc in (documents if documents is not None else [statement_doc_v2()])], } if statements_version is not None: payload["service"]["statements_version"] = statements_version payload.update(overrides or {}) return RecognitionResultV2.model_validate(payload) def result_builder( documents: Sequence[Mapping[str, Any]] | None = None, **kwargs: Any ) -> Callable[..., RecognitionResultV2]: """`result_builder` для `map_golden_snapshot`: путь игнорируется, `job_id`/`file_id` подставляются.""" def build(_path: Path, *, job_id: uuid.UUID, file_id: uuid.UUID) -> RecognitionResultV2: return statement_result_v2(documents, job_id=job_id, file_id=file_id, **kwargs) return build