diff --git a/src/contract_check/api/admin/review.py b/src/contract_check/api/admin/review.py new file mode 100644 index 0000000..ab7d4f4 --- /dev/null +++ b/src/contract_check/api/admin/review.py @@ -0,0 +1,250 @@ +"""Manual-review queue routes for the admin panel. + +List / detail / three operator actions (send to analysis, complete, reject). +All mutations are HTMX-driven and protected by the ``HX-Request`` CSRF guard. +""" + +from __future__ import annotations + +import datetime as dt +import json +import uuid +from decimal import Decimal +from typing import Annotated + +from fastapi import APIRouter, Depends, Form, HTTPException, Request, status +from fastapi.responses import HTMLResponse, RedirectResponse + +from src.contract_check.api.admin.auth import AdminUser, HtmxGuard, require_admin +from src.contract_check.api.admin.templating import templates +from src.contract_check.api.deps import AsyncSessionDep, PublisherDep +from src.contract_check.core.config import get_settings +from src.contract_check.core.db.repositories import ReviewQueueRepository +from src.contract_check.core.logging import get_logger +from src.contract_check.core.review.actions import ( + ReviewActionError, + complete, + reject, + send_to_analysis, +) +from src.contract_check.core.s3.minio_storage import MinioStorage, MinioStorageError + +log = get_logger(__name__) + +router = APIRouter( + prefix="/admin/review", + tags=["admin-review"], + dependencies=[Depends(require_admin)], +) + +PAGE_SIZE = 25 + + +def _storage() -> MinioStorage: + settings = get_settings() + return MinioStorage.from_endpoint_url( + endpoint_url=settings.s3_endpoint_url, + access_key=settings.s3_access_key, + secret_key=settings.s3_secret_key, + bucket=settings.s3_bucket, + region=settings.s3_region, + ) + + +async def _text_preview(extracted_s3_key: str | None) -> str | None: + if not extracted_s3_key: + return None + try: + text_bytes = await _storage().get(extracted_s3_key) + text = text_bytes.decode("utf-8", errors="replace") + return text[:4000] + ("\n…" if len(text) > 4000 else "") + except MinioStorageError: + return None + + +def _date_or_none(value: str) -> dt.date | None: + try: + return dt.date.fromisoformat(value) + except ValueError: + return None + + +def _decimal_or_none(value: str) -> Decimal | None: + value = value.strip() + if not value: + return None + try: + return Decimal(value) + except Exception: + return None + + +@router.get("", response_class=HTMLResponse, include_in_schema=False) +async def list_review_queue( + request: Request, + session: AsyncSessionDep, + date_from: str = "", + date_to: str = "", + conf_min: str = "", + conf_max: str = "", + page: int = 1, +) -> HTMLResponse: + """Paginated manual-review backlog with date/confidence filters.""" + page = max(page, 1) + offset = (page - 1) * PAGE_SIZE + + date_from_parsed = _date_or_none(date_from) + date_to_parsed = _date_or_none(date_to) + conf_min_parsed = _decimal_or_none(conf_min) + conf_max_parsed = _decimal_or_none(conf_max) + + repo = ReviewQueueRepository(session) + rows, total = await repo.list_queue( + limit=PAGE_SIZE, + offset=offset, + date_from=date_from_parsed, + date_to=date_to_parsed, + conf_min=conf_min_parsed, + conf_max=conf_max_parsed, + ) + pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE) + + response = templates.TemplateResponse( + request, + "review_list.html", + { + "request": request, + "items": rows, + "total": total, + "page": page, + "pages": pages, + "date_from": date_from, + "date_to": date_to, + "conf_min": conf_min, + "conf_max": conf_max, + }, + ) + _attach_toast(response, request.query_params.get("toast")) + return response + + +@router.get("/{document_id}", response_class=HTMLResponse, include_in_schema=False) +async def review_detail( + request: Request, + session: AsyncSessionDep, + document_id: uuid.UUID, +) -> HTMLResponse: + """Detail page: prescreen metadata + extracted text preview + actions.""" + item = await ReviewQueueRepository(session).get_item(document_id) + if item is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="document not found") + + preview = await _text_preview(item.extracted_s3_key) + response = templates.TemplateResponse( + request, + "review_detail.html", + { + "request": request, + "item": item, + "preview": preview, + "toast": request.query_params.get("toast"), + }, + ) + _attach_toast(response, request.query_params.get("toast")) + return response + + +def _attach_toast(response: HTMLResponse, toast: str | None) -> None: + if toast: + response.headers["HX-Trigger"] = json.dumps({"showToast": toast}) + + +def _review_action_redirect( + *, + document_id: uuid.UUID, + success_toast: str, + error: ReviewActionError | None = None, +) -> RedirectResponse: + if error is not None: + target = f"/admin/review/{document_id}?toast={error.message}" + else: + target = f"/admin/review?toast={success_toast}" + return RedirectResponse(url=target, status_code=status.HTTP_303_SEE_OTHER) + + +@router.post( + "/{document_id}/send-to-analysis", response_class=HTMLResponse, include_in_schema=False +) +async def send_to_analysis_action( + request: Request, + session: AsyncSessionDep, + publisher: PublisherDep, + document_id: uuid.UUID, + admin: Annotated[AdminUser, Depends(require_admin)], + _: Annotated[None, HtmxGuard], +) -> RedirectResponse: + try: + await send_to_analysis(session, publisher, document_id=document_id) + except ReviewActionError as exc: + return _review_action_redirect(document_id=document_id, success_toast="", error=exc) + log.info( + "admin_review_action", + action="send_to_analysis", + document_id=str(document_id), + admin_user_id=str(admin.user_id), + ) + return _review_action_redirect( + document_id=document_id, + success_toast="Документ отправлен на полный анализ", + ) + + +@router.post("/{document_id}/complete", response_class=HTMLResponse, include_in_schema=False) +async def complete_action( + request: Request, + session: AsyncSessionDep, + document_id: uuid.UUID, + admin: Annotated[AdminUser, Depends(require_admin)], + _: Annotated[None, HtmxGuard], + note: Annotated[str, Form()] = "", +) -> RedirectResponse: + try: + await complete(session, document_id=document_id, note=note) + except ReviewActionError as exc: + return _review_action_redirect(document_id=document_id, success_toast="", error=exc) + log.info( + "admin_review_action", + action="complete", + document_id=str(document_id), + admin_user_id=str(admin.user_id), + ) + return _review_action_redirect( + document_id=document_id, + success_toast="Документ закрыт с пометкой оператора", + ) + + +@router.post("/{document_id}/reject", response_class=HTMLResponse, include_in_schema=False) +async def reject_action( + request: Request, + session: AsyncSessionDep, + document_id: uuid.UUID, + admin: Annotated[AdminUser, Depends(require_admin)], + _: Annotated[None, HtmxGuard], + reason: Annotated[str, Form()] = "", +) -> RedirectResponse: + try: + await reject(session, document_id=document_id, reason=reason) + except ReviewActionError as exc: + return _review_action_redirect(document_id=document_id, success_toast="", error=exc) + log.info( + "admin_review_action", + action="reject", + document_id=str(document_id), + admin_user_id=str(admin.user_id), + reason=reason, + ) + return _review_action_redirect( + document_id=document_id, + success_toast="Документ отклонён, Document Slot возвращён", + ) diff --git a/src/contract_check/api/admin/router.py b/src/contract_check/api/admin/router.py index 5cd1e09..3f8ac84 100644 --- a/src/contract_check/api/admin/router.py +++ b/src/contract_check/api/admin/router.py @@ -14,6 +14,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse from src.contract_check.api.admin.auth import ADMIN_COOKIE, ADMIN_COOKIE_MAX_AGE, resolve_admin from src.contract_check.api.admin.billing import router as billing_router +from src.contract_check.api.admin.review import router as review_router from src.contract_check.api.admin.templating import templates from src.contract_check.api.admin.users import router as users_router from src.contract_check.api.deps import AsyncSessionDep, RateLimiterDep, require_auth_rate_limit @@ -28,6 +29,7 @@ log = get_logger(__name__) router = APIRouter(tags=["admin"]) router.include_router(users_router) router.include_router(billing_router) +router.include_router(review_router) @router.get("/admin", include_in_schema=False) diff --git a/src/contract_check/api/admin/templates/base.html b/src/contract_check/api/admin/templates/base.html index 212b6b4..3d8bc26 100644 --- a/src/contract_check/api/admin/templates/base.html +++ b/src/contract_check/api/admin/templates/base.html @@ -70,6 +70,7 @@
diff --git a/src/contract_check/api/admin/templates/review_detail.html b/src/contract_check/api/admin/templates/review_detail.html new file mode 100644 index 0000000..65e480c --- /dev/null +++ b/src/contract_check/api/admin/templates/review_detail.html @@ -0,0 +1,72 @@ +{% extends "base.html" %} +{% block title %}Ручная проверка — {{ item.filename }}{% endblock %} +{% block content %} +
+ ← Назад к очереди +
+ +
+
+

{{ item.filename }}

+

ID: {{ item.document_id }}

+ +

Реквизиты договора

+ + + + + + + + + + + + + +
Тип{{ item.contract_type or '—' }}
Сторона А{{ item.party_a or '—' }}
Сторона Б{{ item.party_b or '—' }}
Сумма{% if item.total_amount %}{{ item.total_amount }} {{ item.currency or '' }}{% else %}—{% endif %}
Срок{{ item.start_date or '—' }} — {{ item.end_date or '—' }}
Неустойка{{ item.has_penalty_clause | yn }}
Расторжение{{ item.has_termination_clause | yn }}
Арбитраж{{ item.has_arbitration | yn }}
Уверенность{{ (item.confidence_score * 100)|round(1) }}%
Пресскрин{{ item.prescreened_at | dt }}
+ + {% if item.auto_findings %} +

Находки

+ + {% endif %} + + {% if item.auto_summary %} +

Авто-сводка

+

{{ item.auto_summary }}

+ {% endif %} +
+ +
+

Превью текста

+ {% if preview %} + 
{{ preview }}
+ {% else %} +

Превью недоступно.

+ {% endif %} + +

Действия

+ +
+

Отправить документ на полный анализ ИИ.

+ +
+ +
+ + + +
+ +
+ + + +
+
+
+{% endblock %} diff --git a/src/contract_check/api/admin/templates/review_list.html b/src/contract_check/api/admin/templates/review_list.html new file mode 100644 index 0000000..3949bec --- /dev/null +++ b/src/contract_check/api/admin/templates/review_list.html @@ -0,0 +1,59 @@ +{% extends "base.html" %} +{% block title %}Ручная проверка — Админка{% endblock %} +{% block content %} +
+
+

Ручная проверка ({{ total }})

+ +
+ + + + + +
+
+ + + + + + + + + + + + + + + + {% for item in items %} + + + + + + + + + + + {% else %} + + {% endfor %} + +
ДокументПользовательСтороныСуммаУверенностьНаходкиСоздан
{{ item.filename }} + {{ item.email or 'tg ' ~ item.telegram_id }} + {{ item.party_a or '—' }} → {{ item.party_b or '—' }}{% if item.total_amount %}{{ item.total_amount }} {{ item.currency or '' }}{% else %}—{% endif %}{{ (item.confidence_score * 100)|round(1) }}%{{ item.auto_findings | length }}{{ item.created_at | dt }}Открыть
Очередь пуста
+ + {% if pages > 1 %} +
+ {% for p in range(1, pages + 1) %} + {% if p == page %}{{ p }} + {% else %}{{ p }}{% endif %} + {% endfor %} +
+ {% endif %} +
+{% endblock %} diff --git a/src/contract_check/api/admin/templating.py b/src/contract_check/api/admin/templating.py index 35533e7..d9e81c0 100644 --- a/src/contract_check/api/admin/templating.py +++ b/src/contract_check/api/admin/templating.py @@ -25,4 +25,13 @@ def _format_dt(value: object, fmt: str = "%Y-%m-%d %H:%M UTC") -> str: return str(value) +def _yes_no_none(value: object) -> str: + if value is True: + return "да" + if value is False: + return "нет" + return "—" + + templates.env.filters["dt"] = _format_dt +templates.env.filters["yn"] = _yes_no_none diff --git a/src/contract_check/core/db/repositories/__init__.py b/src/contract_check/core/db/repositories/__init__.py index 06ff270..5807081 100644 --- a/src/contract_check/core/db/repositories/__init__.py +++ b/src/contract_check/core/db/repositories/__init__.py @@ -19,6 +19,7 @@ from src.contract_check.core.db.repositories.passkeys import PasskeyRepository from src.contract_check.core.db.repositories.prescreen_results import PrescreenResultRepository from src.contract_check.core.db.repositories.profiles import UserProfileRepository from src.contract_check.core.db.repositories.reports import ReportRepository +from src.contract_check.core.db.repositories.review import ReviewQueueRepository from src.contract_check.core.db.repositories.service_tokens import ServiceTokenRepository from src.contract_check.core.db.repositories.subscriptions import SubscriptionsRepository from src.contract_check.core.db.repositories.users import UserRepository @@ -33,6 +34,7 @@ __all__ = [ "PasskeyRepository", "PrescreenResultRepository", "ReportRepository", + "ReviewQueueRepository", "ServiceTokenRepository", "SubscriptionsRepository", "UserProfileRepository", diff --git a/src/contract_check/core/db/repositories/review.py b/src/contract_check/core/db/repositories/review.py new file mode 100644 index 0000000..e64c6fb --- /dev/null +++ b/src/contract_check/core/db/repositories/review.py @@ -0,0 +1,188 @@ +"""Manual-review queue data access. + +Joins documents in ``manual_review`` with their latest prescreen result and +owning user so an operator can triage them without leaving the admin panel. +""" + +from __future__ import annotations + +import datetime as dt +import json +import uuid +from dataclasses import dataclass +from decimal import Decimal +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + + +@dataclass(frozen=True, slots=True) +class ReviewQueueItem: + """One manual-review backlog row with prescreen metadata and user identity.""" + + document_id: uuid.UUID + user_id: uuid.UUID + email: str | None + telegram_id: int | None + filename: str + status: str + created_at: dt.datetime + extracted_s3_key: str | None + prescreen_result_id: uuid.UUID + correlation_id: uuid.UUID + contract_type: str | None + party_a: str | None + party_b: str | None + total_amount: Decimal | None + currency: str | None + start_date: dt.date | None + end_date: dt.date | None + has_penalty_clause: bool | None + has_termination_clause: bool | None + has_arbitration: bool | None + confidence_score: Decimal | None + auto_summary: str | None + auto_findings: list[dict[str, Any]] + prescreened_at: dt.datetime + + +class ReviewQueueRepository: + """Query the manual-review backlog and load one item for the detail page.""" + + _SELECT_COLUMNS = ( + "d.id, d.user_id, u.email, u.telegram_id, d.filename, d.status, d.created_at, " + "d.extracted_s3_key, " + "pr.id AS prescreen_result_id, pr.correlation_id, pr.contract_type, pr.party_a, " + "pr.party_b, pr.total_amount, pr.currency, pr.start_date, pr.end_date, " + "pr.has_penalty_clause, pr.has_termination_clause, pr.has_arbitration, " + "pr.confidence_score, pr.auto_summary, pr.auto_findings, pr.prescreened_at" + ) + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def list_queue( + self, + *, + limit: int, + offset: int, + date_from: dt.date | None = None, + date_to: dt.date | None = None, + conf_min: Decimal | float | None = None, + conf_max: Decimal | float | None = None, + ) -> tuple[list[ReviewQueueItem], int]: + """Return paginated backlog rows plus total count. + + Ordering is newest-first by document creation time. Confidence and date + filters apply to the selected (latest) prescreen result per document. + """ + params: dict[str, Any] = {"limit": limit, "offset": offset} + date_clauses = ["d.status = 'manual_review'"] + if date_from is not None: + date_clauses.append("CAST(d.created_at AS DATE) >= :date_from") + params["date_from"] = date_from + if date_to is not None: + date_clauses.append("CAST(d.created_at AS DATE) <= :date_to") + params["date_to"] = date_to + + where = " AND ".join(date_clauses) + + # Inner query picks the latest prescreen result per document. + sql = ( + f"SELECT * FROM (" + f" SELECT DISTINCT ON (d.id) {self._SELECT_COLUMNS}" + f" FROM documents d" + f" JOIN users u ON u.id = d.user_id" + f" JOIN prescreen_results pr ON pr.document_id = d.id" + f" WHERE {where}" + f" ORDER BY d.id, pr.prescreened_at DESC" + f") q" + ) + + outer_clauses: list[str] = [] + if conf_min is not None: + outer_clauses.append("q.confidence_score >= :conf_min") + params["conf_min"] = float(conf_min) + if conf_max is not None: + outer_clauses.append("q.confidence_score <= :conf_max") + params["conf_max"] = float(conf_max) + + outer_where = "1 = 1" + if outer_clauses: + outer_where = " AND ".join(outer_clauses) + + rows_result = await self._session.execute( + text( + f"{sql} WHERE {outer_where} ORDER BY q.created_at DESC LIMIT :limit OFFSET :offset" + ), + params, + ) + rows = [self._row_to_item(row) for row in rows_result.all()] + + count_result = await self._session.execute( + text( + f"SELECT count(DISTINCT d.id) FROM documents d " + f"JOIN prescreen_results pr ON pr.document_id = d.id " + f"WHERE {where}" + ), + {k: v for k, v in params.items() if k not in ("limit", "offset")}, + ) + total = int(count_result.scalar_one()) + return rows, total + + async def get_item(self, document_id: uuid.UUID) -> ReviewQueueItem | None: + """Load a single backlog row by document id, or None if not manual-review.""" + result = await self._session.execute( + text( + f"SELECT {self._SELECT_COLUMNS} " + f"FROM documents d " + f"JOIN users u ON u.id = d.user_id " + f"JOIN prescreen_results pr ON pr.document_id = d.id " + f"WHERE d.id = :d AND d.status = 'manual_review' " + f"ORDER BY pr.prescreened_at DESC " + f"LIMIT 1" + ), + {"d": document_id}, + ) + row = result.first() + if row is None: + return None + return self._row_to_item(row) + + @staticmethod + def _row_to_item(row: Any) -> ReviewQueueItem: + raw_findings = row.auto_findings + findings: list[dict[str, Any]] + if isinstance(raw_findings, str): + findings = json.loads(raw_findings) + elif raw_findings is None: + findings = [] + else: + findings = list(raw_findings) + return ReviewQueueItem( + document_id=row.id, + user_id=row.user_id, + email=row.email, + telegram_id=row.telegram_id, + filename=row.filename, + status=row.status, + created_at=row.created_at, + extracted_s3_key=row.extracted_s3_key, + prescreen_result_id=row.prescreen_result_id, + correlation_id=row.correlation_id, + contract_type=row.contract_type, + party_a=row.party_a, + party_b=row.party_b, + total_amount=row.total_amount, + currency=row.currency, + start_date=row.start_date, + end_date=row.end_date, + has_penalty_clause=row.has_penalty_clause, + has_termination_clause=row.has_termination_clause, + has_arbitration=row.has_arbitration, + confidence_score=row.confidence_score, + auto_summary=row.auto_summary, + auto_findings=findings, + prescreened_at=row.prescreened_at, + ) diff --git a/src/contract_check/core/review/__init__.py b/src/contract_check/core/review/__init__.py new file mode 100644 index 0000000..bbf02a9 --- /dev/null +++ b/src/contract_check/core/review/__init__.py @@ -0,0 +1,3 @@ +"""Manual-review queue operator actions (send/complete/reject).""" + +from __future__ import annotations diff --git a/src/contract_check/core/review/actions.py b/src/contract_check/core/review/actions.py new file mode 100644 index 0000000..10bc4a6 --- /dev/null +++ b/src/contract_check/core/review/actions.py @@ -0,0 +1,270 @@ +"""Manual-review queue operator actions. + +Three resolvers for a document stuck in ``manual_review``: + +- ``send_to_analysis`` returns the document to the analyze pipeline. +- ``complete`` closes it as a lightweight admin-reviewed report. +- ``reject`` terminates it as failed and compensates the consumed Document Slot + unconditionally (admin judgment). + +All actions lock the document row, require the current status to be +``manual_review``, and commit their own changes. Publishing happens after the +status change so the analyze worker never sees the message while the document +is still terminal. +""" + +from __future__ import annotations + +import datetime as dt +import uuid + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from src.contract_check.core.billing.quota import compensate_document_slot +from src.contract_check.core.db.repositories import ( + DocumentRepository, + JobRepository, + ReportRepository, + ReviewQueueRepository, +) +from src.contract_check.core.db.repositories.review import ReviewQueueItem +from src.contract_check.core.logging import get_logger +from src.contract_check.core.mq.messages import AnalyzeRequested +from src.contract_check.core.mq.publisher import Publisher +from src.contract_check.core.mq.topology import RK_ANALYZE + +log = get_logger(__name__) + + +class ReviewActionError(Exception): + """User-facing error returned as a toast in the admin panel.""" + + def __init__(self, message: str) -> None: + self.message = message + super().__init__(message) + + +async def send_to_analysis( + session: AsyncSession, + publisher: Publisher, + *, + document_id: uuid.UUID, +) -> None: + """Move a manual-review document back to the analyze pipeline. + + Status is changed to ``analyzing`` before publishing so a racing analyze + worker does not treat the message as a terminal redelivery. If publishing + fails, the status is rolled back to ``manual_review``. + """ + item = await ReviewQueueRepository(session).get_item(document_id) + if item is None: + raise ReviewActionError("Документ не найден в очереди ручной проверки.") + + if not item.extracted_s3_key: + raise ReviewActionError("У документа отсутствует извлечённый текст — отправка невозможна.") + + status = await _require_manual_review_for_update(session, document_id) + if status != "manual_review": + raise ReviewActionError("Документ уже не в статусе ручной проверки.") + + await _set_status(session, document_id, status="analyzing", stage="queued_analyze") + await JobRepository(session).upsert_running( + document_id, + item.correlation_id, + queue="analyze", + ) + await session.commit() + + prescreen_meta = _prescreen_meta_from_item(item) + next_msg = AnalyzeRequested( + correlation_id=item.correlation_id, + document_id=item.document_id, + user_id=item.user_id, + extracted_s3_key=item.extracted_s3_key, + filename=item.filename, + char_count=0, + ocr_used=False, + is_structured=False, + has_tables=False, + prescreen_meta=prescreen_meta, + prescreen_result_id=item.prescreen_result_id, + attempt=0, + ) + + try: + await publisher.publish(next_msg, routing_key=RK_ANALYZE) + except Exception: + await _set_status(session, document_id, status="manual_review", stage="manual_review") + await session.commit() + raise + + log.info( + "admin_review_action", + action="send_to_analysis", + document_id=str(document_id), + user_id=str(item.user_id), + prescreen_result_id=str(item.prescreen_result_id), + correlation_id=str(item.correlation_id), + ) + + +async def complete( + session: AsyncSession, + *, + document_id: uuid.UUID, + note: str, +) -> None: + """Close a manual-review document as ``done`` with an operator note. + + A lightweight report is written so the customer sees a result; no LLM + is invoked. + """ + item = await ReviewQueueRepository(session).get_item(document_id) + if item is None: + raise ReviewActionError("Документ не найден в очереди ручной проверки.") + + status = await _require_manual_review_for_update(session, document_id) + if status != "manual_review": + raise ReviewActionError("Документ уже не в статусе ручной проверки.") + + safe_note = (note or "Закрыто оператором без полного анализа.").strip() + content_json: dict[str, object] = { + "findings": [ + { + "title": "Проверка завершена оператором", + "severity": "low", + "explanation": safe_note, + "recommendation": "", + "citations": [], + } + ] + } + markdown = _operator_note_markdown(item, safe_note) + await ReportRepository(session).upsert_analyze_report( + document_id=document_id, + content_json=content_json, + markdown=markdown, + model_used="admin-complete", + prompt_tokens=0, + eval_tokens=0, + latency_ms=0, + prescreen_meta=_prescreen_meta_from_item(item), + prescreen_result_id=item.prescreen_result_id, + ) + await _set_status(session, document_id, status="done", stage="review_completed") + await session.commit() + + log.info( + "admin_review_action", + action="complete", + document_id=str(document_id), + user_id=str(item.user_id), + prescreen_result_id=str(item.prescreen_result_id), + ) + + +async def reject( + session: AsyncSession, + *, + document_id: uuid.UUID, + reason: str, +) -> None: + """Reject a manual-review document and compensate the Document Slot. + + Compensation is unconditional (``policy='all'``) because this is an + operator judgment call, not an automated failure classification. + """ + item = await ReviewQueueRepository(session).get_item(document_id) + if item is None: + raise ReviewActionError("Документ не найден в очереди ручной проверки.") + + status = await _require_manual_review_for_update(session, document_id) + if status != "manual_review": + raise ReviewActionError("Документ уже не в статусе ручной проверки.") + + await _set_status(session, document_id, status="failed", stage="review_rejected") + await compensate_document_slot( + session, + document_id, + "manual_review_rejected", + "all", + ) + await session.commit() + + log.info( + "admin_review_action", + action="reject", + document_id=str(document_id), + user_id=str(item.user_id), + reason=reason or "", + ) + + +async def _require_manual_review_for_update(session: AsyncSession, document_id: uuid.UUID) -> str: + status = await DocumentRepository(session).get_status_for_update(document_id) + if status is None: + raise ReviewActionError("Документ не найден.") + return status + + +async def _set_status( + session: AsyncSession, + document_id: uuid.UUID, + *, + status: str, + stage: str, +) -> None: + await session.execute( + text("UPDATE documents SET status = :s, stage = :st WHERE id = :d"), + {"s": status, "st": stage, "d": document_id}, + ) + + +def _prescreen_meta_from_item(item: ReviewQueueItem) -> dict[str, object]: + def _iso(d: dt.date | None) -> str | None: + return d.isoformat() if d else None + + return { + "correlation_id": str(item.correlation_id), + "document_id": str(item.document_id), + "user_id": str(item.user_id), + "attempt": 0, + "text_s3_key": item.extracted_s3_key, + "filename": item.filename, + "prescreened_at": item.prescreened_at.isoformat(), + "contract_type": item.contract_type, + "party_a": item.party_a, + "party_b": item.party_b, + "total_amount": float(item.total_amount) if item.total_amount is not None else None, + "currency": item.currency, + "start_date": _iso(item.start_date), + "end_date": _iso(item.end_date), + "has_penalty_clause": item.has_penalty_clause, + "has_termination_clause": item.has_termination_clause, + "has_arbitration": item.has_arbitration, + "confidence_score": float(item.confidence_score) + if item.confidence_score is not None + else 0.0, + "routing_decision": "manual_review", + "auto_summary": item.auto_summary, + "auto_findings": item.auto_findings, + } + + +def _operator_note_markdown(item: ReviewQueueItem, note: str) -> str: + lines: list[str] = [ + f"# Проверка договора: {item.contract_type or 'договор'}", + "", + note, + "", + "**Решение:** проверка завершена оператором без полного анализа ИИ.", + ] + if item.party_a: + lines.append(f"- **Сторона А:** {item.party_a}") + if item.party_b: + lines.append(f"- **Сторона Б:** {item.party_b}") + if item.total_amount is not None: + lines.append(f"- **Сумма:** {item.total_amount} {item.currency or ''}") + lines.extend(["", "---", "", "*Это операторская пометка, а не полный автоматический отчёт.*"]) + return "\n".join(lines) diff --git a/tests/integration/test_admin_review.py b/tests/integration/test_admin_review.py new file mode 100644 index 0000000..5495144 --- /dev/null +++ b/tests/integration/test_admin_review.py @@ -0,0 +1,335 @@ +"""Admin manual-review queue HTTP seam tests (issue 017 + 018). + +Drives the server-rendered admin panel through the ASGI transport with real +Postgres / RabbitMQ / MinIO. Covers list/detail/filters, RBAC, CSRF guard, and +all three review actions. +""" + +from __future__ import annotations + +import datetime as dt +import uuid +from decimal import Decimal +from typing import Any + +import aio_pika +import httpx +import pytest +from sqlalchemy import text + +from contract_check.core.db.session import create_session_factory +from contract_check.core.mq.messages import AnalyzeRequested +from contract_check.core.mq.topology import QUEUE_ANALYZE +from contract_check.core.s3 import extracted_key +from contract_check.core.s3.minio_storage import MinioStorage +from contract_check.core.security.passwords import hash_password + +pytestmark = pytest.mark.integration + + +_TEST_TEXT = "Договор поставки. " * 50 + + +def _storage(infra: dict[str, str]) -> MinioStorage: + return MinioStorage.from_endpoint_url( + endpoint_url=infra["s3_endpoint_url"], + access_key=infra["s3_access_key"], + secret_key=infra["s3_secret_key"], + bucket="contract-check-docs", + ) + + +def _session_factory() -> Any: + return create_session_factory() + + +async def _seed_admin(db_session: Any, *, email: str, password: str) -> str: + result = await db_session.execute( + text( + "INSERT INTO users (email, password_hash, role, is_active) " + "VALUES (:e, :p, 'admin', TRUE) RETURNING id" + ), + {"e": email, "p": hash_password(password)}, + ) + await db_session.commit() + return str(result.scalar_one()) + + +async def _login_as(client: httpx.AsyncClient, email: str, password: str) -> None: + r = await client.post( + "/admin/login", data={"email": email, "password": password}, follow_redirects=False + ) + assert r.status_code == 303 + assert r.headers["location"].startswith("/admin/users") + + +async def _seed_manual_review_doc( + infra: dict[str, str], + *, + telegram_id: int | None = None, + created_at: Any, + confidence: float = 0.5, +) -> tuple[uuid.UUID, uuid.UUID]: + if telegram_id is None: + telegram_id = int(uuid.uuid4().int % 1_000_000_000) + sess = _session_factory() + store = _storage(infra) + async with sess() as session: + result = await session.execute( + text("INSERT INTO users (telegram_id, credits_left) VALUES (:t, 5) RETURNING id"), + {"t": telegram_id}, + ) + user_id = result.scalar_one() + await session.commit() + + document_id = uuid.uuid4() + ext_key = extracted_key(str(user_id), str(document_id)) + await store.put(ext_key, _TEST_TEXT.encode("utf-8"), content_type="text/plain; charset=utf-8") + + async with sess() as session: + await session.execute( + text( + "INSERT INTO documents " + "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) " + "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', :bytes, " + " 'manual_review', 'manual_review')" + ), + { + "id": document_id, + "uid": user_id, + "s3": f"users/{user_id}/docs/{document_id}.pdf", + "ext": ext_key, + "bytes": len(_TEST_TEXT), + }, + ) + await session.execute( + text( + "INSERT INTO prescreen_results " + "(id, document_id, correlation_id, contract_type, party_a, party_b, " + " total_amount, currency, start_date, end_date, has_penalty_clause, " + " has_termination_clause, has_arbitration, confidence_score, routing_decision, " + " prescreened_at, auto_summary, auto_findings, extractor_version) " + "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', " + " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, " + " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')" + ), + { + "id": uuid.uuid4(), + "d": document_id, + "cid": uuid.uuid4(), + "ta": Decimal("100000.00"), + "cs": confidence, + }, + ) + await session.execute( + text("UPDATE documents SET created_at = :ca WHERE id = :id"), + {"ca": created_at, "id": document_id}, + ) + await session.commit() + return user_id, document_id + + +async def test_review_queue_redirects_without_session(client: httpx.AsyncClient) -> None: + r = await client.get("/admin/review", follow_redirects=False) + assert r.status_code == 303 + assert r.headers["location"].startswith("/admin/login") + + +async def test_review_queue_list_and_filters( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + user_id, doc_id = await _seed_manual_review_doc( + infra, + created_at=dt.datetime(2030, 3, 15, tzinfo=dt.UTC), + confidence=0.8, + ) + await _seed_manual_review_doc( + infra, + created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC), + confidence=0.4, + ) + + # Unfiltered list in the unique date range. + r = await client.get( + "/admin/review?date_from=2030-01-01&date_to=2030-12-31", follow_redirects=False + ) + assert r.status_code == 200 + assert b"contract.pdf" in r.content + assert "ООО Продавец".encode() in r.content + + # Date filter excludes the later document. + r = await client.get( + "/admin/review?date_from=2030-03-01&date_to=2030-03-31", follow_redirects=False + ) + assert r.status_code == 200 + assert str(doc_id).encode() in r.content + + # Confidence filter. + r = await client.get( + "/admin/review?date_from=2030-01-01&date_to=2030-12-31&conf_min=0.7", + follow_redirects=False, + ) + assert r.status_code == 200 + assert b"80.0%" in r.content + + +async def test_review_detail_with_preview( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + user_id, doc_id = await _seed_manual_review_doc( + infra, created_at=dt.datetime(2030, 4, 1, tzinfo=dt.UTC) + ) + + r = await client.get(f"/admin/review/{doc_id}", follow_redirects=False) + assert r.status_code == 200 + assert "ООО Продавец".encode() in r.content + assert "Договор поставки.".encode() in r.content + assert f"/admin/review/{doc_id}/send-to-analysis".encode() in r.content + + # Non-existent document returns 404. + r = await client.get(f"/admin/review/{uuid.uuid4()}", follow_redirects=False) + assert r.status_code == 404 + + +async def test_review_action_requires_htmx_header( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + _user_id, doc_id = await _seed_manual_review_doc( + infra, created_at=dt.datetime(2030, 5, 1, tzinfo=dt.UTC) + ) + + r = await client.post( + f"/admin/review/{doc_id}/complete", + data={"note": "note"}, + follow_redirects=False, + ) + assert r.status_code == 400 + + +async def test_review_complete_action( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + _user_id, doc_id = await _seed_manual_review_doc( + infra, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC) + ) + + r = await client.post( + f"/admin/review/{doc_id}/complete", + data={"note": "Проверено оператором."}, + headers={"HX-Request": "true"}, + follow_redirects=False, + ) + assert r.status_code == 303 + assert "/admin/review" in r.headers["location"] + + sess = _session_factory() + async with sess() as session: + status = ( + await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id}) + ).scalar_one() + assert status == "done" + report = ( + await session.execute( + text("SELECT markdown FROM reports WHERE document_id = :d"), {"d": doc_id} + ) + ).scalar_one() + assert "Проверено оператором." in report + + +async def test_review_reject_action( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + user_id, doc_id = await _seed_manual_review_doc( + infra, created_at=dt.datetime(2030, 7, 1, tzinfo=dt.UTC) + ) + + sess = _session_factory() + async with sess() as session: + from contract_check.core.db.repositories import CreditsRepository + + await CreditsRepository(session).reserve(user_id, document_id=doc_id) + await session.commit() + + r = await client.post( + f"/admin/review/{doc_id}/reject", + data={"reason": "Некорректный формат"}, + headers={"HX-Request": "true"}, + follow_redirects=False, + ) + assert r.status_code == 303 + assert "/admin/review" in r.headers["location"] + + async with sess() as session: + status = ( + await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id}) + ).scalar_one() + assert status == "failed" + balance = ( + await session.execute( + text("SELECT credits_left FROM users WHERE id = :u"), {"u": user_id} + ) + ).scalar_one() + assert balance == 5 + + +async def test_review_send_to_analysis_action( + client: httpx.AsyncClient, db_session: Any, infra: dict[str, str] +) -> None: + admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local" + await _seed_admin(db_session, email=admin_email, password="adminpass-123") + await _login_as(client, admin_email, "adminpass-123") + + _user_id, doc_id = await _seed_manual_review_doc( + infra, created_at=dt.datetime(2030, 8, 1, tzinfo=dt.UTC) + ) + + # Clean the shared analyze queue so we can consume exactly one message. + connection = await aio_pika.connect(infra["rabbitmq_url"]) + try: + channel = await connection.channel() + queue = await channel.declare_queue(QUEUE_ANALYZE, passive=True) + await queue.purge() + + r = await client.post( + f"/admin/review/{doc_id}/send-to-analysis", + headers={"HX-Request": "true"}, + follow_redirects=False, + ) + assert r.status_code == 303 + + msg = await queue.get(timeout=5) + body = msg.body.decode("utf-8") + analyze_msg = AnalyzeRequested.model_validate_json(body) + assert analyze_msg.document_id == doc_id + assert analyze_msg.attempt == 0 + assert analyze_msg.prescreen_meta is not None + await msg.ack() + finally: + await connection.close() + + sess = _session_factory() + async with sess() as session: + status = ( + await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id}) + ).scalar_one() + assert status == "analyzing" diff --git a/tests/integration/test_review_actions.py b/tests/integration/test_review_actions.py new file mode 100644 index 0000000..777daf9 --- /dev/null +++ b/tests/integration/test_review_actions.py @@ -0,0 +1,368 @@ +"""Integration tests for manual-review queue actions (issue 018). + +These are the "worker handler seam" tests: actions are driven in-process against +real Postgres / RabbitMQ / MinIO, and the analyze handler is exercised with a stub +LLM provider to prove that ``send_to_analysis`` produces a real Report. +""" + +from __future__ import annotations + +import uuid +from decimal import Decimal +from typing import Any + +import pytest +from sqlalchemy import text + +from contract_check.core.analysis.report_schema import Finding +from contract_check.core.db.repositories import ( + CreditsRepository, + DocumentRepository, + ReportRepository, +) +from contract_check.core.db.session import create_session_factory +from contract_check.core.llm.port import AnalysisResult +from contract_check.core.mq.messages import AnalyzeRequested +from contract_check.core.mq.topology import RK_ANALYZE +from contract_check.core.review.actions import ( + ReviewActionError, + complete, + reject, + send_to_analysis, +) +from contract_check.core.s3 import extracted_key +from contract_check.core.s3.minio_storage import MinioStorage +from contract_check.worker_analyze.handler import AnalyzeHandler + +pytestmark = pytest.mark.integration + + +_TEST_TEXT = "Договор поставки. " * 50 + + +class FakePublisher: + """Captures the message published by ``send_to_analysis``.""" + + def __init__(self) -> None: + self.messages: list[tuple[Any, str]] = [] + + async def publish(self, message: Any, routing_key: str) -> None: + self.messages.append((message, routing_key)) + + +class StubProvider: + """LLMProvider returning one fixed finding.""" + + async def analyze(self, text: str, *, checklist: str) -> AnalysisResult: + return AnalysisResult( + findings=[ + Finding( + checklist_id="penalties", + severity="high", + quote="штраф 0,5%", + section_ref="п. 6.3", + risk="высокая неустойка", + recommendation="ограничить", + ) + ], + model_used="stub-model", + prompt_tokens=10, + eval_tokens=20, + latency_sec=0.1, + ) + + async def extract_prescreen(self, text: str, *, max_chars: int | None = None) -> dict[str, Any]: + raise NotImplementedError + + async def aclose(self) -> None: + pass + + +def _storage(infra: dict[str, str]) -> MinioStorage: + return MinioStorage.from_endpoint_url( + endpoint_url=infra["s3_endpoint_url"], + access_key=infra["s3_access_key"], + secret_key=infra["s3_secret_key"], + bucket="contract-check-docs", + ) + + +def _session_factory() -> Any: + return create_session_factory() + + +async def _seed_user(session: Any, *, telegram_id: int, credits: int = 5) -> uuid.UUID: + result = await session.execute( + text( + "INSERT INTO users (telegram_id, credits_left) VALUES (:t, :c) " + "ON CONFLICT (telegram_id) DO UPDATE SET credits_left = :c " + "RETURNING id" + ), + {"t": telegram_id, "c": credits}, + ) + return result.scalar_one() + + +async def _seed_manual_review_doc( + infra: dict[str, str], + *, + telegram_id: int | None = None, + document_id: uuid.UUID, + confidence: float = 0.5, + total_amount: Decimal = Decimal("100000.00"), +) -> tuple[uuid.UUID, str, uuid.UUID]: + """Create user + extracted text in MinIO + manual_review document + prescreen result.""" + if telegram_id is None: + telegram_id = int(uuid.uuid4().int % 1_000_000_000) + sess = _session_factory() + store = _storage(infra) + async with sess() as session: + user_id = await _seed_user(session, telegram_id=telegram_id) + await session.commit() + + ext_key = extracted_key(str(user_id), str(document_id)) + await store.put(ext_key, _TEST_TEXT.encode("utf-8"), content_type="text/plain; charset=utf-8") + + correlation_id = uuid.uuid4() + async with sess() as session: + await session.execute( + text( + "INSERT INTO documents " + "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) " + "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', :bytes, " + " 'manual_review', 'manual_review')" + ), + { + "id": document_id, + "uid": user_id, + "s3": f"users/{user_id}/docs/{document_id}.pdf", + "ext": ext_key, + "bytes": len(_TEST_TEXT), + }, + ) + await session.execute( + text( + "INSERT INTO prescreen_results " + "(id, document_id, correlation_id, contract_type, party_a, party_b, " + " total_amount, currency, start_date, end_date, has_penalty_clause, " + " has_termination_clause, has_arbitration, confidence_score, routing_decision, " + " prescreened_at, auto_summary, auto_findings, extractor_version) " + "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', " + " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, " + " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')" + ), + { + "id": uuid.uuid4(), + "d": document_id, + "cid": correlation_id, + "ta": total_amount, + "cs": confidence, + }, + ) + await session.commit() + return user_id, ext_key, correlation_id + + +async def _doc_state(session: Any, document_id: uuid.UUID) -> tuple[str, str, bool]: + result = await session.execute( + text("SELECT status, stage, refunded FROM documents WHERE id = :d"), + {"d": document_id}, + ) + return result.one() + + +async def test_reject_credits_refunds_once(infra: dict[str, str]) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + user_id, _ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + async with sess() as session: + await CreditsRepository(session).reserve(user_id, document_id=document_id) + await session.commit() + + async with sess() as session: + await reject(session, document_id=document_id, reason="Некорректный договор") + + async with sess() as session: + status, stage, refunded = await _doc_state(session, document_id) + assert status == "failed" + assert stage == "review_rejected" + assert refunded is True + + balance = await CreditsRepository(session).get_balance(user_id) + assert balance == 5 + + events = await session.execute( + text("SELECT kind, delta FROM credit_events WHERE user_id = :u AND document_id = :d"), + {"u": user_id, "d": document_id}, + ) + rows = events.all() + assert any(kind == "reserve" for kind, _ in rows) + assert any(kind == "refund_auto" and delta == 1 for kind, delta in rows) + + # Second reject is idempotent via status, not compensation again. + async with sess() as session: + with pytest.raises(ReviewActionError): + await reject(session, document_id=document_id, reason="") + + +async def test_reject_quota_restores_slot_no_credit_refund(infra: dict[str, str]) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + user_id, _ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + subscription_id = uuid.uuid4() + async with sess() as session: + await session.execute( + text( + "INSERT INTO subscriptions " + "(id, user_id, plan_code, status, current_period_start, current_period_end) " + "VALUES (:id, :u, 'lite', 'active', now() - interval '1 day', now() + interval '30 days')" + ), + {"id": subscription_id, "u": user_id}, + ) + await session.execute( + text( + "INSERT INTO quota_usage (user_id, subscription_id, document_id) " + "VALUES (:u, :s, :d)" + ), + {"u": user_id, "s": subscription_id, "d": document_id}, + ) + await session.commit() + + async with sess() as session: + await reject(session, document_id=document_id, reason="Неподходящий формат") + + async with sess() as session: + status, stage, refunded = await _doc_state(session, document_id) + assert status == "failed" + assert stage == "review_rejected" + assert refunded is True + + quota = await session.execute( + text("SELECT id FROM quota_usage WHERE document_id = :d"), + {"d": document_id}, + ) + assert quota.first() is None + + events = await session.execute( + text("SELECT kind, delta FROM credit_events WHERE user_id = :u AND document_id = :d"), + {"u": user_id, "d": document_id}, + ) + assert not any(kind == "refund_auto" for kind, _ in events.all()) + + balance = await CreditsRepository(session).get_balance(user_id) + assert balance == 5 # default seeded credits, untouched + + +async def test_complete_writes_report_and_done(infra: dict[str, str]) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + user_id, _ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + async with sess() as session: + await complete(session, document_id=document_id, note="Договор типовой, рисков нет.") + + async with sess() as session: + status, stage, _refunded = await _doc_state(session, document_id) + assert status == "done" + assert stage == "review_completed" + + report = await ReportRepository(session).get_by_document_id(document_id) + assert report is not None + assert report.model_used == "admin-complete" + assert "Договор типовой, рисков нет." in report.markdown + assert report.prescreen_result_id is not None + assert report.prescreen_meta is not None + + +async def test_send_to_analysis_publishes_message_and_handler_finishes( + infra: dict[str, str], +) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + user_id, ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + fake_publisher = FakePublisher() + + async with sess() as session: + await send_to_analysis(session, fake_publisher, document_id=document_id) + + assert len(fake_publisher.messages) == 1 + msg, routing_key = fake_publisher.messages[0] + assert isinstance(msg, AnalyzeRequested) + assert msg.document_id == document_id + assert msg.user_id == user_id + assert msg.extracted_s3_key == ext_key + assert msg.attempt == 0 + assert msg.prescreen_meta is not None + assert msg.prescreen_result_id is not None + assert routing_key == RK_ANALYZE + + async with sess() as session: + status, stage, _refunded = await _doc_state(session, document_id) + assert status == "analyzing" + assert stage == "queued_analyze" + + # Drive the analyze worker with the same message + stub provider. + analyze_handler = AnalyzeHandler(session_factory=sess, provider=StubProvider()) + try: + await analyze_handler.handle(msg) + finally: + await analyze_handler.aclose() + + async with sess() as session: + status, stage, _refunded = await _doc_state(session, document_id) + assert status == "done" + assert stage == "done" + + report = await ReportRepository(session).get_by_document_id(document_id) + assert report is not None + assert len(report.content_json.get("findings", [])) == 1 + + +async def test_send_to_analysis_rolls_back_status_on_publish_failure( + infra: dict[str, str], +) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + _user_id, _ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + class FailingPublisher: + async def publish(self, message: Any, routing_key: str) -> None: + raise RuntimeError("broker down") + + async with sess() as session: + with pytest.raises(RuntimeError): + await send_to_analysis(session, FailingPublisher(), document_id=document_id) + + async with sess() as session: + status, stage, _refunded = await _doc_state(session, document_id) + assert status == "manual_review" + assert stage == "manual_review" + + +async def test_action_requires_manual_review_status(infra: dict[str, str]) -> None: + sess = _session_factory() + document_id = uuid.uuid4() + _user_id, _ext_key, _correlation_id = await _seed_manual_review_doc( + infra, document_id=document_id + ) + + async with sess() as session: + await DocumentRepository(session).update_status(document_id, status="done") + await session.commit() + + async with sess() as session: + with pytest.raises(ReviewActionError): + await complete(session, document_id=document_id, note="note") diff --git a/tests/unit/test_review_repository.py b/tests/unit/test_review_repository.py new file mode 100644 index 0000000..3a28d82 --- /dev/null +++ b/tests/unit/test_review_repository.py @@ -0,0 +1,276 @@ +"""Repository tests for the manual-review queue query. + +Backs issue 017: queue rows are documents in ``manual_review`` joined with the +latest prescreen result, filterable and paginated. These tests hit the Docker +Compose Postgres stack and roll back after each test. +""" + +from __future__ import annotations + +import datetime as dt +import uuid +from decimal import Decimal + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from contract_check.core.db.repositories import ReviewQueueRepository + +pytestmark = [pytest.mark.unit, pytest.mark.usefixtures("db_session")] + + +@pytest.fixture(autouse=True) +async def _clean_queue_before_each(db_session: AsyncSession) -> None: + """Give every queue test a clean manual-review slice inside the savepoint.""" + await _clean_manual_review(db_session) + + +async def _clean_manual_review(session: AsyncSession) -> None: + """Remove leftover manual-review rows from earlier integration runs. + + Runs inside the per-test nested transaction; the enclosing savepoint rolls + back the cleanup at the end of the test, so the shared dev database keeps + whatever was there before. + """ + await session.execute( + text( + "DELETE FROM credit_events WHERE document_id IN " + "(SELECT id FROM documents WHERE status = 'manual_review')" + ) + ) + await session.execute(text("DELETE FROM documents WHERE status = 'manual_review'")) + + +async def _seed_user(session: AsyncSession, *, telegram_id: int = 0) -> uuid.UUID: + if telegram_id == 0: + telegram_id = int.from_bytes(uuid.uuid4().bytes[:4], "big", signed=True) + result = await session.execute( + text("INSERT INTO users (telegram_id, credits_left) VALUES (:t, 5) RETURNING id"), + {"t": telegram_id}, + ) + return result.scalar_one() + + +_TEST_EPOCH = dt.datetime(2030, 1, 1, tzinfo=dt.UTC) + + +async def _seed_manual_review_doc( + session: AsyncSession, + *, + user_id: uuid.UUID, + created_at: dt.datetime | None = None, + confidence: Decimal | float = Decimal("0.5"), + total_amount: Decimal | float | None = Decimal("100000.00"), +) -> tuple[uuid.UUID, uuid.UUID]: + document_id = uuid.uuid4() + correlation_id = uuid.uuid4() + if created_at is None: + created_at = _TEST_EPOCH + await session.execute( + text( + "INSERT INTO documents " + "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) " + "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', 1000, " + " 'manual_review', 'manual_review')" + ), + { + "id": document_id, + "uid": user_id, + "s3": f"users/{user_id}/docs/{document_id}.pdf", + "ext": f"users/{user_id}/docs/{document_id}.txt", + }, + ) + await session.execute( + text( + "INSERT INTO prescreen_results " + "(id, document_id, correlation_id, contract_type, party_a, party_b, " + " total_amount, currency, start_date, end_date, has_penalty_clause, " + " has_termination_clause, has_arbitration, confidence_score, routing_decision, " + " prescreened_at, auto_summary, auto_findings, extractor_version) " + "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', " + " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, " + " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')" + ), + { + "id": uuid.uuid4(), + "d": document_id, + "cid": correlation_id, + "ta": total_amount, + "cs": confidence, + }, + ) + # created_at is server-default now(); override it for ordering tests. + if created_at: + await session.execute( + text("UPDATE documents SET created_at = :ca WHERE id = :id"), + {"ca": created_at, "id": document_id}, + ) + await session.commit() + return document_id, correlation_id + + +async def test_list_returns_manual_review_documents(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + doc_id, _ = await _seed_manual_review_doc(db_session, user_id=user_id) + + rows, total = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + limit=10, + offset=0, + ) + assert total == 1 + assert len(rows) == 1 + assert rows[0].document_id == doc_id + assert rows[0].party_a == "ООО Продавец" + assert rows[0].confidence_score == Decimal("0.5") + + +async def test_list_orders_newest_first(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + older = await _seed_manual_review_doc( + db_session, user_id=user_id, created_at=dt.datetime(2030, 1, 1, tzinfo=dt.UTC) + ) + newer = await _seed_manual_review_doc( + db_session, user_id=user_id, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC) + ) + + rows, _ = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + limit=10, + offset=0, + ) + assert [r.document_id for r in rows] == [newer[0], older[0]] + + +async def test_list_ignores_non_manual_review(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + manual = await _seed_manual_review_doc(db_session, user_id=user_id) + + ignored_id = uuid.uuid4() + await db_session.execute( + text( + "INSERT INTO documents " + "(id, user_id, s3_key, filename, mime, bytes, status) " + "VALUES (:id, :uid, :s3, 'x.pdf', 'application/pdf', 1, 'analyzing')" + ), + {"id": ignored_id, "uid": user_id, "s3": f"users/{user_id}/docs/{ignored_id}.pdf"}, + ) + await db_session.execute( + text( + "INSERT INTO prescreen_results " + "(id, document_id, correlation_id, routing_decision, prescreened_at) " + "VALUES (:id, :d, :cid, 'manual_review', now())" + ), + {"id": uuid.uuid4(), "d": ignored_id, "cid": uuid.uuid4()}, + ) + await db_session.commit() + + rows, total = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + limit=10, + offset=0, + ) + assert total == 1 + assert rows[0].document_id == manual[0] + + +async def test_list_pagination(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + for i in range(3): + await _seed_manual_review_doc( + db_session, + user_id=user_id, + created_at=dt.datetime(2030, 1, 1 + i, tzinfo=dt.UTC), + ) + + rows, total = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + limit=2, + offset=0, + ) + assert total == 3 + assert len(rows) == 2 + + rows2, _ = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + limit=2, + offset=2, + ) + assert len(rows2) == 1 + + +async def test_list_filters_by_date(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + inside = await _seed_manual_review_doc( + db_session, user_id=user_id, created_at=dt.datetime(2030, 3, 15, tzinfo=dt.UTC) + ) + await _seed_manual_review_doc( + db_session, user_id=user_id, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC) + ) + + rows, total = await repo.list_queue( + date_from=dt.date(2030, 3, 1), date_to=dt.date(2030, 3, 31), limit=10, offset=0 + ) + assert total == 1 + assert rows[0].document_id == inside[0] + + +async def test_list_filters_by_confidence(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + low = await _seed_manual_review_doc(db_session, user_id=user_id, confidence=Decimal("0.3")) + high = await _seed_manual_review_doc(db_session, user_id=user_id, confidence=Decimal("0.9")) + + rows, _ = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + conf_min=Decimal("0.5"), + conf_max=Decimal("1.0"), + limit=10, + offset=0, + ) + assert len(rows) == 1 + assert rows[0].document_id == high[0] + + rows, _ = await repo.list_queue( + date_from=dt.date(2030, 1, 1), + date_to=dt.date(2030, 12, 31), + conf_min=Decimal("0.1"), + conf_max=Decimal("0.5"), + limit=10, + offset=0, + ) + assert len(rows) == 1 + assert rows[0].document_id == low[0] + + +async def test_get_item_returns_detail_fields(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + user_id = await _seed_user(db_session) + doc_id, correlation_id = await _seed_manual_review_doc( + db_session, user_id=user_id, total_amount=Decimal("250000.00") + ) + + item = await repo.get_item(doc_id) + assert item is not None + assert item.document_id == doc_id + assert item.correlation_id == correlation_id + assert item.user_id == user_id + assert item.total_amount == Decimal("250000.00") + assert item.extracted_s3_key == f"users/{user_id}/docs/{doc_id}.txt" + + +async def test_get_item_returns_none_for_missing(db_session: AsyncSession) -> None: + repo = ReviewQueueRepository(db_session) + assert await repo.get_item(uuid.uuid4()) is None