"""Transport-neutral invoice-field response schema, parser and grounded re-validation.
A model asked to read an invoice returns strings. Whether it read those strings
off rasterised pixels (:mod:`.evidence_draft_vision`) or off an already-extracted
text layer (:mod:`.evidence_draft_text`) changes nothing about what must happen
next: the response has to be recovered from possibly-chatty completion text,
validated against a strict schema, and then *re-validated field by field* so a
value the model invented is dropped rather than trusted.
That second step is the whole point. The reading model is probabilistic and a
fabricated taxable base or supplier identifier would flow into a Modelo 303/390
filing looking exactly like a read one. Every field here is therefore checked
against an independent authority -- the AEAT checksum algorithm
(:func:`~cadrumo.domain.calculations.registry.tax_id_runtime.validate_runtime_spanish_tax_id`),
the EU VIES structural format
table (:func:`~cadrumo.domain.calculations.registry.nif_iva_catalogue.nif_iva_format_for_country`), the date parser
(:func:`~core.parsing.dates.parse_date`), the finite European-decimal authority
(:func:`~core.decimal.coercion.coerce_finite_european_decimal`) -- and a field that fails
its check becomes ``None``. ``None`` is safe because the confirm path treats a
missing figure as a hard refusal naming the operator override; a guessed figure
would instead be minted silently.
Nothing in this module knows about images, providers or transports, which is why
it lives here rather than beside either reader.
See Also:
:class:`~application.ledger.invoice_draft_records.InvoiceDraft`
Typed draft every grounded reader returns.
:func:`~cadrumo.domain.calculations.registry.tax_id_runtime.validate_runtime_spanish_tax_id`
Spanish NIF/NIE/CIF checksum authority.
:func:`~cadrumo.domain.calculations.registry.nif_iva_catalogue.nif_iva_format_for_country`
EU intra-community NIF-IVA structural format authority.
"""
from __future__ import annotations
import json
import re
from collections.abc import Callable, Mapping
from decimal import Decimal
from typing import cast
from pydantic import BaseModel, Field
from ....application.ledger.evidence_errors import (
PurchaseInvoiceEvidenceReaderError,
)
from ....application.ledger.invoice_draft_records import DraftDiscrepancyFinding, FieldProvenance, InvoiceDraft
from ....core.decimal.coercion import coerce_finite_european_decimal
from ....core.decimal.grammar import european_thousands_reading_is_ambiguous
from ....core.decimal.printed_units import without_currency_unit
from ....core.draft_discrepancy import DraftDiscrepancyKind
from ....core.errors.hierarchy import CoreValidationError
from ....core.field_grounding import FieldGroundingOutcome
from ....core.field_origin import FieldOrigin
from ....core.identity.documents import IdentityError
from ....core.identity.nif_iva import normalise_nif_iva
from ....core.models import STRICT_FROZEN_CONFIG
from ....core.operator_action_enums import ActionEvidenceProvenance
from ....core.parsing.codes import normalise_iso_4217_currency
from ....core.parsing.dates import parse_date
from ....domain.calculations.registry.authority import PinnedAuthorityOperation
from ....domain.calculations.registry.nif_iva_catalogue import nif_iva_format_for_country
from ....domain.calculations.registry.tax_id_runtime import validate_runtime_spanish_tax_id
from ....domain.iva.establishment import country_code_for_printed_country_name
from .invoice_field_contract import (
ANCHOR_KEY_SUFFIX,
INVOICE_FIELD_CONTRACTS,
ROLE_EVIDENCE_KEY_SUFFIX,
InvoiceFieldForm,
contract_for_field,
)
from .preconditions import LLMPreconditionCondition, llm_no_recovery_verdict
__all__ = [
"ExtractedFieldAnchors",
"ExtractedInvoiceFields",
"ExtractedInvoiceResponse",
"ExtractedRoleEvidence",
"ground_extracted_fields",
"parse_invoice_extraction_response",
]
# One JSON object, allowing the model to wrap it in prose or a code fence; the
# first balanced-looking candidate is taken (mirrors the classification parser's
# tolerance for chatty local models).
_JSON_OBJECT_RE = re.compile(r"\{.*\}", re.DOTALL)
class _ExtractedInvoiceFieldClaims(BaseModel):
model_config = STRICT_FROZEN_CONFIG
supplier_tax_id: str | None = Field(default=None)
supplier_name: str | None = Field(default=None)
supplier_postal_code: str | None = Field(default=None)
supplier_country: str | None = Field(default=None)
customer_tax_id: str | None = Field(default=None)
customer_name: str | None = Field(default=None)
customer_postal_code: str | None = Field(default=None)
customer_country: str | None = Field(default=None)
invoice_number: str | None = Field(default=None)
invoice_date: str | None = Field(default=None)
taxable_base: str | None = Field(default=None)
iva_rate: str | None = Field(default=None)
iva_amount: str | None = Field(default=None)
retencion_rate: str | None = Field(default=None)
retencion_amount: str | None = Field(default=None)
grand_total: str | None = Field(default=None)
regime_legend: str | None = Field(default=None)
currency: str | None = Field(default=None)
def _extract_json_object(text: str) -> str | None:
match = _JSON_OBJECT_RE.search(text)
return match.group(0) if match else None
def _grounded_intra_community_tax_id(normalised: str) -> str | None:
"""Return *normalised* if it matches its own country's published NIF-IVA pattern.
The country is read off the number's own two-character IVA prefix, because
that is the only country signal a transcribed identifier carries -- unlike
:func:`~domain.invoices.validators.validate_iva_number`, which validates against a
country declared independently on the invoice record and can therefore fall
back to a permissive generic body check for a non-EU counterparty. That
fallback is unavailable here and would be circular if borrowed: with the
country derived from the string, "prefix plus alphanumeric body" accepts any
two letters followed by anything, which is precisely the fabricate-anything
outcome grounded re-validation exists to prevent.
So the rule is strictly closed: a prefix naming no Member State (or Northern
Ireland), or a body that does not match that State's VIES structure, drops to
``None``.
"""
if len(normalised) < 2:
return None
spec = nif_iva_format_for_country(normalised[:2])
if spec is None or not spec.pattern.match(normalised):
return None
return normalised
def _grounded_tax_id(raw: str | None) -> str | None:
"""Ground a transcribed supplier identifier as a Spanish or EU intra-community id.
Spanish identifiers keep the full AEAT checksum test: a CIF whose control
character does not compute is a misread (or an invention) and must still be
dropped. Only once that test has *rejected* the value is it offered to the
EU structural authority, so nothing about Spanish validation is weakened --
a Spanish-shaped number carries no EU IVA prefix and fails there too.
Routing every identifier through the Spanish validator alone silently
discarded a *correct* read of a valid foreign IVA number, which is the entire
Modelo 349 / intra-EU reverse-charge population: a supplier invoicing from
Ireland or Germany reached the operator with no identifier at all.
"""
if raw is None:
return None
try:
return validate_runtime_spanish_tax_id(raw)
except IdentityError:
return _grounded_intra_community_tax_id(normalise_nif_iva(raw))
def _grounded_free_text(raw: str | None) -> str | None:
"""Return an opaque printed token trimmed of surrounding whitespace, or ``None``.
Named for the FORM it validates rather than for a field, like every other
validator in the two dispatch tables. It was named for the invoice number
when that was the only free-text field declared; it now serves six -- both
postal codes, both printed countries, the invoice number and the regime
legend -- and a validator named after one of the six reads as a rule about
invoice numbers that the other five are borrowing. The tables dispatch on
:class:`~llm.invoice_field_contract.InvoiceFieldForm`, so the form is what
the name must say.
Deliberately the weakest validator of the set, and that is the declared
contract rather than an omission: a free-text field is an opaque token the
document printed, so there is no independent authority to check it against.
A field needing one declares a different form.
"""
if raw is None:
return None
trimmed = raw.strip()
return trimmed or None
def _grounded_date(raw: str | None) -> str | None:
"""Parse *raw* as a day-first (``DD-MM-YYYY`` / ``DD/MM/YYYY``) or ISO-8601 date.
A model transcribing a printed European invoice date returns the day-first
form the document actually shows (mirroring the text-layer heuristic's
``_DATE_RE``); ISO-8601 is tried second in case the model normalises the
printed value itself. Only these two real, registered
:data:`~core.parsing.dates._DateFmt` members are ever passed -- an invented
format string silently degrades to one of the two delegates
(:func:`~core.parsing.dates._parse_date` has no third branch), which would
make a "fallback" attempt a silent no-op duplicate.
"""
if raw is None:
return None
cleaned = raw.strip()
for fmt in ("ddmmyyyy", "iso8601"):
parsed = parse_date(cleaned, fmt=fmt, on_error="none")
if parsed is not None:
return parsed.isoformat()
return None
def _grounded_decimal(raw: str | None) -> Decimal | None:
"""Return the model's transcribed amount as a finite Decimal, or ``None``.
Routed through the canonical finite European-decimal authority rather than
stripping every dot as a thousands separator. Unconditional stripping
corrupted an already dot-decimal transcription -- ``"1234.56"`` became
``Decimal("123456")`` -- silently multiplying a taxable base, IVA amount or
grand total by a hundred, and it admitted ``NaN`` and ``Infinity`` as
amounts. The canonical helper reads a comma as the decimal separator (so
``"1.234,56"`` still parses as ``1234.56``) and refuses non-finite values.
An amount whose dot could equally be a thousands separator or a decimal
point is dropped rather than read one way. The model is instructed to
transcribe what is printed and the schema types every field as a string so
nothing is normalised on the way out, which is deliberate -- it means the
convention in the returned text is the SUPPLIER'S, and a supplier writing
Spanish prints one thousand two hundred and thirty-four euros as
``1.234``. Read as a decimal that is ``1.23``, a thousandfold light on a
taxable base bound for Modelo 303/390.
Dropping to ``None`` is the same call :func:`_grounded_currency` makes on a
symbol it cannot resolve, for the same reason: the confirm path treats a
missing amount as a hard refusal naming the ``--taxable-base`` override, so
the operator is asked for the figure. A guessed one would instead be minted
silently, and the draft review is advisory rather than a gate -- ``confirm``
re-extracts and uses the extracted value for any field the operator did not
explicitly override.
"""
if raw is None:
return None
text = raw.strip()
if european_thousands_reading_is_ambiguous(text):
return None
return coerce_finite_european_decimal(text)
def _grounded_percentage(raw: str | None, currency_unit: str | None = None) -> Decimal | None:
"""Return a transcribed rate as a bare Decimal, dropping the printed unit.
The declared form for a rate is *the bare number*
(:attr:`~llm.invoice_field_contract.InvoiceFieldForm.PERCENTAGE_RATE`), but
a document prints ``IVA (21%)`` and a model obeying "copy exactly as
printed" returns ``"21%"``. Routed through :func:`_grounded_decimal` that
lost the field outright -- measured on a corpus invoice, where the model
that read the rate *more* literally was the one punished for it.
A percent sign is a UNIT MARKER, not a digit, so removing it is unit
normalisation and the number that remains is still the one the document
printed -- copied, never computed. Nothing about anti-fabrication is
weakened: the anchor
(:attr:`ExtractedInvoiceFields.iva_rate`) still holds ``"21%"`` verbatim for
a later closure check to point at, so anchor and value stay explicitly
distinct rather than collapsing into one field that could silently
launder a transcription error into a computed figure. Exactly one
trailing unit is stripped; a value carrying digits
after the sign, or any other stray character, still fails the decimal
authority and drops to ``None``.
Deliberately NOT applied to a monetary amount. A percent sign on an amount
is a misread, not a unit, and tolerating it there would launder a bad
transcription into a filing figure.
"""
del currency_unit # a rate carries no currency; the shared signature is the form table's
if raw is None:
return None
text = raw.strip()
for unit in ("%", "percent", "pct"):
if text.lower().endswith(unit):
text = text[: -len(unit)].strip()
break
return _grounded_decimal(text)
def _grounded_money(raw: str | None, currency_unit: str | None = None) -> Decimal | None:
"""Return a transcribed amount as a Decimal, dropping one printed currency unit.
The declared form for an amount is digits with the printed decimal
separator and no currency sign
(:attr:`~llm.invoice_field_contract.InvoiceFieldForm.MONETARY_AMOUNT`), and
the per-field instruction says so. A model reading a line that prints
``Base imponible: 1200.00 EUR`` nonetheless returns ``"1200.00 EUR"``, and
routed straight through :func:`_grounded_decimal` that lost the taxable
base outright -- measured live on the text lane, where the base, the cuota
and the total all arrived carrying their printed unit.
A currency token beside an amount is a UNIT MARKER, exactly as a percent
sign is beside a rate (:func:`_grounded_percentage`), so removing it is
unit normalisation and the number that remains is the one the document
printed -- copied, never computed. The rule itself is
:func:`~core.decimal.printed_units.without_currency_unit`, shared with the
anchor check so the value half and the anchor half cannot disagree about
the same string. The anchor still holds the printed form
verbatim for the closure check to point at, and an amount whose thousands
separator is genuinely ambiguous is still dropped by
:func:`_grounded_decimal` rather than read one way.
The earlier note on :func:`_grounded_percentage` -- that a percent sign on
an AMOUNT is a misread rather than a unit -- still holds: ``21%`` is not a
currency unit, so it is not removed here and still drops.
"""
if raw is None:
return None
return _grounded_decimal(without_currency_unit(raw, currency_unit))
def _grounded_currency(raw: str | None) -> str | None:
"""Return *raw* as an ISO-4217 code, or ``None`` when it is not one.
A currency the model transcribed as a symbol, a word, or anything other
than a three-letter alphabetic code is dropped rather than guessed: mapping
a bare "$" to USD would invent a fact the document may not support (it is
also CAD, AUD, MXN), and inventing the currency of a filing amount is the
one error the grounded-extraction discipline exists to prevent.
"""
if raw is None:
return None
try:
return normalise_iso_4217_currency(raw)
except CoreValidationError:
return None
_TEXT_GROUNDING_BY_FORM: Mapping[InvoiceFieldForm, Callable[[str | None], str | None]] = {
InvoiceFieldForm.TAX_IDENTIFIER: _grounded_tax_id,
InvoiceFieldForm.FREE_TEXT: _grounded_free_text,
InvoiceFieldForm.CALENDAR_DATE: _grounded_date,
InvoiceFieldForm.CURRENCY_CODE: _grounded_currency,
}
"""Validators for the declared forms whose grounded value stays a string."""
_NUMERIC_GROUNDING_BY_FORM: Mapping[InvoiceFieldForm, Callable[[str | None, str | None], Decimal | None]] = {
InvoiceFieldForm.MONETARY_AMOUNT: _grounded_money,
InvoiceFieldForm.PERCENTAGE_RATE: _grounded_percentage,
}
"""Validators for the declared forms whose grounded value becomes a Decimal.
Split from :data:`_TEXT_GROUNDING_BY_FORM` by RETURN TYPE, not by convenience:
one table would have to be typed as a union and every call site would need a
runtime narrow, trading a static guarantee for a dead branch. Together the two
tables must cover :class:`~llm.invoice_field_contract.InvoiceFieldForm`
exactly and disjointly, which the parity gate asserts -- so a form added to the
enum without a validator fails there rather than falling through at runtime.
"""
def _ground_text(raw: str | None, field_name: str) -> str | None:
"""Ground ``raw`` through the validator ``field_name``'s DECLARED form selects.
Raises:
KeyError: When the field's declared form has no string-valued validator,
which is a declaration error the parity gate exists to catch.
"""
return _TEXT_GROUNDING_BY_FORM[contract_for_field(field_name).form](raw)
def _ground_numeric(raw: str | None, field_name: str, currency_unit: str | None = None) -> Decimal | None:
"""Ground ``raw`` through the validator ``field_name``'s DECLARED form selects.
``currency_unit`` is the code this reply reported for the document, used by
the monetary validator to recognise a printed unit beside an amount and
ignored by the rate validator.
Raises:
KeyError: When the field's declared form has no numeric validator, which
is a declaration error the parity gate exists to catch.
"""
return _NUMERIC_GROUNDING_BY_FORM[contract_for_field(field_name).form](raw, currency_unit)
def _read_provenance(
*,
field_name: str,
value: str | Decimal | None,
anchor: str | None,
role_evidence: str | None,
origin: FieldOrigin,
) -> FieldProvenance | None:
"""Return the envelope recording how ``field_name`` was read, or ``None``.
``None`` when the field carries no grounded value: an envelope describes a
value's provenance, and a field the grounder dropped has no value to
describe. The absence is itself reviewable -- the draft's own contract says
a missing envelope never means the value was exact.
The outcome is always :attr:`~core.field_grounding.FieldGroundingOutcome.UNANCHORED`, and
that is a deliberate under-claim rather than a placeholder. The model
REPORTING an anchor is a claim about the document, not a check against it;
the check belongs to :func:`~application.ledger.grounding_anchor.evaluate_anchor`, which
needs the transcription this stage does not hold. ``UNANCHORED`` is the only
member that is true here: ``ANCHORED`` would assert a search nobody ran,
``RECONCILED`` an independent identity nobody consulted, ``CONTRADICTED`` a
disagreement nobody found, and ``AMBIGUOUS`` is refused outright by the
envelope without two competing candidates.
The anchor still rides along, which is the point of the record: it is the
claim a later stage verifies. Carrying it under an unverified outcome
matches how the anchor check itself reports a contradiction -- the printed
form the operator needs to see survives the outcome that failed.
"""
if value is None:
return None
claimed = anchor.strip() if anchor is not None else ""
claimed_role = role_evidence.strip() if role_evidence is not None else ""
return FieldProvenance(
field=field_name,
origin=origin,
grounding=FieldGroundingOutcome.UNANCHORED,
anchor=claimed or None,
# Recorded as the model REPORTED it, unchecked, exactly like the anchor
# beside it. The check needs the transcription this stage does not hold
# and runs at the grounding stage, which drops an excerpt the document
# does not contain rather than trusting it.
role_evidence=claimed_role or None,
note=(
f"the reading model reported {claimed!r} as the printed form; not yet checked against the document"
if claimed
else "the reading model reported no printed form for this value"
),
)
def _unverified_identity_findings(
response: ExtractedInvoiceResponse,
grounded: Mapping[str, str | Decimal | None],
) -> tuple[DraftDiscrepancyFinding, ...]:
"""Record every identity field the document printed and the checks rejected.
Dropping a value to ``None`` is the right handling and stays; what must not
also be dropped is the FACT that something was there. An identifier failing
its control character is precisely what hides the true counterparty from a
validating read, so an operator needs "we could not verify this" rather than
a slot that looks like the document said nothing.
Without this the two collapse before anything can tell them apart. The
counterparty role resolution reads the DRAFT, so a rejected identifier
reaches it as an absence -- and
:func:`~application.ledger.identity_roles.resolve_counterparty_identity` deliberately
raises no unresolved-role finding on an absence, because a factura
simplificada legitimately has one. The distinction has to be preserved at
the stage that performs the rejection; nothing downstream can reconstruct
it.
Args:
response: The parsed reply, whose ``fields`` half still carries the
verbatim transcribed strings.
grounded: The re-validated values, keyed by field name.
Returns:
One finding per identity field transcribed non-blank and rejected.
"""
findings: list[DraftDiscrepancyFinding] = []
for contract in INVOICE_FIELD_CONTRACTS:
# Keyed on the DECLARED form, like every other identity-scoped decision
# in this module, so a fourth identity field is covered by construction.
if not contract.carries_role_evidence:
continue
transcribed = getattr(response.fields, contract.field_name, None)
if not isinstance(transcribed, str) or not transcribed.strip():
continue
if grounded[contract.field_name] is not None:
continue
findings.append(
DraftDiscrepancyFinding(
kind=DraftDiscrepancyKind.IDENTITY_UNVERIFIED,
field=contract.field_name,
detail=(
f"the document prints {transcribed.strip()!r} as this party's tax identifier, but it "
f"fails its control-character check, so it was not accepted as a verified identity"
),
),
)
return tuple(findings)