"""Modelo IVA wallet gate for calculation and filing lifecycle checks.
Modelo 303 prior-compensation belongs to the IVA wallet authority, not to the
generic previous-filing source mesh. This module uses the calculation
:class:`~aeat.domain.calculations.registry.RegistrySnapshot` and
:class:`~aeat.domain.calculations.registry.ModeloRevision` to route the wallet
decision into the prior-compensation binding, then checks a persisted
:class:`~aeat.domain.iva_compensation._reconciliation.IvaCompensationReconciliationDecision`
against the exported or filed
:class:`~aeat.domain.modelos.CalculationRevision`.
The gate is deliberately repository-backed: transient wallet decisions cannot
feed the Modelo 303 engine unless the same decision is already present in
:class:`~aeat.application.calculations.IvaWalletDecisionRepository` for the
work-unit taxpayer and period. Calculation, verification, internal filing, and
export all replay this authority instead of trusting a caller-provided binding
value for casilla 110. Blocked, missing, stale, target-mismatched, or
amount-mismatched decisions raise
:class:`~aeat.application.modelo.ModeloIvaWalletReconciliationBlockedError`
before a revision, filing record, or fichero-BOE artefact can be persisted.
The only lazy path is local-authority derivation for a bucket-scoped
:class:`~aeat.domain.modelos.WorkUnit`: it can persist a non-blocking local
recurrence decision, and a ``first_period_zero`` decision is accepted only when
profile activity-start evidence and the
:class:`~aeat.domain.calculations.registry.RegistrySnapshot` prove every prior
Modelo 303 compensation dependency is pre-activity.
See Also:
:func:`~aeat.application.calculations.reconcile_modelo_303_iva_compensation`:
Builds and persists the reconciliation decision consumed here.
:class:`~aeat.application.calculations.IvaWalletDecisionSourceResolver`:
Projects a non-blocking decision into calculation binding values.
:func:`~aeat.application.modelo._verification_actions._require_cross_period_clean_state`:
Treats matching IVA-wallet authority as the Modelo 303 compensation gate.
:func:`~aeat.application.modelo._export.export_modelo_revision`:
Replays this gate before writing a Modelo 303 export artefact.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import date
from decimal import Decimal
from typing import Final, Protocol
from ...adapters.persistence.profile.modelos_calculation import CalculationRevisionCatalogueRepository
from ...adapters.persistence.profile.modelos_work_units import WorkUnitCatalogueRepository
from ...core import Modelo
from ...core import Period as _Period
from ...core.i18n import tr
from ...domain.calculations.registry import (
BindingId,
CasillaId,
ModeloRevision,
RegistrySnapshot,
RegistrySnapshotError,
previous_filing_observation_requirements,
validated_casilla_id,
)
from ...domain.iva_compensation import IvaCompensationReconciliationDecision
from ...domain.modelos import (
CalculationRevision,
CalculationRevisionCatalogueRepositoryProtocol,
CalculationRevisionState,
ModeloError,
WorkUnit,
WorkUnitCatalogueRepositoryProtocol,
)
from ..calculations import IvaWalletDecisionRepository
class _IvaWalletBlockedDecision(Protocol):
"""Protocol for decision-like values that can render a wallet-blocked message."""
@property
def divergence(self) -> object: ...
@property
def reason(self) -> object: ...
_M303_PRIOR_COMPENSATION_BINDING_ID: BindingId = "modelo-303-compensacion-pendiente-anteriores"
_M303_PRIOR_COMPENSATION_ORIGIN_IDS: Final[frozenset[str]] = frozenset(
{
_M303_PRIOR_COMPENSATION_BINDING_ID,
"modelo-303-rel-self-compensacion-anteriores",
},
)
def _casilla_id(value: object) -> CasillaId:
"""Validate a static IVA-wallet casilla constant as a :class:`~aeat.domain.calculations.registry.CasillaId`."""
try:
return validated_casilla_id(value, surface="IVA wallet gate casilla constant")
except ValueError as exc:
raise RuntimeError(f"IVA wallet gate casilla constant {value!r} is not a CasillaId") from exc
_M303_PRIOR_COMPENSATION_CASILLA_ID: Final[CasillaId] = _casilla_id(
"iva.compensacion-pendiente-periodos-anteriores",
)
_M303_AVAILABLE_COMPENSATION_CASILLA_ID: Final[CasillaId] = _casilla_id(
"iva.compensacion-disponible-fin-periodo",
)
[docs]
class ModeloIvaWalletReconciliationBlockedError(ModeloError):
"""Raised when Modelo 303 calculation is blocked by IVA wallet reconciliation."""
ModeloIvaWalletReconciliationBlocked = ModeloIvaWalletReconciliationBlockedError
[docs]
def resolve_iva_compensation_decision_for_calculation(
work_unit: WorkUnit,
*,
snapshot: RegistrySnapshot,
supplied_decision: object | None,
repository: IvaWalletDecisionRepository | None,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol | None = None,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol | None = None,
binding_values: Mapping[BindingId, Decimal] | None,
backend_binding_values: Mapping[BindingId, Decimal] | None,
casilla_inputs: Mapping[CasillaId, Decimal] | None,
backend_casilla_inputs: Mapping[CasillaId, Decimal] | None,
) -> object | None:
"""Resolve the Modelo 303 IVA wallet decision that may feed calculation bindings.
The :class:`~aeat.domain.modelos.WorkUnit` fixes the bucket, taxpayer profile
lookup, target period, and registry revision; the
:class:`~aeat.domain.calculations.registry.RegistrySnapshot` is passed to the
lazy reconciliation path when no caller-supplied or persisted wallet decision
exists. A supplied decision must match the persisted
:class:`~aeat.domain.iva_compensation._reconciliation.IvaCompensationReconciliationDecision`.
If the caller supplied a prior-compensation binding or casilla without a
decision, the function tries only the local-authority zero path and otherwise
returns ``None`` so calculation surfaces the seed/reconcile guidance instead
of silently trusting the value.
"""
if supplied_decision is None:
persisted = load_persisted_iva_compensation_decision_for_work_unit(work_unit, repository=repository)
if persisted is not None:
persisted = _refresh_first_period_zero_decision_if_local_evidence_changed(
work_unit,
snapshot=snapshot,
decision=persisted,
repository=repository,
work_unit_repository=work_unit_repository,
calculation_repository=calculation_repository,
)
return _require_first_period_zero_decision_grounded(work_unit, snapshot, persisted)
if caller_supplied_prior_compensation_value(
binding_values=binding_values,
backend_binding_values=backend_binding_values,
casilla_inputs=casilla_inputs,
backend_casilla_inputs=backend_casilla_inputs,
):
supplied_amounts = _supplied_prior_compensation_amounts(
binding_values=binding_values,
backend_binding_values=backend_binding_values,
casilla_inputs=casilla_inputs,
backend_casilla_inputs=backend_casilla_inputs,
)
decision = lazily_reconcile_local_iva_compensation_for_work_unit(
work_unit,
snapshot=snapshot,
repository=repository,
work_unit_repository=work_unit_repository,
calculation_repository=calculation_repository,
persist=False,
)
if decision is not None and not _decision_is_missing_local_authority(decision):
if _decision_has_concrete_zero_authority(decision):
if any(amount != Decimal("0") for amount in supplied_amounts):
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_caller_binding_conflict",
)
decision = _non_blocking_concrete_zero_authority_decision(decision)
if not decision.blocked:
decision = _require_first_period_zero_decision_grounded(work_unit, snapshot, decision)
_save_iva_compensation_decision(decision, repository=repository)
return decision
return None
return lazily_reconcile_local_iva_compensation_for_work_unit(
work_unit,
snapshot=snapshot,
repository=repository,
work_unit_repository=work_unit_repository,
calculation_repository=calculation_repository,
)
return require_persisted_iva_compensation_decision_for_work_unit(
work_unit,
supplied_decision=supplied_decision,
snapshot=snapshot,
repository=repository,
)
[docs]
def apply_iva_compensation_decision_binding(
modelo: str,
filing_year: int,
period: _Period,
*,
bucket_id: str,
revision: ModeloRevision,
taxpayer_nif: str | None = None,
casilla_inputs: Mapping[CasillaId, Decimal] | None = None,
backend_casilla_inputs: Mapping[CasillaId, Decimal] | None = None,
caller_binding_values: dict[BindingId, Decimal],
backend_binding_values: dict[BindingId, Decimal],
decision: object | None,
) -> None:
"""Apply a non-blocking IVA wallet decision to Modelo 303 binding values.
The :class:`~aeat.domain.calculations.registry.ModeloRevision` defines the
binding channel; the decision amount is written only after target period and
taxpayer identity checks pass. Caller and backend inputs for the same binding
or casilla must either match the selected decision amount or are refused as
conflicts. The effective value is then produced through
:class:`~aeat.application.calculations.IvaWalletDecisionSourceResolver`, so
the calculation source mesh records the IVA-wallet provenance instead of a
generic ``previous_filing`` source.
"""
if modelo != Modelo.M303:
return
binding_id = _M303_PRIOR_COMPENSATION_BINDING_ID
bound_casilla_id = _M303_PRIOR_COMPENSATION_CASILLA_ID
caller_casilla_value = dict(casilla_inputs or {}).get(bound_casilla_id)
backend_casilla_value = dict(backend_casilla_inputs or {}).get(bound_casilla_id)
if decision is None:
caller_value = caller_binding_values.get(binding_id)
backend_value = backend_binding_values.get(binding_id)
if (
caller_value is not None
or backend_value is not None
or caller_casilla_value is not None
or backend_casilla_value is not None
):
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_not_seeded",
suggestion="aeat app modelo iva-wallet seed --filing-year YEAR --period PERIOD --amount 0 --confirm",
)
return
if not isinstance(decision, IvaCompensationReconciliationDecision):
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_unsupported_decision_type",
context={"decision_type": type(decision).__name__},
)
if decision.target_period != period:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_target_mismatch",
context={
"target_year": decision.target_year,
"target_period": decision.target_period.registry_token,
"filing_year": filing_year,
"period": period.registry_token,
},
)
if taxpayer_nif is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_taxpayer_identity_missing",
)
if decision.taxpayer_nif.strip().upper() != taxpayer_nif.strip().upper():
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_taxpayer_mismatch",
)
if decision.blocked:
raise ModeloIvaWalletReconciliationBlocked(
"IVA wallet reconciliation blocks automatic Modelo 303 calculation: "
f"{decision.divergence}: {decision.reason}",
translated_message="application.modelo.errors.iva_wallet_blocked",
context={"divergence": str(decision.divergence), "reason": str(decision.reason)},
suggestion=iva_wallet_override_suggestion(decision),
)
if decision.selected_amount is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_selected_amount_missing",
)
selected = Decimal(decision.selected_amount)
caller_value = caller_binding_values.get(binding_id)
if caller_value is not None and Decimal(caller_value) != selected:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_caller_binding_conflict",
)
if caller_casilla_value is not None and Decimal(caller_casilla_value) != selected:
raise ModeloIvaWalletReconciliationBlocked(
"caller casilla input for Modelo 303 prior compensation conflicts with IVA wallet reconciliation decision",
translated_message="application.modelo.errors.iva_wallet_caller_casilla_conflict",
)
if backend_casilla_value is not None and Decimal(backend_casilla_value) != selected:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_backend_casilla_conflict",
)
from ..aggregation import CalculationSourceContext
from ..calculations import IvaWalletDecisionSourceResolver
resolution = IvaWalletDecisionSourceResolver(decision).resolve(
CalculationSourceContext(
bucket_id=bucket_id,
modelo=modelo,
filing_year=filing_year,
period=period,
revision=revision,
),
)
backend_binding_values.update(resolution.binding_values)
[docs]
def require_persisted_iva_compensation_decision_for_work_unit(
work_unit: WorkUnit,
*,
supplied_decision: object,
snapshot: RegistrySnapshot | None = None,
repository: IvaWalletDecisionRepository | None = None,
) -> object:
"""Require a supplied Modelo 303 wallet decision to match the persisted decision.
The optional :class:`~aeat.domain.calculations.registry.RegistrySnapshot`
grounds first-period-zero decisions; when omitted, the function resolves the
snapshot from the supplied :class:`~aeat.domain.modelos.WorkUnit`. This check
prevents a transient or stale
:class:`~aeat.domain.iva_compensation._reconciliation.IvaCompensationReconciliationDecision`
from feeding calculation values unless the repository contains the same
authority record.
"""
if work_unit.modelo != Modelo.M303:
return supplied_decision
persisted = load_persisted_iva_compensation_decision_for_work_unit(work_unit, repository=repository)
if persisted is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_not_seeded",
suggestion="aeat app modelo iva-wallet seed --filing-year YEAR --period PERIOD --amount 0 --confirm",
)
if persisted != supplied_decision:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_supplied_decision_mismatch",
)
if _decision_is_first_period_zero(persisted):
return _require_first_period_zero_decision_grounded(
work_unit,
snapshot or _registry_snapshot_for_work_unit(work_unit),
persisted,
)
return persisted
[docs]
def load_persisted_iva_compensation_decision_for_work_unit(
work_unit: WorkUnit,
*,
repository: IvaWalletDecisionRepository | None = None,
) -> IvaCompensationReconciliationDecision | None:
"""Load the persisted IVA compensation decision for a :class:`~aeat.domain.modelos.WorkUnit`.
Returns:
The persisted :class:`IvaCompensationReconciliationDecision` for Modelo
303 work units, or ``None`` when the work unit or bucket has no wallet
authority record.
"""
if work_unit.modelo != Modelo.M303:
return None
taxpayer_nif = taxpayer_nif_for_bucket(work_unit.bucket_id)
if taxpayer_nif is None:
return None
if repository is None:
from ..calculations import IvaWalletDecisionRepository
repository = IvaWalletDecisionRepository()
return repository.load_decision(
taxpayer_nif,
work_unit.period,
)
[docs]
def caller_supplied_prior_compensation_value(
*,
binding_values: Mapping[BindingId, Decimal] | None,
backend_binding_values: Mapping[BindingId, Decimal] | None,
casilla_inputs: Mapping[CasillaId, Decimal] | None,
backend_casilla_inputs: Mapping[CasillaId, Decimal] | None,
) -> bool:
"""Return whether a Modelo 303 prior-compensation value was explicitly supplied.
The lazy local reconciliation must not fire when the operator or a backend
resolver explicitly asserts the prior-compensation binding/casilla. That
value needs reconciliation against a real wallet/seed decision, and the
seed-verb guidance must surface.
"""
return bool(
_supplied_prior_compensation_amounts(
binding_values=binding_values,
backend_binding_values=backend_binding_values,
casilla_inputs=casilla_inputs,
backend_casilla_inputs=backend_casilla_inputs,
),
)
def _supplied_prior_compensation_amounts(
*,
binding_values: Mapping[BindingId, Decimal] | None,
backend_binding_values: Mapping[BindingId, Decimal] | None,
casilla_inputs: Mapping[CasillaId, Decimal] | None,
backend_casilla_inputs: Mapping[CasillaId, Decimal] | None,
) -> tuple[Decimal, ...]:
"""Return supplied Modelo 303 prior-compensation amounts from all input channels."""
binding_id = _M303_PRIOR_COMPENSATION_BINDING_ID
casilla_id = _M303_PRIOR_COMPENSATION_CASILLA_ID
values = (
dict(binding_values or {}).get(binding_id),
dict(backend_binding_values or {}).get(binding_id),
dict(casilla_inputs or {}).get(casilla_id),
dict(backend_casilla_inputs or {}).get(casilla_id),
)
return tuple(Decimal(value) for value in values if value is not None)
def _profile_path_values_for_bucket(bucket_id: str) -> dict[str, str] | None:
"""Return canonical user-profile path values for ``bucket_id``."""
from ...domain.user_profile import ProfileNotFoundError
from ..user_profile import UserProfileLifecycleRepository, record_to_path_values
try:
record = UserProfileLifecycleRepository(bucket_id=bucket_id).load(bucket_id)
except ProfileNotFoundError:
return None
return record_to_path_values(record)
def _activity_start_date_for_modelo_profile(bucket_id: str) -> date | None:
"""Return the lifecycle profile's declared activity-start date, if parseable."""
from ...core.parsing import parse_iso8601_date
values = _profile_path_values_for_bucket(bucket_id)
if values is None:
return None
try:
return parse_iso8601_date(values.get("censo.activity_start_date"))
except ValueError:
return None
def _registry_snapshot_for_work_unit(work_unit: WorkUnit) -> RegistrySnapshot:
"""Resolve the registry snapshot attached to ``work_unit``."""
from ...core.resources import resources
try:
return resources().modelos.authority.snapshot(
str(work_unit.modelo),
filing_year=work_unit.filing_year,
period=work_unit.period.registry_token,
)
except (FileNotFoundError, RegistrySnapshotError) as exc:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "first_period_zero_unproven",
"reason": "registry snapshot could not be resolved to ground first-period-zero IVA decision",
},
) from exc
def _decision_is_first_period_zero(decision: object) -> bool:
return str(getattr(decision, "divergence", "")) == "first_period_zero"
def _refresh_first_period_zero_decision_if_local_evidence_changed(
work_unit: WorkUnit,
*,
snapshot: RegistrySnapshot,
decision: IvaCompensationReconciliationDecision,
repository: IvaWalletDecisionRepository | None,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol | None,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol | None,
) -> IvaCompensationReconciliationDecision:
if not _decision_is_first_period_zero(decision):
return decision
refreshed = lazily_reconcile_local_iva_compensation_for_work_unit(
work_unit,
snapshot=snapshot,
repository=repository,
work_unit_repository=work_unit_repository,
calculation_repository=calculation_repository,
persist=False,
)
if refreshed is None or _decision_replay_basis(refreshed) == _decision_replay_basis(decision):
return decision
_save_iva_compensation_decision(refreshed, repository=repository)
return refreshed
def _decision_replay_basis(decision: IvaCompensationReconciliationDecision) -> tuple[object, ...]:
return (
decision.selected_authority,
decision.selected_amount,
decision.local_recurrence_amount,
decision.divergence,
decision.blocked,
tuple(
(
source.source_kind,
source.amount,
source.source_locator,
source.source_modelo,
source.source_filing_year,
source.source_periods,
)
for source in decision.authority_sources
),
)
def _decision_is_missing_local_authority(decision: object) -> bool:
return str(getattr(decision, "divergence", "")) == "missing" and getattr(decision, "selected_amount", None) is None
def _decision_has_concrete_zero_authority(decision: object) -> bool:
selected_amount = getattr(decision, "selected_amount", None)
if selected_amount is None or Decimal(selected_amount) != Decimal("0"):
return False
for source in getattr(decision, "authority_sources", ()) or ():
amount = getattr(source, "amount", None)
if amount is None or Decimal(amount) != Decimal("0"):
continue
source_kind = str(getattr(source, "source_kind", ""))
if source_kind == "aeat_wallet" and getattr(source, "captured_at", None) is not None:
return True
if source_kind in {"local_recurrence", "filed_history_observation"} and tuple(
getattr(source, "source_periods", ()) or (),
):
return True
return False
def _non_blocking_concrete_zero_authority_decision(
decision: IvaCompensationReconciliationDecision,
) -> IvaCompensationReconciliationDecision:
return decision.model_copy(
update={
"selected_authority": "local_recurrence",
"selected_amount": Decimal("0"),
"divergence": "match",
"blocked": False,
"stale_wallet": False,
"reason": (
"Caller-supplied zero matches concrete local IVA compensation history; "
"no prior compensation balance is available to apply."
),
},
)
def _save_iva_compensation_decision(
decision: IvaCompensationReconciliationDecision,
*,
repository: IvaWalletDecisionRepository | None,
) -> None:
repo = repository if repository is not None else IvaWalletDecisionRepository()
repo.save_decision(decision)
def _require_first_period_zero_decision_grounded(
work_unit: WorkUnit,
snapshot: RegistrySnapshot,
decision: IvaCompensationReconciliationDecision,
) -> IvaCompensationReconciliationDecision:
"""Fail closed unless a first-period-zero decision is profile/registry-grounded."""
if not _decision_is_first_period_zero(decision):
return decision
if _decision_has_concrete_zero_authority(decision):
return decision
if _activity_start_proves_first_iva_period(work_unit, snapshot):
return decision
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "first_period_zero_unproven",
"reason": (
"first-period-zero IVA compensation requires profile activity-start proof that every "
"Modelo 303 prior-compensation dependency is pre-activity"
),
},
)
def _activity_start_proves_first_iva_period(work_unit: WorkUnit, snapshot: RegistrySnapshot) -> bool:
"""Return whether the profile proves no prior IVA-compensation obligation existed."""
from ..calculations import (
cross_period_dependency_requirements,
partition_cross_period_requirements_by_activity_start,
)
activity_start_date = _activity_start_date_for_modelo_profile(work_unit.bucket_id)
if activity_start_date is None:
return False
requirements = tuple(
requirement
for requirement in cross_period_dependency_requirements(snapshot)
if requirement.source_modelo == Modelo.M303.value
and _M303_PRIOR_COMPENSATION_ORIGIN_IDS.intersection(requirement.origin_ids)
)
if not requirements:
return False
partition = partition_cross_period_requirements_by_activity_start(
requirements,
activity_start_date=activity_start_date,
)
return bool(partition.suppressed) and not partition.in_scope
[docs]
def lazily_reconcile_local_iva_compensation_for_work_unit(
work_unit: WorkUnit,
*,
snapshot: RegistrySnapshot,
repository: IvaWalletDecisionRepository | None = None,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol | None = None,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol | None = None,
persist: bool = True,
) -> IvaCompensationReconciliationDecision | None:
"""Auto-derive and persist the local-authority Modelo 303 compensation decision.
Calculate's prior-compensation gate requires a persisted
:class:`~aeat.domain.iva_compensation._reconciliation.IvaCompensationReconciliationDecision`.
In the seed-only local authority case, the local Modelo 303 recurrence is the
authority, so derive and persist the decision here instead of refusing
calculation.
The :class:`~aeat.domain.calculations.registry.RegistrySnapshot` supplies the
Modelo 303 revision context for the reconciliation service. Missing local
recurrence is treated as ``first_period_zero`` only when the work unit's
profile activity-start date scopes every Modelo 303 compensation dependency
out as pre-activity; otherwise the reconciliation remains a blocking
missing-authority state.
"""
if work_unit.modelo != Modelo.M303:
return None
taxpayer_nif = taxpayer_nif_for_bucket(work_unit.bucket_id)
if taxpayer_nif is None:
return None
from ..calculations import reconcile_modelo_303_iva_compensation
report = reconcile_modelo_303_iva_compensation(
snapshot,
taxpayer_nif=taxpayer_nif,
wallet=None,
decision_repository=repository,
fallback_local_recurrence=_calculated_revision_local_iva_compensation_recurrence(
work_unit,
snapshot=snapshot,
work_unit_repository=work_unit_repository,
calculation_repository=calculation_repository,
),
# Missing local recurrence proves a zero only when the profile's
# activity-start date scopes every Modelo 303 prior-compensation
# dependency out as pre-activity. Otherwise the reconciliation must keep
# failing closed as missing wallet/local history.
treat_absent_recurrence_as_first_period=_activity_start_proves_first_iva_period(work_unit, snapshot),
persist=persist,
)
return report.decision
def _calculated_revision_local_iva_compensation_recurrence(
work_unit: WorkUnit,
*,
snapshot: RegistrySnapshot,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol | None,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol | None,
):
"""Return prior-period calculated M303 carry evidence for wallet reconciliation.
This fallback is deliberately recurrence evidence only: the returned value is
fed through the IVA wallet decision path. A zero amount can unblock the next
period because no compensation is applied; a non-zero amount still becomes a
blocked local-recurrence decision unless wallet evidence or an override exists.
"""
requirements = tuple(
requirement
for requirement in previous_filing_observation_requirements(
snapshot.revision,
filing_year=snapshot.filing_year,
period=snapshot.period,
)
if requirement.source_modelo == Modelo.M303.value
and _M303_PRIOR_COMPENSATION_BINDING_ID in requirement.binding_ids
)
if len(requirements) != 1 or len(requirements[0].periods) != 1:
return None
requirement = requirements[0]
source_period = (
requirement.filing_periods[0]
if requirement.filing_periods
else _Period.from_year_and_code(requirement.filing_year, requirement.periods[0])
)
wu_repo = work_unit_repository if work_unit_repository is not None else WorkUnitCatalogueRepository()
calc_repo = (
calculation_repository if calculation_repository is not None else CalculationRevisionCatalogueRepository()
)
calculation_catalogue = calc_repo.load()
matching_work_units = sorted(
(
candidate
for candidate in wu_repo.load().values()
if candidate.bucket_id == work_unit.bucket_id
and candidate.modelo == Modelo.M303
and candidate.filing_year == requirement.filing_year
and candidate.period == source_period
),
key=lambda candidate: candidate.updated_at,
reverse=True,
)
from ..calculations import LocalIvaCompensationRecurrence
for source_work_unit in matching_work_units:
revision_id = source_work_unit.filed_calculation_revision_id or source_work_unit.current_calculation_revision_id
if revision_id is None:
continue
revision = calculation_catalogue.get(revision_id)
if revision is None or revision.state is CalculationRevisionState.DESCARTADO:
continue
amount = revision.casilla_values.get(_M303_AVAILABLE_COMPENSATION_CASILLA_ID)
if amount is None:
continue
return LocalIvaCompensationRecurrence(
binding_id=_M303_PRIOR_COMPENSATION_BINDING_ID,
amount=Decimal(amount),
source_kind="calculated_revision",
source_modelo=Modelo.M303.value,
source_filing_year=requirement.filing_year,
source_periods=(source_period,),
resolved_at=revision.updated_at,
source_locator=f"calculation_revision:{revision.calculation_revision_id}",
)
return None
[docs]
def require_persisted_iva_compensation_decision_matches_revision(
work_unit: WorkUnit,
revision: CalculationRevision,
*,
repository: IvaWalletDecisionRepository | None = None,
) -> IvaCompensationReconciliationDecision | None:
"""Return the IVA compensation decision when it matches the revision.
The check reads the supplied
:class:`~aeat.domain.modelos.CalculationRevision` and blocks verification,
internal filing, or export when its Modelo 303
prior-compensation amount differs from the persisted wallet decision, when
the decision targets another period, or when the decision is blocked/missing.
Non-Modelo 303 work units return ``None`` because the IVA wallet authority
owns only Modelo 303 prior compensation.
Returns:
The matching :class:`IvaCompensationReconciliationDecision` for Modelo
303, or ``None`` for non-Modelo 303 work units.
"""
if work_unit.modelo != Modelo.M303:
return None
decision = load_persisted_iva_compensation_decision_for_work_unit(work_unit, repository=repository)
if decision is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_not_seeded",
suggestion="aeat app modelo iva-wallet seed --filing-year YEAR --period PERIOD --amount 0 --confirm",
)
if decision.blocked:
raise ModeloIvaWalletReconciliationBlocked(
iva_wallet_blocked_message(decision),
translated_message="application.modelo.errors.iva_wallet_blocked",
context={"divergence": str(decision.divergence), "reason": str(decision.reason)},
suggestion=iva_wallet_override_suggestion(decision),
)
if decision.target_period != work_unit.period:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "authority_target_mismatch",
"reason": "persisted IVA wallet decision target does not match the Modelo 303 work unit",
},
)
if _decision_is_first_period_zero(decision):
_require_first_period_zero_decision_grounded(
work_unit,
_registry_snapshot_for_work_unit(work_unit),
decision,
)
if decision.selected_amount is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "authority_missing_amount",
"reason": "persisted IVA wallet decision has no selected amount",
},
)
revision_amount = revision_iva_compensation_amount(revision)
if revision_amount is None:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "authority_revision_missing_amount",
"reason": "calculation revision does not carry the Modelo 303 prior-compensation amount",
},
)
if Decimal(decision.selected_amount) != revision_amount:
raise ModeloIvaWalletReconciliationBlocked(
translated_message="application.modelo.errors.iva_wallet_blocked",
context={
"divergence": "authority_amount_mismatch",
"reason": "persisted IVA wallet decision does not match the calculation revision",
},
)
return decision
[docs]
def revision_iva_compensation_amount(revision: CalculationRevision) -> Decimal | None:
"""Return the Modelo 303 prior-compensation amount carried by a revision.
Reads the :class:`~aeat.domain.modelos.CalculationRevision` casilla values
first, then binding overrides, matching the persisted calculation payload.
"""
casilla_value = dict(revision.casilla_values).get(_M303_PRIOR_COMPENSATION_CASILLA_ID)
if casilla_value is not None:
return Decimal(casilla_value)
binding_value = dict(revision.binding_overrides).get(_M303_PRIOR_COMPENSATION_BINDING_ID)
if binding_value is not None:
return Decimal(binding_value)
return None
[docs]
def iva_wallet_blocked_message(decision: _IvaWalletBlockedDecision) -> str:
"""Render a localized IVA wallet blocked message from a decision-like object."""
divergence = str(decision.divergence)
reason = str(decision.reason)
return tr("application.modelo.errors.iva_wallet_blocked", divergence=divergence, reason=reason)
[docs]
def iva_wallet_override_suggestion(decision: object) -> str:
"""Return the explicit taxpayer-override command for a blocked wallet decision."""
target_period = getattr(decision, "target_period", None)
filing_year = getattr(decision, "target_year", "YEAR")
period = getattr(target_period, "registry_token", "PERIOD")
amount = "0" if _decision_is_missing_local_authority(decision) else "AMOUNT"
return (
"aeat app modelo iva-wallet override "
f"--filing-year {filing_year} --period {period} --amount {amount} "
'--reason "external evidence reviewed" --evidence-locator "SOURCE" --confirm'
)
[docs]
def taxpayer_nif_for_bucket(bucket_id: str) -> str | None:
"""Return the profile tax id for a bucket, or ``None`` when absent."""
values = _profile_path_values_for_bucket(bucket_id)
if values is None:
return None
value = values.get("identity.tax_id")
if value is None or not value.strip():
return None
return value.strip()
__all__ = [
"ModeloIvaWalletReconciliationBlocked",
"ModeloIvaWalletReconciliationBlockedError",
"apply_iva_compensation_decision_binding",
"caller_supplied_prior_compensation_value",
"iva_wallet_blocked_message",
"iva_wallet_override_suggestion",
"lazily_reconcile_local_iva_compensation_for_work_unit",
"load_persisted_iva_compensation_decision_for_work_unit",
"require_persisted_iva_compensation_decision_for_work_unit",
"require_persisted_iva_compensation_decision_matches_revision",
"resolve_iva_compensation_decision_for_calculation",
"revision_iva_compensation_amount",
"taxpayer_nif_for_bucket",
]