feat(#190): API /quality-metrics — матрица, ряд по месяцам, список, выгрузка

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

api/src/api/dependencies/quality_metrics.py

@@ -0,0 +1,16 @@
+"""Зависимости API «Метрики качества» (R-6: сессия — только `SessionDep`). Видимость по филиалу не применяется (§8)."""
+
+from typing import Annotated
+
+from fastapi import Depends
+
+from src.api.dependencies.db import SessionDep
+from src.services.quality_metrics_service import QualityMetricsService
+
+
+def get_quality_metrics_service(session: SessionDep) -> QualityMetricsService:
+    """Сервис метрик качества: только чтение фактов, без аудита и `VisibilityScope`."""
+    return QualityMetricsService(session)
+
+
+QualityMetricsServiceDep = Annotated[QualityMetricsService, Depends(get_quality_metrics_service)]

api/src/api/endpoints/quality_metrics.py

@@ -0,0 +1,65 @@
+"""`/quality-metrics` — метрики качества распознавания первички (design `quality-metrics` §7, §8).
+
+Все маршруты под правом `QUALITY_METRICS_VIEW`; видимость по филиалу/ГК не применяется. Период — по дате приёма."""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query, Response
+
+from src.api.dependencies.auth import require
+from src.api.dependencies.quality_metrics import QualityMetricsServiceDep
+from src.common.filenames import content_disposition
+from src.models.enums import QualityCause
+from src.schemas.common import Page, PageParams
+from src.schemas.quality_metrics import (
+    ExportFormat,
+    QualityCell,
+    QualityCellDocumentOut,
+    QualityFilters,
+    QualityMatrixOut,
+    QualityMonthlySeriesOut,
+)
+
+router = APIRouter(
+    prefix="/quality-metrics", tags=["quality-metrics"], dependencies=[Depends(require("QUALITY_METRICS_VIEW"))]
+)
+
+
+@router.get("/matrix", summary="Матрица причин за период")
+async def get_matrix(
+    service: QualityMetricsServiceDep, filters: Annotated[QualityFilters, Query()]
+) -> QualityMatrixOut:
+    """Документы по причинам (доли — от подтверждённых), удалённые/заменённые до и после подтверждения, «в работе»,
+    уникальные отчёты и технические отказы по файлам (на них фильтр по типу документа не действует)."""
+    return await service.matrix(filters)
+
+
+@router.get("/series", summary="Ряд по месяцам даты приёма для причины")
+async def get_series(
+    service: QualityMetricsServiceDep,
+    filters: Annotated[QualityFilters, Query()],
+    cause: Annotated[QualityCause, Query()],
+) -> QualityMonthlySeriesOut:
+    return await service.series(filters, cause)
+
+
+@router.get("/cells/{cell}/documents", summary="Документы ячейки матрицы")
+async def get_cell_documents(
+    cell: QualityCell,
+    service: QualityMetricsServiceDep,
+    filters: Annotated[QualityFilters, Query()],
+    page: Annotated[PageParams, Query()],
+) -> Page[QualityCellDocumentOut]:
+    """Ячейка — причина (`QualityCause`), `REMOVED_BEFORE_CONFIRM`, `REMOVED_AFTER_CONFIRM` или `IN_PROGRESS`."""
+    return await service.cell_documents(filters, cell, limit=page.limit, offset=page.offset)
+
+
+@router.get("/export", response_class=Response, summary="Выгрузка фактов за период (CSV / JSON)")
+async def export_facts(
+    service: QualityMetricsServiceDep,
+    filters: Annotated[QualityFilters, Query()],
+    fmt: Annotated[ExportFormat, Query(alias="format")] = "csv",
+) -> Response:
+    content, media_type, ext = await service.export(filters, fmt)
+    header = content_disposition(f"quality-facts.{ext}", ascii_fallback=f"quality-facts.{ext}")
+    return Response(content=content, media_type=media_type, headers={"Content-Disposition": header})

api/src/api/router.py

@@ -8,7 +8,7 @@ dev-эмитент токенов — только при `AUTH_DEV_ISSUER and D
 
 from fastapi import APIRouter
 
-from src.api.endpoints import admin, consolidations, directories, files, me, users
+from src.api.endpoints import admin, consolidations, directories, files, me, quality_metrics, users
 from src.api.endpoints.dev_auth import build_dev_router
 from src.core.config import Settings, get_settings
 
@@ -24,6 +24,7 @@ def build_api_router(settings: Settings) -> APIRouter:
     api_router.include_router(admin.router)
     api_router.include_router(files.router)
     api_router.include_router(consolidations.router)
+    api_router.include_router(quality_metrics.router)
     if settings.AUTH_DEV_ISSUER and settings.DEBUG:
         api_router.include_router(build_dev_router(settings))
     return api_router

api/src/schemas/quality_metrics.py

@@ -0,0 +1,134 @@
+"""Схемы API «Метрики качества» (design `quality-metrics` §6-§8): матрица причин, ряд по месяцам, список документов
+ячейки. Деньги не нужны — только счётчики и доли."""
+
+import datetime
+import enum
+import uuid
+from typing import Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+from src.models.enums import DocumentType, QualityCause, QualityOutcome, RecognitionPipeline
+
+__all__ = [
+    "ExportFormat",
+    "MonthPointOut",
+    "QualityCell",
+    "QualityCellDocumentOut",
+    "QualityFilters",
+    "QualityFixesOut",
+    "QualityMatrixOut",
+    "QualityMonthlySeriesOut",
+    "TechnicalFailuresOut",
+]
+
+ExportFormat = Literal["csv", "json"]
+
+
+class QualityCell(enum.StrEnum):
+    """Ячейка матрицы, из которой раскрывается список документов: пять причин (`QualityCause`), два столбца «удалён /
+    заменён» (в доли не входят, design §6 п.1) и «в работе» (документы без исхода)."""
+
+    SERVICE_ERROR = QualityCause.SERVICE_ERROR.value
+    OUR_ERROR = QualityCause.OUR_ERROR.value
+    CLIENT_DOCUMENT = QualityCause.CLIENT_DOCUMENT.value
+    WITHOUT_PARTICIPATION = QualityCause.WITHOUT_PARTICIPATION.value
+    AS_IS = QualityCause.AS_IS.value
+    REMOVED_BEFORE_CONFIRM = "REMOVED_BEFORE_CONFIRM"
+    REMOVED_AFTER_CONFIRM = "REMOVED_AFTER_CONFIRM"
+    IN_PROGRESS = "IN_PROGRESS"
+
+
+class QualityFilters(BaseModel):
+    """Фильтры из query (`Annotated[QualityFilters, Query()]`). Период — по дате приёма результата (`intake_at`,
+    когорта, design §7), обе границы включительно; `date_to` без времени — до конца этого дня (UTC)."""
+
+    date_from: datetime.date | None = None
+    date_to: datetime.date | None = None
+    group_id: uuid.UUID | None = None
+    legal_entity_id: uuid.UUID | None = None
+    document_type: DocumentType | None = None
+    pipeline: RecognitionPipeline | None = Field(default=None, description="источник: image = PDF, tabular = Excel")
+    is_scan: bool | None = Field(default=None, description="скан (true) / цифровой PDF (false); у Excel признака нет")
+
+    @model_validator(mode="after")
+    def _period_order(self) -> "QualityFilters":
+        if self.date_from and self.date_to and self.date_from > self.date_to:
+            raise ValueError("date_from позже date_to")
+        return self
+
+
+class TechnicalFailuresOut(BaseModel):
+    """Технические отказы считаются по файлам (design §6); фильтр по типу документа на них не действует.
+    `total_files_count` — уникальные файлы с фактом приёма или отказа в выбранных период/ГК/ЮЛ/источнике/скане."""
+
+    failed_files_count: int
+    total_files_count: int
+    share: float
+
+
+class QualityMatrixOut(BaseModel):
+    """Матрица причин за период. `counts`/`shares` — по всем значениям `QualityCause`; доли — от `denominator`
+    (документы с исходом AUTO_CONFIRMED / CONFIRMED). Столбцы «удалён / заменён» и «в работе» в доли не входят.
+    `removed_after_confirm` = `deleted_after_confirm_count` + `superseded_after_confirm_count`."""
+
+    counts: dict[QualityCause, int]
+    shares: dict[QualityCause, float]
+    denominator: int
+    removed_before_confirm: int
+    removed_after_confirm: int
+    deleted_after_confirm_count: int
+    superseded_after_confirm_count: int
+    in_progress_count: int
+    unique_reports_count: int = Field(description="уникальные отчёты: ЮЛ + подтверждённая отчётная дата + тип")
+    technical_failures: TechnicalFailuresOut
+
+
+class MonthPointOut(BaseModel):
+    month: datetime.date
+    count: int
+    denominator: int
+    share: float
+
+
+class QualityMonthlySeriesOut(BaseModel):
+    cause: QualityCause
+    points: list[MonthPointOut]
+
+
+class QualityFixesOut(BaseModel):
+    """Счётчики правок по источнику (design §5): `service` / `ours`; правка периода/счёта ОСВ — всегда «наш разбор»."""
+
+    value_service: int
+    value_ours: int
+    code_service: int
+    code_ours: int
+    added_row_service: int
+    added_row_ours: int
+    header_service: int
+    header_ours: int
+    osv_period_account: int
+
+
+class QualityCellDocumentOut(BaseModel):
+    """Документ ячейки (design §7, §8): `confirmed_by_*` — только здесь, агрегаты фамилий не содержат.
+    `document_link` — путь API карточки документа, `None`, если документ уже удалён. `outcome` — текущий (последний)
+    исход; `previous_outcome` — исход до `*_AFTER_CONFIRM` (только для ячейки «после подтверждения»)."""
+
+    document_id: uuid.UUID
+    document_type: DocumentType
+    legal_entity_id: uuid.UUID
+    legal_entity_name: str
+    group_name: str | None
+    received_at: datetime.datetime
+    outcome: QualityOutcome | None
+    outcome_at: datetime.datetime | None
+    previous_outcome: QualityOutcome | None
+    primary_cause: QualityCause | None
+    report_date: datetime.date | None
+    fixes: QualityFixesOut | None
+    checks_accepted: bool | None
+    issue_count_at_outcome: int | None
+    confirmed_by_id: uuid.UUID | None
+    confirmed_by_name: str | None
+    document_link: str | None

api/src/services/quality_metrics_service.py

@@ -0,0 +1,365 @@
+"""Сервис «Метрики качества» (design `quality-metrics` §6-§8; AURORA-1759): чтение фактов качества, расчёт — в
+`domain.quality_metrics`. Только чтение, без `VisibilityScope` (видимость по филиалу/ГК не применяется, §8), без аудита.
+
+Выборка единая для всех методов (`_base`): факты приёма (`quality_intake_fact`, по одному на документ) LEFT JOIN
+последний факт исхода документа. «Последний» выбирает SQL — `DISTINCT ON (document_id) ORDER BY document_id,
+outcome_at DESC, ранг исхода DESC, id DESC`; ранг (`*_AFTER_CONFIRM` > подтверждение > удаление/замена до него)
+повторяет tie-break домена, поэтому SQL и домен выбирают один и тот же факт. Период — по `intake_at` (когорта).
+Исход без факта приёма (факты пишутся одной транзакцией, но на всякий случай) в метрики не попадает — у него нет
+даты приёма.
+"""
+
+import csv
+import enum
+import io
+import json
+import uuid
+from collections.abc import Iterable, Sequence
+from datetime import UTC, date, datetime, time, timedelta
+from typing import Any
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import Select, case, distinct, func, select, union
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.domain.quality_metrics import (
+    QualityOutcomeFactRow,
+    compute_matrix,
+    compute_monthly_series,
+)
+from src.models.access import AppUser
+from src.models.enums import QualityCause, QualityOutcome
+from src.models.files import Document
+from src.models.quality_facts import QualityFailureFact, QualityIntakeFact, QualityOutcomeFact
+from src.schemas.common import Page
+from src.schemas.quality_metrics import (
+    ExportFormat,
+    MonthPointOut,
+    QualityCell,
+    QualityCellDocumentOut,
+    QualityFilters,
+    QualityFixesOut,
+    QualityMatrixOut,
+    QualityMonthlySeriesOut,
+    TechnicalFailuresOut,
+)
+
+__all__ = ["EXPORT_COLUMNS", "QualityMetricsService"]
+
+_IN_DENOMINATOR = (QualityOutcome.AUTO_CONFIRMED, QualityOutcome.CONFIRMED)
+_BEFORE_CONFIRM = (QualityOutcome.DELETED, QualityOutcome.SUPERSEDED)
+_AFTER_CONFIRM = (QualityOutcome.DELETED_AFTER_CONFIRM, QualityOutcome.SUPERSEDED_AFTER_CONFIRM)
+
+EXPORT_COLUMNS = (
+    "document_id", "file_id", "legal_entity_id", "legal_entity_name", "group_id", "group_name", "document_type",
+    "pipeline", "is_scan", "service_name", "service_version", "intake_at", "intake_status", "report_date_recognized",
+    "outcome", "outcome_at", "primary_cause", "fixes_value_service", "fixes_value_ours", "fixes_code_service",
+    "fixes_code_ours", "fixes_added_row_service", "fixes_added_row_ours", "fixes_header_service", "fixes_header_ours",
+    "fixes_osv_period_account", "checks_accepted", "issue_count_at_outcome", "report_date_confirmed",
+)  # fmt: skip
+
+
+def _start(day: date) -> datetime:
+    return datetime.combine(day, time.min, tzinfo=UTC)
+
+
+def _latest_outcomes():  # noqa: ANN202 - Subquery
+    """Последний факт исхода на `document_id` (см. модульную строку)."""
+    outcome = QualityOutcomeFact.outcome
+    rank = case((outcome.in_(_AFTER_CONFIRM), 2), (outcome.in_(_IN_DENOMINATOR), 1), else_=0)
+    return (
+        select(QualityOutcomeFact)
+        .distinct(QualityOutcomeFact.document_id)
+        .order_by(
+            QualityOutcomeFact.document_id,
+            QualityOutcomeFact.outcome_at.desc(),
+            rank.desc(),
+            QualityOutcomeFact.id.desc(),
+        )
+        .subquery("latest_outcome")
+    )
+
+
+def _intake_conditions(f: QualityFilters, *, with_document_type: bool = True) -> list[Any]:
+    cond: list[Any] = []
+    if f.date_from:
+        cond.append(QualityIntakeFact.intake_at >= _start(f.date_from))
+    if f.date_to:
+        cond.append(QualityIntakeFact.intake_at < _start(f.date_to + timedelta(days=1)))
+    if f.group_id:
+        cond.append(QualityIntakeFact.group_id == f.group_id)
+    if f.legal_entity_id:
+        cond.append(QualityIntakeFact.legal_entity_id == f.legal_entity_id)
+    if f.pipeline:
+        cond.append(QualityIntakeFact.pipeline == f.pipeline)
+    if f.is_scan is not None:
+        cond.append(QualityIntakeFact.is_scan == f.is_scan)
+    if with_document_type and f.document_type:
+        cond.append(QualityIntakeFact.document_type == f.document_type)
+    return cond
+
+
+def _failure_conditions(f: QualityFilters) -> list[Any]:
+    """Те же фильтры по отказам; тип документа не действует (у отказа документов нет)."""
+    cond: list[Any] = []
+    if f.date_from:
+        cond.append(QualityFailureFact.failed_at >= _start(f.date_from))
+    if f.date_to:
+        cond.append(QualityFailureFact.failed_at < _start(f.date_to + timedelta(days=1)))
+    if f.group_id:
+        cond.append(QualityFailureFact.group_id == f.group_id)
+    if f.legal_entity_id:
+        cond.append(QualityFailureFact.legal_entity_id == f.legal_entity_id)
+    if f.pipeline:
+        cond.append(QualityFailureFact.pipeline == f.pipeline)
+    if f.is_scan is not None:
+        cond.append(QualityFailureFact.is_scan == f.is_scan)
+    return cond
+
+
+def _fixes(row: Any) -> QualityFixesOut:
+    return QualityFixesOut(
+        value_service=row.fixes_value_service,
+        value_ours=row.fixes_value_ours,
+        code_service=row.fixes_code_service,
+        code_ours=row.fixes_code_ours,
+        added_row_service=row.fixes_added_row_service,
+        added_row_ours=row.fixes_added_row_ours,
+        header_service=row.fixes_header_service,
+        header_ours=row.fixes_header_ours,
+        osv_period_account=row.fixes_osv_period_account,
+    )
+
+
+class QualityMetricsService:
+    """Матрица, ряд по месяцам, список документов ячейки и выгрузка фактов."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    @staticmethod
+    def _base(f: QualityFilters, latest: Any, *columns: Any) -> Select[Any]:
+        """`SELECT columns FROM intake LEFT JOIN latest_outcome WHERE фильтры`."""
+        return (
+            select(*columns)
+            .select_from(QualityIntakeFact)
+            .outerjoin(latest, latest.c.document_id == QualityIntakeFact.document_id)
+            .where(*_intake_conditions(f))
+        )
+
+    async def _outcome_rows(self, f: QualityFilters) -> list[Any]:
+        """Документы периода с последним исходом (у документов «в работе» все поля исхода `None`)."""
+        latest = _latest_outcomes()
+        stmt = self._base(
+            f,
+            latest,
+            QualityIntakeFact.document_id,
+            QualityIntakeFact.legal_entity_id,
+            QualityIntakeFact.document_type,
+            QualityIntakeFact.intake_at,
+            latest.c.outcome,
+            latest.c.outcome_at,
+            latest.c.primary_cause,
+            latest.c.report_date,
+        ).order_by(QualityIntakeFact.document_id)
+        return list((await self._session.execute(stmt)).all())
+
+    @staticmethod
+    def _fact_rows(rows: Iterable[Any]) -> list[QualityOutcomeFactRow]:
+        return [
+            QualityOutcomeFactRow(
+                document_id=r.document_id,
+                outcome=r.outcome,
+                outcome_at=r.outcome_at,
+                primary_cause=r.primary_cause,
+                received_at=r.intake_at,
+            )
+            for r in rows
+            if r.outcome is not None
+        ]
+
+    async def _technical_failures(self, f: QualityFilters) -> TechnicalFailuresOut:
+        failed_ids = select(QualityFailureFact.file_id).where(*_failure_conditions(f))
+        intake_ids = select(QualityIntakeFact.file_id).where(*_intake_conditions(f, with_document_type=False))
+        failed_stmt = select(func.count(distinct(QualityFailureFact.file_id))).where(*_failure_conditions(f))
+        failed = (await self._session.execute(failed_stmt)).scalar_one()
+        files = union(failed_ids, intake_ids).subquery()
+        total = (await self._session.execute(select(func.count()).select_from(files))).scalar_one()
+        return TechnicalFailuresOut(
+            failed_files_count=failed, total_files_count=total, share=failed / total if total else 0.0
+        )
+
+    async def matrix(self, f: QualityFilters) -> QualityMatrixOut:
+        rows = await self._outcome_rows(f)
+        result = compute_matrix(self._fact_rows(rows))
+        reports = {
+            (r.legal_entity_id, r.report_date, r.document_type)
+            for r in rows
+            if r.outcome in _IN_DENOMINATOR and r.report_date is not None
+        }
+        return QualityMatrixOut(
+            counts=result.counts,
+            shares=result.shares,
+            denominator=result.denominator,
+            removed_before_confirm=result.removed_before_confirm,
+            removed_after_confirm=result.removed_after_confirm,
+            deleted_after_confirm_count=sum(r.outcome is QualityOutcome.DELETED_AFTER_CONFIRM for r in rows),
+            superseded_after_confirm_count=sum(r.outcome is QualityOutcome.SUPERSEDED_AFTER_CONFIRM for r in rows),
+            in_progress_count=sum(r.outcome is None for r in rows),
+            unique_reports_count=len(reports),
+            technical_failures=await self._technical_failures(f),
+        )
+
+    async def series(self, f: QualityFilters, cause: QualityCause) -> QualityMonthlySeriesOut:
+        points = compute_monthly_series(self._fact_rows(await self._outcome_rows(f)), cause)
+        return QualityMonthlySeriesOut(
+            cause=cause,
+            points=[
+                MonthPointOut(month=p.month, count=p.count, denominator=p.denominator, share=p.share) for p in points
+            ],
+        )
+
+    @staticmethod
+    def _cell_condition(cell: QualityCell, latest: Any) -> Any:
+        if cell is QualityCell.IN_PROGRESS:
+            return latest.c.id.is_(None)
+        if cell is QualityCell.REMOVED_BEFORE_CONFIRM:
+            return latest.c.outcome.in_(_BEFORE_CONFIRM)
+        if cell is QualityCell.REMOVED_AFTER_CONFIRM:
+            return latest.c.outcome.in_(_AFTER_CONFIRM)
+        return (latest.c.outcome.in_(_IN_DENOMINATOR)) & (latest.c.primary_cause == QualityCause(cell.value))
+
+    async def cell_documents(
+        self, f: QualityFilters, cell: QualityCell, *, limit: int, offset: int
+    ) -> Page[QualityCellDocumentOut]:
+        latest = _latest_outcomes()
+        condition = self._cell_condition(cell, latest)
+        total = (await self._session.execute(self._base(f, latest, func.count()).where(condition))).scalar_one()
+        stmt = (
+            self._base(
+                f,
+                latest,
+                QualityIntakeFact.document_id,
+                QualityIntakeFact.document_type,
+                QualityIntakeFact.legal_entity_id,
+                QualityIntakeFact.legal_entity_name,
+                QualityIntakeFact.group_name,
+                QualityIntakeFact.intake_at,
+                latest.c.outcome,
+                latest.c.outcome_at,
+                latest.c.primary_cause,
+                latest.c.fixes_value_service,
+                latest.c.fixes_value_ours,
+                latest.c.fixes_code_service,
+                latest.c.fixes_code_ours,
+                latest.c.fixes_added_row_service,
+                latest.c.fixes_added_row_ours,
+                latest.c.fixes_header_service,
+                latest.c.fixes_header_ours,
+                latest.c.fixes_osv_period_account,
+                latest.c.checks_accepted,
+                latest.c.issue_count_at_outcome,
+                latest.c.report_date,
+                latest.c.confirmed_by,
+            )
+            .where(condition)
+            .order_by(QualityIntakeFact.intake_at.desc(), QualityIntakeFact.document_id)
+            .limit(limit)
+            .offset(offset)
+        )
+        rows = list((await self._session.execute(stmt)).all())
+        ids = [r.document_id for r in rows]
+        existing = (
+            set((await self._session.execute(select(Document.id).where(Document.id.in_(ids)))).scalars())
+            if ids
+            else set()
+        )
+        user_ids = {r.confirmed_by for r in rows if r.confirmed_by is not None}
+        names: dict[uuid.UUID, str] = {}
+        if user_ids:
+            found = await self._session.execute(select(AppUser.id, AppUser.full_name).where(AppUser.id.in_(user_ids)))
+            names = {uid: name for uid, name in found.all()}
+        previous = await self._previous_outcomes(ids) if cell is QualityCell.REMOVED_AFTER_CONFIRM else {}
+        items = [
+            QualityCellDocumentOut(
+                document_id=r.document_id,
+                document_type=r.document_type,
+                legal_entity_id=r.legal_entity_id,
+                legal_entity_name=r.legal_entity_name,
+                group_name=r.group_name,
+                received_at=r.intake_at,
+                outcome=r.outcome,
+                outcome_at=r.outcome_at,
+                previous_outcome=previous.get(r.document_id),
+                primary_cause=r.primary_cause,
+                report_date=r.report_date,
+                fixes=_fixes(r) if r.outcome is not None else None,
+                checks_accepted=r.checks_accepted,
+                issue_count_at_outcome=r.issue_count_at_outcome,
+                confirmed_by_id=r.confirmed_by,
+                confirmed_by_name=names.get(r.confirmed_by) if r.confirmed_by else None,
+                document_link=f"/api/v1/documents/{r.document_id}" if r.document_id in existing else None,
+            )
+            for r in rows
+        ]
+        return Page(items=items, total=total, limit=limit, offset=offset)
+
+    async def _previous_outcomes(self, document_ids: Sequence[uuid.UUID]) -> dict[uuid.UUID, QualityOutcome]:
+        """Исход до `*_AFTER_CONFIRM`: самый поздний факт подтверждения документа."""
+        if not document_ids:
+            return {}
+        stmt = (
+            select(QualityOutcomeFact.document_id, QualityOutcomeFact.outcome)
+            .where(QualityOutcomeFact.document_id.in_(document_ids), QualityOutcomeFact.outcome.in_(_IN_DENOMINATOR))
+            .order_by(
+                QualityOutcomeFact.document_id, QualityOutcomeFact.outcome_at.desc(), QualityOutcomeFact.id.desc()
+            )
+            .distinct(QualityOutcomeFact.document_id)
+        )
+        return {doc: outcome for doc, outcome in (await self._session.execute(stmt)).all()}
+
+    async def export_records(self, f: QualityFilters) -> list[dict[str, Any]]:
+        """Факты периода одной таблицей: приём + последний исход (поля исхода пусты у документов «в работе»)."""
+        latest = _latest_outcomes()
+        i = QualityIntakeFact
+        stmt = self._base(
+            f,
+            latest,
+            i.document_id, i.file_id, i.legal_entity_id, i.legal_entity_name, i.group_id, i.group_name,
+            i.document_type, i.pipeline, i.is_scan, i.service_name, i.service_version, i.intake_at, i.intake_status,
+            i.report_date.label("report_date_recognized"),
+            latest.c.outcome, latest.c.outcome_at, latest.c.primary_cause,
+            latest.c.fixes_value_service, latest.c.fixes_value_ours, latest.c.fixes_code_service,
+            latest.c.fixes_code_ours, latest.c.fixes_added_row_service, latest.c.fixes_added_row_ours,
+            latest.c.fixes_header_service, latest.c.fixes_header_ours, latest.c.fixes_osv_period_account,
+            latest.c.checks_accepted, latest.c.issue_count_at_outcome,
+            latest.c.report_date.label("report_date_confirmed"),
+        ).order_by(i.intake_at, i.document_id)  # fmt: skip
+        return [dict(r._mapping) for r in (await self._session.execute(stmt)).all()]
+
+    async def export(self, f: QualityFilters, fmt: ExportFormat) -> tuple[bytes, str, str]:
+        """Файл выгрузки: (содержимое, media type, расширение). CSV — UTF-8 с BOM (для Excel), разделитель — запятая."""
+        records = await self.export_records(f)
+        if fmt == "json":
+            return (
+                json.dumps(jsonable_encoder(records), ensure_ascii=False).encode(),
+                "application/json",
+                "json",
+            )
+        buffer = io.StringIO()
+        writer = csv.DictWriter(buffer, fieldnames=EXPORT_COLUMNS, extrasaction="ignore")
+        writer.writeheader()
+        for record in records:
+            writer.writerow({k: "" if v is None else _cell(v) for k, v in record.items()})
+        return buffer.getvalue().encode("utf-8-sig"), "text/csv; charset=utf-8", "csv"
+
+
+def _cell(value: Any) -> Any:
+    """Значение ячейки CSV: перечисления — их значение, даты — ISO, булевы — true/false."""
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, enum.Enum):
+        return value.value
+    if isinstance(value, datetime | date):
+        return value.isoformat()
+    return value

api/tests/test_api_quality_metrics.py

@@ -0,0 +1,324 @@
+"""API `/quality-metrics` (design `quality-metrics` §6-§8; AURORA-1759, Task 15).
+
+Быстрые тесты (без БД): право `QUALITY_METRICS_VIEW`, валидация query, маршруты в OpenAPI, CSV-ячейки. Интеграционные
+(`integration`, PostgreSQL): факты пишутся напрямую через `insert_*_fact`, без полного цикла маппинга."""
+
+import csv
+import datetime
+import io
+import uuid
+from collections.abc import AsyncIterator
+from types import SimpleNamespace
+from typing import Any
+
+import httpx
+import pytest
+from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.dependencies.auth import current_user
+from src.api.router import router
+from src.core.exception_handlers import register_exception_handlers
+from src.db.db import get_async_session
+from src.models.enums import (
+    DocumentStatus,
+    DocumentType,
+    QualityCause,
+    QualityFailureKind,
+    QualityOutcome,
+    RecognitionPipeline,
+)
+from src.repositories.quality_facts import insert_failure_fact, insert_intake_fact, insert_outcome_fact
+from src.repositories.users import UserWithAccess
+from src.services.quality_metrics_service import EXPORT_COLUMNS
+from tests.conftest import AuthClientFactory
+
+HTTP_OK, HTTP_FORBIDDEN, HTTP_UNPROCESSABLE = 200, 403, 422
+KA_RF = "ka_rf@armka.local"
+BASE = "/quality-metrics"
+UTC = datetime.UTC
+T0 = datetime.datetime(2026, 11, 1, 10, 0, tzinfo=UTC)
+T1 = T0 + datetime.timedelta(hours=1)
+T2 = T0 + datetime.timedelta(hours=2)
+VIEW = "QUALITY_METRICS_VIEW"
+THREE, FOUR = 3, 4
+
+
+# --- быстрые тесты --------------------------------------------------------------------------------------------------
+
+
+async def _forbidden_client(permissions: frozenset[str]) -> AsyncIterator[httpx.AsyncClient]:
+    app = FastAPI()
+    register_exception_handlers(app)
+    app.include_router(router)
+
+    async def no_session() -> AsyncIterator[None]:
+        yield None
+
+    async def principal() -> UserWithAccess:
+        return UserWithAccess(
+            user=SimpleNamespace(),
+            regional_office=SimpleNamespace(),
+            roles=(),
+            permissions=permissions,  # type: ignore[arg-type]
+        )
+
+    app.dependency_overrides[get_async_session] = no_session
+    app.dependency_overrides[current_user] = principal
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test/api/v1") as client:
+        yield client
+
+
+@pytest.fixture
+async def client_without_permission() -> AsyncIterator[httpx.AsyncClient]:
+    async for client in _forbidden_client(frozenset({"DIRECTORY_VIEW"})):
+        yield client
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/matrix",
+        "/series?cause=AS_IS",
+        f"/cells/{QualityCause.AS_IS}/documents",
+        "/export",
+    ],
+)
+async def test_quality_metrics_requires_permission(client_without_permission: httpx.AsyncClient, path: str) -> None:
+    response = await client_without_permission.get(BASE + path)
+    assert response.status_code == HTTP_FORBIDDEN
+    assert VIEW in str(response.json())
+
+
+async def test_quality_metrics_matrix_requires_permission(client_without_permission: httpx.AsyncClient) -> None:
+    response = await client_without_permission.get("/quality-metrics/matrix")
+    assert response.status_code == HTTP_FORBIDDEN
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/matrix?date_from=2026-02-01&date_to=2026-01-01",
+        "/matrix?document_type=NOPE",
+        "/series",  # cause обязателен
+        "/series?cause=NOPE",
+        "/cells/NOPE/documents",
+        "/cells/AS_IS/documents?limit=0",
+        "/export?format=xml",
+    ],
+)
+async def test_quality_metrics_validates_query(path: str) -> None:
+    async for client in _forbidden_client(frozenset({VIEW})):
+        response = await client.get(BASE + path)
+        assert response.status_code == HTTP_UNPROCESSABLE, response.text
+
+
+def test_quality_metrics_routes_in_openapi() -> None:
+    app = FastAPI()
+    app.include_router(router)
+    paths = app.openapi()["paths"]
+    for path in ("/matrix", "/series", "/cells/{cell}/documents", "/export"):
+        assert f"/api/v1{BASE}{path}" in paths
+
+
+# --- интеграционные тесты -------------------------------------------------------------------------------------------
+
+
+def _intake_columns(document_id: uuid.UUID, **over: Any) -> dict[str, Any]:
+    columns: dict[str, Any] = {
+        "document_id": document_id,
+        "file_id": uuid.uuid4(),
+        "job_id": uuid.uuid4(),
+        "attempt": 1,
+        "legal_entity_id": uuid.uuid4(),
+        "legal_entity_name": "ООО Тест",
+        "group_id": None,
+        "group_name": None,
+        "document_type": DocumentType.OSV,
+        "pipeline": RecognitionPipeline.image,
+        "is_scan": True,
+        "intake_at": T0,
+        "intake_status": DocumentStatus.RECOGNIZED,
+        "payload": {},
+    }
+    columns.update(over)
+    return columns
+
+
+async def _doc(
+    session: AsyncSession,
+    outcome: QualityOutcome | None,
+    cause: QualityCause | None,
+    *,
+    at: datetime.datetime = T1,
+    **intake: Any,
+) -> uuid.UUID:
+    """Документ с фактом приёма и (если `outcome`) одним фактом исхода."""
+    document_id = uuid.uuid4()
+    await insert_intake_fact(session, **_intake_columns(document_id, **intake))
+    if outcome is not None:
+        await insert_outcome_fact(
+            session, document_id=document_id, outcome=outcome, outcome_at=at, primary_cause=cause, payload={}
+        )
+    await session.flush()
+    return document_id
+
+
+@pytest.mark.integration
+async def test_quality_metrics_matrix_returns_counts(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.SERVICE_ERROR)
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS)
+    await _doc(seeded, QualityOutcome.AUTO_CONFIRMED, QualityCause.WITHOUT_PARTICIPATION)
+    await _doc(seeded, QualityOutcome.DELETED, None)
+    await _doc(seeded, None, None)  # в работе
+    client = await auth_client(KA_RF)
+    response = await client.get(f"{BASE}/matrix")
+    assert response.status_code == HTTP_OK
+    body = response.json()
+    assert body["counts"]["SERVICE_ERROR"] == 1
+    assert body["counts"]["AS_IS"] == 1
+    assert body["counts"]["OUR_ERROR"] == 0
+    assert body["denominator"] == THREE
+    assert body["shares"]["SERVICE_ERROR"] == pytest.approx(1 / 3)
+    assert body["removed_before_confirm"] == 1
+    assert body["in_progress_count"] == 1
+
+
+@pytest.mark.integration
+async def test_quality_metrics_matrix_dedupes_to_latest_outcome_fact(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    document_id = await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, at=T1)
+    await insert_outcome_fact(
+        seeded,
+        document_id=document_id,
+        outcome=QualityOutcome.SUPERSEDED_AFTER_CONFIRM,
+        outcome_at=T2,
+        primary_cause=None,
+        payload={},
+    )
+    await seeded.flush()
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"{BASE}/matrix")).json()
+    assert body["counts"].get("AS_IS", 0) == 0  # устаревший факт не виден
+    assert body["superseded_after_confirm_count"] == 1
+    assert body["removed_after_confirm"] == 1
+    assert body["denominator"] == 0
+    # series и список ячейки видят тот же последний факт
+    series = (await client.get(f"{BASE}/series", params={"cause": "AS_IS"})).json()
+    assert series["points"] == []
+    cell = (await client.get(f"{BASE}/cells/AS_IS/documents")).json()
+    assert cell["total"] == 0
+    removed = (await client.get(f"{BASE}/cells/REMOVED_AFTER_CONFIRM/documents")).json()
+    assert removed["total"] == 1
+    assert removed["items"][0]["previous_outcome"] == "CONFIRMED"
+    export = (await client.get(f"{BASE}/export", params={"format": "json"})).json()
+    assert [r["outcome"] for r in export] == ["SUPERSEDED_AFTER_CONFIRM"]
+
+
+@pytest.mark.integration
+async def test_quality_metrics_equal_outcome_at_prefers_after_confirm(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    document_id = await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, at=T1)
+    await insert_outcome_fact(
+        seeded,
+        document_id=document_id,
+        outcome=QualityOutcome.DELETED_AFTER_CONFIRM,
+        outcome_at=T1,
+        primary_cause=None,
+        payload={},
+    )
+    await seeded.flush()
+    body = (await (await auth_client(KA_RF)).get(f"{BASE}/matrix")).json()
+    assert body["deleted_after_confirm_count"] == 1
+    assert body["counts"]["AS_IS"] == 0
+
+
+@pytest.mark.integration
+async def test_quality_metrics_filters_and_series(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    group = uuid.uuid4()
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.OUR_ERROR, group_id=group, group_name="ГК")
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, group_id=group, group_name="ГК", is_scan=False)
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, group_id=group, document_type=DocumentType.BALANCE)
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS)  # другая ГК
+    await _doc(  # другой месяц приёма
+        seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, group_id=group, intake_at=T0 + datetime.timedelta(days=40)
+    )
+    client = await auth_client(KA_RF)
+
+    async def denominator(**params: Any) -> int:
+        return (await client.get(f"{BASE}/matrix", params={"group_id": str(group), **params})).json()["denominator"]
+
+    assert await denominator() == FOUR
+    assert await denominator(document_type="BALANCE") == 1
+    assert await denominator(is_scan="false") == 1
+    assert await denominator(pipeline="tabular") == 0
+    assert await denominator(date_from="2026-11-01", date_to="2026-11-01") == THREE  # границы включительно
+    assert await denominator(date_from="2026-11-02") == 1
+    series = (await client.get(f"{BASE}/series", params={"group_id": str(group), "cause": "AS_IS"})).json()
+    assert [(p["month"], p["count"], p["denominator"]) for p in series["points"]][0] == ("2026-11-01", 2, 3)
+    assert len(series["points"]) == 2  # noqa: PLR2004 - ноябрь и декабрь
+
+
+@pytest.mark.integration
+async def test_quality_metrics_technical_failures_by_files(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS)
+    for kind in (QualityFailureKind.SERVICE_ERROR, QualityFailureKind.TIMEOUT):
+        await insert_failure_fact(
+            seeded,
+            job_id=uuid.uuid4(),
+            file_id=uuid.uuid4(),
+            attempt=1,
+            legal_entity_id=uuid.uuid4(),
+            legal_entity_name="ООО Тест",
+            pipeline=RecognitionPipeline.image,
+            is_scan=True,
+            failure_kind=kind,
+            failed_at=T1,
+        )
+    await seeded.flush()
+    client = await auth_client(KA_RF)
+    failures = (await client.get(f"{BASE}/matrix", params={"document_type": "BALANCE"})).json()["technical_failures"]
+    assert failures == {"failed_files_count": 2, "total_files_count": 3, "share": pytest.approx(2 / 3)}
+
+
+@pytest.mark.integration
+async def test_quality_metrics_cell_documents(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    for _ in range(3):
+        await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.SERVICE_ERROR, legal_entity_name="ООО Альфа")
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS)
+    in_progress = await _doc(seeded, None, None)
+    client = await auth_client(KA_RF)
+    page = (await client.get(f"{BASE}/cells/SERVICE_ERROR/documents", params={"limit": 2})).json()
+    assert (page["total"], page["limit"], page["offset"], len(page["items"])) == (3, 2, 0, 2)
+    item = page["items"][0]
+    assert item["primary_cause"] == "SERVICE_ERROR"
+    assert item["legal_entity_name"] == "ООО Альфа"
+    assert item["document_link"] is None  # рабочего документа нет
+    assert set(item["fixes"]) >= {"value_service", "value_ours", "osv_period_account"}
+    second = (await client.get(f"{BASE}/cells/SERVICE_ERROR/documents", params={"limit": 2, "offset": 2})).json()
+    assert len(second["items"]) == 1
+    progress = (await client.get(f"{BASE}/cells/IN_PROGRESS/documents")).json()
+    assert [i["document_id"] for i in progress["items"]] == [str(in_progress)]
+    assert progress["items"][0]["outcome"] is None
+
+
+@pytest.mark.integration
+async def test_quality_metrics_export(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    await _doc(seeded, QualityOutcome.CONFIRMED, QualityCause.AS_IS, legal_entity_name="ООО Бета")
+    await _doc(seeded, None, None)
+    client = await auth_client(KA_RF)
+    response = await client.get(f"{BASE}/export")
+    assert response.status_code == HTTP_OK
+    assert response.headers["content-type"].startswith("text/csv")
+    assert "attachment" in response.headers["content-disposition"]
+    rows = list(csv.DictReader(io.StringIO(response.content.decode("utf-8-sig"))))
+    assert list(rows[0]) == list(EXPORT_COLUMNS)
+    assert len(rows) == 2  # noqa: PLR2004
+    assert {r["outcome"] for r in rows} == {"CONFIRMED", ""}
+    assert rows[0]["is_scan"] == "true"
+    as_json = await client.get(f"{BASE}/export", params={"format": "json", "date_from": "2026-11-01"})
+    assert len(as_json.json()) == 2  # noqa: PLR2004