"""Modelo 369 OSS/IOSS source-mesh resolver and candidate validator.
The ``ledger_oss_aggregation`` source reads OSS/IOSS-tagged issued invoices
through the required application-owned catalogue-read capability and projects
them into substrate-classified :class:`OssIossLedgerCandidate`
rows. Pre-classified callers can also pass candidates directly. Each candidate
is validated against the destination Member State's published IVA rate through
:func:`domain.iva.lookup.lookup_rate` and becomes a registry-ready
:class:`~domain.calculations.registry.ledger_oss_bindings.OssIossLedgerObservation`.
Per the OSS / IOSS regulation suite, the IVA amount on each line MUST
match the destination Member State's published rate for the chosen
rate tier on the supply date. A persisted IVA amount that disagrees
with the lookup is a data-quality blocker: the wrapper rejects it
before the registry resolver sees it, so calculation revisions never
land on inconsistent ledger facts.
The :class:`OssIossLedgerSourceResolver` returns a
:class:`~.source_mesh.CalculationSourceResolution` with resolved binding values,
transaction provenance, and non-blocking diagnostics for empty live catalogues or
declarable OSS observations that no ``ledger_oss_aggregation`` binding consumes.
"""
from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterable, Sequence
from datetime import date
from decimal import Decimal
from typing import Annotated, ClassVar
from pydantic import BaseModel, Field, StringConstraints
from ...core.aggregation import BindingSourceKind, CalculationSourceLineageRole
from ...core.i18n.translatable import Translatable as tr
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.money.rounding import CENT, round_to_cents
from ...core.period import Period
from ...core.time.clock import today_madrid
from ...domain.calculations.registry.authority import PinnedAuthorityOperation, bundled_indexed_authority
from ...domain.calculations.registry.binding_terminal_origin import TerminalOriginClass
from ...domain.calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ...domain.calculations.registry.ids import BindingId
from ...domain.calculations.registry.ledger_oss_bindings import (
OssIossLedgerObservation,
resolve_ledger_oss_aggregation_binding_values,
unsupported_ledger_oss_observations,
)
from ...domain.calculations.registry.manual_input_selector import ManualInputProvider
from ...domain.calculations.registry.schema import ModeloRevision
from ...domain.calculations.registry.schema_base import DateAxis
from ...domain.invoices.enums import iva_rate_kind
from ...domain.invoices.models import Invoice, InvoiceLine
from ...domain.iva.classification import (
InvoiceKind,
TransactionKind,
require_transaction_kind,
resolve_transaction_kind_catalogue,
)
from ...domain.iva.lookup import lookup_rate
from ...domain.iva.oss import OssIossRegime, require_oss_ioss_regime, resolve_oss_ioss_regime_catalogue
from ...domain.iva.schema import EUMemberState, IvaRateKind
from ..invoices.catalogue_reads_ports import InvoiceCatalogueReadPersistenceError, InvoiceCatalogueReadPorts
from .errors import AggregationValidationError
from .invoice_devengo import (
devengo_proxy_attribution_diagnostics,
invoice_devengo_in_period,
resolve_invoice_devengo,
)
from .source_mesh import (
CalculationSourceContext,
CalculationSourceDiagnostic,
CalculationSourceProvenance,
CalculationSourceResolution,
)
from .source_resolution_operations import storage_degradation_resolution
_LedgerId = Annotated[
str,
StringConstraints(strip_whitespace=True, min_length=1, max_length=128),
]
[docs]
class OssIossLedgerCandidate(BaseModel):
"""One un-validated ledger line tagged with substrate classification.
The candidate is the application-layer hand-off shape: a ledger
line carrying the four classification axes the Modelo 369 binding
selectors require, plus the base and IVA amounts the bucket
persists. The :func:`validate_oss_ioss_observation` function
turns a candidate into a registry-ready
:class:`OssIossLedgerObservation` once the persisted IVA amount
has been checked against the destination MS rate.
Attributes:
ledger_id: Stable id of the source ledger line.
transaction_date: When the supply takes place. Drives the rate
lookup.
regime: OSS / IOSS Esquema the line is filed under.
destination_member_state: Member State of consumption per the
OSS / IOSS place-of-supply rules.
rate_kind: Substrate rate tier (general / reduced / etc.).
invoice_direction: Whether the autónomo issued or received the
invoice.
transaction_kind: Substrate
:class:`domain.iva.classification.TransactionKind` the line resolves
to.
base_amount: Taxable base in EUR. Must be non-negative.
iva_amount: IVA amount in EUR persisted on the ledger. Must
be non-negative.
"""
model_config = STRICT_FROZEN_CONFIG
ledger_id: _LedgerId
transaction_date: date
regime: OssIossRegime
destination_member_state: EUMemberState
rate_kind: IvaRateKind
invoice_direction: InvoiceKind
transaction_kind: TransactionKind
base_amount: Decimal = Field(ge=Decimal("0"))
iva_amount: Decimal = Field(ge=Decimal("0"))
def _expected_iva_amount(
candidate: OssIossLedgerCandidate,
*,
operation: PinnedAuthorityOperation,
) -> Decimal:
"""Return the IVA amount derived from the candidate's base and rate.
Looks up the destination Member State's rate for the candidate's
rate tier on the supply date, multiplies by ``base_amount``, and
rounds to two decimal places — the precision at which ledger
amounts are persisted.
Args:
candidate: The substrate-classified ledger line whose base amount
and rate kind are used to derive the expected IVA amount.
operation: Generation-pinned authority operation used for the rate lookup.
Returns:
The expected IVA amount rounded to two decimal places.
"""
rate = lookup_rate(
candidate.destination_member_state,
candidate.rate_kind,
candidate.transaction_date,
operation=operation,
)
derived = candidate.base_amount * rate.pct / Decimal("100")
return round_to_cents(derived)
[docs]
def validate_oss_ioss_observation(
candidate: OssIossLedgerCandidate,
*,
operation: PinnedAuthorityOperation | None = None,
) -> OssIossLedgerObservation:
"""Validate ``candidate`` and return the registry-ready observation.
Looks up the destination Member State's rate at the supply date,
derives the expected IVA amount from ``base_amount`` and the
looked-up rate, and rejects the candidate if the persisted
``iva_amount`` deviates from the derived value by more than
``CENT``.
Args:
candidate: The substrate-classified ledger line to validate.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
Returns:
A registry-ready :class:`OssIossLedgerObservation` carrying
the same identifier, supply date, classification axes, base
amount, and persisted IVA amount as the candidate.
Raises:
AggregationValidationError: When the persisted IVA amount
disagrees with the destination MS rate by more than the
one-cent tolerance.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return validate_oss_ioss_observation(candidate, operation=indexed_operation)
regime = require_oss_ioss_regime(
candidate.regime,
effective_date=candidate.transaction_date,
authority=operation,
)
transaction_kind = require_transaction_kind(
candidate.transaction_kind,
effective_date=candidate.transaction_date,
operation=operation,
)
catalogue = resolve_oss_ioss_regime_catalogue(
effective_date=candidate.transaction_date,
authority=operation,
)
if transaction_kind.value not in catalogue.transaction_kinds_for(regime):
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.invoice_line_rate_kind_unclassifiable"),
)
expected = _expected_iva_amount(candidate, operation=operation)
persisted = round_to_cents(candidate.iva_amount)
# Ledger amounts are rounded to two decimal places at persistence time, so a
# difference of one cent or less is rounding noise rather than a data-quality
# blocker. Larger gaps fail validation and the line is rejected before the
# registry resolver aggregates it. The quantum is CENT because it is the same
# cent the rounding produced, not a threshold this module chose.
if abs(persisted - expected) > CENT:
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.iva_amount_mismatches_destination_rate"),
context={
"ledger_id": candidate.ledger_id,
"destination_member_state": candidate.destination_member_state.value,
"rate_kind": candidate.rate_kind.value,
"transaction_date": candidate.transaction_date.isoformat(),
"base_amount": str(candidate.base_amount),
"persisted_iva_amount": str(persisted),
"expected_iva_amount": str(expected),
},
)
return OssIossLedgerObservation(
ledger_id=candidate.ledger_id,
transaction_date=candidate.transaction_date,
regime=candidate.regime,
destination_member_state=candidate.destination_member_state,
rate_kind=candidate.rate_kind,
invoice_direction=candidate.invoice_direction,
transaction_kind=candidate.transaction_kind,
base_amount=candidate.base_amount,
iva_amount=candidate.iva_amount,
)
[docs]
def validate_oss_ioss_observations(
candidates: Iterable[OssIossLedgerCandidate],
*,
operation: PinnedAuthorityOperation | None = None,
) -> tuple[OssIossLedgerObservation, ...]:
"""Validate every candidate; raise on the first failure.
Args:
candidates: The substrate-classified ledger lines to validate.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
Returns:
A tuple of registry-ready
:class:`OssIossLedgerObservation` records in input order.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return validate_oss_ioss_observations(candidates, operation=indexed_operation)
return tuple(validate_oss_ioss_observation(candidate, operation=operation) for candidate in candidates)
[docs]
def aggregate_oss_ioss_bindings(
revision: ModeloRevision,
candidates: Sequence[OssIossLedgerCandidate],
*,
operation: PinnedAuthorityOperation | None = None,
) -> dict[BindingId, Decimal]:
"""Validate candidates then resolve every ``ledger_oss_aggregation`` binding.
Pipeline:
1. Each candidate is validated through
:func:`validate_oss_ioss_observation`, which checks the
persisted IVA against the destination MS rate.
2. The validated observations are handed off to the registry's
:func:`resolve_ledger_oss_aggregation_binding_values`
resolver, which filters by every binding's selector and
aggregates the matched lines.
Args:
revision: The Modelo 369 :class:`ModeloRevision` whose
``ledger_oss_aggregation`` bindings should be resolved.
candidates: Substrate-classified ledger lines for the period.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
Returns:
A mapping from each binding id on the revision to its
aggregated Decimal value.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_oss_ioss_bindings(revision, candidates, operation=indexed_operation)
observations = validate_oss_ioss_observations(candidates, operation=operation)
return resolve_ledger_oss_aggregation_binding_values(revision, observations)
def _exterior_detail_binding_values(
revision: ModeloRevision,
observations: Sequence[OssIossLedgerObservation],
*,
operation: PinnedAuthorityOperation,
) -> tuple[dict[BindingId, Decimal], dict[BindingId, str]]:
"""Project Exterior service rows from validated invoice observations.
The record-design bindings are generated from AEAT's positional workbook,
so discover their semantic field names from the revision instead of
duplicating generated binding ids here. One row represents one
destination/rate tier; repeated invoice lines in that tier are summed.
"""
if revision.id != "esquema-exterior":
return {}, {}
grouped = _group_exterior_service_observations(observations, operation=operation)
if not grouped:
return {}, {}
declarations = _exterior_projection_declarations(observations[0].transaction_date, operation=operation)
return _bind_exterior_detail_values(revision, grouped, declarations, operation=operation)
_MODELO_369_EXTERIOR_PROJECTION_FACT_ID = "modelo-369-exterior-oss-projection-catalogue"
def _exterior_projection_declarations(
effective_date: date,
*,
operation: PinnedAuthorityOperation,
) -> dict[str, str]:
"""Resolve the dated Modelo 369 Exterior rate-code catalogue."""
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id=_MODELO_369_EXTERIOR_PROJECTION_FACT_ID,
date_axis=DateAxis.TRANSACTION_DATE,
effective_date=effective_date,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.invoice_line_rate_kind_unclassifiable"),
)
declarations = {
entry.key: entry.value
for entry in resolved.payload.entries
if isinstance(entry.key, str) and isinstance(entry.value, str)
}
required = {"rate_code.general", "rate_code.reduced"}
missing = sorted(required - declarations.keys())
if missing:
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.invoice_line_rate_kind_unclassifiable"),
context={"missing": ",".join(missing)},
)
return declarations
def _group_exterior_service_observations(
observations: Sequence[OssIossLedgerObservation],
*,
operation: PinnedAuthorityOperation,
) -> dict[tuple[EUMemberState, IvaRateKind], list[OssIossLedgerObservation]]:
"""Group Exterior service observations by destination and supported rate."""
grouped: dict[tuple[EUMemberState, IvaRateKind], list[OssIossLedgerObservation]] = defaultdict(list)
effective_date = observations[0].transaction_date if observations else today_madrid()
transaction_catalogue = resolve_transaction_kind_catalogue(effective_date, operation=operation)
external_kinds = transaction_catalogue.kinds_for_oss_regime("external_scheme")
external_regime = resolve_oss_ioss_regime_catalogue(
effective_date=effective_date,
authority=operation,
).regime_for_transaction_kind(next(iter(external_kinds)).value)
for observation in observations:
if observation.regime != external_regime:
continue
if observation.transaction_kind not in external_kinds:
continue
_validate_exterior_rate_kind(observation, operation=operation)
grouped[(observation.destination_member_state, observation.rate_kind)].append(observation)
return grouped
def _validate_exterior_rate_kind(
observation: OssIossLedgerObservation,
*,
operation: PinnedAuthorityOperation,
) -> None:
"""Refuse Exterior rate tiers that have no official positional code."""
declarations = _exterior_projection_declarations(observation.transaction_date, operation=operation)
if f"rate_code.{observation.rate_kind.value}" in declarations:
return
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.exterior_rate_kind_unsupported"),
context={
"ledger_id": observation.ledger_id,
"rate_kind": observation.rate_kind.value,
},
)
def _exterior_detail_row_fields(
row: int,
country: EUMemberState,
rate_kind: IvaRateKind,
observations: Sequence[OssIossLedgerObservation],
declarations: dict[str, str],
*,
operation: PinnedAuthorityOperation,
) -> tuple[dict[str, str], dict[str, Decimal]]:
"""Build the workbook field values for one destination/rate row."""
rate = lookup_rate(
country,
rate_kind,
observations[0].transaction_date,
operation=operation,
).pct
# The design writes ISO-3166 alpha-2 codes with its own exceptions ("EL" for
# Greece, "XI" for Northern Ireland), so the wire code is the dated
# declaration's, never the member-state token itself.
country_code = declarations.get(f"country_code.{country.value}")
if country_code is None:
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.exterior_country_code_undeclared"),
context={"member_state": country.value},
)
fields = {
f"3-prestaciones-de-servicios-codigo-de-pais-em-de-consumo-{row}": country_code,
f"3-prestaciones-de-servicios-tipo-iva-{row}": declarations[f"rate_code.{rate_kind.value}"],
}
decimals = {
f"3-prestaciones-de-servicios-tipo-de-iva-{row}": rate,
f"3-prestaciones-de-servicios-base-imponible-{row}": sum(
(item.base_amount for item in observations),
Decimal("0"),
),
f"3-prestaciones-de-servicios-cuota-iva-{row}": sum(
(item.iva_amount for item in observations),
Decimal("0"),
),
}
return fields, decimals
def _bind_exterior_detail_values(
revision: ModeloRevision,
grouped: dict[tuple[EUMemberState, IvaRateKind], list[OssIossLedgerObservation]],
declarations: dict[str, str],
*,
operation: PinnedAuthorityOperation,
) -> tuple[dict[BindingId, Decimal], dict[BindingId, str]]:
"""Resolve generated revision selectors against Exterior row fields."""
decimal_values: dict[BindingId, Decimal] = {}
enum_values: dict[BindingId, str] = {}
for row, ((country, rate_kind), rows) in enumerate(sorted(grouped.items(), key=lambda item: item[0]), start=1):
fields, decimals = _exterior_detail_row_fields(
row,
country,
rate_kind,
rows,
declarations,
operation=operation,
)
_assign_exterior_detail_bindings(revision, fields, decimals, decimal_values, enum_values)
return decimal_values, enum_values
def _assign_exterior_detail_bindings(
revision: ModeloRevision,
fields: dict[str, str],
decimals: dict[str, Decimal],
decimal_values: dict[BindingId, Decimal],
enum_values: dict[BindingId, str],
) -> None:
"""Copy matching generated selector values into their binding channels."""
for binding in revision.bindings:
provider = binding.provider
if not isinstance(provider, ManualInputProvider) or provider.record != "modelo-369-exterior-t36901":
continue
field = provider.field
if field is None:
continue
if field in fields:
enum_values[binding.id] = fields[field]
elif field in decimals:
decimal_values[binding.id] = decimals[field]
def _candidate_for_invoice_line(
invoice: Invoice,
line: InvoiceLine,
*,
line_index: int,
devengo_date: date,
) -> OssIossLedgerCandidate | None:
"""Project one invoice line into an OSS/IOSS candidate, or ``None`` when not applicable.
The regime / destination checks mean the line is not an OSS/IOSS
operation at all -- correctly silent, nothing to declare. A line that
passes both IS a real, declarable OSS/IOSS operation, so a missing
rate_kind or a missing euro conversion from that point on REFUSES rather
than silently drops it: by operator ruling ("aim for explicit red
signals... until schema and api converges"), a would-be-silently-dropped
declarable line is a hard stop, not an advisory beside a smaller-but-green
figure.
"""
if invoice.oss_ioss_regime is None or invoice.oss_transaction_kind is None:
return None
destination = invoice.counterparty_eu_member_state
if destination is None:
return None
rate_kind = line.oss_rate_kind or iva_rate_kind(line.iva_rate)
if rate_kind is None:
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.invoice_line_rate_kind_unclassifiable"),
context={
"invoice_id": invoice.invoice_id,
"invoice_number": invoice.invoice_number,
"line_index": str(line_index),
"iva_rate": str(line.iva_rate.value),
},
)
# OSS/IOSS is cross-border EU B2C by definition, where invoicing in the
# destination Member State's own currency (PLN, SEK, DKK, HUF, CZK, RON,
# BGN...) is the ORDINARY case, not an edge case. line.subtotal /
# line.iva_amount are native to invoice.currency (InvoiceLine carries no
# currency of its own); base_amount/iva_amount on the candidate are
# documented as EUR, so this converts through the same fx_rate resolution
# the invoice-level totals already use (line_amount_eur) rather than the
# invoice-line's raw native fields.
base_amount_eur = invoice.line_amount_eur(line.subtotal)
iva_amount_eur = invoice.line_amount_eur(line.iva_amount)
if base_amount_eur is None or iva_amount_eur is None:
raise AggregationValidationError(
tr("aggregation.oss_ioss.errors.invoice_line_currency_unconverted"),
context={
"invoice_id": invoice.invoice_id,
"invoice_number": invoice.invoice_number,
"currency": invoice.currency,
"line_index": str(line_index),
},
)
return OssIossLedgerCandidate(
ledger_id=f"{invoice.invoice_id}:{line_index}",
transaction_date=devengo_date,
regime=invoice.oss_ioss_regime,
destination_member_state=destination,
rate_kind=rate_kind,
invoice_direction=invoice.kind,
transaction_kind=invoice.oss_transaction_kind,
base_amount=base_amount_eur,
iva_amount=iva_amount_eur,
)
[docs]
class OssIossInvoiceProjection(BaseModel):
"""The Modelo 369 candidates for a period, beside the invoices they came from.
The invoices ride along because the candidate cannot answer for its own
provenance: it carries a resolved devengo date but not whether that date
was declared or substituted, and the advisory the resolver owes the
operator is about exactly that distinction.
Attributes:
candidates: The substrate-classified OSS/IOSS ledger lines.
contributing_invoices: The invoices at least one candidate came from.
"""
model_config = STRICT_FROZEN_CONFIG
candidates: tuple[OssIossLedgerCandidate, ...] = ()
contributing_invoices: tuple[Invoice, ...] = ()
[docs]
def project_oss_ioss_invoices_from_repositories(
*,
period: Period,
ports: InvoiceCatalogueReadPorts,
) -> OssIossInvoiceProjection:
"""Project OSS/IOSS-tagged issued invoices into Modelo 369 ledger candidates.
Attribution is by the LIVA art. 75 devengo date through the shared
:func:`~application.aggregation.invoice_devengo.invoice_devengo_in_period` predicate, so an
operation performed in one quarter and invoiced in the next is declared in
the quarter it was performed, and every path that folds invoices into a
period answers that question the same way.
Args:
period: Filing period whose date span filters issued invoices.
ports: Required application-owned catalogue read capabilities for the
composed profile bucket.
Returns:
The candidates for the period beside the invoices they were projected
from.
"""
projection_period = period
if period.registry_token.startswith("EXT-"):
projection_period = Period.from_year_and_code(
period.filing_year,
period.registry_token.removeprefix("EXT-"),
)
if not projection_period.has_date_span():
return OssIossInvoiceProjection()
candidates: list[OssIossLedgerCandidate] = []
contributing: list[Invoice] = []
for invoice in ports.invoice_reader.load():
if invoice.kind is not InvoiceKind.ISSUED:
continue
if not invoice_devengo_in_period(invoice, period=projection_period):
continue
devengo_date = resolve_invoice_devengo(invoice).devengo_date
projected = [
candidate
for index, line in enumerate(invoice.lines, start=1)
if (candidate := _candidate_for_invoice_line(invoice, line, line_index=index, devengo_date=devengo_date))
is not None
]
if projected:
candidates.extend(projected)
contributing.append(invoice)
return OssIossInvoiceProjection(candidates=tuple(candidates), contributing_invoices=tuple(contributing))
[docs]
def oss_ioss_candidates_from_repositories(
*,
period: Period,
ports: InvoiceCatalogueReadPorts,
) -> tuple[OssIossLedgerCandidate, ...]:
"""Return only the Modelo 369 candidates for the period.
The narrow accessor for callers that aggregate but emit no diagnostics.
Delegates to :func:`project_oss_ioss_invoices_from_repositories` rather
than repeating its selection, so the two cannot attribute a period
differently.
Args:
period: Filing period whose date span filters issued invoices.
ports: Required application-owned catalogue read capabilities for the
composed profile bucket.
Returns:
A tuple of :class:`OssIossLedgerCandidate` rows projected from issued
invoices devengando in the period.
"""
return project_oss_ioss_invoices_from_repositories(
period=period,
ports=ports,
).candidates
[docs]
def aggregate_oss_ioss_from_repositories(
revision: ModeloRevision,
*,
period: Period,
ports: InvoiceCatalogueReadPorts,
operation: PinnedAuthorityOperation | None = None,
) -> dict[BindingId, Decimal]:
"""Resolve Modelo 369 OSS/IOSS bindings from the live invoice catalogue.
Args:
revision: The :class:`ModeloRevision` whose OSS/IOSS bindings are resolved.
period: Filing period whose date span filters issued invoices.
ports: Required application-owned catalogue read capabilities for the
composed profile bucket.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_oss_ioss_from_repositories(
revision,
period=period,
ports=ports,
operation=indexed_operation,
)
return aggregate_oss_ioss_bindings(
revision,
oss_ioss_candidates_from_repositories(period=period, ports=ports),
operation=operation,
)
[docs]
class OssIossLedgerSourceResolver:
"""Source mesh resolver for Modelo 369 OSS / IOSS ledger candidates."""
resolver_id: ClassVar[str] = "ledger_oss_aggregation"
owned_sources: ClassVar[tuple[BindingSourceKind, ...]] = (BindingSourceKind.LEDGER_OSS_AGGREGATION,)
def __init__(
self,
*,
ports: InvoiceCatalogueReadPorts,
candidates: Sequence[OssIossLedgerCandidate] | None = None,
) -> None:
"""Construct the resolver with a pre-classified ledger candidate sequence.
Args:
candidates: The substrate-classified ledger lines for the
current period. The resolver validates and aggregates
these on each :meth:`resolve` call.
ports: Required application-owned catalogue read capabilities used
to project OSS/IOSS-tagged invoices when ``candidates`` is not
supplied.
"""
self._ports = ports
self._candidates = tuple(candidates) if candidates is not None else None
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
"""Validate candidates and return the resolved OSS/IOSS binding values.
Validates every candidate through :func:`validate_oss_ioss_observation`
then delegates to the registry's
``resolve_ledger_oss_aggregation_binding_values`` to aggregate the
matched lines per binding selector.
When the resolver is constructed with explicit ``candidates``, those
candidates are folded directly. When ``candidates`` is omitted, the live
operator path projects OSS/IOSS-tagged issued invoices from the
application catalogue read capability into candidates first. If no
candidate is available, the
resolver still CLAIMS ``ledger_oss_aggregation`` (so the binding
compiles and is not flagged as a novel source) but surfaces one
non-blocking ``oss_no_live_source`` advisory per declared OSS binding.
This keeps a Modelo 369 OSS cuota from resolving to a SILENT
claimed-zero when the catalogue carries no classifiable OSS invoices.
Args:
context: The :class:`CalculationSourceContext` carrying the
``ModeloRevision`` whose ``ledger_oss_aggregation`` bindings
should be resolved.
Returns:
A :class:`CalculationSourceResolution` with resolved
binding values, source transaction ids, and per-observation
provenance records — or, when no candidates were supplied, an
empty resolution carrying one ``oss_no_live_source`` advisory per
declared OSS binding when no candidates can be projected.
Raises:
AggregationValidationError: When any candidate's persisted IVA
amount disagrees with the destination Member State rate by
more than one cent.
"""
try:
with bundled_indexed_authority().operation() as indexed_operation:
projection = (
project_oss_ioss_invoices_from_repositories(
period=context.period,
ports=self._ports,
)
if self._candidates is None
else OssIossInvoiceProjection(candidates=self._candidates)
)
candidates = projection.candidates
observations = validate_oss_ioss_observations(candidates, operation=indexed_operation)
exterior_values, exterior_enum_values = _exterior_detail_binding_values(
context.revision,
observations,
operation=indexed_operation,
)
except InvoiceCatalogueReadPersistenceError as exc:
return storage_degradation_resolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
source_kinds=self.owned_sources,
error=exc,
)
if not candidates:
return CalculationSourceResolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
diagnostics=self._no_live_source_diagnostics(context),
)
# Fail-closed advisory parity with the IVA screen: a non-zero declarable
# OSS line whose classification tuple matches no ledger_oss_aggregation
# binding would otherwise be silently dropped (no-silent-under-declaration).
unrouted = unsupported_ledger_oss_observations(context.revision, observations)
return CalculationSourceResolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
binding_values={
**resolve_ledger_oss_aggregation_binding_values(context.revision, observations),
**exterior_values,
},
enum_binding_values=exterior_enum_values,
source_transaction_ids=tuple(
sorted({observation.ledger_id.split(":", 1)[0] for observation in observations}),
),
diagnostics=devengo_proxy_attribution_diagnostics(
projection.contributing_invoices,
source_kind="ledger_oss_aggregation",
resolver_id=self.resolver_id,
)
+ tuple(
CalculationSourceDiagnostic(
reason="unrouted_observation",
source_kind="ledger_oss_aggregation",
resolver_id=self.resolver_id,
source_ref=f"transaction:{observation.ledger_id}",
message=(
f"declarable OSS observation {observation.ledger_id!r} "
f"(regime={observation.regime.value!r}, "
f"destination={observation.destination_member_state.value!r}, "
f"rate_kind={observation.rate_kind.value!r}, "
f"invoice_direction={observation.invoice_direction.value!r}, "
f"transaction_kind={observation.transaction_kind.value!r}) is not consumed by any "
f"ledger_oss_aggregation binding on revision {context.revision.id!r}; "
"its base/cuota is not declared on this calculation"
),
)
for observation in unrouted
),
provenance=tuple(
CalculationSourceProvenance(
resolver_id=self.resolver_id,
resolved_binding_source=BindingSourceKind.LEDGER_OSS_AGGREGATION,
contributor_source_kind="ledger_oss_aggregation",
contributor_binding_source=BindingSourceKind.LEDGER_OSS_AGGREGATION,
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=f"transaction:{observation.ledger_id}",
parent_source_ref=None,
terminal_origin=TerminalOriginClass.LEDGER_AGGREGATE,
)
for observation in observations
),
)
def _no_live_source_diagnostics(self, context: CalculationSourceContext) -> tuple[CalculationSourceDiagnostic, ...]:
"""Return one ``oss_no_live_source`` advisory per OSS binding on the revision.
Surfaced only when the resolver was constructed with no candidates and
the revision actually declares ``ledger_oss_aggregation`` bindings, so
a modelo with no OSS bindings produces no spurious advisory.
"""
return tuple(
CalculationSourceDiagnostic(
reason="oss_no_live_source",
source_kind="ledger_oss_aggregation",
resolver_id=self.resolver_id,
binding_id=binding.id,
message=(
f"binding {binding.id!r} declares source 'ledger_oss_aggregation' but no "
"OSS/IOSS-tagged issued invoice line was available for the filing period; "
"the OSS cuota is NOT auto-computed from the invoice catalogue"
),
)
for binding in context.revision.bindings
if str(binding.source) == "ledger_oss_aggregation"
)
__all__ = [
"OssIossInvoiceProjection",
"OssIossLedgerCandidate",
"OssIossLedgerSourceResolver",
"aggregate_oss_ioss_bindings",
"aggregate_oss_ioss_from_repositories",
"oss_ioss_candidates_from_repositories",
"project_oss_ioss_invoices_from_repositories",
"validate_oss_ioss_observation",
"validate_oss_ioss_observations",
]