"""Compiled annual Orden projection and snapshot models."""
from __future__ import annotations
from collections.abc import Mapping
from decimal import Decimal
from pydantic import Field, field_validator, model_validator
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.filing_year import FilingYear
from ....core.identity.digest import ContentDigest
from ....core.percentage import Percentage
from ....core.text_bounds import PositiveCount
from ....domain.iva.regimen_simplificado_rows import (
ActividadOrdenAnual,
ActividadOrdenAnualId,
AutoridadAgricolaOrdenAnualNoResuelta,
DificilJustificacionOrdenAnual,
IndiceTemporadaOrdenAnual,
M303RegimenSimplificadoScopeDecision,
PorcentajeIngresoCuentaIaeOrdenAnual,
ReduccionLorcaOrdenAnual,
)
from .errors import RegistryValidationError
from .ids import LegalRefId, RevisionId, SourceRefId
from .m303_orden_constants import (
EXPECTED_ACTIVITY_COUNT,
EXPECTED_NON_AGRICULTURAL_INGRESO_A_CUENTA_COUNT,
EXTRACTOR_VERSION,
validate_percentage_shape,
validate_seasonal_index_shape,
)
from .m303_orden_constants import (
validate_2022_annual_orden_coordinate as _validate_2022_annual_orden_coordinate,
)
from .m303_orden_constants import (
validate_generated_source_axis_shape as _validate_generated_source_axis_shape,
)
from .m303_orden_constants import (
validate_generated_source_counts as _validate_generated_source_counts,
)
from .schema_base import RegistryModel, RegistrySourceKind
from .schema_references import LegalReference, SourceReference, TemporalSupportEnvelope
_M303_2022_RECORD_DESIGN_SOURCE_REF = "aeat-dr-303-2022"
_M303_2022_RECORD_DESIGN_SOURCE_DIGEST = "6648f6b319579e49cd5bfdaae69e7451db75767e7f19da0b90383b25b79b3f60"
def _require_invariant(condition: bool, message: str) -> None:
if not condition:
raise RegistryValidationError(message)
def _has_m303_2022_record_design_source(
source_ref: str,
source_content_digest: str,
*,
record_design_epoch: str | None = None,
) -> bool:
"""Return whether one source axis retains the pinned 2022 design identity."""
return (source_ref, source_content_digest) == (
_M303_2022_RECORD_DESIGN_SOURCE_REF,
_M303_2022_RECORD_DESIGN_SOURCE_DIGEST,
) and record_design_epoch in (None, "2022")
[docs]
class ActividadOrdenAnualRef(RegistryModel):
"""Immutable reference to one source-pinned annual-Orden activity row."""
orden_id: ActividadOrdenAnualId
ejercicio: FilingYear
registry_revision_id: RevisionId
source_ref: SourceRefId
source_content_digest: ContentDigest
[docs]
class M303AnnualOrdenGeneratedSource(RegistryModel):
"""One source-level invariant emitted by the annual Orden generator."""
ejercicio: FilingYear
source_ref: SourceRefId
source_content_digest: ContentDigest
activity_table_count: PositiveCount
module_row_count: PositiveCount
module_distribution: tuple[int, ...] = Field(min_length=7, max_length=7)
agricultural_index_row_count: PositiveCount
agricultural_ingreso_a_cuenta_row_count: PositiveCount
non_agricultural_ingreso_a_cuenta_row_count: PositiveCount
seasonal_index_day_bands: tuple[tuple[int, int], ...] = Field(min_length=3, max_length=3)
seasonal_index_coefficients: tuple[Decimal, ...] = Field(min_length=3, max_length=3)
difficult_justification_pct: Percentage
lorca_reduction_pct: Decimal | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _is_the_exact_supported_annual_quota_shape(self) -> M303AnnualOrdenGeneratedSource:
_validate_generated_source_counts(self)
_validate_generated_source_axis_shape(self)
return self
[docs]
class M303AnnualOrdenGeneratedManifest(RegistryModel):
"""Checked registry artefact that pins the source compiler's exact output."""
extractor_version: str = Field(min_length=1)
sources: tuple[M303AnnualOrdenGeneratedSource, ...] = Field(min_length=1)
@model_validator(mode="after")
@pydantic_validation_boundary
def _has_one_complete_source_per_supported_year(self) -> M303AnnualOrdenGeneratedManifest:
_require_invariant(
self.extractor_version == EXTRACTOR_VERSION,
"annual Orden manifest was generated by an incompatible extractor",
)
_require_invariant(
tuple(item.ejercicio for item in self.sources) == tuple(sorted({item.ejercicio for item in self.sources})),
"annual Orden manifest must list ejercicios exactly once in ascending order",
)
_require_invariant(
len({item.source_ref for item in self.sources}) == len(self.sources),
"annual Orden manifest cannot reuse a source across annual exercises",
)
return self
[docs]
class M303AnnualOrdenCompilation(RegistryModel):
"""Transient compiler output folded directly onto the registry catalogues."""
authority: M303AnnualOrdenAuthority
legal: Mapping[LegalRefId, LegalReference]
[docs]
class M303AnnualOrdenProjection(RegistryModel):
"""Immutable taxonomy for one Modelo 303 year/revision/source coordinate."""
ejercicio: FilingYear
registry_revision_id: RevisionId
source_ref: SourceRefId
source_content_digest: ContentDigest
activities: tuple[ActividadOrdenAnual, ...] = Field(min_length=EXPECTED_ACTIVITY_COUNT)
agricultural_authority: AutoridadAgricolaOrdenAnualNoResuelta
non_agricultural_ingresos_a_cuenta: tuple[PorcentajeIngresoCuentaIaeOrdenAnual, ...] = Field(min_length=1)
seasonal_indexes: tuple[IndiceTemporadaOrdenAnual, ...] = Field(min_length=1)
difficult_justification: DificilJustificacionOrdenAnual
lorca_reduction: ReduccionLorcaOrdenAnual | None
@model_validator(mode="after")
@pydantic_validation_boundary
def _rows_are_complete_and_year_scoped(self) -> M303AnnualOrdenProjection:
_validate_projection_activity_rows(self)
_validate_projection_iae_identity(self)
_validate_projection_axis_shape(self)
return self
def _validate_projection_activity_rows(projection: M303AnnualOrdenProjection) -> None:
_require_invariant(
len(projection.activities) == EXPECTED_ACTIVITY_COUNT,
"annual Orden projection is incomplete",
)
_require_invariant(
not any(activity.ejercicio != projection.ejercicio for activity in projection.activities),
"annual Orden activities must match their projection ejercicio",
)
ids = tuple(activity.orden_id for activity in projection.activities)
_require_invariant(
len(set(ids)) == len(ids),
"annual Orden projection contains duplicate orden_id values",
)
def _validate_projection_iae_identity(projection: M303AnnualOrdenProjection) -> None:
by_iae: dict[str, list[str | None]] = {}
for activity in projection.activities:
if activity.kind == "no_agricola" and activity.iae_epigrafe is not None:
by_iae.setdefault(activity.iae_epigrafe, []).append(activity.auxiliary_activity_indicator)
for indicators in by_iae.values():
_require_invariant(
len(indicators) <= 1 or (None not in indicators and len(set(indicators)) == len(indicators)),
"annual Orden projection contains ambiguous non-agricultural activity identities",
)
def _validate_projection_axis_shape(projection: M303AnnualOrdenProjection) -> None:
_require_invariant(
len(projection.non_agricultural_ingresos_a_cuenta) == EXPECTED_NON_AGRICULTURAL_INGRESO_A_CUENTA_COUNT,
"annual Orden projection has the wrong IAE ingreso-a-cuenta row count",
)
validate_seasonal_index_shape(
tuple((item.minimum_days, item.maximum_days) for item in projection.seasonal_indexes),
tuple(item.coefficient for item in projection.seasonal_indexes),
scope="projection",
)
_require_invariant(
projection.agricultural_authority.annual_orden_source_ref == projection.source_ref,
"annual Orden agricultural refusal must retain its exact annual source reference",
)
_require_invariant(
projection.ejercicio != 2022
or _has_m303_2022_record_design_source(
projection.agricultural_authority.record_design_source_ref,
projection.agricultural_authority.record_design_source_content_digest,
),
"annual Orden 2022 agricultural refusal must retain its exact AEAT design source",
)
reduction = projection.lorca_reduction
if projection.ejercicio == 2022:
_validate_2022_annual_orden_coordinate(
ejercicio=projection.ejercicio,
registry_revision_id=projection.registry_revision_id,
source_ref=projection.source_ref,
source_content_digest=projection.source_content_digest,
scope="projection",
)
if reduction is None:
raise RegistryValidationError("annual Orden 2022 projection lacks its Lorca reduction authority")
if reduction is not None:
validate_percentage_shape(reduction.percentage, scope="projection", subject="Lorca reduction")
_require_invariant(
reduction.source_refs == (projection.source_ref,),
"annual Orden Lorca reduction must retain its exact source reference",
)
[docs]
class M303AnnualOrdenAuthority(RegistryModel):
"""Compiled annual-Orden projections, retained solely by registry snapshots."""
projections: tuple[M303AnnualOrdenProjection, ...] = ()
[docs]
@classmethod
def empty(cls) -> M303AnnualOrdenAuthority:
"""Return an explicitly unresolved state; production M303 resolution refuses it."""
return cls(projections=())
@model_validator(mode="after")
@pydantic_validation_boundary
def _projection_coordinates_are_unique(self) -> M303AnnualOrdenAuthority:
coordinates = tuple((item.ejercicio, item.registry_revision_id) for item in self.projections)
_require_invariant(
len(set(coordinates)) == len(coordinates),
"annual Orden authority contains ambiguous year/revision projections",
)
return self
[docs]
def require_projection(self, *, ejercicio: int, registry_revision_id: RevisionId) -> M303AnnualOrdenProjection:
"""Resolve the sole projection for an exact filing-year/revision coordinate."""
candidates = tuple(
item
for item in self.projections
if item.ejercicio == ejercicio and item.registry_revision_id == registry_revision_id
)
if len(candidates) != 1:
raise RegistryValidationError(
"modelo 303 annual Orden authority requires exactly one projection for "
f"ejercicio {ejercicio} revision {registry_revision_id!r}; found {len(candidates)}",
)
return candidates[0]
[docs]
class M303AnnualOrdenSnapshot(RegistryModel):
"""The source-bound annual Orden resolved for one filing snapshot."""
ejercicio: FilingYear
registry_revision_id: RevisionId
source_ref: SourceRefId
source_content_digest: ContentDigest
activities: tuple[ActividadOrdenAnual, ...] = Field(min_length=EXPECTED_ACTIVITY_COUNT)
activity_refs: tuple[ActividadOrdenAnualRef, ...] = Field(min_length=EXPECTED_ACTIVITY_COUNT)
agricultural_authority: AutoridadAgricolaOrdenAnualNoResuelta
non_agricultural_ingresos_a_cuenta: tuple[PorcentajeIngresoCuentaIaeOrdenAnual, ...] = Field(min_length=1)
seasonal_indexes: tuple[IndiceTemporadaOrdenAnual, ...] = Field(min_length=1)
difficult_justification: DificilJustificacionOrdenAnual
lorca_reduction: ReduccionLorcaOrdenAnual | None
@model_validator(mode="after")
@pydantic_validation_boundary
def _references_match_the_activity_rows(self) -> M303AnnualOrdenSnapshot:
_validate_snapshot_activity_coordinates(self)
_validate_snapshot_source_authority(self)
return self
def _validate_snapshot_activity_coordinates(snapshot: M303AnnualOrdenSnapshot) -> None:
_require_invariant(
tuple(item.orden_id for item in snapshot.activities) == tuple(item.orden_id for item in snapshot.activity_refs),
"annual Orden snapshot refs must exactly match its activity rows",
)
expected_coordinate = (
snapshot.ejercicio,
snapshot.registry_revision_id,
snapshot.source_ref,
snapshot.source_content_digest,
)
_require_invariant(
not any(
(
item.ejercicio,
item.registry_revision_id,
item.source_ref,
item.source_content_digest,
)
!= expected_coordinate
for item in snapshot.activity_refs
),
"annual Orden snapshot refs must retain its exact source coordinate",
)
_require_invariant(
not any(
activity.ejercicio != snapshot.ejercicio or snapshot.source_ref not in activity.source_refs
for activity in snapshot.activities
),
"annual Orden snapshot activities must retain its exact source coordinate",
)
def _validate_snapshot_source_authority(snapshot: M303AnnualOrdenSnapshot) -> None:
reduction = snapshot.lorca_reduction
if snapshot.ejercicio == 2022:
_require_invariant(
reduction is not None and reduction.percentage == Decimal("20"),
"annual Orden 2022 snapshot lacks its Lorca reduction authority",
)
if reduction is not None:
_require_invariant(
reduction.ejercicio == snapshot.ejercicio and reduction.source_refs == (snapshot.source_ref,),
"annual Orden snapshot Lorca reduction must retain its exact year and source reference",
)
_require_invariant(
snapshot.agricultural_authority.annual_orden_source_ref == snapshot.source_ref,
"annual Orden snapshot agricultural refusal must retain its exact source reference",
)
if snapshot.ejercicio == 2022:
_validate_2022_annual_orden_coordinate(
ejercicio=snapshot.ejercicio,
registry_revision_id=snapshot.registry_revision_id,
source_ref=snapshot.source_ref,
source_content_digest=snapshot.source_content_digest,
scope="snapshot",
)
_require_invariant(
_has_m303_2022_record_design_source(
snapshot.agricultural_authority.record_design_source_ref,
snapshot.agricultural_authority.record_design_source_content_digest,
),
"annual Orden 2022 snapshot agricultural refusal must retain its exact AEAT design source",
)
[docs]
class M303RegimenSimplificadoSnapshot(RegistryModel):
"""One resolved M303 annual-Orden, scope, and record-design coordinate."""
filing_year: FilingYear
support: TemporalSupportEnvelope | None = None
registry_revision_id: RevisionId
scope_decision: M303RegimenSimplificadoScopeDecision
orden: M303AnnualOrdenSnapshot
record_design: SourceReference
@field_validator("support", mode="after")
@classmethod
@pydantic_validation_boundary
def _support_is_the_persisted_envelope(
cls, value: TemporalSupportEnvelope | None
) -> TemporalSupportEnvelope | None:
# A richer catalogue subclass would survive in memory but reload as the
# declared envelope, so the stored and computed snapshots would differ.
if value is None or type(value) is TemporalSupportEnvelope:
return value
return TemporalSupportEnvelope(floor=value.floor, horizon=value.horizon, hard_ceiling=value.hard_ceiling)
@model_validator(mode="after")
@pydantic_validation_boundary
def _coordinates_are_complete_and_source_pinned(self) -> M303RegimenSimplificadoSnapshot:
_validate_regimen_simplificado_coordinate(self)
return self
def _validate_regimen_simplificado_coordinate(snapshot: M303RegimenSimplificadoSnapshot) -> None:
_require_invariant(
snapshot.record_design.kind is RegistrySourceKind.RECORD_DESIGN
and snapshot.record_design.record_design_epoch is not None,
"M303 regimen simplificado snapshot requires an epoch-pinned record design",
)
_require_invariant(
(
snapshot.filing_year == snapshot.orden.ejercicio
or (
snapshot.support is not None
and snapshot.support.projection_coordinate(snapshot.filing_year) == snapshot.orden.ejercicio
)
)
and snapshot.registry_revision_id == snapshot.orden.registry_revision_id,
"M303 regimen simplificado snapshot must retain its filing year and revision coordinate",
)
agricultural = snapshot.orden.agricultural_authority
_require_invariant(
agricultural.record_design_source_ref == snapshot.record_design.id
and agricultural.record_design_source_content_digest == snapshot.record_design.sha256,
"M303 regimen simplificado snapshot agricultural refusal must retain its resolved record-design source",
)
if snapshot.filing_year == 2022:
_validate_2022_annual_orden_coordinate(
ejercicio=snapshot.orden.ejercicio,
registry_revision_id=snapshot.orden.registry_revision_id,
source_ref=snapshot.orden.source_ref,
source_content_digest=snapshot.orden.source_content_digest,
scope="resolved snapshot",
)
_require_invariant(
_has_m303_2022_record_design_source(
snapshot.record_design.id,
snapshot.record_design.sha256,
record_design_epoch=snapshot.record_design.record_design_epoch,
),
"M303 regimen simplificado 2022 snapshot must retain its exact AEAT design source",
)