"""Detail-record row-set registry binding helpers."""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from datetime import date
from decimal import Decimal
from typing import TYPE_CHECKING, Annotated, Literal
from pydantic import BaseModel, BeforeValidator, Field, field_validator
from ....core.aggregation import BindingAggregationOp, BindingSourceKind
from ....core.country_code import CountryCodeAlpha2
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.external_constants import DEFAULT_CURRENCY
from ....core.foreign_asset_obligation import M720AssetClassCode
from ....core.identity.tax_id import TaxIdIdentityToken
from ....core.modelo_232_codigos import MetodoValoracion, TipoOperacionVinculada
from ....core.models import STRICT_FROZEN_CONFIG
from .binding_aggregation import binding_aggregation_op
from .binding_selector_utils import (
invariant_diagnostics,
optional_uppercase_alpha_code,
provider_member,
selector_against_model,
uppercase_alpha_code,
)
from .errors import RegistryValidationError
from .ids import BindingId
from .schema_exports import ExportFieldDataType
if TYPE_CHECKING:
from .schema import BindingDefinition, ModeloRevision
__all__ = [
"AtributionMemberObservation",
"Modelo720RowObservation",
"RefundOperationObservation",
"RelatedPartyOperationObservation",
"resolve_atribucion_binding_row_values",
"resolve_foreign_asset_binding_row_values",
"validate_atribucion_binding",
"validate_foreign_asset_binding",
"validate_refund_binding",
"validate_related_party_binding",
]
def _required_detail_record_row_field[RowFieldT](
binding: BindingDefinition,
row_field: RowFieldT | None,
) -> RowFieldT:
"""Return the selector row field, refusing a detail-record binding that names none."""
if row_field is None:
raise RegistryValidationError(f"binding {binding.id!r} fact 'row_field' requires a 'row_field' selector key")
return row_field
def _validate_detail_record_row_field(
binding: BindingDefinition,
selector_fact: object,
selector_row_field: object,
family_label: str,
) -> None:
"""Shared op/fact invariant for the four detail-record families.
Every detail-record family declares exactly the ``row_field`` fact, defaults
to (and requires) the ``rows`` aggregation op, and must name a ``row_field``
selector key. The four families enforced this with byte-identical bodies; the
one shared check raises a family-labelled :class:`RegistryValidationError`.
"""
if selector_fact != "row_field":
raise RegistryValidationError(
f"binding {binding.id!r} declares unsupported {family_label} fact {selector_fact!r}",
)
if binding_aggregation_op(binding) != BindingAggregationOp.ROWS:
raise RegistryValidationError(f"binding {binding.id!r} fact 'row_field' requires aggregation op 'rows'")
if selector_row_field is None:
raise RegistryValidationError(f"binding {binding.id!r} fact 'row_field' requires a 'row_field' selector key")
# Related-party operation source bindings (modelo 232).
#
# Legal authority: LIS art. 18 (operaciones vinculadas), RD 634/2015
# art. 13 (informe-pa�s-por-pa�s y declaraci�n modelo 232), Orden
# HFP/816/2017 Anexo (diseno de registro modelo 232).
# ---------------------------------------------------------------------------
_RelatedPartyRowField = Literal[
"counterparty_tax_id",
"counterparty_legal_name",
"country_code",
"operation_kind_code",
"transfer_pricing_method_code",
"amount",
]
def _hydrate_operation_kind_code(value: object) -> object:
"""Hydrate a resolved binding value into its typed ``TipoOperacionVinculada`` member.
Binding values arrive from the registry as free-form text, so this is the
boundary that turns a token into a member. It is the same code set the
operator-supplied CLI row carries, which is why both read it from ``core``
rather than either side re-spelling the table.
"""
if not isinstance(value, str):
return value
try:
return TipoOperacionVinculada(value.upper())
except ValueError:
accepted = ", ".join(repr(str(member)) for member in TipoOperacionVinculada)
raise ValueError(f"operation_kind_code must be one of {accepted}; got {value!r}") from None
def _hydrate_transfer_pricing_method_code(value: object) -> object:
"""Hydrate a resolved binding value into its typed ``MetodoValoracion`` member."""
if not isinstance(value, str):
return value
try:
return MetodoValoracion(value.upper())
except ValueError:
accepted = ", ".join(repr(str(member)) for member in MetodoValoracion)
raise ValueError(f"transfer_pricing_method_code must be one of {accepted}; got {value!r}") from None
def _validated_related_party_selector(binding: BindingDefinition) -> RelatedPartyOperationProvider:
try:
selector = provider_member(binding, RelatedPartyOperationProvider)
except ValueError as exc:
raise RegistryValidationError(f"binding {binding.id!r} has malformed related-party selector") from exc
_validate_detail_record_row_field(binding, selector.fact, selector.row_field, "related-party")
return selector
# ---------------------------------------------------------------------------
# Foreign asset source bindings (modelo 720).
#
# Legal authority: RD 1065/2007 arts. 42 bis / 42 ter, Orden HAP/72/2013
# Anexo (modelo 720 diseno de registro). Threshold: 50,000 EUR per regulatory
# obligation block (already encoded as a parameter on modelo 720).
# ---------------------------------------------------------------------------
_ForeignAssetRowField = Literal[
"asset_class_code",
"country_code",
"currency_code",
"asset_identifier",
"valuation_amount",
"acquisition_date",
]
[docs]
class Modelo720RowObservation(BaseModel):
"""One foreign asset for modelo 720."""
model_config = STRICT_FROZEN_CONFIG
source_id: str = Field(min_length=1, max_length=128)
asset_class_code: M720AssetClassCode
country_code: CountryCodeAlpha2
currency_code: str = Field(default=DEFAULT_CURRENCY, min_length=3, max_length=3)
asset_identifier: str = Field(default="", max_length=128)
acquisition_date: date
valuation_amount: Decimal
_iso_code_uppercase = field_validator("country_code", "currency_code")(uppercase_alpha_code("ISO code"))
@field_validator("valuation_amount")
@classmethod
@pydantic_validation_boundary
def _decimal_amount(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise RegistryValidationError("foreign asset valuation must be non-negative")
return value
[docs]
class ForeignAssetProvider(BaseModel):
model_config = STRICT_FROZEN_CONFIG
kind: Literal[BindingSourceKind.FOREIGN_ASSET] = BindingSourceKind.FOREIGN_ASSET
fact: Literal["row_field"]
row_field: _ForeignAssetRowField | None = None
asset_classes: tuple[str, ...] = ()
grouping: str | None = Field(default=None, min_length=1, max_length=64)
record: str | None = Field(default=None, min_length=1, max_length=64)
data_type: ExportFieldDataType | None = None
"""Scalar type of the value this row field contributes to the export.
The same fact ``BindingRowExportSelector.data_type`` carries; declared here
so the selector model admits the key, since a source-family selector is
validated whole against its own strict model. Optional while the families
adopt it.
"""
def _validated_foreign_asset_selector(binding: BindingDefinition) -> ForeignAssetProvider:
try:
selector = provider_member(binding, ForeignAssetProvider)
except ValueError as exc:
raise RegistryValidationError(f"binding {binding.id!r} has malformed foreign-asset selector") from exc
_validate_detail_record_row_field(binding, selector.fact, selector.row_field, "foreign-asset")
return selector
[docs]
def validate_foreign_asset_binding(binding: BindingDefinition) -> list[str]:
"""Validate a foreign-asset binding at registry-build time.
Accumulating ``list[str]`` validator: validates the selector against
:class:`ForeignAssetProvider` and lifts the resolve-time op/fact invariant
to build time, preserving the underlying pydantic field error.
"""
failures = selector_against_model(binding, ForeignAssetProvider)
if failures:
return failures
return invariant_diagnostics(binding, "foreign-asset", lambda b: _validated_foreign_asset_selector(b))
def _foreign_asset_binding_members(
revision: ModeloRevision,
) -> tuple[list[tuple[BindingDefinition, ForeignAssetProvider]], set[tuple[str, ...]]]:
"""Collect foreign-asset row bindings and their declared class cohorts."""
members: list[tuple[BindingDefinition, ForeignAssetProvider]] = []
cohort_classes: set[tuple[str, ...]] = set()
for binding in revision.bindings:
if binding.source != BindingSourceKind.FOREIGN_ASSET:
continue
selector = _validated_foreign_asset_selector(binding)
members.append((binding, selector))
cohort_classes.add(tuple(sorted(selector.asset_classes)))
return members, cohort_classes
def _resolve_foreign_asset_rows(
members: list[tuple[BindingDefinition, ForeignAssetProvider]],
rows: tuple[Mapping[str, Decimal | str], ...],
) -> dict[tuple[BindingId, int], Decimal | str]:
"""Project each foreign-asset row field into its binding/index coordinates."""
resolved: dict[tuple[BindingId, int], Decimal | str] = {}
for binding, selector in members:
row_field = _required_detail_record_row_field(binding, selector.row_field)
for row_index, row in enumerate(rows, start=1):
value = row.get(row_field)
if value is None:
raise RegistryValidationError(
f"binding {binding.id!r} row_field {row_field!r} not produced for foreign-asset rows",
)
resolved[(binding.id, row_index)] = value
return resolved
[docs]
def resolve_foreign_asset_binding_row_values(
revision: ModeloRevision,
observations: Iterable[Modelo720RowObservation],
) -> dict[tuple[BindingId, int], Decimal | str]:
"""Resolve row-producer foreign-asset bindings into per-row indexed values.
Args:
revision: The :class:`ModeloRevision` whose foreign-asset bindings are resolved.
observations: Modelo 720 row observations to group into rows.
"""
available = tuple(observations)
members, cohort_classes = _foreign_asset_binding_members(revision)
if not members:
return {}
# All bindings in a cohort share the same asset_classes filter.
sample_classes = next(iter(cohort_classes)) if cohort_classes else ()
class_filter = set(sample_classes)
filtered = tuple(obs for obs in available if not class_filter or obs.asset_class_code in class_filter)
rows = _build_foreign_asset_rows(filtered)
return _resolve_foreign_asset_rows(members, rows)
def _build_foreign_asset_rows(
observations: tuple[Modelo720RowObservation, ...],
) -> tuple[Mapping[str, Decimal | str], ...]:
rows: list[Mapping[str, Decimal | str]] = []
for obs in sorted(
observations,
key=lambda o: (o.country_code, o.asset_class_code, o.asset_identifier, o.acquisition_date.isoformat()),
):
rows.append(
{
"asset_class_code": obs.asset_class_code,
"country_code": obs.country_code,
"currency_code": obs.currency_code,
"asset_identifier": obs.asset_identifier,
"valuation_amount": obs.valuation_amount,
"acquisition_date": obs.acquisition_date.isoformat(),
},
)
return tuple(rows)
# ---------------------------------------------------------------------------
# Atribuci�n member source bindings (modelo 184).
#
# Legal authority: Ley 35/2006 LIRPF arts. 87-90 (r�gimen de atribuci�n de
# rentas), Orden HFP/227/2017 Anexo (modelo 184 diseno de registro).
# ---------------------------------------------------------------------------
_AtributionRowField = Literal[
"member_tax_id",
"member_legal_name",
"country_code",
"share_percentage",
"base_imponible_assigned",
"clave",
"subclave",
"codigo_provincia",
"miembro_a_31_diciembre",
"dias_miembro",
"domicilio_fiscal",
"naturaleza_inmueble",
"situacion_inmueble",
"referencia_catastral",
"clave_declarado",
"porcentaje_titularidad_inmueble",
"dias_arrendamiento",
"reduccion",
"rendimiento_neto_previo_eo",
"rendimiento_neto_minorado_agricola_eo",
]
[docs]
class AtributionMemberObservation(BaseModel):
"""One (member, clave, subclave) atribuci�n row for modelo 184.
Every field below ``base_imponible_assigned`` is the (member, clave,
subclave) row-shape ADR's clave/subclave-conditional fact set, and is
``None`` whenever the declaring clave does not license it -- e.g.
``naturaleza_inmueble`` is only ever populated for a clave-C row. A
``None`` here is a legitimate "this field does not apply to this row",
never a missing value.
"""
model_config = STRICT_FROZEN_CONFIG
source_id: str = Field(min_length=1, max_length=128)
member_tax_id: TaxIdIdentityToken
member_legal_name: str = Field(default="", max_length=200)
country_code: CountryCodeAlpha2 | None = None
"""The party's country, or ``None`` when the source stated none.
Nullable rather than defaulted to ``ES``, because these forms carry the
NON-RESIDENT population by construction -- a perceptor on a withholding
form is routinely foreign, and an attribution member can be -- so a
default silently declares a foreign party Spanish on a filing surface.
Absence propagates as an ABSENT KEY in the built row rather than as a
value, so a binding that needs the country refuses with the shipped
not-produced error naming itself. That is the visible failure the silent
default replaced."""
transaction_date: date
share_percentage: Decimal
base_imponible_assigned: Decimal
clave: str = Field(min_length=1, max_length=1)
subclave: str | None = Field(default=None, min_length=2, max_length=2)
codigo_provincia: str | None = Field(default=None, min_length=2, max_length=2)
miembro_a_31_diciembre: str | None = Field(default=None, min_length=1, max_length=1)
"""``"X"`` when the member remained one at 31 December, else ``None``.
A string flag rather than a bool: the dise�o's own field (position 82)
is text, marked ``"X"`` or left blank -- never a boolean literal -- and
this is the shape the fixed-width renderer's ``data_type = "text"``
field expects.
"""
dias_miembro: int | None = None
domicilio_fiscal: str | None = Field(default=None, max_length=40)
naturaleza_inmueble: str | None = Field(default=None, min_length=1, max_length=1)
situacion_inmueble: str | None = Field(default=None, min_length=1, max_length=1)
referencia_catastral: str | None = Field(default=None, max_length=20)
clave_declarado: str | None = Field(default=None, min_length=1, max_length=1)
porcentaje_titularidad_inmueble: Decimal | None = None
dias_arrendamiento: int | None = None
reduccion: Decimal | None = None
rendimiento_neto_previo_eo: Decimal | None = None
rendimiento_neto_minorado_agricola_eo: Decimal | None = None
_country_code_uppercase = field_validator("country_code")(optional_uppercase_alpha_code("country_code"))
@field_validator("share_percentage")
@classmethod
@pydantic_validation_boundary
def _share_within_bounds(cls, value: Decimal) -> Decimal:
if value < Decimal("0") or value > Decimal("100"):
raise RegistryValidationError("share_percentage must be within [0, 100]")
return value
@field_validator("base_imponible_assigned")
@classmethod
@pydantic_validation_boundary
def _decimal_amount(cls, value: Decimal) -> Decimal:
return value
[docs]
class AtribucionMemberProvider(BaseModel):
model_config = STRICT_FROZEN_CONFIG
kind: Literal[BindingSourceKind.ATRIBUCION_MEMBER] = BindingSourceKind.ATRIBUCION_MEMBER
fact: Literal["row_field"]
row_field: _AtributionRowField | None = None
grouping: str | None = Field(default=None, min_length=1, max_length=64)
record: str | None = Field(default=None, min_length=1, max_length=64)
data_type: ExportFieldDataType | None = None
"""Scalar type of the value this row field contributes to the export.
The same fact ``BindingRowExportSelector.data_type`` carries; declared here
so the selector model admits the key, since a source-family selector is
validated whole against its own strict model. Optional while the families
adopt it.
"""
def _validated_atribucion_selector(binding: BindingDefinition) -> AtribucionMemberProvider:
try:
selector = provider_member(binding, AtribucionMemberProvider)
except ValueError as exc:
raise RegistryValidationError(f"binding {binding.id!r} has malformed atribucion selector") from exc
_validate_detail_record_row_field(binding, selector.fact, selector.row_field, "atribucion")
return selector
[docs]
def validate_atribucion_binding(binding: BindingDefinition) -> list[str]:
"""Validate an atribuci�n-member binding at registry-build time.
Accumulating ``list[str]`` validator: validates the selector against
:class:`AtribucionMemberProvider` and lifts the resolve-time op/fact invariant to
build time, preserving the underlying pydantic field error.
"""
failures = selector_against_model(binding, AtribucionMemberProvider)
if failures:
return failures
return invariant_diagnostics(binding, "atribucion", lambda b: _validated_atribucion_selector(b))
[docs]
def resolve_atribucion_binding_row_values(
revision: ModeloRevision,
observations: Iterable[AtributionMemberObservation],
) -> dict[tuple[BindingId, int], Decimal | str | int | bool]:
"""Resolve row-producer atribucion bindings into per-row indexed values.
Args:
revision: The :class:`ModeloRevision` whose atribucion bindings are resolved.
observations: Attribution member observations to group into rows.
"""
available = tuple(observations)
members: list[tuple[BindingDefinition, AtribucionMemberProvider]] = []
for binding in revision.bindings:
if binding.source != BindingSourceKind.ATRIBUCION_MEMBER:
continue
selector = _validated_atribucion_selector(binding)
members.append((binding, selector))
if not members:
return {}
rows = tuple(
{
"member_tax_id": obs.member_tax_id,
"member_legal_name": obs.member_legal_name,
"country_code": obs.country_code,
"share_percentage": obs.share_percentage,
"base_imponible_assigned": obs.base_imponible_assigned,
"clave": obs.clave,
"subclave": obs.subclave,
"codigo_provincia": obs.codigo_provincia,
"miembro_a_31_diciembre": obs.miembro_a_31_diciembre,
"dias_miembro": obs.dias_miembro,
"domicilio_fiscal": obs.domicilio_fiscal,
"naturaleza_inmueble": obs.naturaleza_inmueble,
"situacion_inmueble": obs.situacion_inmueble,
"referencia_catastral": obs.referencia_catastral,
"clave_declarado": obs.clave_declarado,
"porcentaje_titularidad_inmueble": obs.porcentaje_titularidad_inmueble,
"dias_arrendamiento": obs.dias_arrendamiento,
"reduccion": obs.reduccion,
"rendimiento_neto_previo_eo": obs.rendimiento_neto_previo_eo,
"rendimiento_neto_minorado_agricola_eo": obs.rendimiento_neto_minorado_agricola_eo,
}
for obs in sorted(available, key=lambda o: (o.country_code, o.member_tax_id))
)
resolved: dict[tuple[BindingId, int], Decimal | str | int | bool] = {}
for binding, selector in members:
row_field = _required_detail_record_row_field(binding, selector.row_field)
for row_index, row in enumerate(rows, start=1):
if row_field not in row:
raise RegistryValidationError(
f"binding {binding.id!r} row_field {row_field!r} not produced for atribucion rows",
)
value = row[row_field]
# A clave/subclave-conditional field is legitimately absent for a
# row whose declared clave does not license it (e.g.
# naturaleza_inmueble on a clave-D row) -- not a "not produced"
# authoring defect, so it is skipped rather than refused.
if value is None:
continue
resolved[(binding.id, row_index)] = value
return resolved
# ---------------------------------------------------------------------------
# Refund operation source bindings (modelo 360).
#
# Legal authority: Ley 37/1992 art. 117 bis (devolucion 8a Directiva),
# Orden EHA/789/2010 Anexo (modelo 360 diseno de registro).
# ---------------------------------------------------------------------------
_RefundRowField = Literal[
"member_state_code",
"operation_kind_code",
"operation_date",
"supplier_tax_id",
"refund_amount",
]
[docs]
class RefundOperationObservation(BaseModel):
"""One foreign-MS refund operation for modelo 360."""
model_config = STRICT_FROZEN_CONFIG
source_id: str = Field(min_length=1, max_length=128)
member_state_code: CountryCodeAlpha2
operation_kind_code: str = Field(min_length=1, max_length=4)
operation_date: date
supplier_tax_id: TaxIdIdentityToken
refund_amount: Decimal
_iso_code_uppercase = field_validator("member_state_code")(uppercase_alpha_code("member_state_code"))
@field_validator("refund_amount")
@classmethod
@pydantic_validation_boundary
def _decimal_amount(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise RegistryValidationError("refund_amount must be non-negative")
return value
[docs]
class RefundOperationProvider(BaseModel):
model_config = STRICT_FROZEN_CONFIG
kind: Literal[BindingSourceKind.REFUND_OPERATION] = BindingSourceKind.REFUND_OPERATION
fact: Literal["row_field"]
row_field: _RefundRowField | None = None
grouping: str | None = Field(default=None, min_length=1, max_length=64)
record: str | None = Field(default=None, min_length=1, max_length=64)
data_type: ExportFieldDataType | None = None
"""Scalar type of the value this row field contributes to the export.
The same fact ``BindingRowExportSelector.data_type`` carries; declared here
so the selector model admits the key, since a source-family selector is
validated whole against its own strict model. Optional while the families
adopt it.
"""
AtribucionMemberProvider = AtribucionMemberProvider
ForeignAssetProvider = ForeignAssetProvider
RefundOperationProvider = RefundOperationProvider
RelatedPartyOperationProvider = RelatedPartyOperationProvider
def _validated_refund_selector(binding: BindingDefinition) -> RefundOperationProvider:
try:
selector = provider_member(binding, RefundOperationProvider)
except ValueError as exc:
raise RegistryValidationError(f"binding {binding.id!r} has malformed refund selector") from exc
_validate_detail_record_row_field(binding, selector.fact, selector.row_field, "refund")
return selector
[docs]
def validate_refund_binding(binding: BindingDefinition) -> list[str]:
"""Validate a refund-operation binding at registry-build time.
Accumulating ``list[str]`` validator: validates the selector against
:class:`RefundOperationProvider` and lifts the resolve-time op/fact invariant to
build time, preserving the underlying pydantic field error.
"""
failures = selector_against_model(binding, RefundOperationProvider)
if failures:
return failures
return invariant_diagnostics(binding, "refund", lambda b: _validated_refund_selector(b))