api/src/repositories/quality_facts.py

@@ -0,0 +1,25 @@
+"""Запись фактов качества первички (design §4; AURORA-1759).
+
+Тонкий слой без бизнес-логики: факт добавляется в сессию вызывающего и попадает в ТУ ЖЕ транзакцию, что и событие,
+которое его породило. Функции не делают `flush` и `commit` — решает вызывающий. Аргументы — kwargs по столбцам
+DBML/ORM; колонки, не переданные вызывающим, берут значения по умолчанию БД (`server_default`).
+"""
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.quality_facts import QualityFailureFact, QualityIntakeFact, QualityOutcomeFact
+
+
+async def insert_intake_fact(session: AsyncSession, **columns) -> None:
+    """Добавляет факт «документ как распознано» (design §4.1)."""
+    session.add(QualityIntakeFact(**columns))
+
+
+async def insert_outcome_fact(session: AsyncSession, **columns) -> None:
+    """Добавляет факт исхода документа (design §4.2)."""
+    session.add(QualityOutcomeFact(**columns))
+
+
+async def insert_failure_fact(session: AsyncSession, **columns) -> None:
+    """Добавляет факт технического отказа задания или ошибки маппинга (design §4.3)."""
+    session.add(QualityFailureFact(**columns))

api/tests/test_repositories_quality_facts.py

@@ -0,0 +1,122 @@
+"""Запись фактов качества (§4 design): roundtrip через реальную БД (маркер `integration`)."""
+
+import datetime
+import uuid
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.enums import (
+    DocumentStatus,
+    DocumentType,
+    QualityFailureKind,
+    QualityOutcome,
+    RecognitionPipeline,
+)
+from src.models.quality_facts import QualityFailureFact, QualityIntakeFact, QualityOutcomeFact
+from src.repositories.quality_facts import insert_failure_fact, insert_intake_fact, insert_outcome_fact
+
+INTAKE_AT = datetime.datetime(2026, 11, 5, 10, 0, tzinfo=datetime.UTC)
+LINE_COUNT = 42
+FIXES_VALUE_SERVICE = 3
+FAILURE_ATTEMPT = 2
+
+
+@pytest.mark.integration
+async def test_insert_intake_fact_roundtrip(db_session: AsyncSession) -> None:
+    document_id = uuid.uuid4()
+    file_id = uuid.uuid4()
+    job_id = uuid.uuid4()
+    legal_entity_id = uuid.uuid4()
+
+    await insert_intake_fact(
+        db_session,
+        document_id=document_id,
+        file_id=file_id,
+        job_id=job_id,
+        attempt=1,
+        legal_entity_id=legal_entity_id,
+        legal_entity_name='ООО "Русский бисквит"',
+        group_id=None,
+        group_name=None,
+        document_type=DocumentType.BALANCE,
+        pipeline=RecognitionPipeline.image,
+        is_scan=True,
+        service_name="scan-gateway",
+        intake_at=INTAKE_AT,
+        intake_status=DocumentStatus.RECOGNIZED,
+        line_count=LINE_COUNT,
+        payload={"lines": []},
+    )
+    await db_session.flush()
+
+    row = (
+        await db_session.execute(select(QualityIntakeFact).where(QualityIntakeFact.document_id == document_id))
+    ).scalar_one()
+    assert row.file_id == file_id
+    assert row.job_id == job_id
+    assert row.legal_entity_name == 'ООО "Русский бисквит"'
+    assert row.document_type == DocumentType.BALANCE
+    assert row.pipeline == RecognitionPipeline.image
+    assert row.intake_status == DocumentStatus.RECOGNIZED
+    assert row.line_count == LINE_COUNT
+    assert row.payload == {"lines": []}
+
+
+@pytest.mark.integration
+async def test_insert_outcome_fact_roundtrip(db_session: AsyncSession) -> None:
+    document_id = uuid.uuid4()
+
+    await insert_outcome_fact(
+        db_session,
+        document_id=document_id,
+        outcome=QualityOutcome.CONFIRMED,
+        outcome_at=INTAKE_AT,
+        fixes_value_service=FIXES_VALUE_SERVICE,
+        checks_accepted=True,
+        payload={"checks": []},
+    )
+    await db_session.flush()
+
+    row = (
+        await db_session.execute(select(QualityOutcomeFact).where(QualityOutcomeFact.document_id == document_id))
+    ).scalar_one()
+    assert row.outcome == QualityOutcome.CONFIRMED
+    assert row.outcome_at == INTAKE_AT
+    assert row.fixes_value_service == FIXES_VALUE_SERVICE
+    # не переданные счётчики берутся из server_default БД
+    assert row.fixes_value_ours == 0
+    assert row.checks_accepted is True
+    assert row.payload == {"checks": []}
+
+
+@pytest.mark.integration
+async def test_insert_failure_fact_roundtrip(db_session: AsyncSession) -> None:
+    job_id = uuid.uuid4()
+    file_id = uuid.uuid4()
+
+    await insert_failure_fact(
+        db_session,
+        job_id=job_id,
+        file_id=file_id,
+        attempt=FAILURE_ATTEMPT,
+        legal_entity_id=uuid.uuid4(),
+        legal_entity_name='ООО "Русский бисквит"',
+        group_id=None,
+        group_name=None,
+        pipeline=RecognitionPipeline.tabular,
+        is_scan=False,
+        service_name="tabular-parser",
+        failure_kind=QualityFailureKind.TIMEOUT,
+        error_code="TIMEOUT",
+        failed_at=INTAKE_AT,
+    )
+    await db_session.flush()
+
+    row = (await db_session.execute(select(QualityFailureFact).where(QualityFailureFact.job_id == job_id))).scalar_one()
+    assert row.file_id == file_id
+    assert row.attempt == FAILURE_ATTEMPT
+    assert row.failure_kind == QualityFailureKind.TIMEOUT
+    assert row.error_code == "TIMEOUT"
+    assert row.failed_at == INTAKE_AT