"""Typed CLI row models for multi-row informational modelos.
Provides strictly-validated pydantic row shapes for operator-supplied
detail rows on modelos whose filing content is a list of repeating
records rather than a set of scalar casilla values.
Supported row types:
* ``Modelo184MemberRow`` — atribución member for modelo 184
(``--row miembro nif=X share=Y importe=Z``)
* ``Modelo232VinculadaRow`` — operación vinculada for modelo 232
(``--row vinculada nif=X tipo_vinculacion=Y importe=Z metodo=M pais=P``)
* ``Modelo349OperadorRow`` — operador row for modelo 349
(``--row operador codigo_pais=DE nif_comunitario=DE123456789 razon_social=X clave_operacion=E importe=Y``)
Used when no collectible-invoice ledger exists; maps directly to the
operator record layout.
* ``Modelo349RectificacionRow`` — rectificación intracomunitaria for modelo 349
(``--row rectificacion codigo_pais=DE nif_comunitario=DE123456789 razon_social=X``
``clave_operacion=E ejercicio=2025 periodo=2T base_rectificada=Y base_anterior=Z``)
Used when the operator declares Tipo-2 rectification records directly.
* ``Modelo347ContraparteRow`` — contraparte declarada for modelo 347
(``--row contraparte nif=X nombre=Y importe_Q1=Z clave_operacion=A``)
One row per counterparty. Annual importe threshold check (> €3,005.06)
is performed by the CLI validator, not the model, so partial row sets
accumulate correctly before final validation.
* ``Modelo210AgrupacionRentaRow`` — one component renta in an annual
Modelo 210 agrupación (period ``0A``). The row retains the official
two-digit renta code and the statutory grouping keys; it is evidence
only and never an alternate arithmetic input.
These models are the CLI boundary layer. They validate operator input
before being carried into ``detail_rows`` on the ``CalculationRevision``.
"""
from __future__ import annotations
import re
from collections.abc import Mapping, Sequence
from datetime import date
from decimal import Decimal
from enum import StrEnum
from typing import Annotated, Literal
from pydantic import (
BaseModel,
BeforeValidator,
Field,
StringConstraints,
ValidationInfo,
field_validator,
model_validator,
)
from ...core.errors.hierarchy import CadrumoError, pydantic_validation_boundary
from ...core.irnr import M210PayerMode
from ...core.modelo_232_codigos import MetodoValoracion, TipoOperacionVinculada, TipoVinculacion
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.time.clock import today_madrid
from ...core.unit_proportion import UnitProportion
from ..calculations.registry.authority import PinnedAuthorityOperation
from ..calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact, ResolvedScalarFact
from ..calculations.registry.governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from ..calculations.registry.m347_threshold import m347_threshold_decimal, resolve_m347_counterparty_annual_threshold
from ..calculations.registry.nif_iva_catalogue import nif_iva_format_for_country
from ..calculations.registry.schema_base import DateAxis
from ..transactions.m210_income_classification import resolve_m210_payer_mode
# ---------------------------------------------------------------------------
# Shared type aliases
# ---------------------------------------------------------------------------
_NifStr = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=20)]
_NameStr = Annotated[str, StringConstraints(strip_whitespace=True, max_length=200)]
_RequiredNameStr = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)]
_IsoCountryCode = Annotated[str, StringConstraints(strip_whitespace=True, min_length=2, max_length=2)]
_M210OfficialTipoRentaCode = Annotated[str, StringConstraints(strip_whitespace=True, min_length=2, max_length=2)]
def _registry_detail_catalogue(
*,
effective_date: date,
filing_year: int | None = None,
period: str | None = None,
operation: PinnedAuthorityOperation | None = None,
) -> tuple[Mapping[str, str], frozenset[str]]:
"""Resolve detail-row declarations and an explicitly selected M349 revision.
The mapping fact is date-scoped, while M349 period membership belongs to the
selected filing coordinate. Callers validating M349 therefore provide both
``filing_year`` and ``period``; M210 grouping callers only need the dated
mapping fact and leave the M349 projection absent.
"""
as_of = effective_date
authority: GovernedFactSource | None = operation or governed_facts_in_scope()
if authority is None:
raise ValueError("detail-row registry resolution requires a generation-pinned authority operation or scope")
if (filing_year is None) != (period is None):
raise ValueError("M349 registry selection requires both filing_year and period")
periods = frozenset[str]()
if filing_year is not None and period is not None:
if not isinstance(authority, PinnedAuthorityOperation):
raise ValueError("M349 registry selection requires a generation-pinned authority operation")
m349_revision = authority.revision_for_context(
"349",
filing_year=filing_year,
period=period,
on=as_of,
)
if not m349_revision.period_selector.declared_periods:
raise ValueError("selected M349 registry revision must declare detail-row scope")
periods = frozenset(str(candidate) for candidate in m349_revision.period_selector.declared_periods)
resolved = authority.resolve_governed_fact(
MappingFactQuery(
fact_id="detail-m349-m210-catalogues",
date_axis=DateAxis.FILING_PERIOD,
effective_date=as_of,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise ValueError("detail M349/M210 catalogue must resolve as a mapping fact")
declarations = {str(entry.key): str(entry.value) for entry in resolved.payload.entries}
return declarations, periods
def _required_detail_declaration(declarations: Mapping[str, str], key: str) -> str:
try:
return declarations[key]
except KeyError as exc:
raise ValueError(f"detail registry declaration is missing: {key}") from exc
def _detail_values(declarations: Mapping[str, str], key: str) -> frozenset[str]:
return frozenset(
value.strip() for value in _required_detail_declaration(declarations, key).split(",") if value.strip()
)
[docs]
def resolve_detail_row_owning_modelos(*, effective_date: date) -> Mapping[str, str]:
"""Project registry-declared ownership for every typed detail-row token."""
declarations, _ = _registry_detail_catalogue(effective_date=effective_date)
prefix = "detail_row."
suffix = ".owning_modelo"
ownership: dict[str, str] = {}
for key, value in declarations.items():
if not key.startswith(prefix) or not key.endswith(suffix):
continue
row_type = key[len(prefix) : -len(suffix)]
if not row_type or not value.strip():
raise ValueError(f"detail registry declaration has invalid row ownership: {key!r}")
ownership[row_type] = value.strip()
if not ownership:
raise ValueError("detail registry declaration is missing row ownership")
return ownership
[docs]
def resolve_detail_bearing_modelos(*, effective_date: date) -> frozenset[str]:
"""Project the registry's repeated-detail modelo membership declaration."""
declarations, _ = _registry_detail_catalogue(effective_date=effective_date)
return _detail_values(declarations, "detail_bearing_modelos")
# ---------------------------------------------------------------------------
# Modelo 184 - atribucion de rentas member row
#
# Legal authority: Orden HAP/2250/2015 art. 3; Ley 35/2006 arts. 86-89.
# One row per miembro (socio / comunero / partícipe) of the entity.
# ---------------------------------------------------------------------------
type M184Clave = Literal["A", "C", "D", "E", "F", "G", "I", "J", "K"]
type M184Subclave = Literal["01", "02", "03", "04", "05", "06"]
type M184NaturalezaInmueble = Literal["1", "2"]
type M184SituacionInmueble = Literal["1", "2", "3", "4", "5"]
type M184ClaveDeclarado = Literal["N", "T", "U", "O"]
[docs]
class Modelo184MemberRow(BaseModel):
"""One (member, clave, subclave) atribución row for Modelo 184.
The socio record's own diseño repeats per clave or subclave with an
importe declared, not per member alone (see the accepted row-shape ADR),
so this row's identity is ``(nif, clave, subclave)`` rather than ``nif``
alone -- the same shape 349's ``operador`` row already carries via its
``clave_operacion`` axis.
Fields mirror the per-record Tipo-2 layout declared in the M184
``bindings/0001-bindings.toml`` atribucion_member source block.
Parity assertions:
* ``nif`` → ``member_tax_id`` (binding: modelo-184-member-row-nif)
* ``nombre`` → ``member_legal_name`` (binding: modelo-184-member-row-name)
* ``porcentaje`` → ``share_percentage`` (binding: modelo-184-member-row-share)
* ``importe`` → ``base_imponible_assigned`` (binding: modelo-184-member-row-base-assigned)
* ``pais`` → ``country_code`` (required; never inferred)
* ``clave`` → ``clave`` (binding: modelo-184-member-row-clave)
* ``subclave`` → ``subclave`` (binding: modelo-184-member-row-subclave)
* every clave/subclave-conditional field below mirrors its own
``modelo-184-member-row-<field>`` binding.
Deliberately excluded (see the accepted row-shape ADR): the clave-A
reducción (its governing article is unresolved), provisiones-gastos-
dificil-justificacion (computed from the entity's own régimen fact, not
an operator-declared value), and any clave-E eligibility fact (a tracked
gap, no representation in this tree yet).
"""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["miembro"] = "miembro"
nif: _NifStr
nombre: _NameStr = Field(default="")
# Optional, and deliberately not defaulted to Spain. A default is an
# INFERENCE about a fact nobody stated, and it inferred the one value that
# makes the member domestic -- a foreign member silently declared as
# Spanish is the direction AEAT reconciles against what that member itself
# declared.
#
# Absent rather than required because the profile-driven producer has no
# country to supply: the atribucion socio facts carry nif, name, share and
# base and no territory at all. Demanding one here would refuse every
# profile-resolved row while naming a fact no surface records, which is a
# refusal nobody can answer. Recording the socio's country on the profile
# is the fix that would let this be required.
pais: _IsoCountryCode | None = None
porcentaje: Decimal = Field(description="Share percentage in the entity [0, 100]")
importe: Decimal = Field(description="Attributed income/base imponible in EUR")
# (member, clave, subclave) repetition axis. clave is required -- the
# socio record's own diseño has no row without one; subclave is optional
# because claves C and E carry no subclave table at all.
clave: M184Clave
subclave: M184Subclave | None = None
# Always-present-per-row facts, independent of clave.
codigo_provincia: Annotated[str, StringConstraints(strip_whitespace=True, max_length=2)] | None = None
miembro_a_31_diciembre: bool | None = None
dias_miembro: int | None = Field(default=None, ge=0, le=366)
domicilio_fiscal: Annotated[str, StringConstraints(strip_whitespace=True, max_length=40)] | None = None
# Clave-C inmueble sub-block.
naturaleza_inmueble: M184NaturalezaInmueble | None = None
situacion_inmueble: M184SituacionInmueble | None = None
referencia_catastral: Annotated[str, StringConstraints(strip_whitespace=True, max_length=20)] | None = None
clave_declarado: M184ClaveDeclarado | None = None
porcentaje_titularidad_inmueble: Decimal | None = None
dias_arrendamiento: int | None = Field(default=None, ge=0, le=366)
# Shared clave-C/clave-D reducción field (diseño positions 109-119); the
# clave-A branch of this same physical field is deliberately excluded.
reduccion: Decimal | None = None
# Clave-D subclave 03/04 rendimiento-neto fields (estimación objetiva).
rendimiento_neto_previo_eo: Decimal | None = None
rendimiento_neto_minorado_agricola_eo: Decimal | None = None
@field_validator("pais")
@classmethod
def _pais_uppercase_alpha(cls, value: str | None) -> str | None:
# ``pais`` is optional on THIS row for the reason recorded on the field
# above: the profile-driven producer has no country to supply, so a
# missing one is a declared state rather than a malformed value. The
# shape check therefore applies to a present value only -- calling
# ``.upper()`` on the absent case raised AttributeError instead of
# validating anything, which is a crash rather than a refusal.
if value is None:
return value
if value != value.upper() or not value.replace(" ", "").isalpha():
raise ValueError("pais must be an uppercase two-letter ISO 3166-1 country code (e.g. ES, DE, FR)")
return value
@field_validator("porcentaje")
@classmethod
def _porcentaje_within_bounds(cls, value: Decimal) -> Decimal:
if value < Decimal("0") or value > Decimal("100"):
raise ValueError(f"porcentaje must be within [0, 100]; got {value}")
return value
@field_validator("porcentaje_titularidad_inmueble")
@classmethod
def _porcentaje_titularidad_within_bounds(cls, value: Decimal | None) -> Decimal | None:
if value is not None and (value < Decimal("0") or value > Decimal("100")):
raise ValueError(f"porcentaje_titularidad_inmueble must be within [0, 100]; got {value}")
return value
@field_validator("nif")
@classmethod
def _nif_not_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("nif cannot be blank")
return value.upper()
@model_validator(mode="after")
def _subclave_only_where_the_diseno_declares_one(self) -> Modelo184MemberRow:
if self.subclave is not None and self.clave not in {"A", "D", "F", "G", "I", "J", "K"}:
raise ValueError(f"clave {self.clave!r} carries no subclave table; leave subclave unset")
return self
@model_validator(mode="after")
def _clave_a_reduccion_stays_blocked(self) -> Modelo184MemberRow:
"""Refuse a populated reducción on a clave-A row.
The shared diseño field (positions 109-119) also serves claves C and
D, whose reductions are grounded (LIRPF arts. 23 and 32.1). The
clave-A branch is explicitly BLOCKED by the accepted row-shape ADR:
the diseño's own citation (LIRPF art. 24.2) does not exist as such,
and art. 26.2 is only a strong, unconfirmed candidate. Modelling
reducción as one field (matching the diseño's own physical layout)
makes it reachable under clave A unless refused here explicitly, so
the block lives on the row rather than only in prose.
"""
if self.clave == "A" and self.reduccion is not None:
raise ValueError(
"clave A reducción is blocked pending confirmation of its governing provision "
"(the diseño's own LIRPF art. 24.2 citation does not exist; see the accepted "
"row-shape ADR) -- leave reduccion unset for a clave-A row",
)
return self
# ---------------------------------------------------------------------------
# Modelo 232 - operacion vinculada row
#
# Legal authority: Orden HFP/816/2017 art. 3; Ley 27/2014 art. 18;
# RD 634/2015 art. 13 (LIS transfer pricing).
# One row per related-party transaction group.
# ---------------------------------------------------------------------------
# The three coded fields' value sets are AEAT's published Tablas A, C and B
# of the diseño de registro DR23200. They are declared once in ``core`` --
# see :class:`~core.modelo_232_codigos.TipoVinculacion`, :class:`~core.modelo_232_codigos.TipoOperacionVinculada`
# and :class:`~core.modelo_232_codigos.MetodoValoracion` -- because the registry's own
# related-party observation is typed with the same sets.
def _hydrate_m232_codigo[EnumT: StrEnum](*, field_name: str, value: object, code_set: type[EnumT]) -> EnumT | object:
"""Hydrate the operator's text into its typed DR23200 code set.
The CLI delivers `--row vinculada k=v` as plain strings and the model is
strict, so this is the boundary that turns a token into a member. An
off-catalogue token is refused here rather than by the strict-instance
check, so the message can name the codes AEAT actually publishes instead
of only reporting the wrong type.
"""
if not isinstance(value, str):
return value
try:
return code_set(value.upper())
except ValueError:
accepted = ", ".join(repr(str(member)) for member in code_set)
raise ValueError(f"{field_name} must be one of {accepted}; got {value!r}") from None
[docs]
class Modelo232VinculadaRow(BaseModel):
"""One operación vinculada row for Modelo 232.
Fields mirror the related_party_operation binding source declared in
``232/revisions/2018-y-siguientes/bindings/0218…0223-*.toml``.
The three coded fields carry the closed catalogues AEAT's diseño de
registro DR23200 publishes as Tablas A, C and B — off-catalogue codes are
refused here rather than travelling into a fichero field that cannot hold
them.
Parity assertions:
* ``nif`` → ``counterparty_tax_id`` (binding: modelo-232-related-party-row-nif)
* ``nombre`` → ``counterparty_legal_name`` (binding: modelo-232-related-party-row-name)
* ``pais`` → ``country_code`` (binding: modelo-232-related-party-row-country)
* ``tipo_operacion`` → ``operation_kind_code`` (binding: modelo-232-related-party-row-operation-kind)
* ``metodo`` → ``transfer_pricing_method_code`` (binding: modelo-232-related-party-row-tpr-method)
* ``importe`` → ``amount`` (binding: modelo-232-related-party-row-amount)
"""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["vinculada"] = "vinculada"
nif: _NifStr
nombre: _NameStr = Field(default="")
# Required, and deliberately not defaulted to Spain. The row is built from
# operator-supplied key-value pairs, so a default is an INFERENCE about a
# fact the operator did not state -- and it inferred the one value that
# makes the row domestic. A foreign member or a cross-border related-party
# operation silently declared as Spanish is the direction AEAT reconciles
# against what the counterparty itself declared.
pais: _IsoCountryCode
tipo_vinculacion: Annotated[
TipoVinculacion | str,
BeforeValidator(
lambda v: _hydrate_m232_codigo(field_name="tipo_vinculacion", value=v, code_set=TipoVinculacion),
),
] = TipoVinculacion.NO_DECLARADO
tipo_operacion: Annotated[
TipoOperacionVinculada | str,
BeforeValidator(
lambda v: _hydrate_m232_codigo(field_name="tipo_operacion", value=v, code_set=TipoOperacionVinculada),
),
] = TipoOperacionVinculada.NO_DECLARADO
metodo: Annotated[
MetodoValoracion | str,
BeforeValidator(lambda v: _hydrate_m232_codigo(field_name="metodo", value=v, code_set=MetodoValoracion)),
] = MetodoValoracion.NO_DECLARADO
importe: Decimal
@field_validator("pais")
@classmethod
def _pais_uppercase_alpha(cls, value: str) -> str:
if value != value.upper() or not value.replace(" ", "").isalpha():
raise ValueError("pais must be an uppercase two-letter ISO 3166-1 country code (e.g. ES, DE, FR)")
return value
@field_validator("nif")
@classmethod
def _nif_not_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("nif cannot be blank")
return value.upper()
# ---------------------------------------------------------------------------
# Modelo 349 - operador intracomunitario row (manual-entry path)
#
# The selected registry revision owns operation codes, country-prefix rules,
# and period applicability. This module retains the row/type boundary and
# generic shape validation only.
# ---------------------------------------------------------------------------
Modelo349ClaveOperacionValue = Annotated[
str,
StringConstraints(strip_whitespace=True, min_length=1, max_length=1),
]
"""Typed one-character operation-key shell; the selected registry owns its values."""
# MappingFactQuery consumes the selected detail/M349/M210 declarations.
[docs]
class Modelo349CountryPrefixContextError(CadrumoError):
"""A Modelo 349 country prefix is invalid for the filing context."""
def __init__(
self,
*,
country_code: str,
clave_operacion: str,
filing_year: int,
period: str,
reason: str,
) -> None:
"""Record the prefix, the operation and the period that made it invalid."""
self.country_code = country_code
self.clave_operacion = clave_operacion
self.filing_year = filing_year
self.period = period
self.reason = reason
super().__init__(
translated_message="errors.refused.modelo_349_country_prefix_context",
context={
"country_code": country_code,
"clave_operacion": clave_operacion,
"filing_year": filing_year,
"period": period,
"reason": reason,
},
)
def _validate_m349_codigo_pais(value: str) -> str:
"""Require an uppercase two-letter ISO 3166-1 country code for an M349 row."""
if value != value.upper() or not value.replace(" ", "").isalpha():
raise ValueError("codigo_pais must be an uppercase two-letter ISO 3166-1 country code (e.g. DE, FR, IT)")
return value
def _validate_m349_nif_comunitario(value: str) -> str:
"""Require a non-blank comunitario NIF for an M349 row, normalised to upper case."""
if not value.strip():
raise ValueError("nif_comunitario cannot be blank")
return value.upper()
[docs]
class Modelo349OperadorRow(BaseModel):
"""One typed operator row for Modelo 349 manual-entry input."""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["operador"] = "operador"
codigo_pais: _IsoCountryCode
nif_comunitario: _NifStr
razon_social: _RequiredNameStr
clave_operacion: Modelo349ClaveOperacionValue
importe: Decimal = Field(description="Base imponible o importe de la operacion en EUR")
@field_validator("codigo_pais")
@classmethod
def _codigo_pais_uppercase_alpha(cls, value: str) -> str:
return _validate_m349_codigo_pais(value)
@field_validator("nif_comunitario")
@classmethod
def _nif_comunitario_not_blank(cls, value: str) -> str:
return _validate_m349_nif_comunitario(value)
@field_validator("importe")
@classmethod
def _importe_non_negative(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise ValueError(f"importe must be non-negative; got {value}")
return value
[docs]
class Modelo349RectificacionRow(BaseModel):
"""One typed rectification row for Modelo 349 manual-entry input."""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["rectificacion"] = "rectificacion"
codigo_pais: _IsoCountryCode
nif_comunitario: _NifStr
razon_social: _RequiredNameStr
clave_operacion: Modelo349ClaveOperacionValue
ejercicio: Annotated[str, StringConstraints(strip_whitespace=True, min_length=4, max_length=4)]
periodo: str
base_rectificada: Decimal = Field(description="Base imponible o importe rectificado en EUR")
base_anterior: Decimal = Field(description="Base imponible declarada anteriormente en EUR")
@field_validator("codigo_pais")
@classmethod
def _codigo_pais_uppercase_alpha(cls, value: str) -> str:
return _validate_m349_codigo_pais(value)
@field_validator("nif_comunitario")
@classmethod
def _nif_comunitario_not_blank(cls, value: str) -> str:
return _validate_m349_nif_comunitario(value)
@field_validator("periodo", mode="before")
@classmethod
@pydantic_validation_boundary
def _periodo_uppercase(cls, value: object, info: ValidationInfo) -> object:
if isinstance(value, str):
normalised = value.strip().upper()
ejercicio = info.data.get("ejercicio")
if not isinstance(ejercicio, str) or not ejercicio.isdigit():
# Let the ejercicio validator report its own malformed year;
# period membership cannot be selected without that coordinate.
return normalised
_, periods = _registry_detail_catalogue(
effective_date=date(int(ejercicio), 12, 31),
filing_year=int(ejercicio),
period=normalised,
)
if normalised not in periods:
accepted = ", ".join(repr(member) for member in sorted(periods))
raise ValueError(f"periodo must be one of {accepted}; got {value!r}")
return normalised
return value
@field_validator("ejercicio")
@classmethod
def _ejercicio_four_digit_year(cls, value: str) -> str:
if not value.isdigit():
raise ValueError("ejercicio must be a four-digit year")
return value
@field_validator("base_rectificada", "base_anterior")
@classmethod
def _bases_non_negative(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise ValueError("rectification bases must be non-negative")
return value
[docs]
def validate_m349_country_prefix_context(
*,
country_code: str,
clave_operacion: str,
filing_year: int,
period: str,
is_rectification: bool = False,
rectified_year: int | None = None,
rectified_period: str | None = None,
) -> None:
"""Validate the post-Brexit ``GB`` / ``XI`` prefix rules for one Modelo 349 row.
AEAT's Modelo 349 instructions keep ``XI`` for Northern Ireland goods
operations from the transition year onward and exclude the service keys
from it. ``GB`` stays valid only for rectifications of pre-transition
periods and for the transition year's first periods, never for service
keys. The transition year, first periods and service keys are
registry-declared.
"""
declarations, periods = _registry_detail_catalogue(
effective_date=date(filing_year, 12, 31),
filing_year=filing_year,
period=period,
)
operation_keys = _detail_values(declarations, "m349.operation_keys")
_required_detail_declaration(declarations, "m349.country_prefixes")
service_keys = _detail_values(declarations, "m349.service_keys")
first_periods = _detail_values(declarations, "m349.transition.first_periods")
transition_year = int(_required_detail_declaration(declarations, "m349.transition.year"))
normalized_period = _normalise_m349_period(period)
if normalized_period not in periods:
raise ValueError(f"M349 period is not declared by the selected registry: {period!r}")
clave = clave_operacion.strip().upper()
if clave not in operation_keys:
raise ValueError(f"M349 operation key is not declared by the selected registry: {clave_operacion!r}")
country = country_code.strip().upper()
rectified_period_code = _normalise_m349_period(rectified_period) if rectified_period is not None else None
def refuse(reason: str) -> None:
raise Modelo349CountryPrefixContextError(
country_code=country,
clave_operacion=clave,
filing_year=filing_year,
period=normalized_period,
reason=reason,
)
if country == _required_detail_declaration(declarations, "m349.goods_only_prefix"):
if clave in service_keys:
refuse("Northern Ireland prefix XI is not accepted for service keys")
if is_rectification and rectified_year is not None and rectified_year < transition_year:
refuse("pre-transition rectifications use GB, not XI")
if not is_rectification and filing_year < transition_year:
refuse("XI applies only from the transition year onward")
return
if country != _required_detail_declaration(declarations, "m349.transition.prefix"):
return
if is_rectification:
if rectified_year is not None and rectified_year < transition_year:
return
if rectified_year == transition_year and rectified_period_code in first_periods and clave not in service_keys:
return
refuse("GB is limited to pre-transition rectifications and the transition year's first periods")
if filing_year < transition_year:
return
if filing_year == transition_year and normalized_period in first_periods and clave not in service_keys:
return
refuse("GB is not accepted for post-transition ordinary operations")
def _normalise_m349_period(period: str | None) -> str:
if period is None:
return ""
token = str(period).strip().upper()
if len(token) == 1 and token.isdigit():
return f"0{token}"
return token
[docs]
def m349_nif_number_for_export(nif: str, pais: str) -> str:
"""Return the BOE NIF subfield without the separate country-code prefix.
Modelo 349 operator records split the IVA identifier into ``codigo_pais``
and ``nif_comunitario`` fields. The CLI accepts and validates the full
prefixed IVA identifier for operator ergonomics, but the fixed-width export
must write only the number part into positions 78-92.
"""
normalized_pais = pais.upper()
normalized_nif = nif.upper()
if not validate_m349_nif_format(normalized_nif, normalized_pais):
raise ValueError(
f"nif_comunitario {nif} does not match the expected NIF-IVA format for country {pais}",
)
return normalized_nif[len(normalized_pais) :]
# ---------------------------------------------------------------------------
# Modelo 347 - contraparte declarada row
#
# Legal authority: Orden EHA/3012/2008 art. 1; RD 1065/2007 arts. 31-35
# (reglamento de gestión e inspección tributaria, obligación de informar
# sobre operaciones con terceros); Ley 58/2003 art. 93.
# Threshold: total annual importe > €3,005.06 per counterparty (RD
# 1065/2007 art. 33.1). The threshold check is performed at the CLI
# validator level, not here, so that partial row accumulation works.
# ---------------------------------------------------------------------------
[docs]
class Modelo347ClaveOperacion(StrEnum):
"""Clave de operación declarable on a Modelo 347 counterparty row.
This operation-key type is specific to the M347 row family and is not shared
with the registry-owned M349 operation-key shell.
"""
A = "A"
B = "B"
C = "C"
D = "D"
E = "E"
F = "F"
G = "G"
Modelo347ClaveOperacionValue = Literal[
Modelo347ClaveOperacion.A,
Modelo347ClaveOperacion.B,
Modelo347ClaveOperacion.C,
Modelo347ClaveOperacion.D,
Modelo347ClaveOperacion.E,
Modelo347ClaveOperacion.F,
Modelo347ClaveOperacion.G,
]
"""The same code set for a wire or operation payload field."""
[docs]
class Modelo347ContraparteRow(BaseModel):
"""One contraparte declarada row for Modelo 347.
Fields mirror the per-counterparty Tipo-2 record layout declared in
``347/revisions/2011-2024``.
One row per counterparty. The annual total importe (sum of Q1-Q4)
must exceed €3,005.06 per RD 1065/2007 art. 33.1.
Parity assertions:
* ``nif`` → ``contraparte.nif`` (counterparty tax id)
* ``nombre`` → ``contraparte.nombre`` (legal name)
* ``importe_Q1/Q2/Q3/Q4`` → quarterly importe slots
* ``clave_operacion`` → operation type code
* ``pais_codigo`` → ``contraparte.pais`` (ISO 3166-1; None = domestic)
"""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["contraparte"] = "contraparte"
nif: _NifStr
nombre: _NameStr = Field(default="")
importe_Q1: Decimal = Field(default=Decimal("0"))
importe_Q2: Decimal = Field(default=Decimal("0"))
importe_Q3: Decimal = Field(default=Decimal("0"))
importe_Q4: Decimal = Field(default=Decimal("0"))
clave_operacion: Modelo347ClaveOperacionValue = Modelo347ClaveOperacion.A
pais_codigo: _IsoCountryCode | None = None
@field_validator("nif")
@classmethod
def _nif_not_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("nif cannot be blank")
return value.upper()
@field_validator("pais_codigo")
@classmethod
def _pais_codigo_uppercase_alpha(cls, value: str | None) -> str | None:
if value is None:
return value
v = value.strip().upper()
if not v.isalpha() or len(v) != 2:
raise ValueError("pais_codigo must be an uppercase two-letter ISO 3166-1 country code or None for domestic")
return v
@property
def importe_total(self) -> Decimal:
"""Sum of quarterly importes — used for M347 threshold check."""
return self.importe_Q1 + self.importe_Q2 + self.importe_Q3 + self.importe_Q4
# ---------------------------------------------------------------------------
# Modelo 210 - annual grouped-renta rows
#
# The selected registry revision owns the grouping catalogue and its
# applicability rules. This module retains the row shape and generic
# cross-row mechanics only.
# ---------------------------------------------------------------------------
[docs]
class Modelo210AgrupacionRentaRow(BaseModel):
"""One typed component row in an annual Modelo 210 grouping."""
model_config = STRICT_FROZEN_CONFIG
row_type: Literal["agrupacion_renta"] = "agrupacion_renta"
source_id: _RequiredNameStr
tipo_renta_code: _M210OfficialTipoRentaCode
importe: Decimal = Field(ge=Decimal("0"), description="Non-negative individual renta amount in EUR")
tipo_gravamen: UnitProportion = Field(
description="Applicable tax-rate fraction (for example, 0.24 for 24%)",
)
pagador_mode: M210PayerMode
pagador_id: _RequiredNameStr | None = None
deriva_de_bien_derecho: bool
bien_derecho_id: _RequiredNameStr | None = None
@field_validator("pagador_mode", mode="before")
@classmethod
@pydantic_validation_boundary
def _project_pagador_mode(cls, value: object) -> M210PayerMode:
"""Project the row's payer mode through the selected detail catalogue."""
return resolve_m210_payer_mode(value)
@field_validator("tipo_renta_code")
@classmethod
def _tipo_renta_code_has_shape(cls, value: str) -> str:
code = value.strip()
if not code.isdecimal():
raise ValueError("tipo_renta_code must be a two-character numeric code")
return code
@model_validator(mode="after")
def _typed_identity_contract(self) -> Modelo210AgrupacionRentaRow:
if self.deriva_de_bien_derecho:
if self.bien_derecho_id is None:
raise ValueError("a renta derived from a bien or derecho requires bien_derecho_id")
elif self.bien_derecho_id is not None:
raise ValueError("bien_derecho_id is only valid when deriva_de_bien_derecho is true")
return self
[docs]
class Modelo210AgrupacionRentaRowsError(CadrumoError):
"""A Modelo 210 annual grouped-renta set violates row compatibility."""
def __init__(self, *, reason: str, detail: str) -> None:
"""Record which row compatibility rule the row set broke."""
self.reason = reason
self.detail = detail
super().__init__(f"Modelo 210 annual agrupación rows are invalid ({reason}): {detail}")
def _require_nonempty_agrupacion(rows: Sequence[Modelo210AgrupacionRentaRow]) -> None:
if not rows:
raise Modelo210AgrupacionRentaRowsError(
reason="empty_group",
detail="period 0A requires at least one grouped-renta component",
)
def _require_unique_agrupacion_source_ids(rows: Sequence[Modelo210AgrupacionRentaRow]) -> None:
source_ids = tuple(row.source_id for row in rows)
if len(set(source_ids)) != len(source_ids):
raise Modelo210AgrupacionRentaRowsError(
reason="duplicate_source_id",
detail="each grouped-renta component must carry a unique stable source_id",
)
def _resolve_single_agrupacion_tipo_renta_code(rows: Sequence[Modelo210AgrupacionRentaRow]) -> str:
codes = {row.tipo_renta_code for row in rows}
if len(codes) != 1:
raise Modelo210AgrupacionRentaRowsError(
reason="mixed_tipo_renta_code",
detail=f"components use more than one official code: {', '.join(sorted(codes))}",
)
return next(iter(codes))
def _require_annual_agrupacion_code(code: str, *, effective_date: date) -> None:
declarations, _ = _registry_detail_catalogue(effective_date=effective_date)
if code not in _detail_values(declarations, "m210.annual_grouping_codes"):
raise Modelo210AgrupacionRentaRowsError(
reason="annual_code_not_lease_or_sublease",
detail=f"period 0A groups only the registry-declared lease/sublease codes, got {code}",
)
def _require_single_agrupacion_tipo_gravamen(rows: Sequence[Modelo210AgrupacionRentaRow]) -> None:
rates = {row.tipo_gravamen for row in rows}
if len(rates) != 1:
raise Modelo210AgrupacionRentaRowsError(
reason="mixed_tipo_gravamen",
detail=f"components use more than one tax rate: {', '.join(str(rate) for rate in sorted(rates))}",
)
def _require_shared_agrupacion_bien_derecho(rows: Sequence[Modelo210AgrupacionRentaRow]) -> None:
if not all(row.deriva_de_bien_derecho and row.bien_derecho_id is not None for row in rows):
raise Modelo210AgrupacionRentaRowsError(
reason="missing_bien_derecho",
detail="annual lease/sublease grouping requires an identified shared bien or derecho",
)
bienes_derechos = {row.bien_derecho_id for row in rows}
if len(bienes_derechos) != 1:
raise Modelo210AgrupacionRentaRowsError(
reason="mixed_bien_derecho",
detail="components derive from more than one bien or derecho",
)
def _validate_agrupacion_payer_grouping(
rows: Sequence[Modelo210AgrupacionRentaRow],
code: str,
*,
effective_date: date,
) -> None:
declarations, _ = _registry_detail_catalogue(effective_date=effective_date)
_required_detail_declaration(declarations, "m210.grouping_period")
payer_mode_declarations = {
key.removeprefix("m210.code").removesuffix(".payer_mode"): value
for key, value in declarations.items()
if key.startswith("m210.code") and key.endswith(".payer_mode")
}
required_mode = payer_mode_declarations.get(code)
if required_mode is not None:
if any(row.pagador_mode.value != required_mode for row in rows):
raise Modelo210AgrupacionRentaRowsError(
reason="payer_mode_not_declared",
detail="M210 payer mode does not match the selected registry declaration",
)
return
if any(row.pagador_mode.value in payer_mode_declarations.values() for row in rows):
raise Modelo210AgrupacionRentaRowsError(
reason="payer_mode_not_declared",
detail=f"M210 code {code} declares no multiple-payer exception",
)
payer_ids = {row.pagador_id for row in rows}
if None in payer_ids or len(payer_ids) != 1:
raise Modelo210AgrupacionRentaRowsError(
reason="mixed_pagador",
detail="an annual group without a multiple-payer exception must share one identified payer",
)
[docs]
def validate_m210_agrupacion_renta_rows(
rows: Sequence[Modelo210AgrupacionRentaRow],
*,
effective_date: date,
) -> None:
"""Validate a grouped-renta row set against one dated registry catalogue."""
_require_nonempty_agrupacion(rows)
_require_unique_agrupacion_source_ids(rows)
code = _resolve_single_agrupacion_tipo_renta_code(rows)
_require_annual_agrupacion_code(code, effective_date=effective_date)
_require_single_agrupacion_tipo_gravamen(rows)
_require_shared_agrupacion_bien_derecho(rows)
_validate_agrupacion_payer_grouping(rows, code, effective_date=effective_date)
# ---------------------------------------------------------------------------
# Discriminated union — single type accepted by the CLI --row argument
# ---------------------------------------------------------------------------
ModeloDetailRow = (
Modelo184MemberRow
| Modelo232VinculadaRow
| Modelo349OperadorRow
| Modelo349RectificacionRow
| Modelo347ContraparteRow
| Modelo210AgrupacionRentaRow
)
# Detail-bearing model membership is registry-owned; no Python fallback set is
# retained here.
# ---------------------------------------------------------------------------
# Statutory cross-row / threshold validations (domain-owned)
# ---------------------------------------------------------------------------
[docs]
class Modelo347ThresholdError(CadrumoError):
"""A Modelo 347 contraparte row falls at or below the declarability threshold."""
def __init__(self, *, nif: str, total: Decimal, threshold: ResolvedScalarFact) -> None:
"""Record the counterparty and the total that fell short of the threshold."""
self.nif = nif
self.total = total
self.threshold = threshold
super().__init__(
f"M347 contraparte (nif={nif!r}): importe total {total} does not exceed the "
f"{m347_threshold_decimal(threshold)} threshold required by RD 1065/2007 art. 33.1",
)
[docs]
class Modelo184ShareSumError(CadrumoError):
"""Modelo 184 member share percentages do not sum to exactly 100%."""
def __init__(self, *, total: Decimal, count: int) -> None:
"""Record the share total the miembro rows reached, and over how many rows."""
self.total = total
self.count = count
super().__init__(
f"M184 miembro rows: share percentages must sum to exactly 100%; got {total} across {count} rows",
)
[docs]
def validate_m347_threshold(
rows: Sequence[Modelo347ContraparteRow],
*,
effective_date: date,
) -> None:
"""Enforce the Modelo 347 per-counterparty declarability threshold.
RD 1065/2007 art. 33.1: only counterparties whose annual operations exceed
EUR 3,005.06 are declarable. The threshold applies to the SUM of every
operation with the same person (same NIF), aggregated across all contraparte
rows — not to each row in isolation. A counterparty's operations may be split
across several rows (e.g. entregas and adquisiciones), so a per-row check would
wrongly reject a counterparty whose individual rows are each at/below the
threshold while their annual aggregate exceeds it (a missed declaration), and
would never apply the "same person" threshold the regulation defines.
Raises:
Modelo347ThresholdError: for the first counterparty (in NIF first-appearance
order) whose AGGREGATED annual total is at or below the threshold.
"""
if not rows:
return
threshold = resolve_m347_counterparty_annual_threshold(effective_date=effective_date)
totals_by_nif: dict[str, Decimal] = {}
for row in rows:
totals_by_nif[row.nif] = totals_by_nif.get(row.nif, Decimal("0")) + row.importe_total
for nif, total in totals_by_nif.items():
if total <= m347_threshold_decimal(threshold):
raise Modelo347ThresholdError(nif=nif, total=total, threshold=threshold)
[docs]
def validate_m184_member_share_sum(rows: Sequence[Modelo184MemberRow]) -> None:
"""Enforce that Modelo 184 member share percentages sum to exactly 100%.
Only checked when at least one miembro row is present (partial sets are skipped).
Raises:
Modelo184ShareSumError: when the share percentages do not total exactly 100.
"""
if not rows:
return
total = sum((row.porcentaje for row in rows), Decimal("0"))
if total != Decimal("100"):
raise Modelo184ShareSumError(total=total, count=len(rows))
__all__ = [
"M184Clave",
"M184ClaveDeclarado",
"M184NaturalezaInmueble",
"M184SituacionInmueble",
"M184Subclave",
"Modelo184MemberRow",
"Modelo184ShareSumError",
"Modelo210AgrupacionRentaRow",
"Modelo210AgrupacionRentaRowsError",
"Modelo232VinculadaRow",
"Modelo347ContraparteRow",
"Modelo347ThresholdError",
"Modelo349CountryPrefixContextError",
"Modelo349OperadorRow",
"Modelo349RectificacionRow",
"ModeloDetailRow",
"m349_nif_number_for_export",
"resolve_detail_bearing_modelos",
"resolve_detail_row_owning_modelos",
"validate_m184_member_share_sum",
"validate_m210_agrupacion_renta_rows",
"validate_m347_threshold",
"validate_m349_country_prefix_context",
"validate_m349_nif_format",
]