196 lines
7.6 KiB
Python
196 lines
7.6 KiB
Python
"""Фабрики результата распознавания контракта `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
|