"""Typed projection of the third-party declaration-role fact (0134)."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date
from types import MappingProxyType
from typing import Final
from ....core.aggregation import ThirdPartyDeclarationRole
from ....core.time.clock import today_madrid
from .errors import RegistryValidationError
from .facts.resolution import MappingFactQuery, ResolvedMappingFact, required_mapping_entry, unique_mapping_tokens
from .governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from .schema_base import DateAxis
_ENTRY_SUBJECT: Final = "third-party declaration role catalogue"
_FACT_ID = "third-party-declaration-role-catalogue"
_ORDER_KEY = "role.order"
_ROLE_PREFIX = "role."
_SELECTION_PREFIX = "selection."
_SELECTION_CLAVES = ("C", "D", "E")
[docs]
@dataclass(frozen=True, slots=True)
class ThirdPartyDeclarationRoleDefinition:
"""One registry-declared third-party declaration role and its semantics."""
token: ThirdPartyDeclarationRole
description: str
legal_refs: tuple[str, ...]
[docs]
@dataclass(frozen=True, slots=True)
class ThirdPartyDeclarationRoleCatalogue:
"""Complete typed projection of fact 0134 for one filing date."""
declarations: Mapping[str, str]
definitions: tuple[ThirdPartyDeclarationRoleDefinition, ...]
selections: Mapping[str, tuple[ThirdPartyDeclarationRole, ...]]
@property
def roles(self) -> tuple[ThirdPartyDeclarationRole, ...]:
"""Return the registry-declared role choices in canonical order."""
return tuple(item.token for item in self.definitions)
@property
def choices(self) -> tuple[ThirdPartyDeclarationRole, ...]:
"""Alias used by choice-building consumers."""
return self.roles
[docs]
def require(self, value: object) -> ThirdPartyDeclarationRole:
"""Project one role only when fact 0134 declares it."""
if isinstance(value, ThirdPartyDeclarationRole):
token = value
elif isinstance(value, str):
raw = value.strip()
if not raw:
raise RegistryValidationError("third-party declaration role must be non-empty")
try:
token = ThirdPartyDeclarationRole.from_registry(raw)
except (TypeError, ValueError) as exc:
raise RegistryValidationError(
"third-party declaration role must be a non-empty string",
) from exc
else:
raise RegistryValidationError("third-party declaration role must be a string token")
if token not in self.roles:
raise RegistryValidationError(
f"third-party declaration role {str(token)!r} is not declared by fact {_FACT_ID!r}",
)
return token
[docs]
def definition(self, value: object) -> ThirdPartyDeclarationRoleDefinition:
"""Return the selected role's registry-owned semantics."""
token = self.require(value)
return next(item for item in self.definitions if item.token == token)
[docs]
def roles_for_clave(self, clave: str) -> frozenset[ThirdPartyDeclarationRole]:
"""Return the registry-owned role population for one C/D/E selector."""
normalized = clave.strip().upper()
if normalized not in _SELECTION_CLAVES:
raise RegistryValidationError(f"unsupported third-party declaration selection {clave!r}")
return frozenset(self.selections[normalized])
def _refs(entries: Mapping[str, str], key: str) -> tuple[str, ...]:
return unique_mapping_tokens(entries, key, subject=_ENTRY_SUBJECT)
def _mapping_entries(resolved: ResolvedMappingFact) -> Mapping[str, str]:
entries: dict[str, str] = {}
for entry in resolved.payload.entries:
if not isinstance(entry.key, str) or not isinstance(entry.value, str):
raise RegistryValidationError("third-party declaration role entries must be string-to-string")
if entry.key in entries:
raise RegistryValidationError(
f"duplicate third-party declaration role catalogue key {entry.key!r}",
)
entries[entry.key] = entry.value
return MappingProxyType(entries)
def _resolve_entries(*, effective_date: date, authority: GovernedFactSource) -> Mapping[str, str]:
resolved = authority.resolve_governed_fact(
MappingFactQuery(
fact_id=_FACT_ID,
date_axis=DateAxis.FILING_PERIOD,
effective_date=effective_date,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise RegistryValidationError("third-party declaration role catalogue must resolve as a mapping fact")
return _mapping_entries(resolved)
def _catalogue(entries: Mapping[str, str]) -> ThirdPartyDeclarationRoleCatalogue:
definitions: list[ThirdPartyDeclarationRoleDefinition] = []
for raw_token in unique_mapping_tokens(entries, _ORDER_KEY, subject=_ENTRY_SUBJECT):
token = ThirdPartyDeclarationRole.from_registry(raw_token)
prefix = f"{_ROLE_PREFIX}{raw_token}."
if required_mapping_entry(entries, f"{prefix}value", subject=_ENTRY_SUBJECT) != raw_token:
raise RegistryValidationError(
f"third-party declaration role {raw_token!r} declares a mismatched value",
)
definitions.append(
ThirdPartyDeclarationRoleDefinition(
token=token,
description=required_mapping_entry(entries, f"{prefix}description", subject=_ENTRY_SUBJECT),
legal_refs=_refs(entries, f"{prefix}legal_refs"),
),
)
if len(definitions) != len({item.token for item in definitions}):
raise RegistryValidationError("third-party declaration role catalogue contains duplicate roles")
role_choices = {item.token for item in definitions}
selections: dict[str, tuple[ThirdPartyDeclarationRole, ...]] = {}
for clave in _SELECTION_CLAVES:
raw_roles = unique_mapping_tokens(entries, f"{_SELECTION_PREFIX}{clave}.roles", subject=_ENTRY_SUBJECT)
selected = tuple(ThirdPartyDeclarationRole.from_registry(raw) for raw in raw_roles)
if not set(selected).issubset(role_choices):
raise RegistryValidationError(
f"third-party declaration role selection {clave!r} contains an undeclared role",
)
selections[clave] = selected
return ThirdPartyDeclarationRoleCatalogue(
declarations=entries,
definitions=tuple(definitions),
selections=MappingProxyType(selections),
)
def _selected_catalogue(
*,
effective_date: date,
authority: GovernedFactSource | None,
) -> ThirdPartyDeclarationRoleCatalogue:
selected = authority or governed_facts_in_scope()
if selected is None:
raise RegistryValidationError(
"third-party declaration-role resolution requires a generation-pinned governed-fact source",
)
return _catalogue(_resolve_entries(effective_date=effective_date, authority=selected))
[docs]
def resolve_third_party_declaration_role_catalogue(
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> ThirdPartyDeclarationRoleCatalogue:
"""Resolve the dated third-party declaration-role vocabulary."""
coordinate = effective_date or today_madrid()
return _selected_catalogue(effective_date=coordinate, authority=authority)
[docs]
def require_third_party_declaration_role(
value: object,
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> ThirdPartyDeclarationRole:
"""Project one role through the dated facts-registry catalogue."""
return resolve_third_party_declaration_role_catalogue(
effective_date=effective_date,
authority=authority,
).require(value)
[docs]
def third_party_declaration_role_choices(
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> tuple[ThirdPartyDeclarationRole, ...]:
"""Return registry-declared role choices in canonical order."""
return resolve_third_party_declaration_role_catalogue(
effective_date=effective_date,
authority=authority,
).roles
__all__ = [
"ThirdPartyDeclarationRoleCatalogue",
"ThirdPartyDeclarationRoleDefinition",
"require_third_party_declaration_role",
"resolve_third_party_declaration_role_catalogue",
"third_party_declaration_role_choices",
]