File size: 2,672 Bytes
da6ee5e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """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
|