| """Post-extraction consistency and duplicate checks.""" |
|
|
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from sqlalchemy.ext.asyncio import AsyncSession |
|
|
| from app.models import Document |
| from app.schemas import DocType |
| from app.services.consistency import check_consistency |
| from app.services.doctypes import ARRAY_FIELD_KEYS |
| from app.services.duplicates import find_duplicate, register_invoice |
|
|
|
|
| async def apply_post_extract_checks( |
| session: AsyncSession, |
| *, |
| document_id: str, |
| doc_type: str, |
| data: dict[str, Any], |
| register_if_new: bool = True, |
| ) -> tuple[list[dict[str, Any]], bool, str | None]: |
| """ |
| Run consistency + duplicate checks. |
| |
| Returns (issues, is_duplicate, duplicate_of). |
| """ |
| issues = [ |
| i.model_dump() for i in check_consistency(doc_type, data) |
| ] |
| is_dup = False |
| dup_of: str | None = None |
|
|
| if doc_type != DocType.INVOICE.value: |
| return issues, is_dup, dup_of |
|
|
| dup_of = await find_duplicate( |
| session, |
| supplier=data.get("supplier"), |
| invoice_number=data.get("invoice_number"), |
| exclude_id=document_id, |
| ) |
| if dup_of: |
| is_dup = True |
| issues.append( |
| { |
| "code": "duplicate_invoice", |
| "message": ( |
| "Same supplier + invoice number " |
| f"already processed ({dup_of})" |
| ), |
| "severity": "warning", |
| } |
| ) |
| elif register_if_new: |
| await register_invoice( |
| session, |
| document_id=document_id, |
| supplier=data.get("supplier"), |
| invoice_number=data.get("invoice_number"), |
| ) |
| return issues, is_dup, dup_of |
|
|
|
|
| def mark_fields_confirmed( |
| extracted: dict[str, Any], |
| previous: dict[str, Any] | None = None, |
| ) -> dict[str, dict]: |
| """Mark scalar fields as manually confirmed (score 1.0).""" |
| conf = dict(previous or {}) |
| for key in extracted: |
| if key in ARRAY_FIELD_KEYS: |
| continue |
| conf[key] = {"score": 1.0, "needs_review": False} |
| return conf |
|
|
|
|
| def average_from_detail(confidence: dict[str, Any]) -> float: |
| """Average score from detailed confidence map.""" |
| scores = [ |
| v["score"] |
| for v in confidence.values() |
| if isinstance(v, dict) and "score" in v |
| ] |
| if not scores: |
| return 0.0 |
| return sum(scores) / len(scores) |
|
|
|
|
| def apply_duplicate_flags( |
| doc: Document, |
| *, |
| is_duplicate: bool, |
| duplicate_of: str | None, |
| ) -> None: |
| """Write duplicate flags onto a document row.""" |
| doc.is_duplicate = is_duplicate |
| doc.duplicate_of = duplicate_of |
|
|
