"""Source-mesh resolver for governed invoice records.
:class:`InvoiceCatalogueSourceResolver` reads the
:class:`~domain.invoices.models.InvoiceCatalogue` supplied through its application-owned
read capability. It projects those records into the calculation mesh as
:class:`~application.aggregation.source_mesh.CalculationSourceResolution` values for
:attr:`~core.aggregation.BindingSourceKind.COLLECTIBLE_INVOICE`,
:attr:`~core.aggregation.BindingSourceKind.PAYABLE_INVOICE`, and the combined-direction
:attr:`~core.aggregation.BindingSourceKind.M347_THIRD_PARTY_OPERATION`.
The :class:`~domain.invoices.models.Invoice` aggregate is the sole invoice record and
the reconciliation and link authority. Records reach the mesh only once they can
be represented as registry
:class:`~domain.calculations.registry.invoice_bindings.InvoiceObservation` facts, with Modelo 349
summary bindings, detail rows, transaction ids, and source provenance emitted
through one resolver envelope.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping, Sequence
from datetime import date
from decimal import Decimal
from typing import ClassVar
from ...core.aggregation import (
BindingSourceKind,
CalculationSourceLineageRole,
IntracomOperationType,
ThirdPartyDeclarationRole,
)
from ...core.external_constants import DEFAULT_CURRENCY
from ...core.hashing import sha256_hex
from ...core.identity.bucket import BucketId
from ...core.modelo import Modelo
from ...core.period import Period
from ...domain.calculations.registry.binding_terminal_origin import TerminalOriginClass
from ...domain.calculations.registry.errors import RegistryValidationError
from ...domain.calculations.registry.ids import BindingId
from ...domain.calculations.registry.invoice_bindings import (
InvoiceObservation,
is_m347_declarante_summary_invoice_binding,
m347_operation_clave,
resolve_invoice_binding_row_values,
resolve_invoice_binding_values,
)
from ...domain.calculations.registry.iva_category_catalogue import require_iva_category, resolve_iva_category_catalogue
from ...domain.calculations.registry.third_party_declaration_roles import (
resolve_third_party_declaration_role_catalogue,
)
from ...domain.calculations.registry.travel_agency_mediation import (
is_travel_agency_air_passenger_transport,
)
from ...domain.invoices.decomposition import InvoiceDecomposition, InvoiceDecompositionDefect, decompose_invoice
from ...domain.invoices.models import Invoice
from ...domain.iva.classification import InvoiceKind
from ...domain.iva.schema import IvaCategory
from ...domain.modelos.row_models import Modelo349OperadorRow, validate_m349_country_prefix_context
from ..aggregation.source_mesh import (
CalculationSourceContext,
CalculationSourceDiagnostic,
CalculationSourceProvenance,
CalculationSourceResolution,
)
from ..aggregation.source_resolution_operations import storage_degradation_resolution
from .source_resolver_ports import InvoiceSourcePersistenceError, InvoiceSourceResolverPorts
_OWNED_SOURCES: tuple[BindingSourceKind, ...] = (
BindingSourceKind.COLLECTIBLE_INVOICE,
BindingSourceKind.PAYABLE_INVOICE,
BindingSourceKind.M347_THIRD_PARTY_OPERATION,
)
_ObservedInvoice = tuple[Invoice, InvoiceObservation]
_IncoherentInvoice = tuple[Invoice, InvoiceDecomposition]
_M349_PAYABLE_SUMMARY_BINDING_MIRRORS: dict[str, str] = {
"iva-349-declarante-numero-operadores-adquisicion": "iva-349-declarante-numero-operadores",
"iva-349-declarante-importe-operaciones-adquisicion": "iva-349-declarante-importe-operaciones",
"iva-349-declarante-numero-rectificaciones-adquisicion": "iva-349-declarante-numero-rectificaciones",
"iva-349-declarante-importe-rectificaciones-adquisicion": "iva-349-declarante-importe-rectificaciones",
}
_M349_OPERADOR_ROW_BINDINGS: dict[BindingId, str] = {
"iva-349-operador-row-codigo-pais": "codigo_pais",
"iva-349-operador-row-nif": "nif_comunitario",
"iva-349-operador-row-apellidos": "razon_social",
"iva-349-operador-row-clave": "clave_operacion",
"iva-349-operador-row-base": "importe",
}
_COLLECTIBLE_M349_OPERATION_TYPES: frozenset[IntracomOperationType] = frozenset(
{
IntracomOperationType.E,
IntracomOperationType.H,
IntracomOperationType.M,
IntracomOperationType.S,
IntracomOperationType.T,
IntracomOperationType.R,
IntracomOperationType.D,
IntracomOperationType.C,
},
)
_PAYABLE_M349_OPERATION_TYPES: frozenset[IntracomOperationType] = frozenset(
{
IntracomOperationType.A,
IntracomOperationType.ADQUISICION_SERVICIOS,
IntracomOperationType.T,
},
)
#: The claves an invoice's IVA category alone determines, keyed by side.
#:
#: Values are :class:`~cadrumo.core.aggregation.IntracomOperationType` MEMBERS, never the
#: clave letters, because the member's ``value`` IS the letter the diseño de
#: registro defines: a literal beside the enum is a copy that can drift from
#: the thing it copies with nothing to catch it. The mismatch that makes this
#: concrete is the services acquisition, whose member is named
#: ``ADQUISICION_SERVICIOS`` while its clave is ``I`` -- a literal ``"I"`` here
#: would be reachable from neither the member name nor the letter by search.
#:
#: Membership is deliberately the FOUR entries a category identifies
#: unambiguously, plus triangulation handled separately because it is
#: kind-independent. It is NOT the full ten-clave set, and widening it here
#: would change what gets declared rather than how it is expressed:
#:
#: - ``M``/``H`` (supplies following an exempt importation, LIVA art. 27.12)
#: share the intra-community supply category with ``E``, so no category
#: predicate can separate them; the operator states them via the operation
#: type, and the resolver discloses the ambiguity rather than guessing.
#: - ``R``/``D``/``C`` (the call-off stock claves) report movements that carry
#: no invoice at all, so no invoice-sourced path can reach them.
def _clave_by_kind_and_category() -> dict[tuple[InvoiceKind, IvaCategory], IntracomOperationType]:
"""Project the dated Modelo 349 category-to-clave bindings."""
catalogue = resolve_iva_category_catalogue()
return {
(InvoiceKind.ISSUED, catalogue.require("intra_community_supply")): IntracomOperationType(
catalogue.operation_type("issued.intra_community_supply")
),
(InvoiceKind.ISSUED, catalogue.require("intra_community_service_supply")): IntracomOperationType(
catalogue.operation_type("issued.intra_community_service_supply")
),
(
InvoiceKind.RECEIVED,
catalogue.require("intra_community_service_acquisition_reverse_charge"),
): IntracomOperationType(
catalogue.operation_type("received.intra_community_service_acquisition_reverse_charge")
),
(
InvoiceKind.RECEIVED,
catalogue.require("intra_community_acquisition_reverse_charge"),
): IntracomOperationType(catalogue.operation_type("received.intra_community_acquisition_reverse_charge")),
}
[docs]
def invoice_direction_to_source_kind(kind: InvoiceKind) -> BindingSourceKind:
"""Map an invoice direction to its settlement source kind.
The single contractual home for the direction<->settlement relationship,
consumed by both :class:`InvoiceCatalogueSourceResolver` and the operator
``aeat app ledger invoice`` CLI. An *issued* invoice (we billed a customer)
is *collectible*; a *received* invoice (a vendor billed us) is *payable*.
Returns the canonical :class:`~core.aggregation.BindingSourceKind` member rather than a
locally-declared direction enum: the settlement taxonomy has exactly one
home per ``aeat-registry-bindings``.
Returns:
The :class:`BindingSourceKind` settling ``kind``.
"""
if kind is InvoiceKind.ISSUED:
return BindingSourceKind.COLLECTIBLE_INVOICE
return BindingSourceKind.PAYABLE_INVOICE
def _invoice_sources_in_context(
invoices: Iterable[Invoice],
*,
context: CalculationSourceContext,
active_sources: frozenset[BindingSourceKind],
) -> tuple[Invoice, ...]:
return tuple(
invoice
for invoice in invoices
if _invoice_in_context(invoice, context) and _invoice_source_kind(invoice) in active_sources
)
def _observe_invoice_sources(
source_invoices: Sequence[Invoice],
*,
context: CalculationSourceContext,
) -> tuple[tuple[_ObservedInvoice, ...], tuple[_IncoherentInvoice, ...], tuple[Invoice, ...]]:
observed_items: list[_ObservedInvoice] = []
incoherent: list[_IncoherentInvoice] = []
withheld_for_conversion: list[Invoice] = []
for invoice in source_invoices:
if _is_unconverted_foreign_invoice(invoice):
withheld_for_conversion.append(invoice)
observation = _invoice_observation(invoice, context=context)
if observation is None:
continue
verdict = _m349_incoherent_verdict(invoice, context=context)
if verdict is not None:
incoherent.append((invoice, verdict))
continue
observed_items.append((invoice, observation))
return tuple(observed_items), tuple(incoherent), tuple(withheld_for_conversion)
def _invoice_resolution_from_observations(
*,
context: CalculationSourceContext,
catalogue_invoices: Sequence[Invoice],
source_invoices: Sequence[Invoice],
observed_items: tuple[_ObservedInvoice, ...],
incoherent: tuple[_IncoherentInvoice, ...],
withheld_for_conversion: tuple[Invoice, ...],
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
) -> CalculationSourceResolution:
observations = tuple(observation for _, observation in observed_items)
binding_values = resolve_invoice_binding_values(
context.revision,
observations,
effective_date=date(context.filing_year, 12, 31),
)
declared_invoices = tuple(invoice for invoice, _ in observed_items)
diagnostics = _m349_incoherence_diagnostics(incoherent, resolver_id=resolver_id)
diagnostics += _unconverted_foreign_diagnostics(
withheld_for_conversion,
context=context,
resolver_id=resolver_id,
)
if context.modelo == Modelo("349").value:
diagnostics += _m349_inferred_clave_diagnostics(
declared_invoices,
bucket_invoices=catalogue_invoices,
resolver_id=resolver_id,
)
diagnostics += _m347_role_fact_advisories(source_invoices, context=context, resolver_id=resolver_id)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
binding_values=_m349_declarante_summary_union(context=context, binding_values=binding_values),
detail_rows=_m349_operador_rows_from_observations(context=context, observations=observations),
source_transaction_ids=tuple(
sorted(
{transaction_id for invoice, _ in observed_items for transaction_id in invoice.linked_transaction_ids},
),
),
diagnostics=diagnostics,
provenance=tuple(_invoice_provenance(invoice, observation) for invoice, observation in observed_items),
)
[docs]
class InvoiceCatalogueSourceResolver:
"""Resolve invoice-source bindings and detail rows from persisted invoice records.
The resolver owns both invoice source kinds in the calculation mesh. It
filters records by :class:`CalculationSourceContext`, turns declarable
intracommunity entries into :class:`InvoiceObservation` facts, and returns a
:class:`CalculationSourceResolution` carrying binding values, Modelo 349
detail rows, linked transaction ids, and stable source provenance.
"""
resolver_id: ClassVar[str] = "invoice_catalogue"
owned_sources: ClassVar[tuple[BindingSourceKind, ...]] = _OWNED_SOURCES
def __init__(
self,
*,
ports: InvoiceSourceResolverPorts,
) -> None:
"""Bind the resolver to its explicitly composed invoice read capability.
Args:
ports: Required application-owned capabilities for invoice source
resolution.
"""
self._ports = ports
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
"""Resolve this context's invoice-source bindings against the catalogue.
Args:
context: Modelo, revision, period, and bucket selecting both the
active invoice sources and the invoices in scope.
Returns:
The resolution carrying binding values, Modelo 349 detail rows,
linked transaction ids, diagnostics, and source provenance. Sources
the active revision does not declare resolve to an empty result, and
a degraded catalogue read resolves to a storage-degradation result
rather than a zero total.
"""
active_sources = _invoice_sources_for_revision(context)
if not active_sources:
return CalculationSourceResolution(resolver_id=self.resolver_id, owned_sources=self.owned_sources)
try:
catalogue = self._ports.catalogue_reader.load()
except InvoiceSourcePersistenceError as exc:
return storage_degradation_resolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
source_kinds=tuple(active_sources),
error=exc,
)
catalogue_invoices = tuple(catalogue.values())
source_invoices = _invoice_sources_in_context(
catalogue_invoices,
context=context,
active_sources=active_sources,
)
observed_items, incoherent, withheld_for_conversion = _observe_invoice_sources(
source_invoices,
context=context,
)
return _invoice_resolution_from_observations(
context=context,
catalogue_invoices=catalogue_invoices,
source_invoices=source_invoices,
observed_items=observed_items,
incoherent=incoherent,
withheld_for_conversion=withheld_for_conversion,
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
)
_M349_SELF_CONTRADICTION_DEFECTS: frozenset[InvoiceDecompositionDefect] = frozenset(
{
InvoiceDecompositionDefect.CUOTA_CONTRADICTS_CATEGORY,
InvoiceDecompositionDefect.CATEGORY_IMPOSSIBLE_ON_THIS_KIND,
},
)
"""Decomposition defects where two declarations on one record disagree.
Both members describe a record whose operator made two assertions that cannot
both be true -- an entrega exenta under LIVA art. 25 carrying a repercuted
cuota, or a one-directional category recorded on its impossible side. The
remaining members describe something ABSENT, which on this surface is a gap in
what the record can express rather than evidence that its figures are wrong.
"""
def _m349_incoherent_verdict(
invoice: Invoice,
*,
context: CalculationSourceContext,
) -> InvoiceDecomposition | None:
"""Return the decomposition verdict when it disqualifies an M349 record.
Scoped to Modelo 349 deliberately, and only after the clave is settled.
Modelo 349 is the one invoice-sourced surface whose declared figure is
conditioned on the declared IVA treatment: the clave is CHOSEN from
:attr:`~cadrumo.domain.invoices.models.Invoice.iva_category`, and the base
declared under it is the base imponible of an operation the record asserts
is exenta under LIVA art. 25. A record simultaneously claiming that
exemption and carrying a repercuted cuota contradicts itself, and the
contract cannot tell which of the two declarations is the mistake, so it
grounds neither.
Modelo 347 asks a different question and is deliberately NOT checked here.
Its declared figure is the total contraprestacion of operations with one
third party (RD 1065/2007 art. 34), which the invoice's own totals identity
already bounds and which no IVA category conditions. Running the contract
there would drop real above-threshold operations out of an informativa on
the strength of an unrelated missing field. The OSS/IOSS path is excluded
for the same reason plus a stronger one: no
:class:`~cadrumo.domain.iva.schema.IvaCategory` member names an OSS operation at
all -- the OSS axis is the regime and transaction kind -- so every
legitimate OSS invoice would come back ungrounded, and that path already
runs the coherence check that does apply to it, cross-checking the
persisted cuota against the destination Member State's published rate.
Within Modelo 349 the check is narrowed again, to the defects where the
record CONTRADICTS ITSELF. Absence is deliberately not disqualifying, and
the reason is structural: **an absent category does not mean the operation
was inexpressible, it usually means the clave came from somewhere else.**
:func:`_intracommunity_clave` consults an explicit
:attr:`~cadrumo.domain.invoices.models.Invoice.operation_type` FIRST and returns
without ever reading ``iva_category``, so a record carrying a directly
declared clave legitimately carries no category at all. Since this check
runs only after the clave is settled, treating absence as disqualifying
would drop exactly the records whose clave the operator stated most
explicitly -- the least ambiguous rows in the store.
That reasoning is deliberately independent of what the category enum
happens to contain, because the previous justification was not and went
stale. It asserted that an ordinary prestacion or adquisicion de servicios
intracomunitaria "maps to no :class:`~cadrumo.domain.iva.schema.IvaCategory`
member at all, because the enum names goods, acquisitions and triangulation
but not services". The enum has since gained
``INTRA_COMMUNITY_SERVICE_SUPPLY`` and
``INTRA_COMMUNITY_SERVICE_ACQUISITION_REVERSE_CHARGE``, and
:func:`_intracommunity_clave` maps both to their claves a hundred-odd lines
below -- so the stated ground for weakening a filing-path guard was refuted
by the same module that stated it.
The behaviour is unchanged, and that is a decision rather than an
omission: making absence disqualifying would alter filed M349 output, which
needs its own evidence and its own ruling, not a docstring correction.
Services now being expressible only strengthens the conclusion -- a
services invoice can reach its clave through either route, so absence is
even weaker evidence of an unrepresentable operation than before.
``FX_UNRESOLVED`` is likewise excluded here because the unconverted-foreign
gate upstream already withholds those records.
"""
if context.modelo != Modelo("349").value:
return None
verdict = decompose_invoice(invoice)
contradictions = tuple(defect for defect in verdict.defects if defect in _M349_SELF_CONTRADICTION_DEFECTS)
if not contradictions:
return None
return verdict.model_copy(update={"defects": contradictions})
#: Diagnostic ``reason`` for a Modelo 349 clave the resolver inferred from the
#: invoice's IVA category because the record carried no explicit operation type.
M349_CLAVE_INFERRED_REASON = "m349_clave_inferred_from_category"
def _clave_was_inferred_as_entrega(invoice: Invoice) -> bool:
"""Return whether this invoice's clave E was a guess rather than a statement.
True only for the one ambiguous case: an issued exempt intra-community
supply carrying no operation type, which ``_intracommunity_clave`` resolves
to E. Every other fallback branch maps a category that identifies its clave
unambiguously, and a record carrying an operation type stated its clave
outright.
"""
return (
invoice.operation_type is None
and invoice.kind is InvoiceKind.ISSUED
and invoice.iva_category == require_iva_category("intra_community_supply")
)
def _m349_inferred_clave_diagnostics(
declared: Sequence[Invoice],
*,
bucket_invoices: Sequence[Invoice],
resolver_id: str,
) -> tuple[CalculationSourceDiagnostic, ...]:
"""Return at most ONE advisory disclosing that claves were inferred, not read.
Aggregated rather than per-invoice, and that is the whole design. The
ambiguity this discloses is not separable at this layer -- the prior
importation that distinguishes an art. 27.12 supply from an ordinary art. 25
one appears nowhere on the invoice -- so a per-record advisory would fire on
every ordinary intra-community supply a taxpayer makes. Firing on the correct
majority is what trains an operator to ignore a channel, and it would forfeit
this disclosure exactly when it matters.
One line per calculation states an assumption; N lines per calculation is an
alarm about nothing. The count and the invoice numbers are carried so the
operator can find the records without the advisory having to accuse each one.
"""
inferred = [invoice for invoice in declared if _clave_was_inferred_as_entrega(invoice)]
if not inferred:
return ()
# The discriminator, and the reason this is a disclosure rather than noise.
# Clave M or H requires a PRIOR exempt importation by this same taxpayer
# (LIVA art. 27.12 exempts the importation only because the onward supply is
# art. 25 exempt). A bucket holding no importation at all therefore cannot
# contain a post-importation supply, so the inferred E is not merely likely
# correct there -- it is the only clave available, and saying so would be an
# alarm about nothing on every Modelo 349 an ordinary EU-trading taxpayer
# ever files.
# Read across the whole bucket, not the declared set: an importation is a
# RECEIVED record that produces no Modelo 349 row of its own, so it is absent
# from ``declared`` by construction and invisible to a scan of it.
if not any(invoice.iva_category == require_iva_category("import_third_country") for invoice in bucket_invoices):
return ()
numbers = ", ".join(sorted(invoice.invoice_number for invoice in inferred))
return (
CalculationSourceDiagnostic(
reason=M349_CLAVE_INFERRED_REASON,
source_kind=BindingSourceKind.COLLECTIBLE_INVOICE.value,
resolver_id=resolver_id,
message=(
f"{len(inferred)} intra-community supplies carry no operation type, so their Modelo 349 "
f"clave was inferred as 'E' from the IVA category ({numbers}). That is correct for an "
"ordinary exempt supply under LIVA art. 25, but a supply following an exempt importation "
"(art. 27.12) reports under clave 'M', or 'H' when made through a representante fiscal, "
"and the invoice records no fact that distinguishes the two."
),
remedy=(
"If any of these supplies followed an exempt importation, set its operation type to M or H "
"and recalculate; otherwise the inferred clave is correct and no action is needed."
),
# Advisory-asserted: this module holds no revision, snapshot or
# casilla definition anywhere -- the claim spans two provisions
# (the ordinary art. 25 exemption and the art. 27.12 post-importation
# carve-out) about the invoice catalogue as a whole, not about one
# M349 casilla.
asserted_legal_refs=("ley-37-1992:art-25", "ley-37-1992:art-27"),
),
)
def _m349_incoherence_diagnostics(
incoherent: Sequence[tuple[Invoice, InvoiceDecomposition]],
*,
resolver_id: str,
) -> tuple[CalculationSourceDiagnostic, ...]:
"""Return one advisory per record the decomposition contract disqualified.
Excluded-but-VISIBLE. The record stays in the catalogue and stays editable;
what it must not do is disappear from the recapitulativa without the
operator being told, because a missing intracomunitaria is an
under-declaration whether it was dropped by a contradiction or by silence.
"""
return tuple(
CalculationSourceDiagnostic(
reason="ungrounded_income_substrate",
source_kind=str(_invoice_source_kind(invoice)),
resolver_id=resolver_id,
source_ref=f"invoice:{invoice.invoice_id}",
message=(
f"invoice {invoice.invoice_number!r} declares an intracommunity operation "
f"the decomposition contract could not ground "
f"({', '.join(defect.value for defect in verdict.defects)}), so its base is "
"NOT declared on this Modelo 349"
),
remedy=(
"Reconcile the invoice's declared IVA treatment with the amounts recorded on "
"it, then recalculate so the operation reaches the recapitulativa"
),
)
for invoice, verdict in incoherent
)
def _unconverted_foreign_diagnostics(
invoices: Sequence[Invoice],
*,
context: CalculationSourceContext,
resolver_id: str,
) -> tuple[CalculationSourceDiagnostic, ...]:
"""Return one advisory per invoice withheld for want of a euro conversion.
Excluding the AMOUNT is settled and correct: a foreign invoice with no
resolved rate has an unknown euro value, and declaring its face value as
euro would be a mis-declaration rather than a conservative one.
Excluding it in SILENCE is the defect. The operation is real and declarable
-- the euro figure is what is missing, not the operation -- so an
informativa that simply omits it leaves the operator filing an incomplete
return with nothing on any surface saying so. This is the same principle
:func:`_m349_incoherence_diagnostics` already states: a missing record is an
under-declaration whether it was dropped by a contradiction or by silence.
"""
return tuple(
CalculationSourceDiagnostic(
reason="unconverted_foreign_currency",
source_kind=str(_invoice_source_kind(invoice)),
resolver_id=resolver_id,
source_ref=f"invoice:{invoice.invoice_id}",
message=(
f"invoice {invoice.invoice_number!r} is denominated in {invoice.currency} with no "
f"resolved euro rate, so it is NOT declared on this Modelo {context.modelo}; its "
"euro value is unknown and declaring the foreign amount as euro would misstate it"
),
remedy=(
"Record the euro conversion rate on the invoice, then recalculate so the "
"operation reaches the declaration"
),
)
for invoice in invoices
)
def _m347_role_fact_advisories(
invoices: Sequence[Invoice],
*,
context: CalculationSourceContext,
resolver_id: str,
) -> tuple[CalculationSourceDiagnostic, ...]:
"""Advise on a Modelo 347 clave D/E fact left UNDECLARED, rather than silently deciding it.
Fires only for a filer who actually carries a clave D or E role -- an
empty role set (the overwhelming majority of filers) never triggers
this, so the advisory is proportionate to the population the roles
exist to cover, not every invoice everywhere. Scoped to RECEIVED
invoices for clave D, matching the article's own "adquisiciones" text;
clave E carries no direction restriction in the article, so every
invoice from a public-administration filer is in scope.
"""
if context.modelo != Modelo("347").value:
return ()
declaration_roles = _m347_filer_declaration_roles(context.bucket_id)
if not declaration_roles:
return ()
role_catalogue = resolve_third_party_declaration_role_catalogue()
clave_d_eligible = bool(declaration_roles & role_catalogue.roles_for_clave("D"))
clave_e_eligible = bool(declaration_roles & role_catalogue.roles_for_clave("E"))
diagnostics: list[CalculationSourceDiagnostic] = []
for invoice in invoices:
if clave_d_eligible and invoice.kind is InvoiceKind.RECEIVED and invoice.outside_economic_activity is None:
diagnostics.append(
CalculationSourceDiagnostic(
reason="unclassified_declarant_role_fact",
source_kind=str(_invoice_source_kind(invoice)),
resolver_id=resolver_id,
source_ref=f"invoice:{invoice.invoice_id}",
message=(
f"invoice {invoice.invoice_number!r} is an acquisition from a filer carrying a "
"Modelo 347 clave D role, but whether it is al margen de la actividad "
"empresarial is undeclared, so it is NOT classified as clave D"
),
remedy="Declare outside_economic_activity on this invoice, then recalculate",
),
)
if clave_e_eligible and invoice.is_subvencion_ayuda is None:
diagnostics.append(
CalculationSourceDiagnostic(
reason="unclassified_declarant_role_fact",
source_kind=str(_invoice_source_kind(invoice)),
resolver_id=resolver_id,
source_ref=f"invoice:{invoice.invoice_id}",
message=(
f"invoice {invoice.invoice_number!r} is from a filer carrying the Modelo 347 "
"public-administration role, but whether it is a subvención, auxilio or ayuda "
"is undeclared, so it is NOT classified as clave E"
),
remedy="Declare is_subvencion_ayuda on this invoice, then recalculate",
),
)
return tuple(diagnostics)
def _invoice_sources_for_revision(context: CalculationSourceContext) -> frozenset[BindingSourceKind]:
declared_sources = frozenset(
binding.source for binding in context.revision.bindings if binding.source in _OWNED_SOURCES
)
if any(is_m347_declarante_summary_invoice_binding(binding) for binding in context.revision.bindings):
return frozenset(_OWNED_SOURCES)
return declared_sources
def _invoice_in_context(invoice: Invoice, context: CalculationSourceContext) -> bool:
"""Whether this invoice is declarable for the context's bucket and period.
Only a POPULATED, mismatching bucket excludes. An unattributed invoice
belongs to the store it was loaded from, and that store is opened against
``context.bucket_id`` -- with the composed catalogue reader refusing a
foreign row on read, so nothing another bucket owns reaches here.
Treating ``None`` as a mismatch is what this reads as if the check is
written against the bucket id alone, and it is silent: an unattributed
invoice compares unequal to every real bucket, so it drops out of M347 and
M349 with no defect, no advisory and no refusal. Nothing downstream of the
filter can tell "this taxpayer had no such operations" apart from "the
filter discarded them", which is the shape a declaration must never take.
This is the same rule the persistence guard applies, deliberately: the two
layers previously disagreed about whether an unattributed invoice was
normal, and a disagreement about that between the store and the projection
is resolved in favour of declaring.
"""
if invoice.bucket_id is not None and invoice.bucket_id != context.bucket_id:
return False
return _date_in_period(invoice.issued_at, period=context.period)
def _date_in_period(value: date, *, period: Period) -> bool:
return period.contains(value)
def _invoice_source_kind(invoice: Invoice) -> str:
return invoice_direction_to_source_kind(invoice.kind).value
def _eur(converted: Decimal | None, invoice: Invoice) -> Decimal:
"""Return the euro amount, refusing rather than falling back to face value.
``None`` here means the caller skipped the
:func:`_is_unconverted_foreign_invoice` gate. Falling back to the native
amount would be the exact silent mis-declaration this path exists to
prevent, so the inconsistency is raised instead.
"""
if converted is None:
msg = (
f"invoice {invoice.invoice_id} is denominated in {invoice.currency} with no resolved "
f"euro value; it must be gated out of projection, not declared at face value"
)
raise RegistryValidationError(msg)
return converted
def _is_unconverted_foreign_invoice(invoice: Invoice) -> bool:
"""Return whether *invoice* is foreign-currency with no euro equivalent.
Mirrors the ledger's ``is_non_eur_without_conversion`` gate. Every modelo
amount is declared in euro, so an invoice whose euro value could not be
resolved must be withheld from projection: summing its face value would
declare foreign units as euro.
"""
return invoice.currency != DEFAULT_CURRENCY and invoice.grand_total_eur is None
def _invoice_observation(invoice: Invoice, *, context: CalculationSourceContext) -> InvoiceObservation | None:
if _is_unconverted_foreign_invoice(invoice):
return None
if invoice.counterparty_tax_id is None:
# M347/M349 both declare a third party by their tax id; a factura
# simplificada legitimately carries none (RD 1619/2012 art. 6.1.d), so
# it has nothing these informativas can declare rather than a defect.
return None
if context.modelo == Modelo("347").value:
return _m347_invoice_observation(invoice, context=context)
clave = _intracommunity_clave(invoice)
if clave is None:
return None
if context.modelo == Modelo("349").value:
validate_m349_country_prefix_context(
country_code=invoice.counterparty_country,
clave_operacion=clave,
filing_year=context.filing_year,
period=context.period.registry_token,
)
return InvoiceObservation(
invoice_id=invoice.invoice_id,
source_kind=BindingSourceKind(_invoice_source_kind(invoice)),
party_tax_id=invoice.counterparty_tax_id,
country_code=invoice.counterparty_country,
transaction_date=invoice.issued_at,
base_amount=_eur(invoice.base_total_eur, invoice),
invoice_total_amount=_eur(invoice.grand_total_eur, invoice),
intracommunity_clave=clave,
party_legal_name=invoice.counterparty_name,
)
def _m347_filer_declaration_roles(bucket_id: BucketId) -> frozenset[ThirdPartyDeclarationRole]:
"""Load the filer's :class:`ThirdPartyDeclarationRole` memberships for *bucket_id*.
Mirrors the established bucket-scoped profile-fact loading pattern (see
e.g. ``m111_no_retenciones_periods_for_bucket``): a missing or unset
profile fails closed to an empty role set rather than raising, because
the overwhelming majority of filers legitimately carry none. An empty
set means claves C, D and E simply do not classify for this filer --
never that A, B, F or G are affected, since those read no profile fact.
"""
from ...domain.calculations.registry.authority import bundled_indexed_authority
from ...domain.user_profile.errors import ProfileNotFoundError
from ..user_profile.profile_record_repository import ProfileRecordRepository
from ..user_profile.projections import projection_for_taxpayer
with bundled_indexed_authority().operation() as operation:
try:
profile_decode_context = operation.profile_decode_context()
repository = ProfileRecordRepository.for_current_session(
bucket_id,
profile_decode_context=profile_decode_context,
)
record = repository.load(bucket_id)
except ProfileNotFoundError:
return frozenset[ThirdPartyDeclarationRole]()
return projection_for_taxpayer(record, schema=profile_decode_context.schema).declaration_roles
def _m347_invoice_observation(invoice: Invoice, *, context: CalculationSourceContext) -> InvoiceObservation | None:
"""Build the M347 observation for one invoice, or ``None`` if excluded.
Declares a counterparty regardless of residency: RD 1065/2007 art. 33.2 is
a CLOSED exclusion list, and a counterparty's non-residency is not one of
its nine enumerated items. The diseño de registro's own `pais-codigo`
field (a "XX" alphabetic slot for a non-established non-resident
declarado) is direct evidence AEAT expects some M347 counterparties to be
non-resident.
The one residency-shaped exclusion the article DOES state is art.
33.2.i): an operation already reported through a coincident periodic
informativa. For an invoice, that informativa is Modelo 349's
intracommunity recapitulativa, so an operation `_intracommunity_clave`
classifies as intracommunity is excluded here and routes to M349 instead
-- the same classification M349's own branch of this resolver uses, never
a bare country comparison.
Clave C additionally needs the filer's own
:class:`ThirdPartyDeclarationRole` membership, loaded here via
``context.bucket_id``. When the invoice IS a clave-C collection
(``collected_on_behalf_of_tax_id`` set AND the filer carries the
registry-selected collector role), the declared counterparty is the
BENEFICIARY whose fees were collected (RD 1065/2007 art. 34.g), not
whoever actually paid this invoice -- so ``party_tax_id`` and
``party_legal_name`` are substituted, not merely the clave.
"""
if _intracommunity_clave(invoice) is not None:
return None
if invoice.counterparty_tax_id is None:
# Same reason as the general builder above: M347 declares a third party
# by their tax id, and a factura simplificada legitimately carries none
# (RD 1619/2012 art. 6.1.d). Without this the row reached the observation
# constructor with None and raised there instead of being skipped.
return None
source_kind = BindingSourceKind(_invoice_source_kind(invoice))
declaration_roles = _m347_filer_declaration_roles(context.bucket_id)
clave = _m347_operation_clave(invoice, source_kind=source_kind, declaration_roles=declaration_roles)
is_third_party_collection = clave == "C"
party_tax_id = invoice.collected_on_behalf_of_tax_id if is_third_party_collection else invoice.counterparty_tax_id
party_legal_name = invoice.collected_on_behalf_of_name if is_third_party_collection else invoice.counterparty_name
if party_tax_id is None:
raise RegistryValidationError(
f"invoice {invoice.invoice_id!r} resolves modelo 347 clave {clave!r} with no declaring party tax id",
)
return InvoiceObservation(
invoice_id=invoice.invoice_id,
source_kind=source_kind,
party_tax_id=party_tax_id,
country_code=invoice.counterparty_country,
transaction_date=invoice.issued_at,
base_amount=_eur(invoice.base_total_eur, invoice),
invoice_total_amount=_eur(invoice.grand_total_eur, invoice),
intracommunity_clave=None,
operation_clave=clave,
party_legal_name=party_legal_name,
)
def _m347_role_operation_clave(
invoice: Invoice,
*,
declaration_roles: frozenset[ThirdPartyDeclarationRole],
) -> str | None:
role_catalogue = resolve_third_party_declaration_role_catalogue()
if invoice.collected_on_behalf_of_tax_id is not None and declaration_roles & role_catalogue.roles_for_clave("C"):
return "C"
if invoice.outside_economic_activity is True and declaration_roles & role_catalogue.roles_for_clave("D"):
return "D"
if invoice.is_subvencion_ayuda is True and declaration_roles & role_catalogue.roles_for_clave("E"):
return "E"
return None
def _m347_mediation_operation_clave(invoice: Invoice) -> str | None:
mediation = invoice.travel_agency_mediation
if mediation is not None and invoice.kind is InvoiceKind.ISSUED:
return "F"
if (
mediation is not None
and invoice.kind is InvoiceKind.RECEIVED
and is_travel_agency_air_passenger_transport(
mediation,
effective_date=invoice.issued_at,
)
):
return "G"
return None
def _m347_operation_clave(
invoice: Invoice,
*,
source_kind: BindingSourceKind,
declaration_roles: frozenset[ThirdPartyDeclarationRole] = frozenset(),
) -> str | None:
"""Classify the M347 clave de operacion for one invoice, or ``None``.
Checks clave C first (RD 1065/2007 art. 31.3: the filer collects this
amount on behalf of a socio, asociado or colegiado), then clave D (arts.
31.1's last paragraph / 31.2: an acquisition al margen de la actividad
empresarial by one of four disjoint filer roles), then clave E (art.
31.2's second paragraph: a subvención/ayuda from a public-administration
filer), then the RD 1619/2012 disposición adicional cuarta travel-agency
mediation fact (claves F/G), then falls back to
:func:`m347_operation_clave`'s invoice-direction classification (claves
A/B).
Claves C, D and E all require the SAME conjunction shape: a filer-level
role membership AND a transaction-level fact, NEITHER alone sufficient.
A collecting entity's ordinary sale is not clave C; a D-role filer's
acquisition WITHIN its own economic activity is not clave D; an ordinary
payment from a public administration is not clave E. Every fact is
checked with ``is True`` rather than truthiness, so an undeclared
tri-state ``None`` never silently classifies either way -- it falls
through here and is surfaced as an advisory by the caller instead.
"""
role_clave = _m347_role_operation_clave(invoice, declaration_roles=declaration_roles)
if role_clave is not None:
return role_clave
mediation_clave = _m347_mediation_operation_clave(invoice)
if mediation_clave is not None:
return mediation_clave
return m347_operation_clave(source_kind)
def _intracommunity_clave(invoice: Invoice) -> str | None:
"""Return the Modelo 349 clave de operación for one invoice, or ``None``.
:class:`~cadrumo.core.aggregation.IntracomOperationType` is the clave authority -- its
member VALUES are the letters the diseño de registro defines, which is why
the explicit branch below returns the value directly rather than mapping it.
The category branches are a fallback for an invoice that carries no
operation type, which is the common case: the field is optional on every
creation path.
That fallback reaches five of the ten claves, and the five it omits are
omitted for two different reasons worth separating, because the gap reads
as an oversight otherwise.
R, D and C are unreachable here BY CONSTRUCTION, not by omission. They are
the call-off stock claves -- a transfer of goods under a consignment sales
arrangement, a return of those goods, and a substitution of the intended
acquirer. LIVA art. 9 bis.Dos places the entrega, and its art. 25 exemption,
at the moment the acquirer takes the power of disposal, which is later than
and separate from the movement those three claves report. The movement
transfers no ownership and so carries no invoice at all; art. 9 bis.Uno.d
has the vendor declare the despatch through the libro registro and the
declaración recapitulativa precisely because there is no supply yet to
invoice. An invoice-sourced path therefore cannot produce them, and no
predicate added here would help -- they need a non-invoice record source.
M and H are a real limit rather than a scope boundary. Both are ordinary
invoiced entregas intracomunitarias that happen to FOLLOW an exempt
importation (LIVA art. 27.12, H being the variant made by a representante
fiscal under art. 86.Tres), so they carry the same intra-community supply
category as an art. 25 supply and the fallback cannot tell them apart --
the distinguishing fact is the prior importation, which appears nowhere on
the invoice except in the operation type itself. The diseño defines E as
excluding exactly these, directing them to M or H, so an operator with such
a supply MUST set the operation type. The fallback's E is correct for the
ordinary case and wrong for this one, and refusing instead is not the
remedy: no predicate here separates the two, so a refusal falls on the whole
category and makes the ordinary supply undeclarable -- measured, six
otherwise-passing M349 flows.
"""
operation_type = invoice.operation_type
if operation_type is not None:
return _m349_clave_for_operation_type(
invoice_id=invoice.invoice_id,
source_kind=BindingSourceKind(_invoice_source_kind(invoice)),
operation_type=operation_type,
record_label="catalogue invoice",
)
# Triangulation first, and kind-independent: LIVA art. 26.3 exempts the
# intermediary's adquisición while the onward leg is a supply, so the
# taxpayer files clave T from either side of the operation.
if invoice.iva_category == require_iva_category("intra_community_triangulation"):
return IntracomOperationType.T.value
if invoice.iva_category is None:
return None
derived = _clave_by_kind_and_category().get((invoice.kind, invoice.iva_category))
return None if derived is None else derived.value
def _m349_clave_for_operation_type(
*,
invoice_id: str,
source_kind: BindingSourceKind,
operation_type: IntracomOperationType,
record_label: str,
) -> str:
allowed = (
_COLLECTIBLE_M349_OPERATION_TYPES
if source_kind is BindingSourceKind.COLLECTIBLE_INVOICE
else _PAYABLE_M349_OPERATION_TYPES
)
if operation_type not in allowed:
accepted = ", ".join(item.value for item in sorted(allowed, key=lambda item: item.value))
raise RegistryValidationError(
f"{record_label} {invoice_id!r} uses operation type {operation_type.value!r} "
f"with source kind {source_kind.value!r}; accepted: {accepted}",
)
return operation_type.value
def _m349_declarante_summary_union(
*,
context: CalculationSourceContext,
binding_values: dict[str, Decimal],
) -> dict[str, Decimal]:
if context.modelo != Modelo("349").value:
return binding_values
merged = dict(binding_values)
for payable_binding, public_binding in _M349_PAYABLE_SUMMARY_BINDING_MIRRORS.items():
if payable_binding not in binding_values:
continue
merged[public_binding] = merged.get(public_binding, Decimal("0")) + binding_values[payable_binding]
return merged
def _m349_operador_row_indexes(row_values: Mapping[tuple[BindingId, int], Decimal | str]) -> list[int]:
return sorted(
{row_index for binding_id, row_index in row_values if binding_id in _M349_OPERADOR_ROW_BINDINGS},
)
def _m349_operador_row_values(
row_values: Mapping[tuple[BindingId, int], Decimal | str],
*,
row_index: int,
) -> dict[str, Decimal | str]:
return {
attr: row_values[(binding_id, row_index)]
for binding_id, attr in _M349_OPERADOR_ROW_BINDINGS.items()
if (binding_id, row_index) in row_values
}
def _m349_operador_row_from_values(
values: Mapping[str, Decimal | str],
*,
row_index: int,
) -> Modelo349OperadorRow:
if set(values) != set(_M349_OPERADOR_ROW_BINDINGS.values()):
raise RegistryValidationError(f"Modelo 349 invoice row {row_index} is incomplete")
codigo_pais = values["codigo_pais"]
nif_comunitario = values["nif_comunitario"]
razon_social = values["razon_social"]
clave_operacion = values["clave_operacion"]
importe = values["importe"]
if not (
isinstance(codigo_pais, str)
and isinstance(nif_comunitario, str)
and isinstance(razon_social, str)
and isinstance(clave_operacion, str)
and isinstance(importe, Decimal)
):
raise RegistryValidationError(f"Modelo 349 invoice row {row_index} has invalid field types")
try:
return Modelo349OperadorRow.model_validate(
{
"codigo_pais": codigo_pais,
"nif_comunitario": f"{codigo_pais}{nif_comunitario}",
"razon_social": razon_social,
"clave_operacion": clave_operacion,
"importe": importe,
},
)
except ValueError as exc:
raise RegistryValidationError(str(exc)) from exc
def _m349_operador_rows_from_observations(
*,
context: CalculationSourceContext,
observations: tuple[InvoiceObservation, ...],
) -> tuple[Modelo349OperadorRow, ...]:
if context.modelo != Modelo("349").value or not observations:
return ()
row_values = resolve_invoice_binding_row_values(
context.revision,
observations,
effective_date=date(context.filing_year, 12, 31),
)
return tuple(
_m349_operador_row_from_values(
_m349_operador_row_values(row_values, row_index=row_index),
row_index=row_index,
)
for row_index in _m349_operador_row_indexes(row_values)
)
def _invoice_provenance(invoice: Invoice, observation: InvoiceObservation) -> CalculationSourceProvenance:
payload = observation.model_dump_json()
source_kind = _invoice_source_kind(invoice)
return CalculationSourceProvenance(
resolver_id=InvoiceCatalogueSourceResolver.resolver_id,
resolved_binding_source=BindingSourceKind(source_kind),
contributor_source_kind=source_kind,
contributor_binding_source=BindingSourceKind(source_kind),
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=f"{source_kind}:{observation.invoice_id}",
parent_source_ref=None,
terminal_origin=TerminalOriginClass.INVOICE_CATALOGUE,
fingerprint=f"sha256:{sha256_hex(payload.encode('utf-8'))}",
)
#: The forward reading of the same clave<->category relationship the inverse
#: map above expresses: what IVA treatment an operator has stated by choosing a
#: Modelo 349 clave. It lives beside its inverse deliberately. The two were
#: declared in different layers, agreeing only because both were maintained by
#: hand, and a clave added to one would have been invisible to the other.
#:
#: ``T`` appears here but not in the inverse map, and that asymmetry is real
#: rather than an omission: triangulation is filed from either side of the
#: operation, so it carries no kind and the kind-keyed inverse cannot express
#: it. It is special-cased ahead of that lookup instead.
[docs]
def iva_category_for_operation_type(operation_type: IntracomOperationType | None) -> IvaCategory | None:
"""Return the IVA treatment an operator stated by choosing a 349 clave.
Args:
operation_type: The clave the operator selected, if any.
Returns:
The category that clave declares, or ``None`` when no clave was chosen.
A clave outside the invoice-sourced set also yields ``None`` rather than
a guess: the claves reporting call-off stock movements carry no invoice,
so no category can be inferred from them.
"""
if operation_type is None:
return None
return resolve_iva_category_catalogue().category_for_operation_type(operation_type.value)
__all__ = [
"InvoiceCatalogueSourceResolver",
"invoice_direction_to_source_kind",
"iva_category_for_operation_type",
]