"""Registry reference, source, and temporal coordinate schema models."""
from __future__ import annotations
from datetime import date
from enum import StrEnum, auto
from typing import Annotated, Final, Literal, Self
from typing import override as typing_override
from pydantic import AfterValidator, BeforeValidator, Field, field_validator, model_validator
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.external_constants import (
PDF_EXTENSION,
XLS_EXTENSION,
XLSM_EXTENSION,
XLSX_EXTENSION,
)
from ....core.filing_year import FilingYear
from ....core.identity.digest import ContentDigest
from ....core.period import RegistryPeriodCode, RegistrySelectorPeriodCode
from ....core.record_design_epoch import RECORD_DESIGN_EPOCH_RE
from ....core.revision_review import REVIEWED_REVISION_REVIEW_STATUSES, RevisionReviewStatus
from ....core.url_validation import ANY_HTTP_URL_ADAPTER
from .errors import RegistryValidationError
from .ids import LegalRefId, ModeloId, RevisionId, SourceRefId
from .schema_base import (
CorpusTierField,
DesignAuthority,
EvidenceTier,
EvidenceTierField,
PublishingAuthorityField,
RegistryModel,
RegistrySourceKind,
RegistrySourceKindField,
RevisionReviewStatusField,
coerce_enum_member,
)
__all__ = [
"DateSupportEnvelope",
"LegalReference",
"OrderedSupportEnvelope",
"PeriodOverride",
"PeriodScopedValidityWindow",
"PeriodSelector",
"RegistryExternalLink",
"RegistrySnapshotRef",
"RegistryTemporalBounds",
"RegistryValidityWindow",
"SourceReference",
"TemporalProjectionDirection",
"TemporalSupportEnvelope",
"source_window_applies_across",
]
@pydantic_validation_boundary
def _validate_registry_external_link(value: str) -> str:
"""Validate one authoritative link while retaining registry string semantics."""
parsed = ANY_HTTP_URL_ADAPTER.validate_python(value)
if parsed.scheme != "https":
raise RegistryValidationError(f"registry external link scheme must be https, got {parsed.scheme!r}")
return str(parsed)
RegistryExternalLink = Annotated[str, AfterValidator(_validate_registry_external_link)]
"""Canonical fragment-preserving HTTPS link used by registry evidence records."""
def _validate_legal_text_entries(
entries: tuple[str, ...],
*,
field_name: str,
) -> None:
if any(not item.strip() for item in entries):
raise RegistryValidationError(f"legal reference {field_name} entries must be non-empty")
if len(set(entries)) != len(entries):
raise RegistryValidationError(f"legal reference {field_name} entries must be unique")
def _validate_legal_corpus_ref(reference_id: LegalRefId, corpus_ref: str) -> None:
if "#" not in corpus_ref:
raise RegistryValidationError(
f"legal reference {reference_id!r} corpus_ref must be of the form 'path#anchor' (got {corpus_ref!r})",
)
path_part, _, anchor_part = corpus_ref.partition("#")
if not path_part or not anchor_part:
raise RegistryValidationError(
f"legal reference {reference_id!r} corpus_ref must have non-empty path and anchor"
)
def _validate_legal_review_metadata(
review_status: RevisionReviewStatus,
reviewed_by: str | None,
reviewed_at: date | None,
) -> None:
has_reviewer = reviewed_by is not None
has_review_date = reviewed_at is not None
if review_status is RevisionReviewStatus.PENDING_REVIEW:
if has_reviewer or has_review_date:
raise RegistryValidationError("pending legal reference must not declare reviewed_by or reviewed_at")
elif review_status in REVIEWED_REVISION_REVIEW_STATUSES and not (has_reviewer and has_review_date):
raise RegistryValidationError(f"{review_status.value} legal reference requires reviewed_by and reviewed_at")
def _validate_legal_reference_text(
reference_id: LegalRefId, required: tuple[str, ...], forbidden: tuple[str, ...]
) -> None:
_validate_legal_text_entries(required, field_name="required_text")
_validate_legal_text_entries(forbidden, field_name="forbidden_text")
overlap = set(required) & set(forbidden)
if overlap:
raise RegistryValidationError(
f"legal reference {reference_id!r} required_text and forbidden_text must not overlap: {sorted(overlap)!r}",
)
[docs]
class LegalReferenceKind(StrEnum):
"""The kind of instrument a legal reference cites.
Spanish instrument names are kept as AEAT and the BOE use them; the meanings belong
to those sources and are not paraphrased here.
"""
LEY = auto()
REAL_DECRETO = auto()
REAL_DECRETO_LEGISLATIVO = auto()
REAL_DECRETO_LEY = auto()
ORDEN = auto()
REGLAMENTO = auto()
ACUERDO_INTERNACIONAL = auto()
DIRECTIVA = auto()
MANUAL = auto()
INSTRUCTION = auto()
LegalReferenceKindField = Annotated[LegalReferenceKind, BeforeValidator(coerce_enum_member(LegalReferenceKind))]
"""Registry token hydrated into a LegalReferenceKind member."""
[docs]
class DictionaryCasillaIdGrammar(StrEnum):
"""The casilla-id grammar a dictionary source publishes."""
NUMERIC = "numeric"
NUMERIC_OR_SINGLE_UPPERCASE_LETTER = "numeric_or_single_uppercase_letter"
DictionaryCasillaIdGrammarField = Annotated[
DictionaryCasillaIdGrammar, BeforeValidator(coerce_enum_member(DictionaryCasillaIdGrammar))
]
"""Registry token hydrated into a DictionaryCasillaIdGrammar member."""
[docs]
class RegistrySnapshotRef(RegistryModel):
"""Typed coordinates that identify a registry snapshot."""
modelo: ModeloId
revision_id: RevisionId
modelo_year: FilingYear
period: RegistryPeriodCode
def _validate_period_selector_year_choice(years: tuple[int, ...], year_from: int | None) -> None:
"""Validate the mutually exclusive year-selection forms."""
if years and year_from is not None:
raise RegistryValidationError("period_selector must use either years or year_from/year_to")
if not years and year_from is None:
raise RegistryValidationError("period_selector must declare years or year_from")
def _validate_period_selector_year_uniqueness(years: tuple[int, ...]) -> None:
"""Reject repeated explicit filing years."""
if len(set(years)) != len(years):
raise RegistryValidationError("period_selector years must be unique")
def _validate_period_selector_year_bounds(year_from: int | None, year_to: int | None) -> None:
"""Validate the optional inclusive range endpoints."""
if year_to is not None and year_from is None:
raise RegistryValidationError("period_selector year_to requires year_from")
if year_from is not None and year_to is not None and year_to < year_from:
raise RegistryValidationError("period_selector year_to must be on or after year_from")
def _validate_period_selector_years(
*,
years: tuple[int, ...],
year_from: int | None,
year_to: int | None,
) -> None:
"""Validate year selection in the model's established failure order."""
_validate_period_selector_year_choice(years, year_from)
_validate_period_selector_year_uniqueness(years)
_validate_period_selector_year_bounds(year_from, year_to)
[docs]
class PeriodOverride(RegistryModel):
"""The period surface one covered filing year serves, in place of the flat tuple.
Authored when a legal boundary falls inside a year rather than between two
of them: a design that applies from a monthly period but only from a later
quarterly one serves a strict subset of its own periods in the transition
year, and a single flat tuple cannot state that.
"""
year: int
periods: tuple[RegistrySelectorPeriodCode, ...] = Field(min_length=1)
@field_validator("periods")
@classmethod
@pydantic_validation_boundary
def _periods_unique(cls, value: tuple[str, ...]) -> tuple[str, ...]:
if len(set(value)) != len(value):
raise RegistryValidationError("period_override periods must be unique")
return value
def _validate_period_overrides(selector: PeriodSelector) -> None:
"""Reject repeated override years and any override outside the covered span."""
seen: set[int] = set()
for override in selector.period_overrides:
if override.year in seen:
raise RegistryValidationError(
f"period_selector period_overrides must declare each year once; got {override.year} twice"
)
seen.add(override.year)
if not selector.includes_year(override.year):
raise RegistryValidationError(
f"period_selector period_overrides year {override.year} lies outside the selector's covered years"
)
[docs]
class PeriodSelector(RegistryModel):
"""Select filing years and periods using explicit years or an inclusive range.
``periods`` is the surface every covered year serves. A year named by
``period_overrides`` REPLACES that tuple with its own -- typically a strict
subset, for a mid-year legal boundary -- and the override's periods need not
be a subset of ``periods``, because a transition year can also gain a token
the steady state does not carry. Every override year must itself be covered
by ``years`` or the ``year_from``/``year_to`` range, and no year may be
overridden twice. Read the surface through :meth:`periods_for_year`
wherever the filing year is known; ``declared_periods`` is the union for the
year-less token lookups.
"""
years: tuple[int, ...] = ()
year_from: int | None = None
year_to: int | None = None
periods: tuple[RegistrySelectorPeriodCode, ...] = Field(min_length=1)
period_overrides: tuple[PeriodOverride, ...] = ()
@field_validator("periods")
@classmethod
@pydantic_validation_boundary
def _periods_unique(cls, value: tuple[str, ...]) -> tuple[str, ...]:
if len(set(value)) != len(value):
raise RegistryValidationError("period_selector periods must be unique")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_year_selector(self) -> PeriodSelector:
_validate_period_selector_years(years=self.years, year_from=self.year_from, year_to=self.year_to)
_validate_period_overrides(self)
return self
[docs]
def includes_year(self, year: int) -> bool:
"""Return whether the selector covers a filing year."""
if self.years:
return year in self.years
if self.year_from is None:
return False
return year >= self.year_from and (self.year_to is None or year <= self.year_to)
[docs]
def periods_for_year(self, year: int) -> tuple[RegistrySelectorPeriodCode, ...]:
"""Return the period surface this selector serves in one filing year."""
for override in self.period_overrides:
if override.year == year:
return override.periods
return self.periods
@property
def declared_periods(self) -> tuple[RegistrySelectorPeriodCode, ...]:
"""Every period token this selector declares in any covered year, in declared order."""
if not self.period_overrides:
return self.periods
seen = dict.fromkeys(self.periods)
for override in self.period_overrides:
seen.update(dict.fromkeys(override.periods))
return tuple(seen)
[docs]
class RegistryValidityWindow(RegistryModel):
"""Common inclusive validity window for every revisioned registry declaration."""
valid_from: date
valid_to: date | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_window(self) -> Self:
if self.valid_to is not None and self.valid_to < self.valid_from:
raise RegistryValidationError("valid_to must be on or after valid_from")
return self
[docs]
def contains_date(self, coordinate: date) -> bool:
"""Return whether ``coordinate`` falls inside this inclusive window."""
return coordinate >= self.valid_from and (self.valid_to is None or coordinate <= self.valid_to)
[docs]
class RegistryTemporalBounds(RegistryModel):
"""Optional delta-authored bounds resolved against a containing window."""
valid_from: date | None = None
valid_to: date | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_declared_order(self) -> Self:
if self.valid_from is not None and self.valid_to is not None and self.valid_to < self.valid_from:
raise RegistryValidationError("valid_to must be on or after valid_from")
return self
[docs]
class PeriodScopedValidityWindow(RegistryValidityWindow):
"""A validity window whose filing-year/period coordinate is mandatory."""
period_selector: PeriodSelector
[docs]
class TemporalProjectionDirection(StrEnum):
"""How a resolved coordinate relates to the declaration supplying it."""
AUTHORED = "authored"
BACKWARD = "backward"
FORWARD = "forward"
def _validate_support_bounds[T: (int, date)](floor: T, horizon: T, hard_ceiling: T | None) -> None:
if horizon < floor:
raise ValueError("temporal support horizon must be on or after floor")
if hard_ceiling is not None and hard_ceiling < horizon:
raise ValueError(
"temporal support hard_ceiling must be on or after horizon; a ceiling before "
"the horizon would close a span the corpus already declares coverage for"
)
[docs]
class OrderedSupportEnvelope[CoordinateT: (int, date)]:
"""One implementation of hard-gate admission and newest-authority projection."""
floor: CoordinateT
horizon: CoordinateT
hard_ceiling: CoordinateT | None
[docs]
def admits_coordinate(self, coordinate: CoordinateT) -> bool:
"""Return whether a coordinate lies between the hard support gates."""
if coordinate < self.floor:
return False
return self.hard_ceiling is None or coordinate <= self.hard_ceiling
[docs]
def projection_coordinate(self, coordinate: CoordinateT) -> CoordinateT | None:
"""Map an admitted coordinate to the newest authored coordinate."""
if not self.admits_coordinate(coordinate):
return None
return min(coordinate, self.horizon)
[docs]
def clamp_coordinate(self, coordinate: CoordinateT) -> CoordinateT:
"""Return the admitted coordinate nearest ``coordinate``.
A declaration dated outside the gates is read against the vocabulary
in force at the nearest supported edge, never at an unsupported date.
"""
if coordinate < self.floor:
return self.floor
if self.hard_ceiling is not None and coordinate > self.hard_ceiling:
return self.hard_ceiling
return coordinate
[docs]
class DateSupportEnvelope(RegistryModel, OrderedSupportEnvelope[date]):
"""Hard gates and authored horizon on an effective-date axis."""
floor: date
horizon: date
hard_ceiling: date | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _bounds_are_ordered(self) -> Self:
_validate_support_bounds(self.floor, self.horizon, self.hard_ceiling)
return self
[docs]
class TemporalSupportEnvelope(RegistryModel, OrderedSupportEnvelope[int]):
"""Hard gates and authored horizon shared by forward-projecting registries.
``floor`` and ``hard_ceiling`` are refusal boundaries. ``horizon`` is the
last explicitly authored coordinate, not an implicit ceiling: when no hard
ceiling exists, a consumer may carry the declaration at the horizon
forward. The actual projection remains a consumer decision and is exposed
explicitly by :meth:`projection_coordinate`.
"""
floor: FilingYear
horizon: FilingYear
hard_ceiling: FilingYear | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _bounds_are_ordered(self) -> Self:
_validate_support_bounds(self.floor, self.horizon, self.hard_ceiling)
return self
@property
def years(self) -> tuple[int, ...]:
"""Enumerate the explicitly authored span, both bounds inclusive."""
return tuple(range(self.floor, self.horizon + 1))
[docs]
@typing_override
def admits_coordinate(self, coordinate: int) -> bool:
"""Return whether ``coordinate`` is inside the envelope's hard gates."""
return super().admits_coordinate(coordinate)
[docs]
@typing_override
def projection_coordinate(self, coordinate: int) -> int | None:
"""Map an admitted coordinate to its authored coordinate, carrying the horizon forward."""
projected = super().projection_coordinate(coordinate)
return projected if isinstance(projected, int) else None
[docs]
def date_envelope(self) -> DateSupportEnvelope:
"""Project the filing-year envelope onto the effective-date axis.
Every date-addressed registry declaration shares this one envelope, so a
governed fact never states a floor or ceiling of its own.
"""
return DateSupportEnvelope(
floor=date(self.floor, 1, 1),
horizon=date(self.horizon, 12, 31),
hard_ceiling=None if self.hard_ceiling is None else date(self.hard_ceiling, 12, 31),
)
def _validate_legal_governed_periods(
legal_id: str,
*,
governs_periods_from: date | None,
governs_periods_to: date | None,
effective_from: date,
) -> None:
"""Refuse a governed-period declaration that is not a retroactive reach."""
if governs_periods_to is not None and governs_periods_from is None:
raise RegistryValidationError(
f"legal reference {legal_id!r} declares governs_periods_to without governs_periods_from",
)
if governs_periods_from is None:
return
if governs_periods_from >= effective_from:
raise RegistryValidationError(
f"legal reference {legal_id!r} declares governs_periods_from "
f"{governs_periods_from.isoformat()} on or after effective_from "
f"{effective_from.isoformat()}: the field declares RETROACTIVE reach only, and a "
f"forward value would let a stale citation ground a period its norm never governed",
)
if governs_periods_to is not None and governs_periods_to < governs_periods_from:
raise RegistryValidationError(
f"legal reference {legal_id!r} governs_periods_to must be on or after governs_periods_from",
)
[docs]
class LegalReference(RegistryModel):
"""Legal-authority citation row carried by registry definitions."""
id: LegalRefId
evidence_tier: Annotated[Literal[EvidenceTier.LEGAL_AUTHORITY], BeforeValidator(coerce_enum_member(EvidenceTier))]
authority: PublishingAuthorityField
kind: LegalReferenceKindField
corpus_ref: str
document_id: str
article: str | None = None
section: str | None = None
permalink: RegistryExternalLink
published_at: date | None = None
effective_from: date
effective_to: date | None = None
consolidated_as_of: date | None = None
review_status: RevisionReviewStatusField
reviewed_at: date | None = None
reviewed_by: str | None = Field(default=None, min_length=1)
notes: str | None = None
required_text: tuple[str, ...] = Field(min_length=1)
forbidden_text: tuple[str, ...] = ()
corpus_tier: CorpusTierField | None = None
"""Which kind of corpus evidence ``corpus_ref`` resolves to, when declared.
Deliberately optional and deliberately two-valued. Optional: nothing in
the committed catalogue declares it today, so adding the field cannot
itself introduce a new refusal -- authoring it is opt-in, verified only
when present. Two-valued: a paraphrase with no operative text of its own
must never become a *declarable* tier (a stub calling itself an
excerpt would pass silently); that shape stays a build-time refusal via
the dispositive-content check, never a legal state this field can assert.
Verified against the bundled corpus file, not merely typed, by
``_legal.py``'s grounding check -- this field states a claim; the
verifier is what makes the claim mean something.
"""
"""Phrases the cited corpus document must NOT contain.
``required_text`` alone cannot express "this repealed clause must be
absent" — no set of must-be-present phrases states a negative. This is
the entry's optional negative clause: a corpus excerpt grounding current
law names the repealed text it must not carry, and a deliberately
historical excerpt names the later text that must not have crept in,
which pins its vintage forward as well as backward.
"""
governs_periods_from: date | None = None
"""Earliest devengo this provision governs, when it reaches back before force.
``effective_from`` / ``effective_to`` record when a norm entered and left
FORCE. A retroactive provision governs tax periods that closed before it
existed: RDL 13/2025, in force from 2025-11-27, extends the La Palma
deduction "durante los periodos impositivos 2022, 2023, 2024 y 2025" in its
own operative text. Checking such a citation against the in-force window
rejects a correct grounding, and the cheapest way to silence that refusal is
to backdate ``effective_from`` -- which misstates when the norm came into
force and corrupts every other consumer of that field.
So reach is declared here, separately, and only ever by an author who has
read the clause that states it. Absent, the governed span IS the in-force
span, so adding this field relaxes no existing citation. The declaration is
deliberately retroactive-only: it must precede ``effective_from``, because a
forward-reaching value would let a stale citation ground a period its norm
never governed.
"""
governs_periods_to: date | None = None
"""Latest devengo this provision governs; ``None`` leaves the reach open.
Only meaningful alongside :attr:`governs_periods_from`, which the validator
requires.
"""
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_legal_reference(self) -> LegalReference:
_validate_legal_review_metadata(self.review_status, self.reviewed_by, self.reviewed_at)
if self.effective_to is not None and self.effective_to < self.effective_from:
raise RegistryValidationError("legal reference effective_to must be on or after effective_from")
_validate_legal_governed_periods(
self.id,
governs_periods_from=self.governs_periods_from,
governs_periods_to=self.governs_periods_to,
effective_from=self.effective_from,
)
_validate_legal_reference_text(self.id, self.required_text, self.forbidden_text)
_validate_legal_corpus_ref(self.id, self.corpus_ref)
return self
#: The canonical epoch shape, imported rather than restated. A second copy is how
#: the registry boundary and the filing-evidence boundary drift apart -- and that
#: drift is what let a value the registry would refuse reach a filed artefact.
_RECORD_DESIGN_EPOCH: Final = RECORD_DESIGN_EPOCH_RE
[docs]
def source_window_applies_across(
*,
applies_from: date | None,
applies_to: date | None,
span_from: date,
span_to: date | None,
) -> bool:
"""Report whether a source applicability window overlaps one date span.
The ONE definition of the overlap rule, shared by every carrier of an
applicability window so that a diagnostic copy of a source cannot answer
the question differently from the source itself. An open bound means the
window is open in that direction, never that it is unknown; deciding
whether an absent bound is admissible belongs to the caller, before it asks.
"""
if applies_to is not None and applies_to < span_from:
return False
return not (applies_from is not None and span_to is not None and applies_from > span_to)
def _validate_source_window(*, applies_from: date | None, applies_to: date | None) -> None:
"""Validate the source's inclusive date window."""
if applies_to is not None and applies_from is not None and applies_to < applies_from:
raise RegistryValidationError("source reference applies_to must be on or after applies_from")
def _validate_source_dictionary_grammar(
*,
kind: RegistrySourceKind,
dictionary_casilla_id_grammar: DictionaryCasillaIdGrammar,
) -> None:
"""Keep the extended casilla grammar exclusive to dictionary evidence."""
if kind is not RegistrySourceKind.DICTIONARY and dictionary_casilla_id_grammar != "numeric":
raise RegistryValidationError(
"dictionary_casilla_id_grammar other than 'numeric' is only valid for kind='dictionary'",
)
def _validate_source_period_selector(
*,
source_id: SourceRefId,
period_selector: PeriodSelector | None,
applies_from: date | None,
applies_to: date | None,
) -> None:
"""Ensure declared selector years cover both bounded source years."""
if period_selector is None or applies_from is None:
return
if not period_selector.includes_year(applies_from.year):
raise RegistryValidationError(
f"source reference {source_id!r} declares period_selector that does not cover "
f"applies_from's year {applies_from.year}",
)
if applies_to is not None and not period_selector.includes_year(applies_to.year):
raise RegistryValidationError(
f"source reference {source_id!r} declares period_selector that does not cover "
f"applies_to's year {applies_to.year}",
)
def _validate_source_corpus_path(corpus_path: str) -> None:
"""Require repository-relative POSIX corpus paths."""
if "\\" in corpus_path or corpus_path.startswith(("/", ".")):
raise RegistryValidationError("source reference corpus_path must be repository-relative POSIX style")
def _validate_source_kind_path_and_epoch(
*,
source_id: SourceRefId,
kind: RegistrySourceKind,
corpus_path: str,
record_design_epoch: str | None,
) -> None:
"""Validate kind-specific corpus extension and epoch ownership."""
if kind is RegistrySourceKind.RECORD_DESIGN:
allowed_record_design_suffixes = (
PDF_EXTENSION,
XLS_EXTENSION,
XLSX_EXTENSION,
XLSM_EXTENSION,
)
suffix = corpus_path.rsplit(".", 1)
extension = "." + suffix[1].lower() if len(suffix) == 2 else ""
if extension not in allowed_record_design_suffixes:
raise RegistryValidationError(
f"source reference {source_id!r} declares kind='record_design' but corpus_path "
f"{corpus_path!r} has unsupported extension {extension!r}; the record-design "
f"extractor accepts only .pdf / .xls / .xlsx / .xlsm — reclassify the source "
f"(e.g. kind='form_spec' for an AEAT/BOE landing page HTML) or ingest the real "
f"Diseño workbook",
)
elif record_design_epoch is not None:
raise RegistryValidationError("record_design_epoch is only valid for kind='record_design'")
def _validate_record_design_epoch(source_id: SourceRefId, record_design_epoch: str | None) -> None:
"""Validate the optional epoch after its record-design ownership is known."""
if record_design_epoch is None:
return
if not record_design_epoch.strip():
raise RegistryValidationError("record_design_epoch must contain non-whitespace text")
if not _RECORD_DESIGN_EPOCH.fullmatch(record_design_epoch):
raise RegistryValidationError(
f"source reference {source_id!r} declares record_design_epoch "
f"{record_design_epoch!r}, which is not a design EPOCH. An epoch names the "
"filing period a design governs -- a four-digit ejercicio, optionally with a "
"lower-case sub-year label where AEAT re-laid the form out mid-ejercicio "
"('2024-early', '2024-late'). It is NOT the document's version: "
"'aeat-dr-111-2019-v18' is epoch '2019', because v18 is which revision of the "
"PDF AEAT published and says nothing about which filings it governs. Two "
"designs differing only by version are the same epoch and must not both claim "
"one; two designs governing different periods are different epochs",
)
[docs]
class SourceReference(RegistryModel):
"""Official-source evidence row with bundled-corpus integrity metadata."""
id: SourceRefId
evidence_tier: EvidenceTierField
authority: PublishingAuthorityField
kind: RegistrySourceKindField
corpus_path: str
sha256: ContentDigest
bytes: int = Field(gt=0)
retrieved_at: date
published_at: date | None = None
applies_from: date | None = None
applies_to: date | None = None
dictionary_casilla_id_grammar: DictionaryCasillaIdGrammarField = DictionaryCasillaIdGrammar.NUMERIC
"""Exact casilla-id grammar published by a dictionary source.
The default accepts the numbered ids used by every ordinary AEAT dictionary.
The extension is an evidence claim tied to the hash-pinned source row, not a
parser-side modelo, source-id, or year exception: when a reviewed dictionary
publishes its one-letter annex boxes, that source declares the wider grammar.
Its ``applies_from`` / ``applies_to`` window remains the temporal authority
for the source itself.
"""
record_design_epoch: str | None = Field(default=None, min_length=1, max_length=128)
design_authority: DesignAuthority = "authoritative"
source_url: RegistryExternalLink
review_status: RevisionReviewStatusField
period_selector: PeriodSelector | None = None
"""Disambiguates two designs that share one ``applies_from``/``applies_to`` window.
Deliberately optional and deliberately reuses :class:`PeriodSelector` --
the registry's one period-selector shape, already carrying the AEAT-token
vocabulary (:data:`RegistrySelectorPeriodCode`) -- rather than inventing a
second grammar. ``applies_from``/``applies_to`` alone cannot express two
real AEAT designs published for the SAME year that govern different
PERIODS within it: M303 published one whole-year 2018 record design and a
second covering every period except the last (quarter 4 / month 12), and
a 2021 pair split at period 07 (one design "hasta periodo 06", the
successor "desde periodo 07"). Both members of each pair carry an
identical nominal date window, so nothing before this field could tell
a record-design resolver's caller which one to pick for a
given filing period. Declare it only for a design whose own filename or
published text states the period boundary; leave it undeclared rather
than infer one.
"""
corpus_tier: CorpusTierField | None = None
"""Mirrors :attr:`LegalReference.corpus_tier` in philosophy, recalibrated here.
Same two-valued, verified-not-merely-typed contract: optional so adding
the field cannot itself introduce a refusal, and deliberately two-valued
so a paraphrase can never become a *declarable* tier -- that shape stays
a build-time refusal, never a state this field can assert. The
calibration is NOT reused from :attr:`LegalReference.corpus_tier`: this
model's ``corpus_path`` is a bare repository-relative path with no
``#anchor`` convention (``LegalReference.corpus_ref`` is ``path#anchor``),
so a resolved-anchor read does not apply here, and the size floor is
calibrated against THIS model's own observed population, not
``LegalReference``'s. The excerpt/full-text duality only exists for
``corpus_path`` entries under ``corpus/normatives/`` -- the same BOE/AEAT
norm-text tree ``LegalReference.corpus_ref`` resolves into, which is why a
``form_spec`` source can legitimately point at the exact file a
``LegalReference`` entry also cites. A design workbook, manual PDF, XSD or
data dictionary under ``corpus/aeat_official/`` has no such duality --
verified by ``_corpus_catalogue.py``'s check, which refuses the field
outside ``corpus/normatives/`` rather than silently accepting a claim the
concept does not apply to.
"""
[docs]
def applies_across(self, span_from: date, span_to: date | None) -> bool:
"""Report whether this source's applicability window overlaps one date span.
Delegates to :func:`source_window_applies_across`, which is the one
definition of the overlap rule. Callers keep their own policy on whether
a missing bound is admissible -- a record-design binary must declare
``applies_from``, an evidence-backed source cell must declare both --
and apply it before asking this question.
"""
return source_window_applies_across(
applies_from=self.applies_from,
applies_to=self.applies_to,
span_from=span_from,
span_to=span_to,
)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_source_reference(self) -> SourceReference:
_validate_source_window(applies_from=self.applies_from, applies_to=self.applies_to)
_validate_source_dictionary_grammar(
kind=self.kind,
dictionary_casilla_id_grammar=self.dictionary_casilla_id_grammar,
)
_validate_source_period_selector(
source_id=self.id,
period_selector=self.period_selector,
applies_from=self.applies_from,
applies_to=self.applies_to,
)
_validate_source_corpus_path(self.corpus_path)
_validate_source_kind_path_and_epoch(
source_id=self.id,
kind=self.kind,
corpus_path=self.corpus_path,
record_design_epoch=self.record_design_epoch,
)
_validate_record_design_epoch(self.id, self.record_design_epoch)
return self
@property
def supports_single_uppercase_letter_casilla_ids(self) -> bool:
"""Whether this exact reviewed dictionary declares the annex-id grammar."""
return self.dictionary_casilla_id_grammar == "numeric_or_single_uppercase_letter"
@field_validator("sha256")
@classmethod
@pydantic_validation_boundary
def _sha256_lower_hex(cls, value: str) -> str:
lowered = value.lower()
if lowered != value or any(char not in "0123456789abcdef" for char in value):
raise RegistryValidationError("sha256 must be lowercase hexadecimal")
return value
[docs]
def governed_period_span(reference: LegalReference) -> tuple[date, date | None]:
"""Return the devengo span ``reference`` governs, which may precede its force.
Defaults to the in-force window, so a reference that declares nothing is
tested exactly as before -- the retroactive fields cannot relax a citation
by existing. A reference that DOES declare reach is tested against the
declared span, because that is the axis its citation defends: RDL 13/2025 is
in force only from 2025-11-27 yet governs periods 2022 through 2025 by its
own operative text, so the 2024 devengo it grounds is inside its reach and
outside its force.
The declaration is validated retroactive-only at the model boundary, so this
can widen a span backwards but never forwards.
"""
if reference.governs_periods_from is None:
return reference.effective_from, reference.effective_to
return reference.governs_periods_from, reference.governs_periods_to