"""Binding prefill: resolve ``previous_filing`` bindings from prior filings.
Used by :class:`~._multi_year.PreviousFilingSourceResolver` for the source-mesh
calculation path and by
:func:`~application.calculations.extract_modelo_303_local_iva_compensation_recurrence`
for the IVA wallet comparison path.
One of three distinct prefill tiers, NOT to be merged: this is the
PREVIOUS-FILING direct-carry tier. The other two are the relation tier
(:mod:`~application.calculations._relation_prefill`) and the AEAT borrador
pre-fill tier (the registry ``aeat_prefilled`` flag, an AEAT-live source). They
share only the word "prefill"; each routes a different source through a
different mechanism.
Sister module to :mod:`~._relation_prefill`. The runtime distinguishes
``relation`` leaves (cross-revision aggregations declared as
:class:`~domain.calculations.registry.RelationDefinition` records) from
``previous_filing`` bindings (declared as
:class:`~domain.calculations.registry.DataBindingDefinition` with
``source = "previous_filing"``).
Modelo 390 uses bindings — modelo 200 uses relations — both express
"sum a prior modelo's casilla across periods" but route through
different schema entities.
Prior-filing values are gathered as :class:`CasillaObservation` records and
merged inside
:class:`~domain.calculations.registry.RegistryModeloObservation` rows before
the binding is resolved.
The strict registry boundary remains
:func:`~domain.calculations.registry.previous_filing_observation_requirements`
and
:func:`~domain.calculations.registry.resolve_previous_filing_binding_values`;
this module is the application reader that supplies local observations from
:class:`~._observations_repository.CalculationObservationRepository` and returns
:class:`BindingPrefillReport` coverage.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import date, datetime
from decimal import Decimal
from typing import Final
from pydantic import BaseModel
from ...core import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core import Modelo, Period
from ...core.resources import resources
from ...core.time import now
from ...domain.calculations.registry import (
MODELO_303_IVA_COMPENSATION_BINDING_ID,
BindingId,
CasillaDefinition,
CasillaId,
CasillaObservation,
FormulaDefinition,
RegistryFoldRequirement,
RegistryModeloObservation,
RegistrySnapshot,
binding_source_casilla_ids,
expression_casilla_refs,
previous_filing_observation_requirements,
resolve_previous_filing_binding_values,
validated_casilla_id,
)
from ...domain.iva_compensation import IvaCompensationCasillaReferenceError, IvaCompensationPeriodState
from ._errors import BindingPrefillTypeError
from ._iva_compensation_history import IvaCompensationHistoryRepository
from ._observations_repository import CalculationObservationRepository, _ObservationEnvelopePayload
from ._revision_carry_gate import revision_carry_outcome
def _selector_year_delta(value: object) -> int:
"""Narrow a binding-selector ``filing_year_delta`` to ``int``.
Selectors flow through pydantic with a union value type, so static
analysis loses the per-key shape; an explicit guard restores it and
rejects unexpected payloads at runtime.
"""
if isinstance(value, int) and not isinstance(value, bool):
return value
if isinstance(value, str):
return int(value)
raise BindingPrefillTypeError(f"binding selector 'filing_year_delta' must be int|str, got {type(value).__name__}")
def _selector_periods(value: object) -> tuple[str, ...]:
"""Normalise a binding-selector ``source_periods`` into a tuple of strings."""
if isinstance(value, str):
return (value,)
if isinstance(value, tuple) and all(isinstance(item, str) for item in value):
return tuple(str(item) for item in value)
raise BindingPrefillTypeError(
f"binding selector 'source_periods' must be str|tuple[str,...], got {type(value).__name__}",
)
_LOCAL_FILING_PROVENANCE: Final = "local_filing"
_PRE_ACTIVITY_NO_PRIOR_OBLIGATION_SOURCE_KIND: Final = "pre_activity_no_prior_obligation"
_IVA_COMPENSATION_HISTORY_SOURCE_KIND: Final = "aeat_sede_iva_compensation_history"
_MIXED_OBSERVATION_SOURCE_KIND: Final = "mixed_observation_sources"
def _casilla_id(value: object) -> CasillaId:
try:
return validated_casilla_id(value, surface="binding-prefill casilla constant")
except ValueError as exc:
raise RuntimeError(f"binding-prefill casilla constant {value!r} is not a CasillaId") from exc
_M303_COMPENSACION_PENDIENTE_ANTERIORES_CASILLA: Final[CasillaId] = _casilla_id(
"iva.compensacion-pendiente-periodos-anteriores"
)
_M303_COMPENSACION_APLICADA_CASILLA: Final[CasillaId] = _casilla_id("iva.compensacion-aplicada-periodo")
_M303_POSTERIOR_CASILLA: Final[CasillaId] = _casilla_id("iva.compensacion-pendiente-periodos-posteriores")
_M303_RESULTADO_CASILLA: Final[CasillaId] = _casilla_id("iva.resultado")
_M303_RESULTADO_FINAL_CASILLA: Final[CasillaId] = _casilla_id("71")
_M303_GENERADA_CASILLA: Final[CasillaId] = _casilla_id("iva.compensacion-generada-periodo")
_M303_DISPONIBLE_CASILLA: Final[CasillaId] = _casilla_id("iva.compensacion-disponible-fin-periodo")
def _revision_carry_outcome(payload: _ObservationEnvelopePayload) -> bool:
"""Return whether a payload's revision stamp must be refused.
Thin adapter over the single shared
:func:`~application.calculations._revision_carry_gate.revision_carry_outcome`
gate (ADR 2026-06-10-period-revision-resolution-adr, Ruling 3 / R2): it
extracts the source context off the payload's observation and delegates the
refusal decision so the binding-prefill, cross-period
clean-state, and relation-prefill carry reads share one law-determined
re-confirmation rather than three parallel copies.
"""
obs = payload.observation
return revision_carry_outcome(
payload.stamped_revision_id,
source_modelo=obs.modelo,
source_filing_year=obs.filing_year,
source_period=obs.period,
)
def _revision_prefill_divergence(payload: _ObservationEnvelopePayload) -> bool:
"""Return True when the payload's stamped revision cannot be re-confirmed.
See :func:`_revision_carry_outcome` — a divergent or unreconfirmable stamp
means the carry must be refused (the caller drops the observation).
"""
return _revision_carry_outcome(payload)
class _GatheredObservation(BaseModel):
"""Registry observation plus the persisted source channel that produced it."""
model_config = _STRICT_FROZEN
observation: RegistryModeloObservation
source_kind: str
casilla_source_kinds: Mapping[CasillaId, str]
def _gathered_observation(
observation: RegistryModeloObservation,
*,
source_kind: str,
) -> _GatheredObservation:
return _GatheredObservation(
observation=observation,
source_kind=source_kind,
casilla_source_kinds={item.casilla_id: source_kind for item in observation.observations},
)
def _merge_gathered_observations(
primary: _GatheredObservation,
overlay: _GatheredObservation,
) -> _GatheredObservation:
"""Merge same-filing observations while preserving casilla-level provenance.
The registry previous-filing resolver accepts one observation for each
single-filer ``(modelo, filing_year, period)``. Secure IVA-history
projections therefore have to be folded into the app-filing observation
before resolution, otherwise one source shadows the other.
"""
primary_key = (primary.observation.modelo, primary.observation.filing_year, primary.observation.period)
overlay_key = (overlay.observation.modelo, overlay.observation.filing_year, overlay.observation.period)
if primary_key != overlay_key:
raise BindingPrefillTypeError(
"cannot merge previous_filing observations with different modelo/year/period keys",
)
observations_by_casilla = {item.casilla_id: item for item in primary.observation.observations}
observations_by_casilla.update({item.casilla_id: item for item in overlay.observation.observations})
casilla_source_kinds = {
**dict(primary.casilla_source_kinds),
**dict(overlay.casilla_source_kinds),
}
source_kinds = {primary.source_kind, overlay.source_kind}
source_kind = primary.source_kind if len(source_kinds) == 1 else _MIXED_OBSERVATION_SOURCE_KIND
return _GatheredObservation(
observation=primary.observation.model_copy(update={"observations": tuple(observations_by_casilla.values())}),
source_kind=source_kind,
casilla_source_kinds=casilla_source_kinds,
)
[docs]
class PrefilledBinding(BaseModel):
"""One resolved previous-filing binding with local-source provenance.
Emitted by :func:`resolve_bindings_from_local_store` and collected in
:class:`BindingPrefillReport`. The ``source_*`` fields point back to the
:class:`~domain.calculations.registry.RegistryModeloObservation` rows
that satisfied the registry previous-filing requirement.
"""
model_config = _STRICT_FROZEN
binding_id: BindingId
value: Decimal
provenance: str = _LOCAL_FILING_PROVENANCE
source_kind: str = _LOCAL_FILING_PROVENANCE
source_modelo: str
source_filing_year: int
source_periods: tuple[str, ...]
resolved_at: datetime
[docs]
class BindingPrefillReport(BaseModel):
"""Outcome of one direct previous-filing binding-prefill pass.
``binding_values`` is the mapping passed to
:func:`~domain.calculations.registry.calculate_registry_snapshot`;
``prefilled`` keeps the :class:`PrefilledBinding` provenance used by
:class:`~._multi_year.PreviousFilingSourceResolver` when stamping source-mesh
results.
"""
model_config = _STRICT_FROZEN
prefilled: tuple[PrefilledBinding, ...]
binding_values: Mapping[BindingId, Decimal]
[docs]
class LocalIvaCompensationRecurrence(BaseModel):
"""Local Modelo 303 recurrence evidence extracted for wallet reconciliation only.
This is comparison evidence. It does not choose the effective casilla `110`
value; the :class:`~._iva_wallet_reconciliation.IvaWalletDecisionSourceResolver`
and wallet reconciliation decision remain the only selectors.
"""
model_config = _STRICT_FROZEN
binding_id: BindingId
amount: Decimal
source_kind: str = _LOCAL_FILING_PROVENANCE
source_modelo: str
source_filing_year: int
source_periods: tuple[Period, ...]
resolved_at: datetime
source_locator: str | None = None
def _gather_grouped_member_observations(
req_key: tuple[str, int, str],
*,
repository: CalculationObservationRepository,
needed: dict[tuple[str, int, str, int], _GatheredObservation],
seen_member: dict[tuple[str, int, str], int],
) -> None:
"""Fold every member's filing for ``req_key`` into ``needed``, member-distinct.
Cross-member fan-in for the 353<-322 aggregation: enumerate every stored row
for the modelo (including member-NIF-widened keys) via ``iter_modelo`` rather
than loading one observation by key, so the resolver can sum across members.
Mutates the shared ``needed`` and ``seen_member`` accumulators in place to
preserve the caller's member-index sequencing.
"""
requirement_modelo = req_key[0]
for payload in repository.iter_modelo(requirement_modelo):
obs = payload.observation
if (obs.modelo, obs.filing_year, obs.period) != req_key:
continue
# R2 carry gate: divergent or unreconfirmable stamp -> skip.
if _revision_carry_outcome(payload):
continue
member_idx = seen_member.get(req_key, 0)
seen_member[req_key] = member_idx + 1
needed[(obs.modelo, obs.filing_year, obs.period, member_idx)] = _gathered_observation(
obs,
source_kind=payload.source_kind,
)
def _gathered_from_payload(payload: _ObservationEnvelopePayload | None) -> _GatheredObservation | None:
"""Apply the R2 carry gate to a single-key payload.
Divergent or unreconfirmable source revision stamps refuse the carry.
"""
if payload is None:
return None
if _revision_carry_outcome(payload):
return None
return _gathered_observation(
payload.observation,
source_kind=payload.source_kind,
)
def _gather_single_key_observation(
requirement_modelo: str,
requirement_filing_year: int,
requirement_period: str,
*,
repository: CalculationObservationRepository,
iva_history_repository: IvaCompensationHistoryRepository | None,
) -> _GatheredObservation | None:
"""Load one observation by key, folding in any secure Modelo 303 IVA history.
The single-filer path: one observation per ``(modelo, filing_year, period)``.
For Modelo 303 a secure IVA-compensation-history projection is merged into the
app-filing observation (when both exist) so neither source shadows the other.
"""
gathered = _gathered_from_payload(
repository.load_observation(
requirement_modelo,
Period.from_year_and_code(requirement_filing_year, requirement_period),
),
)
if requirement_modelo == Modelo.M303.value and iva_history_repository is not None:
state = iva_history_repository.load_period(
Period.from_year_and_code(requirement_filing_year, requirement_period),
)
if state is not None:
history_gathered = _gathered_observation(
_observation_from_iva_compensation_history(state),
source_kind=_IVA_COMPENSATION_HISTORY_SOURCE_KIND,
)
gathered = (
_merge_gathered_observations(gathered, history_gathered) if gathered is not None else history_gathered
)
return gathered
def _gather_observations(
snapshot: RegistrySnapshot,
*,
repository: CalculationObservationRepository,
iva_history_repository: IvaCompensationHistoryRepository | None = None,
excluded_binding_ids: frozenset[BindingId] | None = None,
) -> tuple[_GatheredObservation, ...]:
"""Walk every previous_filing binding in the revision and pull matching observations from the local store.
For ordinary single-filer requirements one observation per
``(modelo, filing_year, period)`` is loaded by key
(:func:`_gather_single_key_observation`). For a requirement whose binding
declares ``grouping = "per_grupo_member"`` (the 353<-322 cross-member
aggregation), EVERY member's filing for that ``(modelo, filing_year, period)``
must be gathered (:func:`_gather_grouped_member_observations`), so the resolver
can sum across members.
"""
grouped_keys = _per_grupo_member_requirement_keys(snapshot.revision, snapshot)
excluded = excluded_binding_ids or frozenset()
needed: dict[tuple[str, int, str, int], _GatheredObservation] = {}
seen_member: dict[tuple[str, int, str], int] = {}
for requirement in previous_filing_observation_requirements(
snapshot.revision,
filing_year=snapshot.filing_year,
period=snapshot.period,
):
if requirement.binding_ids and all(binding_id in excluded for binding_id in requirement.binding_ids):
continue
req_key = (requirement.source_modelo, requirement.filing_year, requirement.periods[0])
if req_key in grouped_keys:
_gather_grouped_member_observations(
req_key,
repository=repository,
needed=needed,
seen_member=seen_member,
)
continue
gathered = _gather_single_key_observation(
requirement.source_modelo,
requirement.filing_year,
requirement.periods[0],
repository=repository,
iva_history_repository=iva_history_repository,
)
if gathered is None:
continue
obs = gathered.observation
needed.setdefault(
(obs.modelo, obs.filing_year, obs.period, 0),
gathered,
)
return tuple(needed.values())
def _per_grupo_member_requirement_keys(revision: object, snapshot: RegistrySnapshot) -> set[tuple[str, int, str]]:
"""Return the (modelo, filing_year, period) requirement keys whose binding declares per_grupo_member.
These keys must be gathered by enumeration (every member's filing), not by
single-key load — the cross-member fan-in for the 353<-322 aggregation.
"""
def _is_per_grupo_member(binding: object) -> bool:
if getattr(binding, "source", None) != "previous_filing":
return False
selector = getattr(binding, "selector", None)
if isinstance(selector, dict):
return selector.get("grouping") == "per_grupo_member"
return getattr(selector, "grouping", None) == "per_grupo_member"
grouped_binding_ids = {binding.id for binding in snapshot.revision.bindings if _is_per_grupo_member(binding)}
if not grouped_binding_ids:
return set()
keys: set[tuple[str, int, str]] = set()
for requirement in previous_filing_observation_requirements(
snapshot.revision,
filing_year=snapshot.filing_year,
period=snapshot.period,
):
if any(bid in grouped_binding_ids for bid in requirement.binding_ids):
keys.add((requirement.source_modelo, requirement.filing_year, requirement.periods[0]))
return keys
def _observation_from_iva_compensation_history(
state: IvaCompensationPeriodState,
) -> RegistryModeloObservation:
"""Project secure IVA compensation history into the registry resolver contract."""
snapshot = resources().modelos.authority.snapshot(
Modelo.M303.value,
filing_year=state.filing_year,
period=state.period.registry_token,
)
casillas = {item.id: item for item in snapshot.revision.casillas}
formulas = {item.target_casilla_id: item for item in snapshot.revision.formulas}
def observed(casilla_id: CasillaId, value: Decimal | None) -> tuple[CasillaObservation, ...]:
if value is None:
return ()
operand_refs: tuple[CasillaId, ...] = ()
operand_values: tuple[Decimal, ...] = ()
if casilla_id == _M303_POSTERIOR_CASILLA and (
state.prior_pending_amount is not None and state.applied_amount is not None
):
operand_refs = (_M303_COMPENSACION_PENDIENTE_ANTERIORES_CASILLA, _M303_COMPENSACION_APLICADA_CASILLA)
operand_values = (state.prior_pending_amount, state.applied_amount)
elif casilla_id == _M303_GENERADA_CASILLA and state.period_result_amount is not None:
operand_refs = (_M303_RESULTADO_CASILLA,)
operand_values = (state.period_result_amount,)
elif casilla_id == _M303_DISPONIBLE_CASILLA and (
state.pending_for_later_amount is not None and state.generated_amount is not None
):
operand_refs = (_M303_POSTERIOR_CASILLA, _M303_GENERADA_CASILLA)
operand_values = (state.pending_for_later_amount, state.generated_amount)
return _iva_compensation_history_observation(
snapshot=snapshot,
casillas=casillas,
formulas=formulas,
casilla_id=casilla_id,
value=value,
operand_refs=operand_refs,
operand_values=operand_values,
)
return RegistryModeloObservation(
modelo=Modelo.M303.value,
filing_year=state.filing_year,
period=state.period.registry_token,
observations=(
*observed(_M303_COMPENSACION_PENDIENTE_ANTERIORES_CASILLA, state.prior_pending_amount),
*observed(_M303_COMPENSACION_APLICADA_CASILLA, state.applied_amount),
*observed(_M303_POSTERIOR_CASILLA, state.pending_for_later_amount),
*observed(_M303_RESULTADO_CASILLA, state.period_result_amount),
*observed(_M303_RESULTADO_FINAL_CASILLA, state.final_result_amount),
*observed(_M303_GENERADA_CASILLA, state.generated_amount),
*observed(_M303_DISPONIBLE_CASILLA, state.available_end_amount),
),
)
def _iva_compensation_history_observation(
*,
snapshot: RegistrySnapshot,
casillas: Mapping[CasillaId, CasillaDefinition],
formulas: Mapping[CasillaId, FormulaDefinition],
casilla_id: CasillaId,
value: Decimal,
operand_refs: tuple[CasillaId, ...] = (),
operand_values: tuple[Decimal, ...] = (),
) -> tuple[CasillaObservation, ...]:
"""Project one IVA history casilla, failing if registry provenance is missing."""
casilla = casillas.get(casilla_id)
if casilla is None:
raise IvaCompensationCasillaReferenceError(
f"IVA compensation history casilla {casilla_id!r} is not declared in "
f"modelo {snapshot.modelo.id} revision {snapshot.revision.id}; refusing "
"to emit a CasillaObservation without legal_refs/source_refs",
context={
"modelo": snapshot.modelo.id,
"revision_id": snapshot.revision.id,
"casilla_id": casilla_id,
},
)
formula = formulas.get(casilla_id)
formula_id = None
if len(operand_values) != len(operand_refs):
raise IvaCompensationCasillaReferenceError(
f"IVA compensation history casilla {casilla_id!r} has {len(operand_refs)} operand refs "
f"but {len(operand_values)} operand values; refusing to emit partial formula provenance",
context={
"modelo": snapshot.modelo.id,
"revision_id": snapshot.revision.id,
"casilla_id": casilla_id,
},
)
if operand_refs and formula is None:
raise IvaCompensationCasillaReferenceError(
f"IVA compensation history casilla {casilla_id!r} supplied operand refs but has no "
f"formula in modelo {snapshot.modelo.id} revision {snapshot.revision.id}",
context={
"modelo": snapshot.modelo.id,
"revision_id": snapshot.revision.id,
"casilla_id": casilla_id,
},
)
if formula is not None:
expected_operand_refs = expression_casilla_refs(formula.expression)
if operand_refs:
if operand_refs != expected_operand_refs:
raise IvaCompensationCasillaReferenceError(
f"IVA compensation history casilla {casilla_id!r} supplied operand refs "
f"{operand_refs!r} but formula {formula.id!r} projects to {expected_operand_refs!r}",
context={
"modelo": snapshot.modelo.id,
"revision_id": snapshot.revision.id,
"casilla_id": casilla_id,
"formula_id": formula.id,
},
)
formula_id = formula.id
elif not expected_operand_refs:
formula_id = formula.id
return (
CasillaObservation(
casilla_id=casilla_id,
value=value,
formula_id=formula_id,
operand_refs=operand_refs,
operand_casilla_refs=operand_refs,
operand_values=operand_values,
legal_refs=tuple(casilla.legal_refs),
source_refs=tuple(casilla.source_refs),
),
)
def _requirements_by_binding(
snapshot: RegistrySnapshot,
) -> dict[str, tuple[str, int, tuple[str, ...]]]:
grouped: dict[str, tuple[str, int, set[str]]] = {}
for requirement in previous_filing_observation_requirements(
snapshot.revision,
filing_year=snapshot.filing_year,
period=snapshot.period,
):
for binding_id in requirement.binding_ids:
current = grouped.setdefault(binding_id, (requirement.source_modelo, requirement.filing_year, set()))
current[2].add(requirement.periods[0])
return {
binding_id: (source_modelo, source_year, tuple(sorted(periods)))
for binding_id, (source_modelo, source_year, periods) in grouped.items()
}
def _requirement_strictly_before_activity_start(
requirement: RegistryFoldRequirement,
activity_start_date: date,
) -> bool:
"""Return whether every source period in a previous-filing requirement is pre-activity."""
filing_periods = requirement.filing_periods or tuple(
Period.from_year_and_code(requirement.filing_year, token) for token in requirement.periods
)
if not filing_periods:
return False
return all(period.has_date_span() and period.end_date < activity_start_date for period in filing_periods)
def _pre_activity_scoped_binding_ids(
snapshot: RegistrySnapshot,
activity_start_date: date | None,
) -> frozenset[BindingId]:
"""Binding ids whose every required previous-filing source period is pre-activity."""
if activity_start_date is None:
return frozenset()
requirements_by_binding: dict[BindingId, list[RegistryFoldRequirement]] = {}
for requirement in previous_filing_observation_requirements(
snapshot.revision,
filing_year=snapshot.filing_year,
period=snapshot.period,
):
for binding_id in requirement.binding_ids:
requirements_by_binding.setdefault(binding_id, []).append(requirement)
return frozenset(
binding_id
for binding_id, requirements in requirements_by_binding.items()
if requirements
and all(
_requirement_strictly_before_activity_start(requirement, activity_start_date)
for requirement in requirements
)
)
def _selector_value(selector: object, key: str, default: object) -> object:
if isinstance(selector, dict):
# items() yields (Unknown, object) pairs; filter by key to preserve None-valued entries.
return next((v for k, v in selector.items() if k == key), default)
return getattr(selector, key, default)
def _source_kind_for_binding(
gathered: tuple[_GatheredObservation, ...],
*,
source_modelo: str,
source_filing_year: int,
source_periods: tuple[str, ...],
source_casilla_ids: tuple[CasillaId, ...] = (),
) -> str:
required_periods = set(source_periods)
matched_source_kinds: set[str] = set()
for item in gathered:
if (
item.observation.modelo != source_modelo
or item.observation.filing_year != source_filing_year
or item.observation.period not in required_periods
):
continue
if source_casilla_ids:
matched_source_kinds.update(
item.casilla_source_kinds.get(casilla_id, item.source_kind)
for casilla_id in source_casilla_ids
if casilla_id in item.observation.casilla_values
)
else:
matched_source_kinds.add(item.source_kind)
if len(matched_source_kinds) == 1:
return next(iter(matched_source_kinds))
if matched_source_kinds:
return _MIXED_OBSERVATION_SOURCE_KIND
return _LOCAL_FILING_PROVENANCE
[docs]
def resolve_bindings_from_local_store(
snapshot: RegistrySnapshot,
*,
repository: CalculationObservationRepository | None = None,
iva_history_repository: IvaCompensationHistoryRepository | None = None,
captured_at: datetime | None = None,
activity_start_date: date | None = None,
excluded_binding_ids: frozenset[BindingId] | None = None,
) -> BindingPrefillReport:
"""Resolve every ``previous_filing`` binding the revision declares against observations in the local store.
The caller provides a :class:`RegistrySnapshot`; this function asks the
registry for
:class:`~domain.calculations.registry.RegistryFoldRequirement` records,
loads matching :class:`RegistryModeloObservation` rows, then delegates the
final value calculation to
:func:`~domain.calculations.registry.resolve_previous_filing_binding_values`.
Args:
snapshot: The :class:`RegistrySnapshot` whose revision's ``previous_filing``
bindings are resolved from the local calculation observation store.
repository: Optional :class:`CalculationObservationRepository`;
defaults to the active-bucket repository when ``None``.
iva_history_repository: Optional
:class:`IvaCompensationHistoryRepository` consulted for the IVA
compensation prior-balance bindings; defaults to the
active-bucket repository when ``None``.
captured_at: Optional capture timestamp recorded on the produced
prefill records; defaults to the canonical clock when ``None``.
activity_start_date: Optional operator-declared activity start. Source
periods strictly before this date are no-prior-obligation periods
and resolve to a neutral zero instead of requiring observed filings.
excluded_binding_ids: Optional previous-filing binding ids to leave
unresolved because another authority owns them.
Returns a :class:`BindingPrefillReport` carrying the resolved
``binding_values`` mapping (suitable for passing through
``calculate_registry_snapshot``'s ``binding_values=`` argument) plus
a tuple of ``PrefilledBinding`` records with provenance per entry.
Bindings the local store cannot satisfy are skipped silently —
the engine emits blank cells the operator fills by hand. Strict
enforcement (refusing the export when prior filings are missing)
is the caller's choice via the prefill report's coverage.
See Also:
:class:`~._multi_year.PreviousFilingSourceResolver` adapts this report
to :class:`~application.aggregation.CalculationSourceResolution`;
:func:`~._relation_prefill.resolve_relations_from_local_store` resolves
the separate ``relation_prefill`` source family over the same
observation repository.
"""
repo = repository if repository is not None else CalculationObservationRepository()
# The Modelo 303 IVA-compensation-history merge is NO LONGER an implicit
# default: the live calculate path's compensación value is owned exclusively
# by the iva-wallet decision (ruling D3), so the previous_filing gather stays
# pure (registry observations only). Only the explicit wallet-feeding path
# (extract_modelo_303_local_iva_compensation_recurrence) passes the history
# repository to reconstruct the local recurrence the reconciliation consumes.
when = captured_at if captured_at is not None else now()
observations = _gather_observations(
snapshot,
repository=repo,
iva_history_repository=iva_history_repository,
excluded_binding_ids=excluded_binding_ids,
)
if not observations and activity_start_date is None:
return BindingPrefillReport(prefilled=(), binding_values={})
resolved_map = resolve_previous_filing_binding_values(
snapshot.revision,
tuple(item.observation for item in observations),
filing_year=snapshot.filing_year,
period=snapshot.period,
activity_start_date=activity_start_date,
excluded_binding_ids=excluded_binding_ids,
)
prefilled: list[PrefilledBinding] = []
binding_index = {binding.id: binding for binding in snapshot.revision.bindings}
requirement_index = _requirements_by_binding(snapshot)
pre_activity_zero_binding_ids = _pre_activity_scoped_binding_ids(snapshot, activity_start_date)
for binding_id, value in resolved_map.items():
binding = binding_index.get(binding_id)
if binding is None:
continue
selector = binding.selector
source_modelo, source_filing_year, source_periods = requirement_index.get(
binding_id,
(
str(_selector_value(selector, "source_modelo", "") or ""),
snapshot.filing_year + _selector_year_delta(_selector_value(selector, "filing_year_delta", 0)),
_selector_periods(_selector_value(selector, "source_periods", ())),
),
)
source_casilla_ids = binding_source_casilla_ids(binding)
source_kind = (
_PRE_ACTIVITY_NO_PRIOR_OBLIGATION_SOURCE_KIND
if binding_id in pre_activity_zero_binding_ids
else _source_kind_for_binding(
observations,
source_modelo=source_modelo,
source_filing_year=source_filing_year,
source_periods=source_periods,
source_casilla_ids=source_casilla_ids,
)
)
prefilled.append(
PrefilledBinding(
binding_id=binding_id,
value=Decimal(value),
source_kind=source_kind,
source_modelo=source_modelo,
source_filing_year=source_filing_year,
source_periods=source_periods,
resolved_at=when,
),
)
return BindingPrefillReport(
prefilled=tuple(prefilled),
binding_values=dict(resolved_map),
)
__all__ = [
"BindingPrefillReport",
"LocalIvaCompensationRecurrence",
"PrefilledBinding",
"extract_modelo_303_local_iva_compensation_recurrence",
"resolve_bindings_from_local_store",
]