"""Modelo 347/349 counterpart aggregation for informativa declarations.
Used by :mod:`~application.aggregation.service`, the per-modelo
aggregation service. This is an operator preview surface, not the live
calculation source mesh: standalone ``CounterpartObservation`` rows are
operator-supplied, while calculation uses enrolled repository-backed sources.
Modelo 347 covers annual operations with third parties whose total exceeds the
declaration floor. Modelo 349 covers intra-EU operations by member-state
operation kind. Both aggregate per ``(source_kind, counterparty_nif,
operation_kind)`` using the counterpart subset of the canonical source-kind
taxonomy.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from pydantic import BaseModel, Field, InstanceOf, NonNegativeInt, field_validator, model_validator
from ...core.aggregation import (
CounterpartSourceKind,
counterpart_source_kind,
)
from ...core.country_code import CountryCodeAlpha2
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.modelo import Modelo
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.parsing.dates import IsoDateString
from ...core.period import FilingPeriodCode, Period
from ...domain.calculations.registry.authority import PinnedAuthorityOperation, bundled_indexed_authority
from ...domain.calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ...domain.calculations.registry.schema_base import DateAxis
from ._grouping import assert_rollup_totals_match, filter_observations_for_modelo, group_and_collect_names
def _validate_source_kind(value: str) -> CounterpartSourceKind:
return counterpart_source_kind(value)
@dataclass(frozen=True, slots=True)
class _CounterpartRegistryCatalogue:
"""Generic counterpart projection of the selected registry mapping."""
model_kinds: Mapping[str, frozenset[str]]
groi_countries: frozenset[str]
nif_iva_for_non_groi_country: bool
@property
def operation_kinds(self) -> frozenset[str]:
"""Return the union used by the operator-boundary vocabulary check."""
operation_kinds: set[str] = set()
for model_kinds in self.model_kinds.values():
operation_kinds.update(model_kinds)
return frozenset(operation_kinds)
[docs]
def counterpart_operation_catalogue_entries(
effective_date: date,
*,
operation: PinnedAuthorityOperation,
) -> dict[str, str]:
"""Return the dated M347/M349 counterpart operation mapping as unique string pairs."""
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id="m347-m349-counterpart-operation-catalogue",
date_axis=DateAxis.FILING_PERIOD,
effective_date=effective_date,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise TypeError("counterpart operation declarations must resolve as a mapping fact")
entries: dict[str, str] = {}
for entry in resolved.payload.entries:
if not isinstance(entry.key, str) or not isinstance(entry.value, str):
raise TypeError("counterpart operation mapping entries must be string-to-string")
if entry.key in entries:
raise ValueError(f"duplicate counterpart registry mapping key {entry.key!r}")
entries[entry.key] = entry.value
return entries
# Registry authority: selected M347/M349 counterpart declarations are consumed through
# one generation-pinned operation and the dated mapping fact.
def _registry_counterpart_catalogue(
effective_date: date,
*,
operation: PinnedAuthorityOperation,
) -> _CounterpartRegistryCatalogue:
"""Resolve counterpart kinds and readiness gates from registry authority."""
for modelo in (Modelo("347").value, Modelo("349").value):
directory = operation.modelo_directory(modelo)
if not any(metadata.contains_date(effective_date) for metadata in directory.revisions):
raise ValueError(f"{modelo} has no registry revision in force on {effective_date.isoformat()}")
entries = counterpart_operation_catalogue_entries(effective_date, operation=operation)
def required(key: str) -> str:
value = entries.get(key)
if value is None or not value.strip():
raise ValueError(f"counterpart registry mapping is missing {key!r}")
return value
def tokens(key: str) -> frozenset[str]:
result = frozenset(token.strip() for token in required(key).split(",") if token.strip())
if not result:
raise ValueError(f"counterpart registry mapping {key!r} has no tokens")
return result
readiness = required("modelo.349.readiness.nif_iva_for_non_groi_country").casefold()
if readiness not in {"true", "false"}:
raise ValueError("counterpart registry readiness flag must be true or false")
return _CounterpartRegistryCatalogue(
model_kinds={
Modelo("347").value: tokens("modelo.347.operation_kinds"),
Modelo("349").value: tokens("modelo.349.operation_kinds"),
},
groi_countries=frozenset({required("modelo.349.readiness.groi_country").strip()}),
nif_iva_for_non_groi_country=readiness == "true",
)
def _validate_operation_kind(value: str) -> str:
"""Validate the operation token's boundary shape.
Registry membership is checked at the aggregation boundary, where the
caller supplies the pinned operation. Pydantic construction has no safe
authority-generation input and must not reopen a global catalogue.
"""
if not value.strip():
raise ValueError("operation_kind must not be blank")
return value
def _validate_country(value: str, *, field_name: str) -> str:
if len(value) != 2 or any(char < "A" or char > "Z" for char in value):
raise ValueError(f"{field_name} must be uppercase ISO-3166 alpha-2, got {value!r}")
return value
class _CounterpartBoundaryModel(BaseModel):
"""Shared Pydantic boundary validation for counterpart records."""
model_config = STRICT_FROZEN_CONFIG
@field_validator("source_kind", mode="before", check_fields=False)
@classmethod
@pydantic_validation_boundary
def _source_kind_is_canonical(cls, value: object) -> CounterpartSourceKind:
if not isinstance(value, str):
raise ValueError("source_kind must be a string")
return _validate_source_kind(value)
@field_validator("operation_kind", check_fields=False)
@classmethod
@pydantic_validation_boundary
def _operation_kind_is_canonical(cls, value: str) -> str:
return _validate_operation_kind(value)
@field_validator("counterparty_country", check_fields=False)
@classmethod
@pydantic_validation_boundary
def _country_is_uppercase(cls, value: str) -> str:
return _validate_country(value, field_name="counterparty_country")
[docs]
class CounterpartObservation(_CounterpartBoundaryModel):
"""One typed observation for a 347 or 349 aggregator pass.
This model is the operator boundary: ``aeat app modelo aggregate`` validates
each ``--counterpart-observation`` JSON object directly against it, so
whatever it admits reaches the preview rollups unchecked.
``accrued_on`` and ``operation_period`` are therefore admitted by the same
authorities the rest of the system uses. Bounding them by string length
alone let an impossible calendar date (``2026-99-99`` and ``2026-02-30`` are
both ten characters) and an arbitrary period token into a rollup the
operator reads as a preview of a real declaration, while the adjacent
registry counterpart binding types the same concepts as a
:class:`~datetime.date` and a registry period code.
"""
source_kind: CounterpartSourceKind
source_object_id: str = Field(min_length=1)
counterparty_nif: str = Field(min_length=1, max_length=20)
counterparty_name: str = Field(default="", max_length=200)
# Required, and deliberately not defaulted to Spain. This is an operator
# boundary, so a default is an INFERENCE about a fact the operator did not
# state -- and it was the one inference the Modelo 349 readiness rule turns
# on. That rule asks for a GROI check when the country is Spain and a
# NIF-IVA check when it is not, so a row omitting the country was read as
# domestic and the NIF-IVA verification an intra-community counterparty
# must pass was never required of it. The declaration is the recapitulativa
# de operaciones INTRACOMUNITARIAS, where a Spanish counterparty is the one
# thing the row cannot be.
#
# Refusing is right rather than admitting an absent value, because every
# consumer here has to branch on the country: an optional field would move
# the same guess into each of them, and the shape of the mistake would
# survive the fix.
counterparty_country: CountryCodeAlpha2
operation_kind: str = Field(min_length=1)
operation_period: FilingPeriodCode
taxable_base: Decimal = Field(ge=Decimal("0"))
invoice_total: Decimal = Field(ge=Decimal("0"))
accrued_on: IsoDateString = Field(min_length=10, max_length=10)
groi_verified: bool = False
nif_iva_verified: bool = False
[docs]
class CounterpartRollup(_CounterpartBoundaryModel):
"""One (source_kind, counterparty_nif, operation_kind) rollup row."""
source_kind: CounterpartSourceKind
counterparty_nif: str = Field(min_length=1, max_length=20)
counterparty_name: str = Field(default="", max_length=200)
counterparty_country: CountryCodeAlpha2
operation_kind: str = Field(min_length=1)
observations_count: NonNegativeInt
total_taxable_base: Decimal = Field(ge=Decimal("0"))
total_invoice_total: Decimal = Field(ge=Decimal("0"))
requires_groi_check: bool = False
requires_nif_iva_check: bool = False
groi_ready: bool = True
nif_iva_ready: bool = True
declarable_readiness_satisfied: bool = True
[docs]
class CounterpartAggregation(BaseModel):
"""347 / 349 counterpart aggregation output."""
model_config = STRICT_FROZEN_CONFIG
modelo: str = Field(min_length=1)
period: InstanceOf[Period]
rollups: tuple[CounterpartRollup, ...] = Field(default_factory=tuple)
total_counterparties: NonNegativeInt
total_taxable_base: Decimal = Field(ge=Decimal("0"))
total_invoice_total: Decimal = Field(ge=Decimal("0"))
@model_validator(mode="after")
@pydantic_validation_boundary
def _totals_match_rollups(self) -> CounterpartAggregation:
assert_rollup_totals_match(
self.rollups,
checks=(
("total_taxable_base", self.total_taxable_base, lambda row: row.total_taxable_base),
("total_invoice_total", self.total_invoice_total, lambda row: row.total_invoice_total),
),
)
unique_counterparties = {row.counterparty_nif for row in self.rollups}
if len(unique_counterparties) != self.total_counterparties:
raise ValueError(
f"total_counterparties {self.total_counterparties} != distinct NIFs {len(unique_counterparties)}",
)
return self
def _aggregate_for_modelo(
observations: tuple[CounterpartObservation, ...],
*,
modelo: str,
period: Period,
operation: PinnedAuthorityOperation,
) -> CounterpartAggregation:
registry_catalogue = _registry_counterpart_catalogue(period.end_date, operation=operation)
# A clave of the other modelo is filtered below; only a token in neither
# vocabulary would silently vanish from both passes.
undeclared = sorted(
{observation.operation_kind for observation in observations} - registry_catalogue.operation_kinds
)
if undeclared:
raise ValueError(
f"operation_kind is not a declared 347/349 clave: {', '.join(undeclared)}",
)
filtered = filter_observations_for_modelo(
observations,
modelo=modelo,
catalogue=registry_catalogue.model_kinds,
attribute_fn=lambda obs: obs.operation_kind,
aggregator_label="counterpart aggregator",
)
grouped, names = group_and_collect_names(
filtered,
group_key_fn=lambda obs: (obs.source_kind, obs.counterparty_nif, obs.operation_kind),
identity_key_fn=lambda obs: (obs.source_kind, obs.counterparty_nif),
name_fn=lambda obs: obs.counterparty_name,
)
rollups: list[CounterpartRollup] = []
for (source_kind, nif, op_kind), group in sorted(grouped.items(), key=lambda kv: (kv[0][0], kv[0][1], kv[0][2])):
total_base = sum((g.taxable_base for g in group), Decimal("0"))
total_invoice = sum((g.invoice_total for g in group), Decimal("0"))
country = _single_counterparty_country(
observations=tuple(group),
source_kind=source_kind,
counterparty_nif=nif,
operation_kind=op_kind,
)
readiness = _counterpart_readiness_for_modelo(
modelo=modelo,
country=country,
observations=tuple(group),
registry_catalogue=registry_catalogue,
)
rollups.append(
CounterpartRollup(
source_kind=source_kind,
counterparty_nif=nif,
counterparty_name=names.get((source_kind, nif), ""),
counterparty_country=country,
operation_kind=op_kind,
observations_count=len(group),
total_taxable_base=total_base,
total_invoice_total=total_invoice,
**readiness,
),
)
counterparties = {row.counterparty_nif for row in rollups}
return CounterpartAggregation(
modelo=modelo,
period=period,
rollups=tuple(rollups),
total_counterparties=len(counterparties),
total_taxable_base=sum((row.total_taxable_base for row in rollups), Decimal("0")),
total_invoice_total=sum((row.total_invoice_total for row in rollups), Decimal("0")),
)
def _single_counterparty_country(
*,
observations: tuple[CounterpartObservation, ...],
source_kind: str,
counterparty_nif: str,
operation_kind: str,
) -> str:
countries = frozenset(observation.counterparty_country for observation in observations)
if len(countries) != 1:
raise ValueError(
"conflicting counterparty_country values for counterpart aggregation cohort "
f"source_kind={source_kind!r}, counterparty_nif={counterparty_nif!r}, "
f"operation_kind={operation_kind!r}: {sorted(countries)!r}",
)
return next(iter(countries))
[docs]
def aggregate_counterpart_347(
observations: tuple[CounterpartObservation, ...],
*,
period: Period,
operation: PinnedAuthorityOperation | None = None,
) -> CounterpartAggregation:
"""Aggregate Modelo 347 (operaciones con terceros, annual).
Filters to 347 operation kinds. Threshold gating (the €3,005.06
declaration floor) belongs to the modelo binding consumer; this
aggregator returns raw per-counterparty totals.
Returns a :class:`CounterpartAggregation`.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_counterpart_347(observations, period=period, operation=indexed_operation)
return _aggregate_for_modelo(
observations,
modelo=Modelo("347").value,
period=period,
operation=operation,
)
[docs]
def aggregate_counterpart_349(
observations: tuple[CounterpartObservation, ...],
*,
period: Period,
operation: PinnedAuthorityOperation | None = None,
) -> CounterpartAggregation:
"""Aggregate Modelo 349 (operaciones intracomunitarias).
Filters to 349 intracomunitarias operation kinds. Rollups carry
the additional NIF-IVA / GROI readiness gates: Spanish
counterparties require GROI readiness and non-Spanish
counterparties require NIF-IVA readiness.
Returns a :class:`CounterpartAggregation` with rollups sorted by
``(source_kind, counterparty_nif, operation_kind)``.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_counterpart_349(observations, period=period, operation=indexed_operation)
return _aggregate_for_modelo(
observations,
modelo=Modelo("349").value,
period=period,
operation=operation,
)
def _counterpart_readiness_for_modelo(
*,
modelo: str,
country: str,
observations: tuple[CounterpartObservation, ...],
registry_catalogue: _CounterpartRegistryCatalogue,
) -> dict[str, bool]:
if modelo != Modelo("349").value:
return {
"requires_groi_check": False,
"requires_nif_iva_check": False,
"groi_ready": True,
"nif_iva_ready": True,
"declarable_readiness_satisfied": True,
}
requires_groi = country in registry_catalogue.groi_countries
requires_nif_iva = (
registry_catalogue.nif_iva_for_non_groi_country and country not in registry_catalogue.groi_countries
)
groi_ready = (not requires_groi) or all(obs.groi_verified for obs in observations)
nif_iva_ready = (not requires_nif_iva) or all(obs.nif_iva_verified for obs in observations)
return {
"requires_groi_check": requires_groi,
"requires_nif_iva_check": requires_nif_iva,
"groi_ready": groi_ready,
"nif_iva_ready": nif_iva_ready,
"declarable_readiness_satisfied": groi_ready and nif_iva_ready,
}
__all__ = [
"CounterpartAggregation",
"CounterpartObservation",
"CounterpartRollup",
"aggregate_counterpart_347",
"aggregate_counterpart_349",
"counterpart_operation_catalogue_entries",
]