"""Файл результата в S3 (`results/{file_id}/{job_id}/result.json`): `recognition-result.schema.json` v1.0. Один файл → один job → N документов (ОСВ по счёту / ББ / ОФР / прочее). В Срезе 0 — модели и валидация примеров; маппинг в таблицы — Срез 2. Суммы — `Decimal` в модели и строка в JSON (`"106311116.46"`, без `float`). Нормализация счетов, резолв контрагентов и проверки — в arm-core, не здесь. """ from enum import StrEnum from typing import Annotated, Literal, Self from pydantic import Field, model_validator from src.contract.common import ( Confidence, ContractDate, ContractModel, ContractUUID, DataType, HeaderVariant, IntLike, LaxErrorCode, Metrics, Money, OptionalNotNull, SchemaVersion, ServiceInfo, ) __all__ = [ "FailedRelation", "OsvChecks", "OsvDocument", "OsvGroup", "OsvHeader", "OsvRow", "RecognitionResult", "ResultError", "ResultStatus", "StatementChecks", "StatementColumn", "StatementDocument", "StatementHeader", "StatementLine", "StatementValue", ] class ResultStatus(StrEnum): DONE = "done" NEEDS_CHECK = "needs_check" ERROR = "error" LaxResultStatus = Annotated[ResultStatus, Field(strict=False)] class ResultError(ContractModel): """`error` файла результата: в отличие от конверта, `code` и `message` необязательны.""" code: OptionalNotNull[LaxErrorCode] = None message: OptionalNotNull[str] = None # --- ОСВ ------------------------------------------------------------------------------------------------------------ class OsvHeader(ContractModel): organization_name: str # как в r1 файла organization_inn: str | None = None account_code_raw: str # «60», «58.03» period_text: str # «1 квартал 2025 г.» period_year: IntLike | None = None period_quarter: Literal["Q1", "Q2", "Q3", "Q4"] | None = None is_cumulative: bool | None = None # по тексту; ядро может переопределить по данным (OQ-02) period_start: ContractDate | None = None # начало периода по шапке (v1.2, OQ-131) period_end: ContractDate | None = None is_from_year_start: bool | None = None # period_start — 1 января; is_cumulative не переопределяет data_type: DataType | None = None currency: str | None = None header_variant: OptionalNotNull[HeaderVariant] = None class OsvGroup(ContractModel): """Группировка из шапки (порядок — после первого ACCOUNT).""" type: Literal["ACCOUNT", "DEPARTMENT", "COUNTERPARTY", "CONTRACT", "OTHER"] name: OptionalNotNull[str] = None # исходный текст из шапки has_inn: OptionalNotNull[bool] = None class OsvRow(ContractModel): row_number: IntLike level: IntLike parent_row_number: IntLike | None = None row_value: str group_type: Literal["ACCOUNT", "DEPARTMENT", "COUNTERPARTY", "CONTRACT", "OTHER", "TOTAL"] group_name: str | None = None account_code_raw: str | None = None # код текущей ветки, как в файле (60.1) counterparty_name: str | None = None counterparty_inn: str | None = None data_type: DataType | None = None opening_debit: Money = None opening_credit: Money = None turnover_debit: Money = None turnover_credit: Money = None closing_debit: Money = None closing_credit: Money = None extra: dict[str, str | None] | None = None # нераспознанные колонки таблицы: заголовок → текст ячейки (§1.2) class OsvChecks(ContractModel): total_row_matches_root: OptionalNotNull[bool] = None balance_identity_violations: OptionalNotNull[list[IntLike]] = None # row_number, где opening−turnover≠closing parent_sum_violations: OptionalNotNull[list[IntLike]] = None class OsvDocument(ContractModel): document_type: Literal["OSV"] locator: str # «sheet:Лист_1» header: OsvHeader group_schema: list[OsvGroup] columns: OptionalNotNull[dict[str, str]] = None # диагностика: какие колонки приняты за шесть сумм и ИНН rows: list[OsvRow] checks: OptionalNotNull[OsvChecks] = None # --- ББ / ОФР / прочее ---------------------------------------------------------------------------------------------- class StatementHeader(ContractModel): form_name: OptionalNotNull[str] = None organization_name: str | None = None organization_name_confidence: Confidence = None inn: str | None = None inn_confidence: Confidence = None report_date_text: str | None = None report_date: ContractDate | None = None units: str | None = None # «тыс. руб.» okud: str | None = None okpo: str | None = None okopf: str | None = None okfs: str | None = None address: str | None = None class StatementColumn(ContractModel): column_index: IntLike label: str # «На 30 сентября 2025 г.» period_date: ContractDate | None = None class StatementValue(ContractModel): column_index: IntLike amount: Money = None string_value: str | None = None confidence: Confidence = None is_recognized: OptionalNotNull[bool] = None class StatementLine(ContractModel): line_code: str line_code_confidence: Confidence = None line_name: str line_name_confidence: Confidence = None values: list[StatementValue] class FailedRelation(ContractModel): """Нарушенное соотношение §7 по одной колонке. `expected` — левая часть как в файле, `actual` — правая по строкам. """ column_index: IntLike relation: str # «1600 = 1100 + 1200» — текст из словаря statement_checks.yaml expected: Money actual: Money class StatementChecks(ContractModel): """Самопроверка формы (v1.2): отсутствие объекта — «не проверялось», пустые списки — «сошлось».""" relations_failed: list[FailedRelation] = Field(default_factory=list) soft_relations_failed: list[FailedRelation] = Field(default_factory=list) duplicate_line_codes: list[str] = Field(default_factory=list) class StatementDocument(ContractModel): document_type: Literal["BALANCE", "INCOME", "DISCLOSURE", "OTHER"] locator: str # «pages:1-2» header: StatementHeader columns: list[StatementColumn] lines: list[StatementLine] checks: OptionalNotNull[StatementChecks] = None ResultDocument = Annotated[OsvDocument | StatementDocument, Field(discriminator="document_type")] class RecognitionResult(ContractModel): schema_version: SchemaVersion job_id: ContractUUID file_id: ContractUUID service: ServiceInfo status: LaxResultStatus error: ResultError | None = None metrics: OptionalNotNull[Metrics] = None raw_s3_key: str | None = None # scan-gateway: raw/{job_id}.json documents: list[ResultDocument] @model_validator(mode="after") def _check_documents_for_success(self) -> Self: """`if status in (done, needs_check) then documents.minItems = 1` — буквально из схемы.""" if self.status in (ResultStatus.DONE, ResultStatus.NEEDS_CHECK) and not self.documents: raise ValueError(f"при status = {self.status.value} список documents не может быть пустым") return self