"""Resolve the regulatory values an invoice-reading prompt enumerates.
The reading prompt names Spanish IVA rates, statutory retención rates, the
categories of document that print no tax at all, and the regime mentions an
issuer is obliged to print. Every one of those is a regulatory value, so
``aeat-registry-authority-flow`` puts its home in the registry rather than in the
module that happens to consume it -- they are versioned by filing year, and a
copy taken anywhere else bakes one year's law into that call site.
**Why the resolution lives HERE and not beside the prompt.** The prompt is
rendered in the ``llm`` package, which is an adapter over a model transport. An
adapter that reaches into :mod:`~domain.iva` and :mod:`~domain.transactions` to
look up a rate has made itself a second consumer of the calculation authorities,
and the hexagonal direction this project keeps says the application layer
resolves and the adapter receives. So the resolution is done once here and
handed down as ONE typed value object; the renderer substitutes what it is given
and can no longer reach an authority at all.
That is a stronger guarantee than "we remembered not to write ``21``". A literal
in a prompt is the least-audited literal in the codebase -- nothing type-checks
it, no gate reads it, and a stale rate keeps steering a reading model silently.
Removing the renderer's *access* to the authorities means the only rate it can
print is one this function resolved.
**Every overlapping record, not the one in force on a chosen day.** A period is a
span, and RD-ley 4/2024 stepped part of the reducido and super-reducido tiers
mid-year. Reading the table on one date inside such a period would omit a rate
that documents dated inside the same period genuinely print, and the prompt would
then be telling a model that a rate it can see on the page is not registered.
See Also:
:class:`~core.period.Period`
The law-determined coordinate rates are resolved against. It is the
caller's ``(filing_year, code)``, never a stored revision id fed back
into resolution.
:class:`~domain.iva.schema.IvaCategory`
The closed set the no-printed-tax vocabulary is derived from.
"""
from __future__ import annotations
from contextlib import suppress
from datetime import timedelta
from decimal import Decimal
from typing import TYPE_CHECKING
from pydantic import BaseModel, Field
from ...core.decimal.constants import HUNDRED, ZERO
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.period import Period
from ...domain.calculations.registry.governed_fact_scope import (
cache_governed_projection,
governed_facts_in_scope,
validating_governed_facts,
)
from ...domain.iva.schema import IvaCategory
if TYPE_CHECKING:
from collections.abc import Iterable
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
__all__ = [
"InvoiceExtractionAuthorityValues",
"default_invoice_extraction_period",
"resolve_invoice_extraction_authority_values",
]
def _overlapping_iva_rate_pcts(
period: Period,
*,
operation: PinnedAuthorityOperation,
) -> tuple[Decimal, ...]:
"""Return every registered Spanish IVA percentage overlapping ``period``.
Every lookup goes through the retained typed IVA facade at a representative
devengo day for each authority validity interval. Registry variants are
piecewise constant between their declared boundaries, so evaluating those
boundaries preserves every mid-period transition without repeating the same
resolution for every calendar day.
"""
from ...domain.calculations.registry.iva_rate_kind_catalogue import resolve_iva_rate_kind_catalogue
from ...domain.invoices.enums import iva_rate_percentage, resolve_iva_rate_slot
from ...domain.iva.errors import IvaRateNotFoundError
from ...domain.iva.lookup import coexisting_tier_rates, lookup_rate
from ...domain.iva.rates import load_iva_rate_table
from ...domain.iva.schema import spanish_eu_member_state
overlapping: set[Decimal] = set()
boundary_dates = {period.start_date}
spanish_state = spanish_eu_member_state(effective_date=period.start_date, authority=operation)
for record in load_iva_rate_table(operation=operation).get(spanish_state, ()):
if period.start_date <= record.effective_from <= period.end_date:
boundary_dates.add(record.effective_from)
if record.effective_until is not None and record.effective_until < period.end_date:
following = record.effective_until + timedelta(days=1)
if period.start_date <= following <= period.end_date:
boundary_dates.add(following)
for fact_id in ("iva-rate-slot-catalogue", "eu-member-state-catalogue"):
fact = operation.governed_fact(fact_id)
for variant in fact.variants:
if variant.valid_from is not None and period.start_date <= variant.valid_from <= period.end_date:
boundary_dates.add(variant.valid_from)
if variant.valid_to is not None and variant.valid_to < period.end_date:
following = variant.valid_to + timedelta(days=1)
if period.start_date <= following <= period.end_date:
boundary_dates.add(following)
for on_date in sorted(boundary_dates):
catalogue = resolve_iva_rate_kind_catalogue(effective_date=on_date, authority=operation)
kinds = tuple(definition.token for definition in catalogue.definitions)
for kind in kinds:
if kind == catalogue.zero_token:
# RATE_0 is a permanent registry slot, while 0062's zero rows
# are date-bounded temporary measures. Project the slot's
# zero value through 0094 instead of treating the absence of a
# temporary 0062 row as proof that zero is unavailable.
zero_slot = resolve_iva_rate_slot(ZERO, on_date)
zero_fraction = iva_rate_percentage(zero_slot, on_date)
if zero_fraction is not None:
overlapping.add(zero_fraction * HUNDRED)
continue
with suppress(IvaRateNotFoundError):
overlapping.add(
lookup_rate(
spanish_eu_member_state(effective_date=on_date, authority=operation),
kind,
on_date,
operation=operation,
).pct,
)
overlapping.update(
rate.pct
for rate in coexisting_tier_rates(
spanish_eu_member_state(effective_date=on_date, authority=operation),
kind,
on_date,
operation=operation,
)
)
return tuple(sorted(overlapping))
def _as_pcts(fractions: Iterable[Decimal]) -> tuple[Decimal, ...]:
"""Return ``fractions`` as ascending percentages.
The retención parameters are stored as fractions (``0.15``) because that is
how a transaction carries one; a printed invoice states the percentage, so
the prompt is given percentages and the conversion happens once, here.
"""
return tuple(sorted(fraction * HUNDRED for fraction in fractions))
@cache_governed_projection(maxsize=64)
def _resolve_scoped_invoice_extraction_authority_values(period: Period) -> InvoiceExtractionAuthorityValues:
"""Resolve one period once for the immutable authority generation in scope."""
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
operation = governed_facts_in_scope()
if not isinstance(operation, PinnedAuthorityOperation):
raise TypeError("invoice extraction authority requires a pinned published authority operation")
from ...domain.iva.components import registry_category_projection
from ...domain.iva.regime_legend import regime_legend_phrases, resolve_regime_legends
from ...domain.transactions.retencion_facts import statutory_activity_retencion_rates
return InvoiceExtractionAuthorityValues(
period=period,
iva_rate_pcts=_overlapping_iva_rate_pcts(period, operation=operation),
retencion_rate_pcts=_as_pcts(statutory_activity_retencion_rates(effective_date=period.end_date)),
no_printed_tax_categories=tuple(
sorted(
registry_category_projection(
"no_printed_tax",
effective_date=period.end_date,
authority=operation,
),
key=lambda member: member.value,
),
),
regime_legend_phrases=regime_legend_phrases(
resolve_regime_legends(operation=operation, effective_date=period.end_date),
),
)