# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : loader.py # Description : ALSL v0.1 YAML loader — strict parsing, floats and uncited rates rejected. # ============================================================================= """Load ALSL v0.1 policy sets from YAML. Strictness rules: - Rates, thresholds and any decimal value MUST be YAML strings ("0.05"), never bare numbers — bare YAML floats lose precision and are rejected. - Every tax policy MUST cite a source (research doc or official URL). """ from __future__ import annotations from decimal import Decimal from pathlib import Path from typing import Any import yaml from alsl.model import ( ALSL_VERSION, ClassificationPolicy, PolicySet, RoundingPolicy, TaxComponent, TaxPolicy, WhenClause, parse_account_type, ) from core.journal import Account from core.money import RoundingMode class AlslLoadError(ValueError): """Raised when a policy file violates ALSL v0.1 rules.""" def _decimal(value: Any, context: str) -> Decimal: if isinstance(value, float): raise AlslLoadError( f"{context}: decimal values must be YAML strings (e.g. \"0.05\"), " f"got float {value!r} — floats are forbidden in ALSL" ) if isinstance(value, (int, str)): return Decimal(str(value)) raise AlslLoadError(f"{context}: cannot parse decimal from {value!r}") def _when(raw: dict[str, Any] | None, context: str) -> WhenClause: raw = raw or {} jurisdiction_in = raw.get("jurisdiction_in") return WhenClause( jurisdiction=raw.get("jurisdiction"), jurisdiction_in=tuple(jurisdiction_in) if jurisdiction_in else None, event_type=raw.get("event_type"), min_amount=( _decimal(raw["min_amount"], context) if "min_amount" in raw else None ), currency=raw.get("currency"), ) def load_policy_set(path: str | Path) -> PolicySet: path = Path(path) data = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(data, dict): raise AlslLoadError(f"{path}: not a mapping") version = str(data.get("alsl_version", "")) if version != ALSL_VERSION: raise AlslLoadError( f"{path}: alsl_version {version!r} unsupported (expected {ALSL_VERSION!r})" ) accounts: dict[str, Account] = {} for role, spec in (data.get("accounts") or {}).items(): accounts[role] = Account( code=str(spec["code"]), name=str(spec["name"]), type=parse_account_type(str(spec["type"])), ) rounding_raw = data.get("rounding") or {} rounding = RoundingPolicy( mode=RoundingMode(rounding_raw.get("mode", "half_up")), source=str(rounding_raw.get("source", "")), ) tax_policies: list[TaxPolicy] = [] classification_policies: list[ClassificationPolicy] = [] for raw in data.get("policies") or []: name = str(raw.get("name", "")) kind = raw.get("kind") ctx = f"{path}: policy '{name}'" if kind == "tax": source = str(raw.get("source", "")).strip() if not source: raise AlslLoadError( f"{ctx}: tax policies must cite a source (research doc or " "official URL); uncited rates are forbidden" ) components = tuple( TaxComponent( code=str(c["code"]), rate=_decimal(c["rate"], f"{ctx} component {c.get('code')}"), base=str(c.get("base", "subtotal")), payable_role=str(c.get("payable_role", "")), receivable_role=str(c.get("receivable_role", "")), recoverable_on_purchase=bool(c.get("recoverable_on_purchase", True)), ) for c in raw.get("apply") or [] ) if not components: raise AlslLoadError(f"{ctx}: tax policy applies no components") tax_policies.append(TaxPolicy( name=name, when=_when(raw.get("when"), ctx), components=components, source=source, )) elif kind == "classification": then = raw.get("then") or {} classification_policies.append(ClassificationPolicy( name=name, when=_when(raw.get("when"), ctx), classify_as=str(then.get("classify", "expense")), account_role=str(then.get("account_role", "")), source=str(raw.get("source", "")), )) else: raise AlslLoadError(f"{ctx}: unknown policy kind {kind!r}") return PolicySet( alsl_version=version, name=str(data.get("policy_set", path.stem)), version=str(data.get("version", "")), description=str(data.get("description", "")), functional_currency=str(data.get("functional_currency", "CAD")), rounding=rounding, accounts=accounts, tax_policies=tuple(tax_policies), classification_policies=tuple(classification_policies), )