"""Typed shape and normalization boundary for ledger IRPF category tokens."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from ...core.time.clock import today_madrid
from ...domain.calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ...domain.calculations.registry.governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from ...domain.calculations.registry.schema_base import DateAxis
from .enums import TransactionDirection
[docs]
@dataclass(frozen=True, slots=True)
class LedgerIrpfCategoryDescriptor:
"""One registry-projected category descriptor."""
id: str
purpose: str
directions: tuple[TransactionDirection, ...]
net_paid: bool
related_ids: tuple[str, ...]
def _registry_taxonomy_declarations(*, authority: GovernedFactSource | None = None) -> Mapping[str, str]:
"""Resolve the dated IRPF ledger category taxonomy."""
authority = authority or governed_facts_in_scope()
if authority is None:
raise ValueError("IRPF ledger category taxonomy requires an explicit authority operation or scope")
resolved = authority.resolve_governed_fact(
MappingFactQuery(
fact_id="irpf-ledger-category-taxonomy",
date_axis=DateAxis.FILING_PERIOD,
effective_date=today_madrid(),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise ValueError("IRPF ledger category taxonomy must resolve as a mapping fact")
return {str(entry.key): str(entry.value) for entry in resolved.payload.entries}
def _required_taxonomy_declaration(declarations: Mapping[str, str], key: str) -> str:
try:
return declarations[key]
except KeyError as exc:
raise ValueError(f"IRPF ledger taxonomy declaration is missing: {key}") from exc
def _split_declaration(declarations: Mapping[str, str], key: str) -> tuple[str, ...]:
return tuple(
value.strip() for value in _required_taxonomy_declaration(declarations, key).split(",") if value.strip()
)
def _descriptor_from_registry(category_id: str, declarations: Mapping[str, str]) -> LedgerIrpfCategoryDescriptor:
prefix = f"category.{category_id}"
directions = tuple(
TransactionDirection(value.upper()) for value in _split_declaration(declarations, f"{prefix}.directions")
)
net_paid_field = "_".join(("net", "paid", "invoice"))
related_field = "_".join(("related", "category", "ids"))
net_paid = _required_taxonomy_declaration(declarations, f"{prefix}.{net_paid_field}")
if net_paid not in {"true", "false"}:
raise ValueError(f"invalid IRPF net-paid declaration: {net_paid}")
return LedgerIrpfCategoryDescriptor(
id=category_id,
purpose=_required_taxonomy_declaration(declarations, f"{prefix}.purpose"),
directions=directions,
net_paid=net_paid == "true",
related_ids=_split_declaration(declarations, f"{prefix}.{related_field}"),
)
[docs]
def ledger_irpf_category_catalogue(
*, authority: GovernedFactSource | None = None
) -> tuple[LedgerIrpfCategoryDescriptor, ...]:
"""Return the dated registry-projected IRPF category descriptors."""
declarations = _registry_taxonomy_declarations(authority=authority)
return tuple(
_descriptor_from_registry(category_id, declarations)
for category_id in _split_declaration(declarations, "catalogue.ids")
)
[docs]
def normalize_irpf_category(value: str | None) -> str | None:
"""Normalize a raw category token without resolving registry meaning."""
if value is None:
return None
return value.strip().casefold() or None
[docs]
def ledger_irpf_category(
value: str | None,
*,
direction: TransactionDirection | None = None,
authority: GovernedFactSource | None = None,
) -> LedgerIrpfCategoryDescriptor | None:
"""Resolve a normalized token through the registry-owned taxonomy."""
normalized = normalize_irpf_category(value)
if normalized is None:
return None
declarations = _registry_taxonomy_declarations(authority=authority)
descriptors = {
descriptor.id: descriptor
for descriptor in (
_descriptor_from_registry(category_id, declarations)
for category_id in _split_declaration(declarations, "catalogue.ids")
)
}
descriptor = descriptors.get(normalized)
if descriptor is None or (direction is not None and direction not in descriptor.directions):
return None
return descriptor
[docs]
def has_non_work_irpf_category(
value: str | None, *, direction: TransactionDirection, authority: GovernedFactSource | None = None
) -> bool:
"""Resolve the registry-owned non-employment predicate."""
descriptor = ledger_irpf_category(value, direction=direction, authority=authority)
return descriptor is not None and not descriptor.purpose.endswith("_income")
[docs]
def has_activity_irpf_category(
value: str | None, *, direction: TransactionDirection, authority: GovernedFactSource | None = None
) -> bool:
"""Resolve the registry-owned activity-income predicate."""
descriptor = ledger_irpf_category(value, direction=direction, authority=authority)
return descriptor is not None and descriptor.purpose.endswith("_income_withholding")
[docs]
def has_rent_irpf_category(
value: str | None, *, direction: TransactionDirection, authority: GovernedFactSource | None = None
) -> bool:
"""Resolve the registry-owned rental predicate."""
descriptor = ledger_irpf_category(value, direction=direction, authority=authority)
return descriptor is not None and descriptor.purpose.endswith("_expense_withholding")
[docs]
def has_employment_irpf_category(
value: str | None, *, direction: TransactionDirection, authority: GovernedFactSource | None = None
) -> bool:
"""Resolve the registry-owned employment predicate."""
descriptor = ledger_irpf_category(value, direction=direction, authority=authority)
return descriptor is not None and descriptor.purpose.endswith("_income")