feat(#190): факт приёма документа при маппинге (§14.1)

Хук в _map_candidate (result_mapping.py): после map_file, до auto_confirm — для
каждого документа BALANCE/INCOME/OSV строит снимок build_intake_payload и пишет
quality_intake_fact через insert_intake_fact. Чисто добавочный шаг: не меняет
документы, статусы и прочий результат маппинга; DISCLOSURE/OTHER факта не получают.

Счётчики строк/значений — из financial_statement_* (BALANCE/INCOME) и document_issue;
warning_count/osv_warnings — предупреждения источника service (document_issue.source =
service): для BALANCE/ОФР 1.x таких записей нет, для ОСВ это коды OSV_*. Для ОСВ
построчные счётчики не считаются (их в ядре нет), только row_count в payload.

Тесты в test_result_mapping.py (маркер integration, не запускались — нет PostgreSQL
в песочнице, подтверждена только коллекция) + один DB-free тест на множество типов
документов хука.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

api/src/services/result_mapping.py

@@ -53,6 +53,7 @@ from src.domain.osv_mapping import (
     document_status,
     opening_matches,
 )
+from src.domain.quality_facts import build_intake_payload
 from src.domain.statement_issues import IssueDraft, mapping_warning
 from src.domain.statement_mapping import StatementDraft, build_statement, statement_status, values_ok
 from src.domain.statement_status import duplicate_period_forms, has_error_warning, osv_status_v2
@@ -73,11 +74,14 @@ from src.models.enums import (
     OsvGroupType,
     PeriodQuarter,
     RecognitionStatus,
+    StatementLineStatus,
 )
 from src.models.files import Document, RecognitionJob, S3Object, StoredFile
 from src.models.osv import OsvEntry
+from src.models.statements import DocumentIssue
 from src.repositories import statements as statements_repo
 from src.repositories.counterparties import CounterpartyIndex
+from src.repositories.quality_facts import insert_intake_fact
 from src.services import consolidation_documents
 from src.services.audit import AuditService
 from src.services.file_status import set_file_status_for_job
@@ -119,6 +123,8 @@ MAPPABLE_JOB_STATUSES = (RecognitionStatus.done, RecognitionStatus.needs_check)
 CUMULATIVE_CHECK_QUARTERS = frozenset({PeriodQuarter.Q2, PeriodQuarter.Q3, PeriodQuarter.Q4})
 RESOLVED_DATA_TYPES = frozenset({OsvDataType.BU, OsvDataType.SUM})  # резолв контрагента — §3 п. 4, R-3
 ROOT_LEVEL = 0
+# §14.1 (design quality-metrics §4.1): факт приёма пишется только для этих типов; DISCLOSURE/OTHER — без факта
+INTAKE_FACT_DOCUMENT_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME, DocumentType.OSV})
 STORED_FILE_OBJECT_TYPE = "stored_file"  # `audit_event.object_type` события FILE_REPLACE
 SUPERSEDE_LOCK_PREFIX = "supersede"  # пространство ключей advisory-блокировки автозамены в пределах ЮЛ
 _FORBIDDEN_KEY_SEGMENTS = frozenset({"", ".", ".."})
@@ -1094,6 +1100,123 @@ async def _register_failure(
     return exhausted
 
 
+# --- факт приёма (quality_intake_fact, §14.1, design quality-metrics §4.1, AURORA-1759) ------------------------------
+
+
+async def _s3_key(session: AsyncSession, object_id: uuid.UUID | None) -> str | None:
+    """Ключ `s3_object` по id для факта качества (хранится строкой, без FK — design §4); нет id или записи — `None`."""
+    if object_id is None:
+        return None
+    return (await session.execute(select(S3Object.key).where(S3Object.id == object_id))).scalar_one_or_none()
+
+
+async def _statement_intake_fields(
+    session: AsyncSession, document: Document, issues: Sequence[DocumentIssue]
+) -> dict[str, Any]:
+    """Счётчики и `payload` факта приёма формы ББ/ОФР: строки и значения — `financial_statement_*` формы документа,
+    нарушения/предупреждения — уже прочитанные `document_issue`. `warning_count` — предупреждения СЕРВИСА
+    (`document_issue.source = service`; у `1.x` таких записей нет, формирует только маппер `2.0`).
+
+    Формы без шапки (`meta is None`) из текущего `map_file` не возникают — каждая форма получает `financial_statement_
+    meta` в той же транзакции (`_insert_statements`/`insert_forms_v2`); на случай, если это всё же не так, снимок без
+    построчных данных дешевле, чем прерванная транзакция маппинга."""
+    meta = await statements_repo.get_meta(session, document.id)
+    if meta is None:
+        return {"payload": build_intake_payload(document, issues=issues)}
+    _, pairs = await statements_repo.load_form(session, meta.id)
+    lines = [line for line, _ in pairs]
+    values = [value for _, line_values in pairs for value in line_values]
+    return {
+        "line_count": len(lines),
+        "value_count": len(values),
+        "value_needs_review_count": sum(1 for value in values if value.needs_review),
+        "line_unreadable_count": sum(1 for line in lines if line.line_status is StatementLineStatus.unreadable),
+        "line_missing_count": sum(1 for line in lines if line.line_status is StatementLineStatus.missing),
+        "issue_relation_count": sum(1 for issue in issues if issue.kind is DocumentIssueKind.relation),
+        "issue_soft_count": sum(1 for issue in issues if issue.kind is DocumentIssueKind.soft),
+        "warning_count": sum(
+            1 for issue in issues if issue.kind is DocumentIssueKind.warning and issue.source is IssueSource.service
+        ),
+        "payload": build_intake_payload(
+            document, lines=lines, values=values, issues=issues, header_source=meta.header_source
+        ),
+    }
+
+
+async def _osv_intake_fields(
+    session: AsyncSession, document: Document, issues: Sequence[DocumentIssue]
+) -> dict[str, Any]:
+    """`payload` факта приёма ОСВ: без построчного снимка (строк/значений §7 правок у ОСВ в ядре нет) — только число
+    строк `osv_entry` и коды предупреждений сервиса `OSV_*` (`document_issue` вида `warning`, `source = service`,
+    см. `_osv_warning_issue`; предупреждения маппинга ядра — `MAPPING_WARNING`, `source = core` — в этот список не
+    входят, design §4.1 «коды предупреждений OSV_*»). Счётчики строк/значений формы для ОСВ не считаются (`None`)."""
+    row_count = (
+        await session.execute(select(func.count()).select_from(OsvEntry).where(OsvEntry.document_id == document.id))
+    ).scalar_one()
+    osv_warnings = [
+        issue.code
+        for issue in issues
+        if issue.kind is DocumentIssueKind.warning and issue.source is IssueSource.service
+    ]
+    return {"payload": build_intake_payload(document, osv_warnings=osv_warnings, osv_row_count=row_count)}
+
+
+async def _write_intake_facts(  # noqa: PLR0913 - все входы факта приёма именованные
+    session: AsyncSession,
+    *,
+    job: RecognitionJob,
+    file: StoredFile,
+    owner: LegalEntity,
+    documents: Sequence[Document],
+    now: datetime,
+) -> None:
+    """Факт приёма «как распознано» (§14.1, design §4.1): один `quality_intake_fact` на каждый документ
+    BALANCE/INCOME/OSV файла (`DISCLOSURE`/`OTHER` — без факта); пишется в транзакции маппинга, после `map_file`
+    (документы, строки форм и `document_issue` уже вставлены), до `auto_confirm`. Чисто добавочный шаг: не меняет
+    `documents`, статусы и прочий результат маппинга."""
+    group_name: str | None = None
+    if owner.company_group_id is not None:
+        group = await session.get(CompanyGroup, owner.company_group_id)
+        group_name = group.name if group is not None else None
+    raw_s3_key = await _s3_key(session, job.raw_s3_object_id)
+    result_s3_key = await _s3_key(session, job.result_s3_object_id)
+    upload_s3_key = await _s3_key(session, file.s3_object_id)
+    for document in documents:
+        if document.document_type not in INTAKE_FACT_DOCUMENT_TYPES:
+            continue
+        issues = await statements_repo.issues(session, document.id)
+        fields = (
+            await _osv_intake_fields(session, document, issues)
+            if document.document_type is DocumentType.OSV
+            else await _statement_intake_fields(session, document, issues)
+        )
+        await insert_intake_fact(
+            session,
+            document_id=document.id,
+            file_id=file.id,
+            job_id=job.id,
+            attempt=job.attempt,
+            legal_entity_id=file.legal_entity_id,
+            legal_entity_name=owner.name,
+            group_id=owner.company_group_id,
+            group_name=group_name,
+            document_type=document.document_type,
+            pipeline=job.pipeline,
+            is_scan=file.is_scan,
+            service_name=job.service_name,
+            service_version=job.service_version,
+            statements_version=job.statements_version,
+            result_schema_version=job.result_schema_version,
+            report_date=document.reporting_date,
+            intake_at=now,
+            intake_status=document.status,
+            raw_s3_key=raw_s3_key,
+            result_s3_key=result_s3_key,
+            upload_s3_key=upload_s3_key,
+            **fields,
+        )
+
+
 async def _map_candidate(
     session_factory: async_sessionmaker[AsyncSession],
     candidate: Candidate,
@@ -1128,6 +1251,11 @@ async def _map_candidate(
         # автоподтверждение чистых документов (привязка и переход версии, OQ-157, design D-1), затем переход по
         # `NEEDS_CHECK` (D-6) — он уже видит версию после автоподтверждения и второго перехода не делает.
         documents = list((await session.execute(select(Document).where(Document.stored_file_id == file.id))).scalars())
+        # §14.1: факт приёма — ПОСЛЕ map_file (документы и формы уже вставлены), ДО auto_confirm (снимок «как
+        # распознано», не «как подтверждено»); owner уже есть в БД — map_file выше не дал бы MappingFatal иначе
+        owner = await session.get(LegalEntity, file.legal_entity_id)
+        if owner is not None:
+            await _write_intake_facts(session, job=job, file=file, owner=owner, documents=documents, now=now)
         # отложенный импорт: цикл documents → result_mapping → auto_confirm → statements → documents
         from src.services.auto_confirm import auto_confirm  # noqa: PLC0415
 

api/tests/test_result_mapping.py

@@ -54,6 +54,7 @@ from src.models.enums import (
 )
 from src.models.files import Document, RecognitionJob, StoredFile
 from src.models.osv import OsvEntry
+from src.models.quality_facts import QualityIntakeFact
 from src.models.statements import (
     DocumentIssue,
     FinancialStatementLine,
@@ -2104,3 +2105,88 @@ async def test_unreadable_result_json_is_fatal(
     assert "Expecting" not in job.metrics[MAPPING_ERROR_KEY]
     assert MAPPING_ATTEMPTS_KEY not in job.metrics
     assert await process_once(maker, storage, _settings()) is False
+
+
+# --- факт приёма (quality_intake_fact, §14.1, design quality-metrics §4.1, AURORA-1759/#190) ------------------------
+
+STATEMENT_LINE_COUNT = 2  # result-statement-minimal.json: строки 1100, 1600
+STATEMENT_VALUE_COUNT = 4  # 2 строки × 2 колонки
+
+
+async def _intake_fact(maker: async_sessionmaker[AsyncSession], document_id: uuid.UUID) -> QualityIntakeFact:
+    async with maker() as session:
+        return (
+            await session.execute(select(QualityIntakeFact).where(QualityIntakeFact.document_id == document_id))
+        ).scalar_one()
+
+
+async def test_map_candidate_writes_intake_fact_for_statement_document(
+    committed_sessionmaker: async_sessionmaker[AsyncSession],
+):
+    """Т5 плана quality-metrics (§14.1): маппинг формы ББ пишет один `quality_intake_fact` со снимком «как
+    распознано» — построчный снимок, счётчики и ключи S3 загрузки/результата."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    file_id, job_id, entity_id = await _prepare(maker, storage, _statement_result("BALANCE"), owner_inn="7700000000")
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    [document] = await _documents(maker, file_id)
+    assert document.document_type is DocumentType.BALANCE
+    fact = await _intake_fact(maker, document.id)
+    assert fact.document_type is DocumentType.BALANCE
+    assert (fact.file_id, fact.job_id, fact.legal_entity_id) == (file_id, job_id, entity_id)
+    assert fact.legal_entity_name == OWNER_NAME
+    assert (fact.group_id, fact.group_name) == (None, None)  # ЮЛ без ГК
+    assert fact.pipeline is (await _job(maker, job_id)).pipeline
+    assert fact.attempt == (await _job(maker, job_id)).attempt
+    assert fact.report_date == document.reporting_date
+    # «как распознано» — ДО auto_confirm (design §4.1): CONFIRMED сюда попасть не может, даже если документ чистый
+    # и `auto_confirm` подтвердит его позже в той же транзакции (`document` выше прочитан уже ПОСЛЕ auto_confirm)
+    assert fact.intake_status in (DocumentStatus.RECOGNIZED, DocumentStatus.NEEDS_CHECK)
+    assert fact.line_count == STATEMENT_LINE_COUNT
+    assert fact.value_count == STATEMENT_VALUE_COUNT
+    assert fact.raw_s3_key is None  # `raw_s3_object_id` не заполняется до приёма image pipeline (Task 11/12)
+    assert fact.result_s3_key is not None
+    assert fact.upload_s3_key is not None
+    assert fact.payload["header"]["document_type"] == "BALANCE"
+    assert len(fact.payload["lines"]) == STATEMENT_LINE_COUNT
+    assert len(fact.payload["values"]) == STATEMENT_VALUE_COUNT
+    assert "osv_warnings" not in fact.payload
+
+
+async def test_map_candidate_writes_intake_fact_for_osv_without_lines(
+    committed_sessionmaker: async_sessionmaker[AsyncSession],
+):
+    """ОСВ не имеет построчного снимка правок (их в ядре нет): `payload` без `lines`/`values`, только число строк
+    `osv_entry` (`row_count`) и коды предупреждений сервиса; `line_count` и прочие построчные счётчики — `None`
+    (счётчики формы ББ/ОФР к ОСВ не относятся)."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    file_id, job_id, entity_id = await _prepare(maker, storage, _minimal())
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    [document] = await _documents(maker, file_id)
+    assert document.document_type is DocumentType.OSV
+    fact = await _intake_fact(maker, document.id)
+    assert fact.document_type is DocumentType.OSV
+    assert (fact.file_id, fact.job_id, fact.legal_entity_id) == (file_id, job_id, entity_id)
+    assert "lines" not in fact.payload
+    assert "values" not in fact.payload
+    assert fact.payload["row_count"] == len(await _entries(maker, document.id))
+    assert fact.payload["osv_warnings"] == []  # результат 1.x: предупреждений сервиса `OSV_*` нет
+    assert fact.line_count is None
+    assert fact.value_count is None
+    assert fact.value_needs_review_count is None
+    assert fact.line_unreadable_count is None
+    assert fact.line_missing_count is None
+    assert fact.issue_relation_count is None
+    assert fact.issue_soft_count is None
+    assert fact.warning_count is None
+
+
+def test_intake_fact_document_types_excludes_disclosure_and_other() -> None:
+    """§14.1 (design §4.1): факт приёма пишется только для BALANCE/INCOME/OSV; `DISCLOSURE`/`OTHER` исключены —
+    чистая проверка множества типов хука, без БД (в отличие от прочих тестов `_map_candidate` в этом файле)."""
+    assert rm.INTAKE_FACT_DOCUMENT_TYPES == {DocumentType.BALANCE, DocumentType.INCOME, DocumentType.OSV}
+    assert DocumentType.DISCLOSURE not in rm.INTAKE_FACT_DOCUMENT_TYPES
+    assert DocumentType.OTHER not in rm.INTAKE_FACT_DOCUMENT_TYPES