← Files VeraARCHIVED FILE
modules/bilancio-xbrl-it/scripts/xbrl_case.py
331 KB · Oct 2, 2026 · 00:29 UTC
#!/usr/bin/env python3
"""Deterministic case engine for Italian OIC annual accounts and XBRL drafts.
The module deliberately limits deterministic logic to mechanically verifiable
work: exact Decimal arithmetic, explicit rule-pack conditions, provenance,
revision hashes, workflow gates, and XML construction. Account meaning,
materiality, legal interpretation, and professional approval remain explicit
review decisions.
"""
from __future__ import annotations
import argparse
import base64
import binascii
import csv
import hashlib
import html
import json
import logging
import math
import re
import shutil
import sys
import tempfile
import zipfile
from copy import deepcopy
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
try:
from enum import StrEnum
except ImportError: # Python 3.10 in Cowork
from strenum import StrEnum
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Sequence
from accounts_preview import render_accounts_preview
from client_history import (
client_history_suggestions,
remember_approved_client_history,
)
from defusedxml.ElementTree import fromstring
from disclosure_engine import (
build_disclosure_coverage,
disclosure_answer_complete,
disclosure_rule_pack_hash,
manual_disclosure_flags,
narrative_redline,
normalize_narrative_blocks,
note_outline,
prior_narrative_suggestions,
)
from external_validation import record_external_validation_result
from intelligence_contract import (
IntelligenceTask,
build_intelligence_packet,
intelligence_packet_hash,
validate_intelligence_output,
)
from lxml import etree
from mapping_memory import mapping_candidates, remember_approved_mappings
from pdf_trial_balance import extract_pdf_tables
from prior_xbrl import parse_prior_xbrl
from schedule_engine import (
SCHEDULE_TYPES,
normalize_schedule,
required_schedule_types,
schedule_fact_records,
schedule_template_fields,
schedule_template_text_fields,
)
from schedule_taxonomy_adapter import compile_schedule_taxonomy_adapter
from statutory_presentation import (
build_primary_presentation_inventory,
build_statutory_presentation_coverage,
)
from validate_xbrl import validate_instance
__all__ = [
"CaseState",
"EvidenceStatus",
"ParserConvention",
"ValidationIssue",
"approve_case",
"archive_case",
"attach_supporting_document",
"activate_disclosures",
"apply_mapping_decisions",
"build_statements",
"create_case",
"determine_forms",
"export_case",
"generate_mapping_candidates",
"ingest_prior_xbrl",
"ingest_pdf_trial_balance",
"ingest_trial_balance",
"ingest_schedule_file",
"load_client_history",
"migrate_regulatory_versions",
"record_intelligence_suggestion",
"record_issue_reviews",
"record_adjustments",
"record_taxonomy_facts",
"record_taxonomy_catalogue_build",
"record_statutory_presentation",
"record_taxonomy_mapping_index",
"record_taxonomy_representation",
"record_micro_reporting",
"record_pdf_trial_balance_review",
"load_case",
"normalize_decimal",
"prepare_xbrl_review",
"record_schedule",
"record_schedule_taxonomy_adapter",
"record_disclosure_trigger_decisions",
"record_comparative_reconciliation_decisions",
"record_narrative_blocks",
"record_external_validation",
"record_file_security_scan",
"record_artifact_access",
"remember_client_history",
"remember_mappings",
"render_xbrl",
"save_case",
"validate_case",
]
LOGGER = logging.getLogger(__name__)
CASE_FILE = "case.json"
CASE_CHECKSUM_FILE = "case.json.sha256"
SCHEMA_VERSION = 1
RULE_PACK_ROOT = Path(__file__).resolve().parents[1] / "rulepacks" / "it"
XML_NS = "http://www.w3.org/XML/1998/namespace"
XBRLI_NS = "http://www.xbrl.org/2003/instance"
LINK_NS = "http://www.xbrl.org/2003/linkbase"
XLINK_NS = "http://www.w3.org/1999/xlink"
ISO4217_NS = "http://www.xbrl.org/2003/iso4217"
XBRLDI_NS = "http://xbrl.org/2006/xbrldi"
XSI_NS = "http://www.w3.org/2001/XMLSchema-instance"
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
SAFE_QNAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*:[A-Za-z_][A-Za-z0-9_.-]*$")
XBRL_DECIMAL_LEXICAL = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)$")
NORMALIZED_MONEY_LEXICAL = re.compile(
r"^(?P<sign>[+-]?)(?P<integer>\d+)(?:\.(?P<fraction>\d+))?$"
)
MONEY_LITERAL = re.compile(
r"(?:€|euro)\s*(\(?-?(?:\d{1,3}(?:[.\s]\d{3})+|\d+)(?:[,.]\d+)?\)?)",
flags=re.IGNORECASE,
)
EXPORTABLE_STATUSES = {"OBSERVED", "DERIVED", "USER_CONFIRMED"}
ANNUAL_NEGATIVE_CONFIRMATION_KEYS = {
"guarantees_and_commitments",
"contingent_liabilities",
"related_party_transactions",
"off_balance_sheet_arrangements",
"derivatives",
"post_closing_events",
"accounting_policy_changes",
"prior_period_errors",
"going_concern_uncertainties",
"non_market_transactions",
"double_format_events",
}
SUPPORTED_LEGAL_FORMS = {"SRL", "SPA", "SAPA"}
MAX_INPUT_BYTES = 100 * 1024 * 1024
MAX_XLSX_UNCOMPRESSED_BYTES = 250 * 1024 * 1024
MAX_TEMPLATE_ROWS = 20_000
MAX_MONEY_INTEGER_DIGITS = 18
MAX_MONEY_FRACTION_DIGITS = 6
PDF_REVIEW_DECLARATIONS = {
"headers_and_columns_reviewed",
"account_rows_reviewed",
"monetary_values_reviewed",
"excluded_rows_reviewed",
}
TRIAL_BALANCE_COLUMNS = {
"account_code",
"account_description",
"opening_debit",
"opening_credit",
"opening_signed",
"period_debit",
"period_credit",
"closing_debit",
"closing_credit",
"closing_signed",
"prior_closing_debit",
"prior_closing_credit",
"prior_closing_signed",
}
class CaseState(StrEnum):
"""Supported workflow states from the product specification."""
DRAFT = "DRAFT"
INGESTING = "INGESTING"
INPUT_REVIEW = "INPUT_REVIEW"
MAPPING_REVIEW = "MAPPING_REVIEW"
STATEMENT_REVIEW = "STATEMENT_REVIEW"
DATA_GAPS = "DATA_GAPS"
NOTE_DRAFT = "NOTE_DRAFT"
VALIDATION_FAILED = "VALIDATION_FAILED"
READY_FOR_REVIEW = "READY_FOR_REVIEW"
APPROVED = "APPROVED"
EXPORTED = "EXPORTED"
ARCHIVED = "ARCHIVED"
UNSUPPORTED = "UNSUPPORTED"
class EvidenceStatus(StrEnum):
"""Evidence states preserved by every canonical and XBRL fact."""
OBSERVED = "OBSERVED"
DERIVED = "DERIVED"
USER_CONFIRMED = "USER_CONFIRMED"
MODEL_SUGGESTED = "MODEL_SUGGESTED"
ASSUMED = "ASSUMED"
MISSING = "MISSING"
NOT_APPLICABLE = "NOT_APPLICABLE"
class ParserConvention(StrEnum):
"""Supported meanings of debit and credit progressives."""
TURNOVER_EXCLUDES_OPENING = "TURNOVER_EXCLUDES_OPENING"
TURNOVER_INCLUDES_OPENING = "TURNOVER_INCLUDES_OPENING"
TURNOVER_INCLUDES_CLOSING_ENTRIES = "TURNOVER_INCLUDES_CLOSING_ENTRIES"
SIGNED_BALANCE_ONLY = "SIGNED_BALANCE_ONLY"
UNKNOWN = "UNKNOWN"
@dataclass(frozen=True)
class ValidationIssue:
"""One deterministic validation result."""
issue_id: str
severity: str
rule_id: str
message: str
affected_facts: tuple[str, ...] = ()
source_refs: tuple[str, ...] = ()
override_allowed: bool = False
def as_dict(self) -> dict[str, Any]:
"""Return a JSON-compatible issue object."""
payload = {
"issue_id": self.issue_id,
"severity": self.severity,
"rule_id": self.rule_id,
"message": self.message,
"affected_facts": list(self.affected_facts),
"source_refs": list(self.source_refs),
"override_allowed": self.override_allowed,
}
payload["fingerprint"] = _issue_fingerprint(payload)
return payload
def _now() -> str:
return datetime.now(tz=timezone.utc).replace(microsecond=0).isoformat()
def _canonical_json(value: Any) -> bytes:
return json.dumps(
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
def _sha256_bytes(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def _issue_fingerprint(issue: Mapping[str, Any]) -> str:
"""Bind a review decision to the exact issue content it considered."""
return _sha256_bytes(
_canonical_json(
{
"severity": issue["severity"],
"rule_id": issue["rule_id"],
"message": issue["message"],
"affected_facts": list(issue.get("affected_facts", [])),
"source_refs": list(issue.get("source_refs", [])),
}
)
)
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _reject_symlink_path_components(path: Path, label: str) -> None:
"""Reject a path when it or any existing ancestor is a symbolic link."""
absolute = path.absolute()
for component in (absolute, *absolute.parents):
if component.is_symlink():
raise ValueError(f"{label} must not contain symbolic-link components")
def _controlled_output_path(output_dir: Path, label: str) -> Path:
"""Validate an output destination without resolving through symlinks."""
output = output_dir.absolute()
_reject_symlink_path_components(output, label)
resolved = output.resolve(strict=False)
if resolved in {Path("/"), Path.home().resolve()}:
raise ValueError(f"Refusing a broad {label.lower()}")
output.parent.mkdir(parents=True, exist_ok=True)
_reject_symlink_path_components(output, label)
if output.exists() and not output.is_dir():
raise ValueError(f"{label} must be a directory")
return output
def _directory_file_manifest(directory: Path, label: str) -> dict[str, str]:
"""Hash a flat artifact directory while refusing links and nested entries."""
manifest: dict[str, str] = {}
for path in sorted(directory.iterdir(), key=lambda item: item.name):
if path.is_symlink() or not path.is_file():
raise ValueError(f"{label} may contain regular files only")
manifest[path.name] = _sha256_file(path)
return manifest
def _publish_output_directory(staging: Path, output: Path, label: str) -> None:
"""Atomically publish complete artifacts or accept an exact retry replay."""
_reject_symlink_path_components(output, label)
if output.exists():
existing = _directory_file_manifest(output, label)
staged = _directory_file_manifest(staging, label)
if existing:
if existing != staged:
raise ValueError(
f"{label} already contains different or incomplete artifacts"
)
return
output.rmdir()
staging.replace(output)
def _decimal_text(value: Decimal) -> str:
_validate_monetary_decimal(value)
text = format(value, "f")
if value == value.to_integral():
return f"{text.split('.', maxsplit=1)[0]}.00"
return text
def _validate_monetary_decimal(value: Decimal) -> Decimal:
"""Reject non-finite or operationally unbounded monetary values."""
if not value.is_finite():
raise ValueError("Monetary values must be finite")
if value.is_zero():
integer_digits = 1
else:
integer_digits = max(value.adjusted() + 1, 1)
fractional_digits = max(-value.as_tuple().exponent, 0)
if integer_digits > MAX_MONEY_INTEGER_DIGITS:
raise ValueError(
f"Monetary values support at most {MAX_MONEY_INTEGER_DIGITS} integer digits"
)
if fractional_digits > MAX_MONEY_FRACTION_DIGITS:
raise ValueError(
f"Monetary values support at most {MAX_MONEY_FRACTION_DIGITS} fractional digits"
)
return value
def _decimal_from_normalized_text(text: str, raw: Any, label: str) -> Decimal:
"""Parse one bounded, non-exponent monetary lexical representation."""
descriptor = f"{label} monetary".strip()
match = NORMALIZED_MONEY_LEXICAL.fullmatch(text)
if not match:
raise ValueError(f"Invalid {descriptor} value: {raw!r}")
integer = match.group("integer").lstrip("0") or "0"
fraction = match.group("fraction") or ""
if len(integer) > MAX_MONEY_INTEGER_DIGITS:
raise ValueError(
f"{descriptor.capitalize()} values support at most "
f"{MAX_MONEY_INTEGER_DIGITS} integer digits"
)
if len(fraction) > MAX_MONEY_FRACTION_DIGITS:
raise ValueError(
f"{descriptor.capitalize()} values support at most "
f"{MAX_MONEY_FRACTION_DIGITS} fractional digits"
)
try:
return _validate_monetary_decimal(Decimal(text))
except InvalidOperation as exc:
raise ValueError(f"Invalid {descriptor} value: {raw!r}") from exc
def _reported_decimal(value: Decimal, precision: int) -> Decimal:
"""Round a presentation value without changing canonical precision."""
quantum = Decimal(1).scaleb(-precision)
return value.quantize(quantum, rounding=ROUND_HALF_UP)
def _previous_year_date(value: date) -> date:
"""Return the corresponding prior-year date, including a leap-day fallback."""
try:
return value.replace(year=value.year - 1)
except ValueError:
if value.month == 2 and value.day == 29:
return value.replace(year=value.year - 1, day=28)
raise
def _xbrl_fact_id(kind: str, stable_id: object, context_ref: str) -> str:
"""Build a stable XML NCName for one rendered fact occurrence."""
raw = f"fact_{kind}_{stable_id}_{context_ref}"
normalized = re.sub(r"[^A-Za-z0-9_.-]+", "_", raw).strip("_.-")
if not normalized or not re.match(r"[A-Za-z_]", normalized):
normalized = f"fact_{normalized}"
return normalized
def normalize_decimal(raw: Any) -> Decimal:
"""Parse locale-neutral or Italian-formatted money as an exact Decimal."""
if isinstance(raw, Decimal):
return _validate_monetary_decimal(raw)
if isinstance(raw, bool) or raw is None:
raise ValueError("A monetary value is required")
if isinstance(raw, int):
return _validate_monetary_decimal(Decimal(raw))
if isinstance(raw, float) and not math.isfinite(raw):
raise ValueError("Monetary values must be finite")
text = str(raw).strip().replace("\u00a0", "").replace(" ", "")
if not text:
raise ValueError("A monetary value is required; blank is not zero")
has_opening_parenthesis = text.startswith("(")
has_closing_parenthesis = text.endswith(")")
if has_opening_parenthesis != has_closing_parenthesis:
raise ValueError(f"Invalid monetary value: {raw!r}")
negative = has_opening_parenthesis and has_closing_parenthesis
if negative:
text = text[1:-1]
if text.startswith(("+", "-")):
raise ValueError(
f"Invalid monetary value with both parentheses and sign: {raw!r}"
)
text = re.sub(r"^(EUR|€)", "", text, flags=re.IGNORECASE)
text = re.sub(r"(EUR|€)$", "", text, flags=re.IGNORECASE)
if "," in text and "." in text:
if text.rfind(",") > text.rfind("."):
text = text.replace(".", "").replace(",", ".")
else:
text = text.replace(",", "")
elif "," in text:
text = text.replace(".", "").replace(",", ".")
value = _decimal_from_normalized_text(text, raw, "")
return -value if negative else value
def _normalize_narrative_money_literal(raw: str, language: str) -> Decimal:
"""Parse a prose monetary literal using the approved output locale."""
text = raw.strip().replace("\u00a0", "").replace(" ", "")
has_opening_parenthesis = text.startswith("(")
has_closing_parenthesis = text.endswith(")")
if has_opening_parenthesis != has_closing_parenthesis:
raise ValueError(f"Invalid narrative monetary value: {raw!r}")
negative = has_opening_parenthesis and has_closing_parenthesis
if negative:
text = text[1:-1]
if text.startswith(("+", "-")):
raise ValueError(
"Invalid narrative monetary value with both parentheses and sign: "
f"{raw!r}"
)
if language == "it":
if re.fullmatch(r"-?\d{1,3}(?:\.\d{3})+", text):
text = text.replace(".", "")
elif "," in text:
text = text.replace(".", "").replace(",", ".")
elif language == "en":
if re.fullmatch(r"-?\d{1,3}(?:,\d{3})+", text):
text = text.replace(",", "")
elif "." in text:
text = text.replace(",", "")
else:
raise ValueError("Narrative monetary parsing supports Italian or English")
value = _decimal_from_normalized_text(text, raw, "narrative")
return -value if negative else value
def _case_payload_for_hash(case: Mapping[str, Any]) -> dict[str, Any]:
payload = deepcopy(dict(case))
payload.pop("_pending_before_hash", None)
payload.pop("approval", None)
payload.pop("approval_snapshots", None)
payload.pop("audit_events", None)
payload.pop("updated_at", None)
return payload
def _manifest_hash(case: Mapping[str, Any]) -> str:
documents = [
{"document_id": item["document_id"], "sha256": item["sha256"]}
for item in case.get("source_documents", [])
]
return _sha256_bytes(_canonical_json(documents))
def _available_evidence_refs(case: Mapping[str, Any]) -> set[str]:
"""Return evidence identifiers already anchored in the current case."""
refs = {
str(document["document_id"]) for document in case.get("source_documents", [])
}
refs.update(
str(ref)
for document in case.get("source_documents", [])
for ref in document.get("source_refs", [])
)
refs.update(
str(anchor["source_ref"])
for anchor in (case.get("trial_balance") or {}).get("source_anchors", [])
)
refs.update(
str((fact.get("source_anchor") or {}).get("source_ref"))
for fact in (case.get("prior_xbrl") or {}).get("facts", [])
if (fact.get("source_anchor") or {}).get("source_ref")
)
refs.update(
str(fact["fact_id"])
for fact in [
*case.get("canonical_facts", []),
*case.get("taxonomy_facts", []),
*case.get("schedule_taxonomy_facts", []),
]
)
refs.update(
str(ref)
for schedule in case.get("schedules", [])
for row in schedule.get("rows", [])
for ref in row.get("source_refs", [])
)
refs.update(
str(ref)
for answer in case.get("disclosure_answers", [])
for ref in answer.get("source_refs", [])
)
return refs
def _computation_context(
case: Mapping[str, Any],
template_version: str,
*,
model_version: str | None = None,
) -> dict[str, Any]:
"""Bind a derived output to its exact inputs, decisions, and versions."""
return {
"case_id": case["case_id"],
"revision_id": case["revision_id"],
"input_manifest_hash": _manifest_hash(case),
"mapping_version": _sha256_bytes(_canonical_json(case.get("mappings", []))),
"rule_pack_versions": deepcopy(dict(case["rule_pack_versions"])),
"regulatory_rule_pack_checksums": {
"statutory_rule_pack": case.get("rule_pack_checksum"),
"oic_rule_pack": case.get("oic_rule_pack_checksum"),
"filing_instruction_pack": case.get("filing_instruction_pack_checksum"),
"disclosure_rule_pack": case.get("disclosure_rule_pack_checksum"),
"statutory_presentation_rule_pack": case.get(
"statutory_presentation_rule_pack_checksum"
),
"schedule_taxonomy_adapter_rule_pack": case.get(
"schedule_taxonomy_adapter_rule_pack_checksum"
),
},
"filing_campaign_year": case.get("filing_campaign_year"),
"taxonomy_checksum": case.get("taxonomy_checksum"),
"model_version": model_version,
"template_version": template_version,
"computed_at": _now(),
}
def _review_content_hash(case: Mapping[str, Any]) -> str:
"""Hash the substantive review content independently of workflow metadata."""
payload = deepcopy(dict(case))
payload.pop("_pending_before_hash", None)
for key in (
"approval",
"approval_snapshots",
"artifacts",
"audit_events",
"external_validation",
"preview",
"xbrl_review",
"revision_id",
"state",
"updated_at",
"validation",
):
payload.pop(key, None)
return _sha256_bytes(_canonical_json(payload))
def _reviewed_preview_bytes(preview: Mapping[str, Any]) -> bytes:
"""Decode and verify the exact HTML bytes shown during professional review."""
encoded = preview.get("content_base64")
if not isinstance(encoded, str) or not encoded:
raise ValueError("The reviewed preview does not contain bound HTML bytes")
try:
content = base64.b64decode(encoded, validate=True)
except (binascii.Error, ValueError) as exc:
raise ValueError(
"The reviewed preview contains invalid base64 content"
) from exc
if len(content) != preview.get("size_bytes"):
raise ValueError("The reviewed preview byte length does not match its receipt")
if _sha256_bytes(content) != preview.get("sha256"):
raise ValueError("The reviewed preview checksum does not match its receipt")
return content
def _next_document_id(case: Mapping[str, Any]) -> str:
numbers = [
int(match.group(1))
for item in case.get("source_documents", [])
if (match := re.fullmatch(r"doc_(\d+)", str(item.get("document_id", ""))))
]
return f"doc_{(max(numbers, default=0) + 1):04d}"
def _record_event(
case: dict[str, Any],
action: str,
actor: str,
details: Mapping[str, Any] | None = None,
*,
after_hash: str | None = None,
) -> None:
before_hash = case.pop("_pending_before_hash", None)
resolved_after_hash = after_hash or _sha256_bytes(
_canonical_json(_case_payload_for_hash(case))
)
case.setdefault("audit_events", []).append(
{
"event_id": f"evt_{len(case.get('audit_events', [])) + 1:06d}",
"action": action,
"actor": actor,
"at": _now(),
"case_id": case["case_id"],
"tenant_id": case["tenant_id"],
"revision_id": case["revision_id"],
"details": dict(details or {}),
"before_hash": before_hash,
"after_hash": resolved_after_hash,
"originating_interface": "DIRECT_LIBRARY",
}
)
def _record_evidence_attached(
case: dict[str, Any], actor: str, document: Mapping[str, Any]
) -> None:
"""Record the specification event for checksum-bound case evidence."""
_record_event(
case,
"evidence_attached",
actor,
{
"document_id": document["document_id"],
"purpose": document["purpose"],
"sha256": document["sha256"],
},
)
def _mapping_audit_summary(value: Mapping[str, Any] | None) -> dict[str, Any] | None:
"""Return a bounded mapping-change summary plus its exact content hash."""
if value is None:
return None
allocations = list(value.get("allocations", []))
return {
"decision": value.get("decision"),
"candidate_source": value.get("candidate_source"),
"canonical_lines": sorted(
{str(item.get("canonical_line")) for item in allocations}
),
"xbrl_concepts": sorted(
{
str(item["xbrl_concept"])
for item in allocations
if item.get("xbrl_concept")
}
),
"content_sha256": _sha256_bytes(_canonical_json(value)),
}
def _mutate(case: dict[str, Any], actor: str, action: str) -> None:
before_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
prior_approval = case.get("approval")
if prior_approval:
archived = deepcopy(prior_approval)
archived["invalidated_at"] = _now()
archived["invalidated_by_action"] = action
case.setdefault("approval_snapshots", []).append(archived)
case["approval"] = None
_record_event(case, "snapshot_invalidated", actor, {"reason": action})
review_metadata_actions = {
"preview_rendered",
"validation_run",
"xbrl_review_prepared",
}
if case.get("preview") and action not in review_metadata_actions:
case["preview"] = None
_record_event(case, "preview_invalidated", actor, {"reason": action})
if case.get("xbrl_review") and action not in review_metadata_actions:
case["xbrl_review"] = None
_record_event(case, "xbrl_review_invalidated", actor, {"reason": action})
revision_number = int(str(case["revision_id"]).split("_")[-1]) + 1
case["revision_id"] = f"rev_{revision_number}"
case["updated_at"] = _now()
case["_pending_before_hash"] = before_hash
def _clear_accounting_dependent_reviews(
case: dict[str, Any], *, preserve_adjustments: bool = False
) -> None:
"""Invalidate reviewed outputs whose meaning depends on accounting amounts."""
replacements: dict[str, Any] = {
"adjustments": [],
"comparative_reconciliation_decisions": [],
"canonical_facts": [],
"statements": None,
"taxonomy_facts": [],
"taxonomy_fact_context": None,
"schedule_taxonomy_adapter": None,
"schedule_taxonomy_facts": [],
"statutory_presentation": None,
"taxonomy_representation": None,
"micro_reporting": None,
"schedules": [],
"disclosure_answers": [],
"disclosure_trigger_flags": [],
"disclosure_trigger_decisions": [],
"disclosure_coverage": None,
"questionnaire": [],
"note_outline": [],
"note_outline_context": None,
"narrative_blocks": [],
"narrative_context": None,
"model_mapping_suggestions": [],
"disclosure_activation_suggestions": [],
"narrative_suggestions": [],
"review_decisions": [],
"validation": None,
}
for key, value in replacements.items():
if key == "adjustments" and preserve_adjustments:
continue
case[key] = value
def _clear_narrative_reviews(case: dict[str, Any]) -> None:
"""Invalidate accepted prose after any supporting structured fact changes."""
case["narrative_blocks"] = []
case["narrative_context"] = None
case["narrative_suggestions"] = []
case["review_decisions"] = []
case["validation"] = None
def _clear_schedule_taxonomy_adapter(case: dict[str, Any]) -> None:
"""Invalidate note-table output after any adapter input changes."""
case["schedule_taxonomy_adapter"] = None
case["schedule_taxonomy_facts"] = []
case["validation"] = None
def _ensure_revision(case: Mapping[str, Any], expected_revision: str) -> None:
if case["revision_id"] != expected_revision:
raise ValueError(
f"Stale revision: expected {expected_revision}, current {case['revision_id']}"
)
def load_case(case_dir: Path) -> dict[str, Any]:
"""Load a case only after its durable JSON checksum is verified."""
if case_dir.is_symlink():
raise ValueError("Case directory must not be a symbolic link")
path = case_dir.resolve() / CASE_FILE
checksum_path = case_dir.resolve() / CASE_CHECKSUM_FILE
if path.is_symlink() or checksum_path.is_symlink():
raise ValueError("Case record and checksum must not be symbolic links")
if not path.is_file():
raise FileNotFoundError(f"Case file not found: {path}")
if not checksum_path.is_file():
raise ValueError("Case checksum metadata is missing")
record = path.read_bytes()
expected_checksum = checksum_path.read_text(encoding="ascii").strip()
if len(expected_checksum) != 64 or any(
character not in "0123456789abcdef" for character in expected_checksum
):
raise ValueError("Case checksum metadata is invalid")
if _sha256_bytes(record) != expected_checksum:
raise ValueError("Case record integrity verification failed")
payload = json.loads(record)
if payload.get("schema_version") != SCHEMA_VERSION:
raise ValueError("Unsupported case schema version")
return payload
def save_case(case_dir: Path, case: Mapping[str, Any]) -> Path:
"""Atomically persist a case and checksum without following symlinks."""
if case_dir.is_symlink():
raise ValueError("Case directory must not be a symbolic link")
resolved = case_dir.resolve()
resolved.mkdir(parents=True, exist_ok=True)
target = resolved / CASE_FILE
checksum_target = resolved / CASE_CHECKSUM_FILE
if target.is_symlink() or checksum_target.is_symlink():
raise ValueError("Refusing to write a case or checksum through a symbolic link")
temporary = resolved / f".{CASE_FILE}.tmp"
checksum_temporary = resolved / f".{CASE_CHECKSUM_FILE}.tmp"
if temporary.is_symlink() or checksum_temporary.is_symlink():
raise ValueError("Refusing to write a case through a temporary symbolic link")
record = _canonical_json(case) + b"\n"
digest = _sha256_bytes(record)
temporary.write_bytes(record)
checksum_temporary.write_text(digest + "\n", encoding="ascii")
temporary.replace(target)
checksum_temporary.replace(checksum_target)
if _sha256_file(target) != digest:
raise OSError("Persisted case checksum verification failed")
return target
def _validate_entity(entity: Mapping[str, Any]) -> list[str]:
reasons: list[str] = []
if str(entity.get("legal_form", "")).upper() not in SUPPORTED_LEGAL_FORMS:
reasons.append("UNSUPPORTED_LEGAL_FORM")
if str(entity.get("accounting_framework", "")).upper() != "OIC":
reasons.append("UNSUPPORTED_ACCOUNTING_FRAMEWORK")
if entity.get("listed") is not False:
reasons.append("LISTED_OR_UNCONFIRMED")
if entity.get("regulated_sector") is not False:
reasons.append("REGULATED_SECTOR_OR_UNCONFIRMED")
if entity.get("consolidated") is not False:
reasons.append("CONSOLIDATED_OR_UNCONFIRMED")
if entity.get("final_liquidation") is not False:
reasons.append("FINAL_LIQUIDATION_OR_UNCONFIRMED")
return reasons
def _validate_required_entity_profile(entity: Mapping[str, Any]) -> None:
"""Reject incomplete identity/profile data required by every MVP case."""
missing = [
key
for key in ("legal_name", "tax_identifier", "registered_office")
if not str(entity.get(key, "")).strip()
]
if missing:
raise ValueError(
"Entity profile is missing required fields: " + ", ".join(missing)
)
if not isinstance(entity.get("first_financial_year"), bool):
raise ValueError(
"Entity profile requires an explicit first-financial-year flag"
)
if entity["first_financial_year"] is False and str(
entity.get("prior_year_form", "")
).upper() not in {"ORDINARY", "ABBREVIATED", "MICRO"}:
raise ValueError("A non-first-year case requires the prior-year statutory form")
micro_exclusions = entity.get("micro_exclusion_flags")
if not isinstance(micro_exclusions, list) or any(
not isinstance(item, str) or not item.strip() for item in micro_exclusions
):
raise ValueError(
"Entity profile requires reviewed micro-exclusion flags, including an empty list"
)
def _taxonomy_output_contracts(rule_pack: Mapping[str, Any]) -> dict[str, Any]:
"""Validate and copy versioned non-statement taxonomy output mappings."""
raw_contracts = rule_pack.get("taxonomy_output_contracts")
if not isinstance(raw_contracts, Mapping):
raise ValueError("Statutory rule pack requires taxonomy output contracts")
micro = raw_contracts.get("MICRO_FOOTER_TEXT")
if not isinstance(micro, Mapping) or not SAFE_QNAME.fullmatch(
str(micro.get("xbrl_concept", ""))
):
raise ValueError("Statutory rule pack requires a micro-footer XBRL concept")
labels = micro.get("labels")
expected_keys = {
"guarantees_commitments_contingencies",
"director_auditor_compensation",
"own_and_parent_shares",
}
if not isinstance(labels, Mapping) or any(
not isinstance(labels.get(language), Mapping)
or set(labels[language]) != expected_keys
or any(not str(value).strip() for value in labels[language].values())
for language in ("it", "en")
):
raise ValueError("Micro-footer output labels are incomplete")
not_applicable = micro.get("not_applicable_text")
if not isinstance(not_applicable, Mapping) or any(
not str(not_applicable.get(language, "")).strip() for language in ("it", "en")
):
raise ValueError("Micro-footer negative-confirmation text is incomplete")
return deepcopy(dict(raw_contracts))
def _validate_source_register(rule_pack: Mapping[str, Any], pack_label: str) -> None:
"""Require auditable official-source metadata for a regulatory pack."""
sources = rule_pack.get("source_register")
if not isinstance(sources, list) or not sources:
raise ValueError(f"{pack_label} requires a non-empty source register")
for source in sources:
if not isinstance(source, Mapping) or not all(
str(source.get(key, "")).strip()
for key in ("id", "kind", "title", "url", "retrieved_on")
):
raise ValueError(f"{pack_label} contains an incomplete source record")
if not str(source["url"]).startswith("https://"):
raise ValueError(f"{pack_label} source URLs must use HTTPS")
date.fromisoformat(str(source["retrieved_on"]))
def _validate_effective_rule_pack(
rule_pack: Mapping[str, Any],
expected_kind: str,
period_start: date,
early_adoption_flags: Sequence[str] = (),
) -> None:
"""Validate fixed regulatory applicability required for auditability."""
if (
rule_pack.get("schema_version") != 1
or str(rule_pack.get("kind")) != expected_kind
):
raise ValueError(f"Expected a schema-1 {expected_kind} rule pack")
if not SAFE_ID.fullmatch(str(rule_pack.get("id", ""))):
raise ValueError(f"{expected_kind} rule pack requires a safe identifier")
effective_from = date.fromisoformat(str(rule_pack["effective_from"]))
effective_to = date.fromisoformat(str(rule_pack["effective_to"]))
if (
effective_from > effective_to
or not effective_from <= period_start <= effective_to
):
raise ValueError(
f"{expected_kind} rule pack is not effective for the reporting period"
)
flags = {str(item) for item in early_adoption_flags}
recognized = {
str(item) for item in rule_pack.get("recognized_early_adoption_flags", [])
}
if expected_kind == "OIC_ACCOUNTING_RULES":
if flags - recognized:
raise ValueError(
"OIC rule pack does not recognize the requested early-adoption flags"
)
mandatory_from = date.fromisoformat(
str(rule_pack.get("mandatory_from", rule_pack["effective_from"]))
)
early_flag = str(rule_pack.get("early_adoption_flag", ""))
if period_start < mandatory_from and (
not early_flag or early_flag not in flags
):
raise ValueError("OIC rule pack requires its reviewed early-adoption flag")
if period_start >= mandatory_from and flags:
raise ValueError(
"Early-adoption flags are invalid after the OIC pack becomes mandatory"
)
_validate_oic_professional_review_rules(rule_pack)
_validate_source_register(rule_pack, expected_kind)
def _validate_oic_professional_review_rules(
rule_pack: Mapping[str, Any],
) -> None:
"""Require explicit reviewer-owned behavior from every OIC rule pack."""
rules = rule_pack.get("professional_review_rules")
if not isinstance(rules, list):
raise ValueError("OIC rule pack requires professional review rules")
rule_ids: set[str] = set()
answer_keys: set[str] = set()
for rule in rules:
if not isinstance(rule, Mapping):
raise ValueError("OIC professional review rules must be objects")
rule_id = str(rule.get("id", ""))
if not SAFE_ID.fullmatch(rule_id) or rule_id in rule_ids:
raise ValueError("OIC professional review rule IDs must be safe and unique")
rule_ids.add(rule_id)
if str((rule.get("trigger") or {}).get("kind", "")).upper() != "ALWAYS":
raise ValueError(
"OIC professional review applicability must remain reviewer-owned"
)
requirements = rule.get("requirements")
if not isinstance(requirements, list) or not requirements:
raise ValueError("OIC professional review rules require answer gates")
required_answers = {
str(item.get("key", ""))
for item in requirements
if isinstance(item, Mapping)
and str(item.get("kind", "")).upper() == "ANSWER"
}
question = rule.get("question")
questions = list(rule.get("questions", []))
if not questions and isinstance(question, Mapping):
questions = [question]
question_answers = {
str(item.get("answer_key", ""))
for item in questions
if isinstance(item, Mapping)
}
if not required_answers or required_answers != question_answers:
raise ValueError(
"Each OIC professional review answer gate requires one question"
)
if answer_keys & required_answers:
raise ValueError("OIC professional review answer keys must be unique")
answer_keys.update(required_answers)
def _effective_disclosure_rule_pack(
case: Mapping[str, Any], rule_pack: Mapping[str, Any]
) -> dict[str, Any]:
"""Combine statutory disclosures with the locked OIC review checklist."""
effective = deepcopy(dict(rule_pack))
statutory_rules = list(effective.get("rules", []))
oic_rules = deepcopy(list(case.get("oic_professional_review_rules", [])))
statutory_ids = {str(item.get("id")) for item in statutory_rules}
oic_ids = {str(item.get("id")) for item in oic_rules}
if len(oic_ids) != len(oic_rules) or statutory_ids & oic_ids:
raise ValueError("OIC and statutory disclosure rule IDs must be unique")
effective["rules"] = [*statutory_rules, *oic_rules]
return effective
def _validate_filing_instruction_pack(
rule_pack: Mapping[str, Any], campaign_year: int
) -> None:
"""Validate filing instructions against the explicitly selected campaign."""
if (
rule_pack.get("schema_version") != 1
or rule_pack.get("kind") != "FILING_INSTRUCTIONS"
or not SAFE_ID.fullmatch(str(rule_pack.get("id", "")))
):
raise ValueError("Expected a schema-1 FILING_INSTRUCTIONS rule pack")
campaign_from = date.fromisoformat(str(rule_pack["campaign_from"]))
campaign_to = date.fromisoformat(str(rule_pack["campaign_to"]))
if (
campaign_from > campaign_to
or int(rule_pack.get("campaign_year", -1)) != campaign_year
):
raise ValueError("Filing-instruction pack does not match the selected campaign")
_validate_source_register(rule_pack, "FILING_INSTRUCTIONS")
def _validate_disclosure_rule_pack(
rule_pack: Mapping[str, Any], period_start: date
) -> None:
"""Validate disclosure-pack identity and applicability before mutation."""
if (
rule_pack.get("schema_version") != 1
or rule_pack.get("kind") != "DISCLOSURE_RULES"
or not SAFE_ID.fullmatch(str(rule_pack.get("id", "")))
or not isinstance(rule_pack.get("rules"), list)
or not rule_pack["rules"]
):
raise ValueError("Expected a non-empty schema-1 disclosure rule pack")
effective_from = date.fromisoformat(str(rule_pack["effective_from"]))
effective_to = date.fromisoformat(str(rule_pack["effective_to"]))
if (
effective_from > effective_to
or not effective_from <= period_start <= effective_to
):
raise ValueError("Disclosure rule pack is not effective for the case period")
def _bundled_regulatory_pack(pack_id: str, expected_kind: str) -> dict[str, Any]:
"""Resolve an identifier only to a controlled bundled regulatory pack."""
if not SAFE_ID.fullmatch(pack_id):
raise ValueError(f"Invalid {expected_kind} rule-pack identifier")
matches: list[dict[str, Any]] = []
for path in sorted(RULE_PACK_ROOT.glob("*.json")):
if path.is_symlink() or not path.is_file():
continue
payload = json.loads(path.read_bytes())
if (
isinstance(payload, dict)
and str(payload.get("id")) == pack_id
and str(payload.get("kind")) == expected_kind
):
matches.append(payload)
if len(matches) != 1:
raise ValueError(
f"{expected_kind} identifier must resolve to exactly one controlled pack"
)
return matches[0]
def _resolve_regulatory_pack(raw: Any, expected_kind: str) -> dict[str, Any]:
"""Resolve a caller-supplied identifier only to a bundled controlled pack."""
if isinstance(raw, Mapping):
raise ValueError(
f"{expected_kind} must be selected by controlled rule-pack identifier"
)
return _bundled_regulatory_pack(str(raw), expected_kind)
def create_case(
case_dir: Path,
payload: Mapping[str, Any],
rule_pack: Mapping[str, Any],
actor: str,
) -> dict[str, Any]:
"""Create a scope-checked case locked to explicit rule and taxonomy versions."""
entity = dict(payload.get("entity", {}))
period = dict(payload.get("period", {}))
case_id = str(payload.get("case_id") or "")
if not SAFE_ID.fullmatch(case_id):
raise ValueError("case_id must be a safe stable identifier")
start = date.fromisoformat(str(period["start"]))
end = date.fromisoformat(str(period["end"]))
if start > end:
raise ValueError("Reporting period start must not follow its end")
prior_start_raw = str(entity.get("prior_period_start") or "")
prior_end_raw = str(entity.get("prior_period_end") or "")
if entity.get("first_financial_year") is True and (
prior_start_raw or prior_end_raw
):
raise ValueError("A first financial year cannot declare a comparative period")
if bool(prior_start_raw) != bool(prior_end_raw):
raise ValueError(
"An explicit comparative period requires both prior start and end"
)
if entity.get("first_financial_year") is not True and not prior_start_raw:
raise ValueError(
"A non-first-year case requires an explicit comparative period"
)
if prior_start_raw:
prior_start_date = date.fromisoformat(prior_start_raw)
prior_end_date = date.fromisoformat(prior_end_raw)
if prior_start_date > prior_end_date or prior_end_date >= start:
raise ValueError(
"The comparative period is invalid or overlaps the current year"
)
if str(payload.get("currency", "EUR")) != "EUR":
raise ValueError("MVP supports EUR only")
taxonomy_checksum = str(payload.get("taxonomy_checksum") or "")
if not re.fullmatch(r"[0-9a-f]{64}", taxonomy_checksum):
raise ValueError("A lowercase SHA-256 taxonomy package checksum is required")
unsupported_reasons = _validate_entity(entity)
state = CaseState.UNSUPPORTED if unsupported_reasons else CaseState.DRAFT
reporting_precision = int(payload.get("reporting_precision", 0))
if reporting_precision not in {0, 1, 2}:
raise ValueError("Reporting precision must be zero, one, or two decimals")
output_language = str(payload.get("output_language", "it")).strip().lower()
if output_language not in {"it", "en"}:
raise ValueError("Output language must be it or en")
_validate_required_entity_profile(entity)
raw_early_adoption_flags = payload.get("early_adoption_flags", [])
if not isinstance(raw_early_adoption_flags, list) or not all(
isinstance(item, str) and item.strip() for item in raw_early_adoption_flags
):
raise ValueError("Early-adoption flags must be a list of non-empty strings")
early_adoption_flags = sorted(set(raw_early_adoption_flags))
filing_campaign_year = int(payload["filing_campaign_year"])
if filing_campaign_year < 2016 or filing_campaign_year > 9999:
raise ValueError("Filing campaign year is outside the supported range")
oic_rule_pack = _resolve_regulatory_pack(
payload["oic_rule_pack"], "OIC_ACCOUNTING_RULES"
)
filing_instruction_pack = _resolve_regulatory_pack(
payload.get("filing_instruction_pack", "RI_2026.1"),
"FILING_INSTRUCTIONS",
)
_validate_effective_rule_pack(rule_pack, "STATUTORY_FORM_RULES", start)
_validate_effective_rule_pack(
oic_rule_pack,
"OIC_ACCOUNTING_RULES",
start,
early_adoption_flags,
)
_validate_filing_instruction_pack(filing_instruction_pack, filing_campaign_year)
taxonomy_output_contracts = _taxonomy_output_contracts(rule_pack)
case: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"case_id": case_id,
"tenant_id": str(payload["tenant_id"]),
"revision_id": "rev_1",
"state": state.value,
"entity": entity,
"period": {"start": start.isoformat(), "end": end.isoformat()},
"currency": "EUR",
"reporting_precision": reporting_precision,
"output_language": output_language,
"requested_form": str(payload.get("requested_form", "AUTO_RECOMMEND")),
"rule_pack_versions": {
"jurisdiction": "IT",
"accounting_framework": "OIC",
"statutory_rule_pack": rule_pack["id"],
"oic_rule_pack": str(oic_rule_pack["id"]),
"taxonomy_id": str(payload.get("taxonomy_id", "PCI_2018-11-04")),
"filing_instruction_pack": str(filing_instruction_pack["id"]),
"early_adoption_flags": early_adoption_flags,
},
"rule_pack_checksum": _sha256_bytes(_canonical_json(rule_pack)),
"oic_rule_pack_checksum": _sha256_bytes(_canonical_json(oic_rule_pack)),
"oic_professional_review_rules": deepcopy(
list(oic_rule_pack["professional_review_rules"])
),
"filing_instruction_pack_checksum": _sha256_bytes(
_canonical_json(filing_instruction_pack)
),
"filing_campaign_year": filing_campaign_year,
"taxonomy_output_contracts": taxonomy_output_contracts,
"statutory_presentation_rule_pack_checksum": None,
"taxonomy_checksum": taxonomy_checksum,
"unsupported_reasons": unsupported_reasons,
"source_documents": [],
"file_security_scans": [],
"prior_xbrl": None,
"comparative_reconciliation_decisions": [],
"trial_balance": None,
"pdf_trial_balance_candidate": None,
"form_analysis": None,
"selected_form": None,
"mappings": [],
"adjustments": [],
"taxonomy_facts": [],
"taxonomy_fact_context": None,
"schedule_taxonomy_adapter": None,
"schedule_taxonomy_adapter_rule_pack_checksum": None,
"schedule_taxonomy_facts": [],
"taxonomy_catalogue_build": None,
"statutory_presentation_required": True,
"statutory_presentation": None,
"taxonomy_representation": None,
"micro_reporting": None,
"mapping_candidates": [],
"taxonomy_mapping_index": None,
"canonical_facts": [],
"statements": None,
"schedules": [],
"disclosure_answers": [],
"disclosure_trigger_flags": [],
"disclosure_trigger_decisions": [],
"disclosure_rule_pack": None,
"disclosure_coverage": None,
"questionnaire": [],
"note_outline": [],
"note_outline_context": None,
"prior_narrative_suggestions": [],
"narrative_blocks": [],
"narrative_context": None,
"intelligence_runs": [],
"latest_workflow_guidance": None,
"model_mapping_suggestions": [],
"disclosure_activation_suggestions": [],
"narrative_suggestions": [],
"client_history_suggestions": None,
"review_decisions": [],
"regulatory_migrations": [],
"preview": None,
"xbrl_review": None,
"external_validation": None,
"validation": None,
"approval": None,
"approval_snapshots": [],
"artifacts": [],
"audit_events": [],
"created_at": _now(),
"updated_at": _now(),
}
_record_event(case, "case_created", actor)
if unsupported_reasons:
_record_event(
case, "unsupported_case_detected", actor, {"reasons": unsupported_reasons}
)
save_case(case_dir, case)
return case
def migrate_regulatory_versions(
case: dict[str, Any],
migration: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Explicitly migrate an open case and invalidate all regulated outputs."""
_ensure_revision(case, expected_revision)
if case.get("approval") or case.get("state") in {
CaseState.APPROVED,
CaseState.EXPORTED,
CaseState.ARCHIVED,
}:
raise ValueError("Approved, exported, and archived cases cannot be migrated")
if case.get("state") == CaseState.UNSUPPORTED:
raise ValueError("Unsupported cases cannot be migrated")
required = {
"reason",
"statutory_rule_pack",
"oic_rule_pack",
"taxonomy_id",
"taxonomy_checksum",
"filing_instruction_pack",
"filing_campaign_year",
"early_adoption_flags",
}
allowed = required | {"disclosure_rule_pack"}
if set(migration) - allowed or not required.issubset(migration):
raise ValueError(
"Regulatory migration requires exactly the controlled version fields"
)
reason = str(migration["reason"]).strip()
if len(reason) < 10:
raise ValueError("Regulatory migration requires a specific reason")
statutory_rule_pack = _resolve_regulatory_pack(
migration["statutory_rule_pack"], "STATUTORY_FORM_RULES"
)
period_start = date.fromisoformat(case["period"]["start"])
_validate_effective_rule_pack(
statutory_rule_pack, "STATUTORY_FORM_RULES", period_start
)
taxonomy_checksum = str(migration["taxonomy_checksum"])
if not re.fullmatch(r"[0-9a-f]{64}", taxonomy_checksum):
raise ValueError("Migrated taxonomy checksum must be a lowercase SHA-256")
early_adoption_flags = migration["early_adoption_flags"]
if not isinstance(early_adoption_flags, list) or not all(
isinstance(item, str) and item.strip() for item in early_adoption_flags
):
raise ValueError("Early-adoption flags must be a list of non-empty strings")
early_adoption_flags = sorted(set(early_adoption_flags))
oic_rule_pack = _resolve_regulatory_pack(
migration["oic_rule_pack"], "OIC_ACCOUNTING_RULES"
)
filing_instruction_pack = _resolve_regulatory_pack(
migration["filing_instruction_pack"], "FILING_INSTRUCTIONS"
)
filing_campaign_year = int(migration["filing_campaign_year"])
_validate_effective_rule_pack(
oic_rule_pack,
"OIC_ACCOUNTING_RULES",
period_start,
early_adoption_flags,
)
_validate_filing_instruction_pack(filing_instruction_pack, filing_campaign_year)
disclosure_rule_pack = migration.get("disclosure_rule_pack")
if disclosure_rule_pack is not None:
disclosure_rule_pack = _resolve_regulatory_pack(
disclosure_rule_pack, "DISCLOSURE_RULES"
)
_validate_disclosure_rule_pack(disclosure_rule_pack, period_start)
previous_versions = deepcopy(case["rule_pack_versions"])
target_versions = {
"jurisdiction": "IT",
"accounting_framework": "OIC",
"statutory_rule_pack": str(statutory_rule_pack["id"]),
"oic_rule_pack": str(oic_rule_pack["id"]),
"taxonomy_id": str(migration["taxonomy_id"]),
"filing_instruction_pack": str(filing_instruction_pack["id"]),
"early_adoption_flags": list(early_adoption_flags),
}
if disclosure_rule_pack is not None:
target_versions["disclosure_rule_pack"] = str(disclosure_rule_pack["id"])
statutory_checksum = _sha256_bytes(_canonical_json(statutory_rule_pack))
oic_checksum = _sha256_bytes(_canonical_json(oic_rule_pack))
filing_checksum = _sha256_bytes(_canonical_json(filing_instruction_pack))
disclosure_checksum = (
disclosure_rule_pack_hash(disclosure_rule_pack)
if disclosure_rule_pack is not None
else None
)
version_changes = {
key: {"from": previous_versions.get(key), "to": target_versions.get(key)}
for key in sorted(set(previous_versions) | set(target_versions))
if previous_versions.get(key) != target_versions.get(key)
}
checksum_changes = {
"statutory_rule_pack": {
"from": case.get("rule_pack_checksum"),
"to": statutory_checksum,
},
"taxonomy": {
"from": case.get("taxonomy_checksum"),
"to": taxonomy_checksum,
},
"oic_rule_pack": {
"from": case.get("oic_rule_pack_checksum"),
"to": oic_checksum,
},
"filing_instruction_pack": {
"from": case.get("filing_instruction_pack_checksum"),
"to": filing_checksum,
},
"filing_campaign_year": {
"from": case.get("filing_campaign_year"),
"to": filing_campaign_year,
},
"disclosure_rule_pack": {
"from": case.get("disclosure_rule_pack_checksum"),
"to": disclosure_checksum,
},
}
checksum_changes = {
key: value
for key, value in checksum_changes.items()
if value["from"] != value["to"]
}
if not version_changes and not checksum_changes:
raise ValueError("Regulatory migration target is identical to the locked case")
reset_values: dict[str, Any] = {
"form_analysis": None,
"selected_form": None,
"mappings": [],
"adjustments": [],
"taxonomy_facts": [],
"taxonomy_fact_context": None,
"schedule_taxonomy_adapter": None,
"schedule_taxonomy_facts": [],
"taxonomy_catalogue_build": None,
"statutory_presentation": None,
"taxonomy_representation": None,
"micro_reporting": None,
"mapping_candidates": [],
"canonical_facts": [],
"statements": None,
"schedules": [],
"disclosure_answers": [],
"disclosure_trigger_flags": [],
"disclosure_trigger_decisions": [],
"disclosure_rule_pack": None,
"disclosure_coverage": None,
"questionnaire": [],
"note_outline": [],
"note_outline_context": None,
"prior_narrative_suggestions": [],
"narrative_blocks": [],
"narrative_context": None,
"intelligence_runs": [],
"latest_workflow_guidance": None,
"model_mapping_suggestions": [],
"disclosure_activation_suggestions": [],
"narrative_suggestions": [],
"review_decisions": [],
"preview": None,
"xbrl_review": None,
"external_validation": None,
"validation": None,
"artifacts": [],
}
invalidated = [
{
"component": key,
"prior_sha256": _sha256_bytes(_canonical_json(case.get(key))),
}
for key, empty_value in reset_values.items()
if case.get(key) not in (None, [], {}) and case.get(key) != empty_value
]
_mutate(case, actor, "regulatory_versions_migrated")
case["rule_pack_versions"] = target_versions
case["rule_pack_checksum"] = statutory_checksum
case["oic_rule_pack_checksum"] = oic_checksum
case["oic_professional_review_rules"] = deepcopy(
list(oic_rule_pack["professional_review_rules"])
)
case["filing_instruction_pack_checksum"] = filing_checksum
case["filing_campaign_year"] = filing_campaign_year
case["taxonomy_output_contracts"] = _taxonomy_output_contracts(statutory_rule_pack)
case["statutory_presentation_rule_pack_checksum"] = None
case["schedule_taxonomy_adapter_rule_pack_checksum"] = None
case["rule_pack_versions"].pop("statutory_presentation_rule_pack", None)
case["taxonomy_checksum"] = taxonomy_checksum
case["disclosure_rule_pack_checksum"] = disclosure_checksum
for key, empty_value in reset_values.items():
case[key] = empty_value
case["state"] = (
CaseState.INPUT_REVIEW
if (case.get("trial_balance") or {}).get("confirmed_convention")
else CaseState.DRAFT
)
report = {
"migration_id": f"migration_{len(case.get('regulatory_migrations', [])) + 1:04d}",
"requested_by": actor,
"requested_at": _now(),
"reason": reason,
"from_revision_id": expected_revision,
"to_revision_id": case["revision_id"],
"version_changes": version_changes,
"checksum_changes": checksum_changes,
"invalidated_components": invalidated,
"retained_evidence": {
"source_document_count": len(case.get("source_documents", [])),
"trial_balance_retained": case.get("trial_balance") is not None,
"prior_xbrl_retained": case.get("prior_xbrl") is not None,
},
"required_recomputation": [
"FORM_ELIGIBILITY",
"MAPPING",
"STATEMENTS",
"SCHEDULES",
"DISCLOSURES",
"NOTES",
"STATUTORY_PRESENTATION",
"LOCAL_XBRL_REVIEW",
],
"revalidation_status": "REQUIRED",
"revalidation_runs": [],
}
case.setdefault("regulatory_migrations", []).append(report)
_record_event(
case,
"regulatory_versions_migrated",
actor,
{
"migration_id": report["migration_id"],
"changed_version_keys": sorted(version_changes),
"changed_checksum_keys": sorted(checksum_changes),
"invalidated_components": [item["component"] for item in invalidated],
},
)
return case
def ingest_prior_xbrl(
case: dict[str, Any], source: Path, actor: str, expected_revision: str
) -> dict[str, Any]:
"""Attach source-anchored prior XBRL after exact entity and period checks."""
_ensure_revision(case, expected_revision)
if case["state"] == CaseState.UNSUPPORTED:
raise ValueError("Prior XBRL cannot be added to an unsupported case")
if case["entity"].get("first_financial_year") is True:
raise ValueError("A first financial year cannot have a prior filed XBRL")
parsed = parse_prior_xbrl(source)
allowed_identifiers = {
str(case["entity"].get("tax_identifier", "")).strip(),
str(case["entity"].get("prior_xbrl_identifier", "")).strip(),
} - {""}
if parsed["entity_identifier"] not in allowed_identifiers:
raise ValueError("Prior-XBRL entity identifier does not match the case")
expected_prior_end = str(case["entity"].get("prior_period_end") or "")
if not expected_prior_end:
raise ValueError("The case has no explicit comparative period end")
matching_contexts = [
context
for context in parsed["contexts"]
if context["period"].get("end") == expected_prior_end
]
if not matching_contexts:
raise ValueError("Prior-XBRL periods do not align with the comparative period")
document_id = _next_document_id(case)
for fact in parsed["facts"]:
fact["source_anchor"]["document_id"] = document_id
fact["source_anchor"]["source_ref"] = f"{document_id}_{fact['fact_id']}"
document = {
"document_id": document_id,
"purpose": "PRIOR_XBRL",
"file_name": parsed["file_name"],
"media_type": "application/xbrl+xml",
"sha256": parsed["sha256"],
"size_bytes": source.resolve().stat().st_size,
"parser_profile": "prior_xbrl_v4",
"parsed_at": _now(),
}
_mutate(case, actor, "prior_xbrl_parsed")
case["source_documents"] = [
item
for item in case.get("source_documents", [])
if item.get("purpose") != "PRIOR_XBRL"
] + [document]
case["prior_xbrl"] = {
**parsed,
"document_id": document_id,
"expected_prior_end": expected_prior_end,
"matching_context_ids": [item["context_id"] for item in matching_contexts],
"computation_context": _computation_context(case, "prior-xbrl-parser-v4"),
}
case["mapping_candidates"] = []
case["taxonomy_mapping_index"] = None
case["comparative_reconciliation_decisions"] = []
case["validation"] = None
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
_record_event(
case,
"document_parsed",
actor,
{"document_id": document_id, "parser_profile": "prior_xbrl_v4"},
)
_record_event(
case,
"prior_xbrl_parsed",
actor,
{"document_id": document_id, "fact_count": len(parsed["facts"])},
)
return case
def attach_supporting_document(
case: dict[str, Any],
source: Path,
purpose: str,
description: str,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Attach checksum-bound supporting evidence without interpreting its content."""
_ensure_revision(case, expected_revision)
if source.is_symlink() or not source.is_file():
raise ValueError("Supporting evidence must be a regular local file")
resolved = source.resolve()
if resolved.stat().st_size > MAX_INPUT_BYTES:
raise ValueError("Supporting evidence exceeds the 100 MiB size limit")
allowed_suffixes = {
".csv": "text/csv",
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
".pdf": "application/pdf",
".txt": "text/plain",
".md": "text/markdown",
".json": "application/json",
".xml": "application/xml",
".html": "text/html",
".htm": "text/html",
}
suffix = resolved.suffix.lower()
if suffix not in allowed_suffixes:
raise ValueError("Unsupported supporting-evidence file type")
normalized_purpose = purpose.strip().upper()
allowed_purposes = {
"SUPPORTING_EVIDENCE",
"CORPORATE_RESOLUTION",
"TAX_COMPUTATION",
"CONTRACT",
"QUESTIONNAIRE_EVIDENCE",
"RESTATEMENT_WORKPAPER",
"GOING_CONCERN_ASSESSMENT",
}
if normalized_purpose not in allowed_purposes:
raise ValueError("Unsupported supporting-evidence purpose")
normalized_description = description.strip()
if not normalized_description:
raise ValueError("Supporting evidence requires a description")
document_id = _next_document_id(case)
document = {
"document_id": document_id,
"purpose": normalized_purpose,
"description": normalized_description,
"file_name": resolved.name,
"media_type": allowed_suffixes[suffix],
"sha256": _sha256_file(resolved),
"size_bytes": resolved.stat().st_size,
"source_refs": [document_id],
"content_treated_as_untrusted": True,
"attached_at": _now(),
}
_mutate(case, actor, "supporting_evidence_attached")
case.setdefault("source_documents", []).append(document)
case["validation"] = None
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
return case
def _load_table(
path: Path, sheet: str | None
) -> tuple[list[str], list[list[Any]], str]:
if path.stat().st_size > MAX_INPUT_BYTES:
raise ValueError("Input table exceeds the 100 MiB size limit")
suffix = path.suffix.lower()
if suffix == ".csv":
with path.open("r", encoding="utf-8-sig", newline="") as handle:
sample = handle.read(8192)
handle.seek(0)
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
rows = list(csv.reader(handle, dialect))
if not rows:
raise ValueError("The CSV file is empty")
if len(rows) - 1 > MAX_TEMPLATE_ROWS:
raise ValueError("Input table exceeds the 20,000-row MVP limit")
return [str(value) for value in rows[0]], rows[1:], "csv"
if suffix == ".xlsx":
try:
from openpyxl import load_workbook
except ImportError as exc:
raise RuntimeError("openpyxl is required for XLSX ingestion") from exc
try:
with zipfile.ZipFile(path) as archive:
total_uncompressed = 0
for member in archive.infolist():
member_path = Path(member.filename)
if member.flag_bits & 0x1:
raise ValueError("Encrypted XLSX members are not supported")
if member_path.is_absolute() or ".." in member_path.parts:
raise ValueError("Unsafe XLSX member path")
total_uncompressed += member.file_size
if (
member.compress_size > 0
and member.file_size / member.compress_size > 200
):
raise ValueError("Suspicious XLSX compression ratio")
lowered = member.filename.lower()
if "vbaproject" in lowered:
raise ValueError("Spreadsheet macros are not supported")
if lowered.startswith("xl/externallinks/"):
raise ValueError("XLSX external links are not supported")
if total_uncompressed > MAX_XLSX_UNCOMPRESSED_BYTES:
raise ValueError("XLSX expanded content exceeds the safety limit")
corrupt_member = archive.testzip()
if corrupt_member:
raise ValueError(f"Corrupt XLSX member: {corrupt_member}")
except zipfile.BadZipFile as exc:
raise ValueError("XLSX input is not a valid ZIP package") from exc
workbook = load_workbook(path, read_only=True, data_only=True, keep_links=False)
selected = workbook[sheet] if sheet else workbook[workbook.sheetnames[0]]
selected_title = selected.title
rows = list(selected.iter_rows(values_only=True))
if not rows:
raise ValueError("The XLSX worksheet is empty")
if len(rows) - 1 > MAX_TEMPLATE_ROWS:
raise ValueError("Input table exceeds the 20,000-row MVP limit")
formula_workbook = load_workbook(
path, read_only=True, data_only=False, keep_links=False
)
formula_sheet = formula_workbook[selected_title]
for value_row, formula_row in zip(
rows, formula_sheet.iter_rows(), strict=False
):
for cached_value, formula_cell in zip(value_row, formula_row, strict=False):
if formula_cell.data_type == "f" and cached_value is None:
raise ValueError(
"XLSX formula has no trusted cached value for deterministic import"
)
formula_workbook.close()
workbook.close()
return (
["" if value is None else str(value) for value in rows[0]],
[list(row) for row in rows[1:]],
selected_title,
)
raise ValueError("Only CSV and XLSX trial balances are supported")
def _normalized_header(value: str) -> str:
text = re.sub(r"[^a-z0-9]+", "_", value.strip().lower())
aliases = {
"codice": "account_code",
"cod_conto": "account_code",
"codice_conto": "account_code",
"conto": "account_code",
"descrizione": "account_description",
"descrizione_conto": "account_description",
"denominazione_conto": "account_description",
"saldo_iniziale_dare": "opening_debit",
"saldo_iniziale_avere": "opening_credit",
"saldo_iniziale": "opening_signed",
"movimenti_dare": "period_debit",
"movimenti_avere": "period_credit",
"progressivo_dare": "period_debit",
"progressivo_avere": "period_credit",
"progressivi_dare": "period_debit",
"progressivi_avere": "period_credit",
"saldo_finale_dare": "closing_debit",
"saldo_finale_avere": "closing_credit",
"saldo_finale": "closing_signed",
"saldo_precedente_dare": "prior_closing_debit",
"saldo_precedente_avere": "prior_closing_credit",
"saldo_precedente": "prior_closing_signed",
}
return aliases.get(text.strip("_"), text.strip("_"))
def _column_reference(position: int) -> str:
"""Return the one-based spreadsheet column letter for a zero-based position."""
if position < 0:
raise ValueError("Column positions cannot be negative")
value = position + 1
letters = ""
while value:
value, remainder = divmod(value - 1, 26)
letters = chr(65 + remainder) + letters
return letters
def _normalized_header_columns(headers: Sequence[str]) -> list[dict[str, Any]]:
"""Normalize headers while preserving coordinates and rejecting collisions."""
columns: list[dict[str, Any]] = []
seen: dict[str, dict[str, Any]] = {}
for position, original in enumerate(headers):
normalized = _normalized_header(original)
coordinate = _column_reference(position)
if not normalized:
raise ValueError(f"Column {coordinate} has a blank or unsupported header")
prior = seen.get(normalized)
if prior:
raise ValueError(
"Duplicate normalized column "
f"{normalized!r}: {prior['column']} ({prior['column_header']!r}) "
f"and {coordinate} ({original!r})"
)
column = {
"column": coordinate,
"column_index": position + 1,
"column_header": original,
"normalized_column": normalized,
}
seen[normalized] = column
columns.append(column)
return columns
def _row_decimal(row: Mapping[str, Any], key: str) -> Decimal:
if key not in row:
raise ValueError(f"Required monetary field is missing: {key}")
return normalize_decimal(row[key])
def _calibrate(rows: Sequence[Mapping[str, Any]], tolerance: Decimal) -> dict[str, Any]:
matches_excludes = 0
matches_includes = 0
tested = 0
samples: list[dict[str, str]] = []
for row in rows:
opening = _row_decimal(row, "opening_signed")
closing = _row_decimal(row, "closing_signed")
period_net = _row_decimal(row, "period_debit") - _row_decimal(
row, "period_credit"
)
excludes_diff = opening + period_net - closing
includes_diff = period_net - closing
if abs(excludes_diff) <= tolerance:
matches_excludes += 1
if abs(includes_diff) <= tolerance:
matches_includes += 1
tested += 1
if len(samples) < 5:
samples.append(
{
"account_code": str(row["account_code"]),
"opening_signed": _decimal_text(opening),
"period_net": _decimal_text(period_net),
"closing_signed": _decimal_text(closing),
"excludes_opening_difference": _decimal_text(excludes_diff),
"includes_opening_difference": _decimal_text(includes_diff),
}
)
if not rows or all(
_row_decimal(row, "period_debit") == 0
and _row_decimal(row, "period_credit") == 0
for row in rows
):
detected = ParserConvention.SIGNED_BALANCE_ONLY
elif tested and matches_excludes == tested and matches_includes != tested:
detected = ParserConvention.TURNOVER_EXCLUDES_OPENING
elif tested and matches_includes == tested and matches_excludes != tested:
detected = ParserConvention.TURNOVER_INCLUDES_OPENING
else:
detected = ParserConvention.UNKNOWN
return {
"detected_convention": detected.value,
"tested_rows": tested,
"matches_turnover_excludes_opening": matches_excludes,
"matches_turnover_includes_opening": matches_includes,
"unmatched_rows": tested - max(matches_excludes, matches_includes),
"tolerance": _decimal_text(tolerance),
"samples": samples,
"closing_entries_assessment": {
"appears_included": None,
"status": "REQUIRES_PROFESSIONAL_CONFIRMATION",
"reason": (
"The supported numeric columns do not mechanically distinguish "
"ordinary turnover from closing entries."
),
},
}
def _trial_balance_layout(
normalized_headers: Sequence[str], first_year: bool
) -> tuple[bool, bool]:
"""Return mechanically supported layout flags or reject the source shape."""
required = {"account_code", "account_description"}
if not required.issubset(normalized_headers):
raise ValueError("Trial balance requires account_code and account_description")
separate_current_layout = {
"opening_debit",
"opening_credit",
"period_debit",
"period_credit",
"closing_debit",
"closing_credit",
}.issubset(normalized_headers)
signed_current_layout = {
"opening_signed",
"period_debit",
"period_credit",
"closing_signed",
}.issubset(normalized_headers)
comparative_columns = {
"prior_closing_debit",
"prior_closing_credit",
"prior_closing_signed",
}
supplied_comparative_columns = comparative_columns & set(normalized_headers)
if first_year and supplied_comparative_columns:
raise ValueError(
"A first-financial-year trial balance cannot contain comparative columns"
)
separate_layout = separate_current_layout and (
first_year
or {"prior_closing_debit", "prior_closing_credit"}.issubset(normalized_headers)
)
signed_layout = signed_current_layout and (
first_year or "prior_closing_signed" in normalized_headers
)
if not separate_layout and not signed_layout:
raise ValueError(
"Trial balance does not match supported separate or signed layouts"
)
return separate_layout, signed_layout
def _parse_trial_balance_rows(
case: Mapping[str, Any],
headers: Sequence[str],
raw_rows: Sequence[Sequence[Any]],
*,
document_id: str,
sheet_name: str,
parser_profile: str,
tolerance: Decimal,
anchor_metadata: Mapping[tuple[int, int], Mapping[str, Any]] | None = None,
) -> dict[str, Any]:
"""Normalize one reviewed table while retaining source-specific coordinates."""
header_columns = _normalized_header_columns(headers)
normalized_headers = [item["normalized_column"] for item in header_columns]
first_year = case["entity"].get("first_financial_year") is True
separate_layout, _ = _trial_balance_layout(normalized_headers, first_year)
entries: list[dict[str, Any]] = []
seen: set[str] = set()
anchors: list[dict[str, Any]] = []
for row_offset, values in enumerate(raw_rows):
index = row_offset + 2
row = {
name: values[pos] if pos < len(values) else None
for pos, name in enumerate(normalized_headers)
}
code = str(row.get("account_code") or "").strip()
description = str(row.get("account_description") or "").strip()
if not code and not description:
continue
if not code:
raise ValueError(f"Missing account identifier at row {index}")
if code in seen:
raise ValueError(
f"Duplicate account identifier without aggregation key: {code}"
)
seen.add(code)
if separate_layout:
opening_debit = _row_decimal(row, "opening_debit")
opening_credit = _row_decimal(row, "opening_credit")
closing_debit = _row_decimal(row, "closing_debit")
closing_credit = _row_decimal(row, "closing_credit")
prior_debit = (
None if first_year else _row_decimal(row, "prior_closing_debit")
)
prior_credit = (
None if first_year else _row_decimal(row, "prior_closing_credit")
)
opening_signed = opening_debit - opening_credit
closing_signed = closing_debit - closing_credit
prior_signed = (
None
if prior_debit is None or prior_credit is None
else prior_debit - prior_credit
)
else:
opening_signed = _row_decimal(row, "opening_signed")
closing_signed = _row_decimal(row, "closing_signed")
prior_signed = (
None if first_year else _row_decimal(row, "prior_closing_signed")
)
opening_debit = max(opening_signed, Decimal("0"))
opening_credit = max(-opening_signed, Decimal("0"))
closing_debit = max(closing_signed, Decimal("0"))
closing_credit = max(-closing_signed, Decimal("0"))
prior_debit = (
None if prior_signed is None else max(prior_signed, Decimal("0"))
)
prior_credit = (
None if prior_signed is None else max(-prior_signed, Decimal("0"))
)
period_debit = _row_decimal(row, "period_debit")
period_credit = _row_decimal(row, "period_credit")
source_refs: list[str] = []
account_id = f"acc_{len(entries) + 1:06d}"
for column in header_columns:
position = int(column["column_index"]) - 1
column_name = str(column["normalized_column"])
anchor_id = f"src_{len(anchors) + 1:07d}"
source_refs.append(anchor_id)
raw_value = values[position] if position < len(values) else None
metadata = dict((anchor_metadata or {}).get((row_offset, position), {}))
anchor: dict[str, Any] = {
"source_ref": anchor_id,
"document_id": document_id,
"account_id": account_id,
"sheet": sheet_name,
"row": index,
"column": column["column"],
"column_header": column["column_header"],
"normalized_column": column_name,
"raw_value": metadata.pop(
"raw_value", "" if raw_value is None else str(raw_value)
),
"parser_profile": parser_profile,
"confidence": metadata.pop("confidence", "HIGH"),
}
for key in (
"page",
"table_id",
"source_row",
"source_column",
"bbox",
"extraction_method",
"extraction_confidence",
"candidate_source_ref",
"correction",
"evidence_status",
):
if key in metadata:
anchor[key] = metadata[key]
if column_name not in {"account_code", "account_description"}:
anchor["normalized_value"] = _decimal_text(
_row_decimal(row, column_name)
)
else:
anchor["normalized_value"] = str(row[column_name] or "").strip()
anchors.append(anchor)
entries.append(
{
"account_id": account_id,
"account_code": code,
"account_description": description,
"opening_debit": _decimal_text(opening_debit),
"opening_credit": _decimal_text(opening_credit),
"opening_signed": _decimal_text(opening_signed),
"period_debit": _decimal_text(period_debit),
"period_credit": _decimal_text(period_credit),
"closing_debit": _decimal_text(closing_debit),
"closing_credit": _decimal_text(closing_credit),
"closing_signed": _decimal_text(closing_signed),
"prior_closing_debit": (
None if prior_debit is None else _decimal_text(prior_debit)
),
"prior_closing_credit": (
None if prior_credit is None else _decimal_text(prior_credit)
),
"prior_closing_signed": (
None if prior_signed is None else _decimal_text(prior_signed)
),
"source_refs": source_refs,
}
)
calibration = _calibrate(entries, tolerance)
debit_total = sum(Decimal(row["closing_debit"]) for row in entries)
credit_total = sum(Decimal(row["closing_credit"]) for row in entries)
calibration["closing_debit_total"] = _decimal_text(debit_total)
calibration["closing_credit_total"] = _decimal_text(credit_total)
calibration["closing_difference"] = _decimal_text(debit_total - credit_total)
return {
"layout": "SEPARATE_DEBIT_CREDIT" if separate_layout else "SIGNED_BALANCES",
"comparative_status": (
"NOT_APPLICABLE_FIRST_FINANCIAL_YEAR" if first_year else "PRESENT"
),
"entries": entries,
"source_anchors": anchors,
"calibration": calibration,
"header_columns": header_columns,
"parser_profile": parser_profile,
}
def _install_trial_balance(
case: dict[str, Any],
parsed: Mapping[str, Any],
document: Mapping[str, Any],
actor: str,
action: str,
*,
pdf_review: Mapping[str, Any] | None = None,
pdf_candidate: Mapping[str, Any] | None = None,
) -> None:
"""Install a validated table into the canonical case in one mutation."""
_mutate(case, actor, action)
document_id = str(document["document_id"])
case["source_documents"] = [
item
for item in case.get("source_documents", [])
if item.get("purpose") not in {"TRIAL_BALANCE", "TRIAL_BALANCE_PDF"}
and str(item.get("document_id")) != document_id
] + [dict(document)]
case["trial_balance"] = {
"layout": parsed["layout"],
"comparative_status": parsed["comparative_status"],
"entries": list(parsed["entries"]),
"source_anchors": list(parsed["source_anchors"]),
"calibration": dict(parsed["calibration"]),
"confirmed_convention": None,
"computation_context": _computation_context(
case, str(parsed["parser_profile"])
),
}
if pdf_review is not None:
case["trial_balance"]["pdf_extraction_review"] = dict(pdf_review)
case["pdf_trial_balance_candidate"] = (
None if pdf_candidate is None else dict(pdf_candidate)
)
_clear_accounting_dependent_reviews(case)
case["state"] = CaseState.INPUT_REVIEW
case["mappings"] = []
case["mapping_candidates"] = []
case["taxonomy_mapping_index"] = None
case["canonical_facts"] = []
case["statements"] = None
case["statutory_presentation"] = None
case["schedules"] = []
case["validation"] = None
def ingest_trial_balance(
case: dict[str, Any],
source_path: Path,
actor: str,
expected_revision: str,
sheet: str | None = None,
tolerance: Decimal = Decimal("0.01"),
) -> dict[str, Any]:
"""Parse one source-anchored CSV/XLSX trial balance."""
_ensure_revision(case, expected_revision)
if case["state"] == CaseState.UNSUPPORTED:
raise ValueError("Unsupported cases cannot ingest accounting data")
if source_path.is_symlink() or not source_path.is_file():
raise ValueError("Trial balance must be a regular local file")
path = source_path.resolve()
headers, raw_rows, sheet_name = _load_table(path, sheet)
document_id = _next_document_id(case)
parsed = _parse_trial_balance_rows(
case,
headers,
raw_rows,
document_id=document_id,
sheet_name=sheet_name,
parser_profile="generic-it-trial-balance-parser-v1",
tolerance=tolerance,
)
document = {
"document_id": document_id,
"purpose": "TRIAL_BALANCE",
"file_name": path.name,
"media_type": (
"text/csv"
if path.suffix.lower() == ".csv"
else "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
),
"sha256": _sha256_file(path),
"size_bytes": path.stat().st_size,
"sheet": sheet_name,
"columns": parsed["header_columns"],
"parser_profile": "generic-it-trial-balance-parser-v1",
"parsed_at": _now(),
}
_install_trial_balance(case, parsed, document, actor, "document_parsed")
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
_record_event(case, "document_parsed", actor, {"document_id": document_id})
return case
def _pdf_row_values(row: Mapping[str, Any]) -> list[str]:
"""Return whitespace-normalized values from one extracted PDF row."""
return [
" ".join(str(cell.get("raw_value") or "").split())
for cell in row.get("cells", [])
]
def _pdf_header_candidate(
table: Mapping[str, Any], *, first_year: bool
) -> dict[str, Any] | None:
"""Select a plausible header row using only exact aliases and geometry."""
candidates: list[dict[str, Any]] = []
for row in list(table.get("rows", []))[:12]:
values = _pdf_row_values(row)
nonempty = sum(bool(value) for value in values)
if nonempty < 4:
continue
normalized = [_normalized_header(value) for value in values]
recognized = [value for value in normalized if value in TRIAL_BALANCE_COLUMNS]
required = {"account_code", "account_description"} & set(recognized)
complete = False
if len(recognized) == len(set(recognized)):
try:
_trial_balance_layout(recognized, first_year)
complete = True
except ValueError:
pass
score = (
(1000 if complete else 0)
+ len(required) * 100
+ len(set(recognized)) * 10
+ nonempty
)
candidates.append(
{
"row": row,
"values": values,
"normalized": normalized,
"recognized": len(set(recognized)),
"complete": complete,
"score": score,
}
)
if not candidates:
return None
candidates.sort(
key=lambda item: (int(item["score"]), -int(item["row"]["row_index"])),
reverse=True,
)
return candidates[0]
def _aligned_pdf_cells(
row: Mapping[str, Any], header_row: Mapping[str, Any]
) -> list[dict[str, Any]]:
"""Align visual cells to header geometry without assigning accounting meaning."""
cells = [dict(item) for item in row.get("cells", [])]
headers = [dict(item) for item in header_row.get("cells", [])]
if len(cells) == len(headers):
return cells
header_boxes = [item.get("bbox") for item in headers]
if not headers or any(not box for box in header_boxes):
return cells
centers = [
(float(box[0]) + float(box[2])) / 2
for box in header_boxes
if isinstance(box, Sequence) and len(box) == 4
]
if len(centers) != len(headers):
return cells
boundaries = [
(centers[index] + centers[index + 1]) / 2 for index in range(len(centers) - 1)
]
buckets: list[list[dict[str, Any]]] = [[] for _ in headers]
for cell in cells:
box = cell.get("bbox")
if not isinstance(box, Sequence) or len(box) != 4:
return cells
center = (float(box[0]) + float(box[2])) / 2
position = sum(center > boundary for boundary in boundaries)
buckets[position].append(cell)
aligned: list[dict[str, Any]] = []
for bucket, header in zip(buckets, headers, strict=True):
if not bucket:
aligned.append(
{"raw_value": "", "bbox": header.get("bbox"), "confidence": 0.0}
)
continue
bucket.sort(key=lambda item: float((item.get("bbox") or [0])[0]))
boxes = [
tuple(float(value) for value in item["bbox"])
for item in bucket
if isinstance(item.get("bbox"), Sequence) and len(item["bbox"]) == 4
]
aligned.append(
{
"raw_value": " ".join(
str(item.get("raw_value") or "").strip()
for item in bucket
if str(item.get("raw_value") or "").strip()
),
"bbox": list(
(
min(item[0] for item in boxes),
min(item[1] for item in boxes),
max(item[2] for item in boxes),
max(item[3] for item in boxes),
)
),
"confidence": min(
float(item.get("confidence", 0.0)) for item in bucket
),
}
)
return aligned
PDF_CANDIDATE_HASH_FIELDS = (
"source_document_id",
"page_count",
"page_methods",
"methods",
"ocr_used",
"columns",
"rows",
"table_coverage",
"issues",
)
def _pdf_candidate_hashed_content(candidate: Mapping[str, Any]) -> dict[str, Any]:
"""Return every review-relevant extraction field bound by the candidate hash."""
return {key: candidate[key] for key in PDF_CANDIDATE_HASH_FIELDS}
def _pdf_candidate_from_extraction(
extraction: Mapping[str, Any], *, document_id: str, first_year: bool
) -> dict[str, Any]:
"""Build a complete, review-only trial-balance candidate from PDF geometry."""
tables = [
table for table in extraction.get("tables", []) if isinstance(table, Mapping)
]
table_headers = [
(table, candidate)
for table in tables
if (candidate := _pdf_header_candidate(table, first_year=first_year))
is not None
]
if not table_headers:
raise ValueError("No plausible tabular header was found in the PDF")
table_headers.sort(key=lambda item: int(item[1]["score"]), reverse=True)
selected_table, selected = table_headers[0]
selected_values = list(selected["values"])
selected_signature = tuple(selected["normalized"])
selected_header_row = selected["row"]
columns = []
header_cells = list(selected_header_row.get("cells", []))
for position, raw_header in enumerate(selected_values, start=1):
normalized = _normalized_header(raw_header)
columns.append(
{
"column_index": position,
"raw_header": raw_header,
"proposed_normalized_column": (
normalized if normalized in TRIAL_BALANCE_COLUMNS else None
),
"bbox": (
header_cells[position - 1].get("bbox")
if position <= len(header_cells)
else None
),
}
)
header_by_table = {
str(table["table_id"]): header for table, header in table_headers
}
candidate_rows: list[dict[str, Any]] = []
table_coverage: list[dict[str, Any]] = []
for table in tables:
table_id = str(table["table_id"])
header = header_by_table.get(table_id)
if table_id == str(selected_table["table_id"]):
disposition = "INCLUDED_HEADER"
alignment_header = selected_header_row
first_data_row = int(selected_header_row["row_index"]) + 1
elif header is not None and tuple(header["normalized"]) == selected_signature:
disposition = "INCLUDED_HEADER"
alignment_header = header["row"]
first_data_row = int(header["row"]["row_index"]) + 1
elif header is not None and (
bool(header["complete"]) or int(header["recognized"]) > 0
):
table_coverage.append(
{
"table_id": table_id,
"page": int(table["page"]),
"disposition": "BLOCKED_INCOMPATIBLE_HEADER",
"source_row_count": len(table.get("rows", [])),
"candidate_row_count": 0,
"unresolved_row_ids": [],
}
)
continue
else:
# Reusing selected column geometry for a headerless continuation is
# mechanical layout alignment. All resulting rows remain untrusted
# and require the same professional review as the first page.
disposition = "INCLUDED_CONTINUATION"
alignment_header = selected_header_row
first_data_row = 1
unresolved_row_ids: list[str] = []
blank_row_count = 0
included_count = 0
for row in table.get("rows", []):
source_row = int(row["row_index"])
if source_row < first_data_row:
continue
cells = _aligned_pdf_cells(row, alignment_header)
row_id = f"{table_id}_r{source_row:04d}"
if len(cells) != len(columns):
unresolved_row_ids.append(row_id)
continue
if not any(str(item.get("raw_value") or "").strip() for item in cells):
blank_row_count += 1
continue
candidate_cells = [
{
"candidate_source_ref": f"{row_id}_c{position:03d}",
"column_index": position,
"raw_value": str(cell.get("raw_value") or "").strip(),
"bbox": cell.get("bbox"),
"confidence": float(cell.get("confidence", 0.0)),
}
for position, cell in enumerate(cells, start=1)
]
candidate_rows.append(
{
"row_id": row_id,
"page": int(table["page"]),
"table_id": table_id,
"source_row": source_row,
"extraction_method": str(table["method"]),
"cells": candidate_cells,
}
)
included_count += 1
if len(candidate_rows) > MAX_TEMPLATE_ROWS:
raise ValueError("PDF table exceeds the 20,000-row MVP limit")
table_coverage.append(
{
"table_id": table_id,
"page": int(table["page"]),
"disposition": (
"BLOCKED_ROW_ALIGNMENT" if unresolved_row_ids else disposition
),
"source_row_count": len(table.get("rows", [])),
"candidate_row_count": included_count,
"blank_row_count": blank_row_count,
"unresolved_row_ids": unresolved_row_ids,
}
)
if not candidate_rows:
raise ValueError("No candidate accounting rows followed the PDF header")
page_count = int(extraction["page_count"])
page_methods = [dict(item) for item in extraction["page_methods"]]
if [int(item.get("page", 0)) for item in page_methods] != list(
range(1, page_count + 1)
):
raise ValueError("PDF extraction must account for every page exactly once")
actual_table_counts = {
page: sum(int(table["page"]) == page for table in tables)
for page in range(1, page_count + 1)
}
if any(
int(item.get("table_count", -1)) != actual_table_counts[int(item["page"])]
for item in page_methods
):
raise ValueError("PDF page table counts do not match extracted table coverage")
proposed = {
int(item["column_index"]): item.get("proposed_normalized_column")
for item in columns
}
account_code_index = next(
(index for index, name in proposed.items() if name == "account_code"), None
)
low_confidence_refs = [
str(cell["candidate_source_ref"])
for row in candidate_rows
for cell in row["cells"]
if float(cell["confidence"])
< float(extraction.get("ocr_review_threshold", 0.90))
and str(cell.get("raw_value") or "").strip()
]
rows_without_account = []
if account_code_index is not None:
rows_without_account = [
str(row["row_id"])
for row in candidate_rows
if not str(
row["cells"][account_code_index - 1].get("raw_value") or ""
).strip()
]
issues: list[dict[str, Any]] = []
if extraction.get("ocr_used"):
issues.append(
{
"code": "OCR_REVIEW_REQUIRED",
"count": len(candidate_rows),
"sample_refs": [str(item["row_id"]) for item in candidate_rows[:20]],
}
)
if low_confidence_refs:
issues.append(
{
"code": "LOW_CONFIDENCE_CELL",
"count": len(low_confidence_refs),
"sample_refs": low_confidence_refs[:50],
}
)
if rows_without_account:
issues.append(
{
"code": "ROW_WITHOUT_ACCOUNT_CODE",
"count": len(rows_without_account),
"sample_refs": rows_without_account[:50],
}
)
proposed_headers = [
str(item["proposed_normalized_column"])
for item in columns
if item.get("proposed_normalized_column")
]
mapping_complete = bool(selected["complete"])
if not mapping_complete:
issues.append(
{
"code": "COLUMN_MAPPING_REQUIRED",
"count": 1,
"sample_refs": [],
}
)
continuation_tables = [
item["table_id"]
for item in table_coverage
if item["disposition"] == "INCLUDED_CONTINUATION"
]
if continuation_tables:
issues.append(
{
"code": "CONTINUATION_HEADER_INHERITED",
"count": len(continuation_tables),
"sample_refs": continuation_tables[:50],
}
)
pages_requiring_disposition = [
int(item["page"]) for item in page_methods if int(item["table_count"]) == 0
]
if pages_requiring_disposition:
issues.append(
{
"code": "PAGE_DISPOSITION_REQUIRED",
"count": len(pages_requiring_disposition),
"sample_refs": [str(page) for page in pages_requiring_disposition[:50]],
}
)
blocked_tables = [
str(item["table_id"])
for item in table_coverage
if str(item["disposition"]).startswith("BLOCKED_")
]
if blocked_tables:
issues.append(
{
"code": "TABLE_COVERAGE_BLOCKED",
"count": len(blocked_tables),
"sample_refs": blocked_tables[:50],
}
)
candidate = {
"schema_version": 1,
"status": "PENDING_REVIEW",
"source_document_id": document_id,
"columns": columns,
"rows": candidate_rows,
"parser_profile": str(extraction["parser_profile"]),
"page_count": page_count,
"page_methods": page_methods,
"methods": list(extraction["methods"]),
"ocr_used": bool(extraction["ocr_used"]),
"row_count": len(candidate_rows),
"table_coverage": table_coverage,
"coverage_status": (
"BLOCKED"
if blocked_tables
else (
"PENDING_PAGE_DISPOSITIONS"
if pages_requiring_disposition
else "COMPLETE"
)
),
"proposed_headers": proposed_headers,
"proposed_mapping_complete": mapping_complete,
"issues": issues,
"created_at": _now(),
}
candidate["content_sha256"] = _sha256_bytes(
_canonical_json(_pdf_candidate_hashed_content(candidate))
)
return candidate
def ingest_pdf_trial_balance(
case: dict[str, Any],
source_path: Path,
actor: str,
expected_revision: str,
*,
ocr_enabled: bool = True,
ocr_language: str = "it",
) -> dict[str, Any]:
"""Extract a review-only trial-balance candidate from a readable or scanned PDF."""
_ensure_revision(case, expected_revision)
if case["state"] == CaseState.UNSUPPORTED:
raise ValueError("Unsupported cases cannot ingest accounting data")
if ocr_language not in {"it", "en"}:
raise ValueError("OCR language must be 'it' or 'en'")
_reject_symlink_path_components(source_path, "PDF trial balance")
if source_path.is_symlink() or not source_path.is_file():
raise ValueError("PDF trial balance must be a regular local file")
path = source_path.resolve()
extraction = extract_pdf_tables(
path,
ocr_enabled=ocr_enabled,
ocr_language=ocr_language,
max_bytes=MAX_INPUT_BYTES,
)
document_id = _next_document_id(case)
candidate = _pdf_candidate_from_extraction(
extraction,
document_id=document_id,
first_year=case["entity"].get("first_financial_year") is True,
)
document = {
"document_id": document_id,
"purpose": "TRIAL_BALANCE_PDF",
"file_name": path.name,
"media_type": "application/pdf",
"sha256": _sha256_file(path),
"size_bytes": path.stat().st_size,
"page_count": int(extraction["page_count"]),
"extraction_methods": list(extraction["methods"]),
"ocr_used": bool(extraction["ocr_used"]),
"parser_profile": str(extraction["parser_profile"]),
"parsed_at": _now(),
}
_mutate(case, actor, "pdf_trial_balance_extracted")
case["source_documents"] = [
item
for item in case.get("source_documents", [])
if item.get("purpose") not in {"TRIAL_BALANCE", "TRIAL_BALANCE_PDF"}
] + [document]
case["trial_balance"] = None
case["pdf_trial_balance_candidate"] = candidate
_clear_accounting_dependent_reviews(case)
case["state"] = CaseState.INPUT_REVIEW
case["mappings"] = []
case["mapping_candidates"] = []
case["taxonomy_mapping_index"] = None
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
_record_event(
case,
"pdf_trial_balance_extracted",
actor,
{
"document_id": document_id,
"candidate_sha256": candidate["content_sha256"],
"row_count": candidate["row_count"],
"ocr_used": candidate["ocr_used"],
},
)
return case
def _review_list(payload: Mapping[str, Any], key: str) -> list[Mapping[str, Any]]:
value = payload.get(key, [])
if isinstance(value, (str, bytes, bytearray)) or not isinstance(value, Sequence):
raise ValueError(f"PDF review {key} must be a list")
if any(not isinstance(item, Mapping) for item in value):
raise ValueError(f"PDF review {key} entries must be objects")
return [item for item in value if isinstance(item, Mapping)]
def record_pdf_trial_balance_review(
case: dict[str, Any],
payload: Mapping[str, Any],
reviewer: str,
expected_revision: str,
*,
tolerance: Decimal = Decimal("0.01"),
) -> dict[str, Any]:
"""Accept or reject one extraction candidate under explicit professional review."""
_ensure_revision(case, expected_revision)
candidate = case.get("pdf_trial_balance_candidate")
if (
not isinstance(candidate, Mapping)
or candidate.get("status") != "PENDING_REVIEW"
):
raise ValueError("No pending PDF trial-balance extraction is available")
decision = str(payload.get("decision") or "").upper()
if decision not in {"ACCEPTED", "REJECTED"}:
raise ValueError("PDF extraction review decision must be ACCEPTED or REJECTED")
reason = str(payload.get("reason") or "").strip()
if not reason:
raise ValueError("PDF extraction review requires a reason")
if _sha256_bytes(
_canonical_json(_pdf_candidate_hashed_content(candidate))
) != candidate.get("content_sha256"):
raise ValueError("PDF extraction candidate content hash does not match")
if decision == "REJECTED":
_mutate(case, reviewer, "pdf_trial_balance_reviewed")
rejected = deepcopy(candidate)
rejected.update(
{
"status": "REJECTED",
"review_reason": reason,
"reviewed_by": reviewer,
"reviewed_at": _now(),
}
)
case["pdf_trial_balance_candidate"] = rejected
_record_event(
case,
"pdf_trial_balance_reviewed",
reviewer,
{
"decision": decision,
"candidate_sha256": candidate["content_sha256"],
},
)
return case
declarations = payload.get("declarations")
if (
not isinstance(declarations, Mapping)
or set(declarations) != PDF_REVIEW_DECLARATIONS
):
raise ValueError(
"PDF acceptance requires the exact extraction review declarations"
)
missing_declarations = sorted(
key for key in PDF_REVIEW_DECLARATIONS if declarations.get(key) is not True
)
if missing_declarations:
raise ValueError(
"PDF extraction declarations are not confirmed: "
+ ", ".join(missing_declarations)
)
blocked_tables = [
str(item["table_id"])
for item in candidate.get("table_coverage", [])
if str(item.get("disposition", "")).startswith("BLOCKED_")
]
if blocked_tables:
raise ValueError(
"PDF extraction contains unresolved table coverage: "
+ ", ".join(blocked_tables)
)
required_page_dispositions = {
int(item["page"])
for item in candidate.get("page_methods", [])
if int(item.get("table_count", 0)) == 0
}
page_dispositions = _review_list(payload, "page_dispositions")
reviewed_pages: dict[int, dict[str, Any]] = {}
for item in page_dispositions:
page = int(item.get("page", 0))
page_decision = str(item.get("decision") or "").upper()
page_reason = str(item.get("reason") or "").strip()
if (
page not in required_page_dispositions
or page in reviewed_pages
or page_decision != "EXCLUDED_NON_ACCOUNTING"
or not page_reason
):
raise ValueError("PDF review contains invalid page dispositions")
reviewed_pages[page] = {
"page": page,
"decision": page_decision,
"reason": page_reason,
}
if set(reviewed_pages) != required_page_dispositions:
raise ValueError(
"PDF acceptance requires exact page dispositions for every page "
"without a reviewable table"
)
candidate_columns = {
int(item["column_index"]): item for item in candidate.get("columns", [])
}
mappings = _review_list(payload, "column_mapping")
if mappings:
selected_mapping = []
for item in mappings:
index = int(item.get("column_index", 0))
normalized = str(item.get("normalized_column") or "").strip()
if index not in candidate_columns:
raise ValueError("PDF review maps an unknown source column")
if normalized not in TRIAL_BALANCE_COLUMNS:
raise ValueError(
f"Unsupported reviewed trial-balance column: {normalized}"
)
selected_mapping.append(
{"column_index": index, "normalized_column": normalized}
)
else:
selected_mapping = [
{
"column_index": index,
"normalized_column": str(column["proposed_normalized_column"]),
}
for index, column in candidate_columns.items()
if column.get("proposed_normalized_column") in TRIAL_BALANCE_COLUMNS
]
selected_indices = [int(item["column_index"]) for item in selected_mapping]
selected_names = [str(item["normalized_column"]) for item in selected_mapping]
if len(selected_indices) != len(set(selected_indices)) or len(
selected_names
) != len(set(selected_names)):
raise ValueError("PDF review column mapping must be one-to-one")
_trial_balance_layout(
selected_names,
case["entity"].get("first_financial_year") is True,
)
rows = {str(item["row_id"]): item for item in candidate.get("rows", [])}
exclusions = _review_list(payload, "excluded_rows")
excluded: dict[str, dict[str, Any]] = {}
for item in exclusions:
row_id = str(item.get("row_id") or "")
row_kind = str(item.get("row_kind") or "").upper()
summarized = item.get("summarizes_row_ids", [])
exclusion_reason = str(item.get("reason") or "").strip()
if (
row_id not in rows
or row_id in excluded
or row_kind not in {"NON_ACCOUNT", "SUMMARY"}
or isinstance(summarized, (str, bytes, bytearray))
or not isinstance(summarized, Sequence)
or any(not isinstance(value, str) for value in summarized)
or len(summarized) != len(set(summarized))
or not exclusion_reason
):
raise ValueError("PDF review contains an invalid or duplicate excluded row")
summarized_ids = [str(value) for value in summarized]
if (row_kind == "SUMMARY") != bool(summarized_ids):
raise ValueError(
"PDF summary exclusions require named accepted rows; non-account "
"exclusions must not summarize rows"
)
excluded[row_id] = {
"row_id": row_id,
"row_kind": row_kind,
"summarizes_row_ids": summarized_ids,
"reason": exclusion_reason,
}
corrections = _review_list(payload, "corrections")
corrected: dict[tuple[str, int], dict[str, str]] = {}
for item in corrections:
row_id = str(item.get("row_id") or "")
index = int(item.get("column_index", 0))
correction_reason = str(item.get("reason") or "").strip()
value = item.get("value")
if (
row_id not in rows
or index not in selected_indices
or (row_id, index) in corrected
or row_id in excluded
or not correction_reason
or isinstance(value, (Mapping, Sequence))
and not isinstance(value, str)
):
raise ValueError("PDF review contains an invalid or duplicate correction")
corrected[(row_id, index)] = {
"value": "" if value is None else str(value),
"reason": correction_reason,
}
monetary_columns = [
(source_index, normalized_name)
for source_index, normalized_name in zip(
selected_indices, selected_names, strict=True
)
if normalized_name not in {"account_code", "account_description"}
]
def reviewed_decimal(row_id: str, source_index: int) -> Decimal:
"""Read one reviewed monetary value for mechanical exclusion checks."""
row = rows[row_id]
cell_lookup = {int(item["column_index"]): item for item in row["cells"]}
if source_index not in cell_lookup:
raise ValueError("PDF review row is missing a selected monetary column")
correction = corrected.get((row_id, source_index))
raw_value = str(cell_lookup[source_index].get("raw_value") or "")
reviewed_value = correction["value"] if correction else raw_value
return normalize_decimal(reviewed_value if str(reviewed_value).strip() else "0")
accepted_ids = set(rows) - set(excluded)
for row_id, exclusion in excluded.items():
if exclusion["row_kind"] == "NON_ACCOUNT":
if any(
reviewed_decimal(row_id, source_index) != Decimal("0")
for source_index, _normalized_name in monetary_columns
):
raise ValueError("PDF review cannot exclude a non-zero non-account row")
continue
summarized_ids = set(exclusion["summarizes_row_ids"])
if not summarized_ids <= accepted_ids:
raise ValueError(
"PDF summary exclusions may reference only accepted candidate rows"
)
for source_index, normalized_name in monetary_columns:
summary_value = reviewed_decimal(row_id, source_index)
accepted_total = sum(
(reviewed_decimal(item, source_index) for item in summarized_ids),
Decimal("0"),
)
if abs(summary_value - accepted_total) > tolerance:
raise ValueError(
"PDF summary exclusion does not reconcile for " + normalized_name
)
raw_rows: list[list[Any]] = []
anchor_metadata: dict[tuple[int, int], dict[str, Any]] = {}
accepted_row_ids: list[str] = []
for row in candidate.get("rows", []):
row_id = str(row["row_id"])
if row_id in excluded:
continue
cell_lookup = {int(item["column_index"]): item for item in row["cells"]}
values: list[Any] = []
output_row = len(raw_rows)
for output_column, source_index in enumerate(selected_indices):
cell = cell_lookup[source_index]
correction = corrected.get((row_id, source_index))
raw_value = str(cell.get("raw_value") or "")
value = correction["value"] if correction else raw_value
values.append(value)
metadata: dict[str, Any] = {
"raw_value": raw_value,
"page": int(row["page"]),
"table_id": str(row["table_id"]),
"source_row": int(row["source_row"]),
"source_column": source_index,
"bbox": cell.get("bbox"),
"extraction_method": str(row["extraction_method"]),
"extraction_confidence": float(cell.get("confidence", 0.0)),
"candidate_source_ref": str(cell["candidate_source_ref"]),
"confidence": "REVIEWED",
"evidence_status": EvidenceStatus.USER_CONFIRMED,
}
if correction:
metadata["correction"] = {
"confirmed_value": correction["value"],
"reason": correction["reason"],
"reviewed_by": reviewer,
}
anchor_metadata[(output_row, output_column)] = metadata
raw_rows.append(values)
accepted_row_ids.append(row_id)
if not raw_rows:
raise ValueError("PDF review excluded every extracted row")
parsed = _parse_trial_balance_rows(
case,
selected_names,
raw_rows,
document_id=str(candidate["source_document_id"]),
sheet_name="PDF",
parser_profile="reviewed-pdf-it-trial-balance-v1",
tolerance=tolerance,
anchor_metadata=anchor_metadata,
)
document = next(
(
deepcopy(item)
for item in case.get("source_documents", [])
if item.get("document_id") == candidate["source_document_id"]
),
None,
)
if document is None:
raise ValueError("PDF source document is missing from the case")
reviewed_at = _now()
review_summary = {
"status": "ACCEPTED",
"candidate_sha256": candidate["content_sha256"],
"reason": reason,
"declarations": dict(declarations),
"column_mapping": selected_mapping,
"accepted_row_count": len(accepted_row_ids),
"page_dispositions": [reviewed_pages[page] for page in sorted(reviewed_pages)],
"excluded_rows": [excluded[row_id] for row_id in sorted(excluded)],
"correction_count": len(corrected),
"reviewed_by": reviewer,
"reviewed_at": reviewed_at,
}
document.update(
{
"purpose": "TRIAL_BALANCE",
"columns": parsed["header_columns"],
"parser_profile": "reviewed-pdf-it-trial-balance-v1",
"pdf_extraction_review": review_summary,
}
)
candidate_summary = {
"schema_version": candidate["schema_version"],
"status": "ACCEPTED",
"source_document_id": candidate["source_document_id"],
"content_sha256": candidate["content_sha256"],
"row_count": candidate["row_count"],
"page_count": candidate["page_count"],
"page_methods": candidate["page_methods"],
"methods": candidate["methods"],
"ocr_used": candidate["ocr_used"],
"table_coverage": candidate["table_coverage"],
"coverage_status": "REVIEWED_COMPLETE",
"page_dispositions": review_summary["page_dispositions"],
"issues": candidate["issues"],
"review": review_summary,
}
_install_trial_balance(
case,
parsed,
document,
reviewer,
"pdf_trial_balance_reviewed",
pdf_review=review_summary,
pdf_candidate=candidate_summary,
)
_record_event(
case,
"pdf_trial_balance_reviewed",
reviewer,
{
"decision": decision,
"candidate_sha256": candidate["content_sha256"],
"accepted_row_count": len(accepted_row_ids),
"excluded_row_count": len(excluded),
"correction_count": len(corrected),
},
)
_record_event(
case,
"document_parsed",
reviewer,
{"document_id": candidate["source_document_id"]},
)
return case
def confirm_parser(
case: dict[str, Any], convention: str, actor: str, expected_revision: str
) -> dict[str, Any]:
"""Record the explicit human confirmation required before computation."""
_ensure_revision(case, expected_revision)
if not case.get("trial_balance"):
raise ValueError("No parsed trial balance is available")
selected = ParserConvention(convention)
if selected is ParserConvention.UNKNOWN:
raise ValueError("UNKNOWN cannot be confirmed")
closing_difference = Decimal(
str(case["trial_balance"]["calibration"]["closing_difference"])
)
tolerance = Decimal(str(case["trial_balance"]["calibration"]["tolerance"]))
if abs(closing_difference) > tolerance:
raise ValueError(
"Trial-balance closing debit and credit totals do not reconcile"
)
_mutate(case, actor, "parser_convention_confirmed")
case["trial_balance"]["confirmed_convention"] = selected.value
if selected is ParserConvention.TURNOVER_INCLUDES_CLOSING_ENTRIES:
appears_included: bool | None = True
closing_reason = "The professional explicitly confirmed that turnover includes closing entries."
elif selected is ParserConvention.SIGNED_BALANCE_ONLY:
appears_included = None
closing_reason = (
"The professional confirmed a signed-balance-only source, so closing-entry "
"inclusion cannot be assessed from turnover columns."
)
else:
appears_included = False
closing_reason = (
"The professional explicitly confirmed a turnover convention that excludes "
"closing entries."
)
closing_review = {
"appears_included": appears_included,
"status": "USER_CONFIRMED",
"reason": closing_reason,
"confirmed_convention": selected.value,
"confirmed_by": actor,
"confirmed_at": _now(),
}
case["trial_balance"]["closing_entries_review"] = closing_review
case["trial_balance"]["calibration"]["closing_entries_assessment"] = deepcopy(
closing_review
)
case["state"] = CaseState.MAPPING_REVIEW
case["validation"] = None
_record_event(
case,
"parser_convention_confirmed",
actor,
{
"convention": selected.value,
"closing_entries_review": closing_review,
},
)
return case
def _threshold_result(
year: Mapping[str, Any], thresholds: Mapping[str, Any], actor: str
) -> dict[str, Any]:
values = {
"assets": normalize_decimal(year["assets"]),
"revenue": normalize_decimal(year["revenue"]),
"employees": normalize_decimal(year["employees"]),
}
negative_metrics = sorted(key for key, value in values.items() if value < 0)
if negative_metrics:
raise ValueError(
"Form metrics cannot be negative: " + ", ".join(negative_metrics)
)
breached = {key: values[key] > normalize_decimal(thresholds[key]) for key in values}
raw_source_refs = year.get("source_refs", {})
if raw_source_refs is not None and not isinstance(raw_source_refs, Mapping):
raise ValueError("Form metric source_refs must be keyed by metric name")
raw_statuses = year.get("evidence_status", {})
if raw_statuses is not None and not isinstance(raw_statuses, Mapping):
raise ValueError("Form metric evidence_status must be keyed by metric name")
provenance: dict[str, Any] = {}
for key in values:
supplied_refs = (raw_source_refs or {}).get(key, [])
if isinstance(supplied_refs, (str, bytes)) or not isinstance(
supplied_refs, Sequence
):
raise ValueError(f"Form metric source refs for {key} must be a list")
source_refs = sorted(
{str(item).strip() for item in supplied_refs if str(item).strip()}
)
status = str((raw_statuses or {}).get(key, "USER_CONFIRMED")).upper()
if status not in {
EvidenceStatus.OBSERVED,
EvidenceStatus.DERIVED,
EvidenceStatus.USER_CONFIRMED,
}:
raise ValueError(f"Form metric {key} has a non-reviewable evidence status")
if not source_refs:
source_refs = [f"professional_input:{actor}:{int(year['year'])}:{key}"]
status = EvidenceStatus.USER_CONFIRMED
provenance[key] = {
"status": str(status),
"source_refs": source_refs,
"confirmed_by": actor,
}
return {
"year": int(year["year"]),
"values": {key: _decimal_text(value) for key, value in values.items()},
"breached": breached,
"breach_count": sum(breached.values()),
"within_thresholds": sum(breached.values()) <= 1,
"provenance": provenance,
}
def _prior_xbrl_reconciliation(case: Mapping[str, Any]) -> dict[str, Any]:
"""Compare current comparative facts with an attached prior filing exactly.
This is deterministic because both sides are reviewed monetary facts with
explicit periods; it does not decide whether a difference is a valid
restatement, which remains a separate professional decision.
"""
prior_xbrl = case.get("prior_xbrl")
if not prior_xbrl:
return {"status": "NOT_AVAILABLE", "checks": [], "issues": []}
matching_context_ids = set(prior_xbrl.get("matching_context_ids", []))
contexts = {
str(item["context_id"]): item for item in prior_xbrl.get("contexts", [])
}
observed: dict[str, Decimal] = {}
observed_refs: dict[str, str] = {}
invalid_unit_qnames: set[str] = set()
invalid_value_qnames: set[str] = set()
issues: list[dict[str, Any]] = []
for fact in prior_xbrl.get("facts", []):
context_ref = str(fact.get("context_ref", ""))
if (
context_ref not in matching_context_ids
or (contexts.get(context_ref) or {}).get("has_dimensions")
or fact.get("nil")
or fact.get("value") in {None, ""}
):
continue
qname = str(fact["qname"])
unit = fact.get("unit") or {}
measure = unit.get("measure") or {}
if not (
unit.get("kind") == "MEASURE"
and measure.get("namespace") == ISO4217_NS
and measure.get("local_name") == "EUR"
):
invalid_unit_qnames.add(qname)
continue
raw_value = str(fact["value"]).strip()
if not XBRL_DECIMAL_LEXICAL.fullmatch(raw_value):
invalid_value_qnames.add(qname)
continue
value = Decimal(raw_value)
if qname in observed and observed[qname] != value:
issues.append(
{
"code": "PRIOR_XBRL_DUPLICATE_CONFLICT",
"xbrl_concept": qname,
}
)
continue
observed[qname] = value
observed_refs[qname] = str(
(fact.get("source_anchor") or {}).get("source_ref", "")
)
expected: dict[str, Decimal] = {}
def add_expected(qname: str, value: Decimal) -> None:
if qname in expected and expected[qname] != value:
issues.append(
{
"code": "CURRENT_COMPARATIVE_DUPLICATE_CONFLICT",
"xbrl_concept": qname,
}
)
return
expected[qname] = value
precision = int(case.get("reporting_precision", 0))
for fact in case.get("canonical_facts", []):
qname = fact.get("xbrl_concept")
if qname and fact.get("prior_value") is not None:
add_expected(
str(qname),
_reported_decimal(
Decimal(str(fact["prior_value"]))
* Decimal(str(fact["xbrl_sign_multiplier"])),
precision,
),
)
presentation = case.get("statutory_presentation") or {}
for fact in presentation.get("output_facts", []):
if fact.get("prior_value") is not None:
add_expected(
str(fact["xbrl_concept"]),
_reported_decimal(Decimal(str(fact["prior_value"])), precision),
)
for fact in case.get("taxonomy_facts", []):
if (
fact.get("fact_type") == "MONETARY"
and not fact.get("dimensions")
and str(fact.get("period", "")).startswith("prior_")
):
add_expected(
str(fact["xbrl_concept"]),
_reported_decimal(Decimal(str(fact["value"])), precision),
)
inventory = presentation.get("inventory") or {}
relevant = {
str(item["xbrl_concept"])
for item in [
*inventory.get("requirements", []),
*inventory.get("totals", []),
]
} | set(expected)
for qname in sorted(invalid_unit_qnames & relevant):
issues.append(
{
"code": "PRIOR_XBRL_MONETARY_UNIT_INVALID",
"xbrl_concept": qname,
}
)
for qname in sorted(invalid_value_qnames & relevant):
issues.append(
{
"code": "PRIOR_XBRL_MONETARY_VALUE_INVALID",
"xbrl_concept": qname,
}
)
checks: list[dict[str, Any]] = []
decisions = {
str(item["xbrl_concept"]): item
for item in case.get("comparative_reconciliation_decisions", [])
}
for qname in sorted(relevant & (set(expected) | set(observed))):
expected_value = expected.get(qname)
observed_value = observed.get(qname)
check = {
"xbrl_concept": qname,
"current_comparative": (
None if expected_value is None else _decimal_text(expected_value)
),
"prior_filed": (
None if observed_value is None else _decimal_text(observed_value)
),
"prior_source_ref": observed_refs.get(qname),
}
if expected_value == observed_value:
check["status"] = "PASS"
elif expected_value == 0 and observed_value is None:
check["status"] = "PASS_OMITTED_ZERO"
else:
decision = decisions.get(qname)
if (
decision
and decision.get("action") == "RESTATEMENT_CONFIRMED"
and decision.get("current_comparative") == check["current_comparative"]
and decision.get("prior_filed") == check["prior_filed"]
):
check["status"] = "RESTATEMENT_CONFIRMED"
check["decision_id"] = decision["decision_id"]
else:
check["status"] = "FAIL"
issues.append(
{
"code": "PRIOR_XBRL_COMPARATIVE_MISMATCH",
**{
key: value
for key, value in check.items()
if key != "status"
},
}
)
checks.append(check)
return {
"status": "PASS" if not issues else "FAIL",
"checks": checks,
"issues": issues,
}
def record_comparative_reconciliation_decisions(
case: dict[str, Any],
decisions: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record evidence-backed professional restatements of prior filed amounts."""
_ensure_revision(case, expected_revision)
if not case.get("prior_xbrl"):
raise ValueError("Comparative decisions require an attached prior XBRL")
probe = deepcopy(case)
probe["comparative_reconciliation_decisions"] = []
reconciliation = _prior_xbrl_reconciliation(probe)
failed = {
str(item["xbrl_concept"]): item
for item in reconciliation["checks"]
if item["status"] == "FAIL"
}
if not decisions:
raise ValueError("At least one comparative reconciliation decision is required")
normalized: list[dict[str, Any]] = []
seen: set[str] = set()
for raw in decisions:
if set(raw) != {"xbrl_concept", "action", "reason", "source_refs"}:
raise ValueError(
"Comparative decisions require concept, action, reason, and source_refs"
)
qname = str(raw["xbrl_concept"])
action = str(raw["action"]).upper()
reason = str(raw["reason"]).strip()
raw_source_refs = raw["source_refs"]
if not isinstance(raw_source_refs, list):
raise ValueError("Comparative decision source_refs must be a list")
source_refs = sorted(
{str(item) for item in raw_source_refs if str(item).strip()}
)
if qname in seen or qname not in failed:
raise ValueError(f"Unknown or duplicate comparative difference: {qname}")
if action != "RESTATEMENT_CONFIRMED" or not reason or not source_refs:
raise ValueError(
"Comparative restatements require explicit evidence and a reason"
)
if not set(source_refs) <= _available_evidence_refs(case):
raise ValueError(
"Comparative restatement references evidence outside the case"
)
seen.add(qname)
check = failed[qname]
normalized.append(
{
"decision_id": f"comparative_{len(normalized) + 1:06d}",
"xbrl_concept": qname,
"action": action,
"current_comparative": check["current_comparative"],
"prior_filed": check["prior_filed"],
"reason": reason,
"source_refs": source_refs,
"reviewed_by": actor,
"reviewed_at": _now(),
}
)
retained = [
item
for item in case.get("comparative_reconciliation_decisions", [])
if item["xbrl_concept"] not in seen
]
_mutate(case, actor, "comparative_reconciliation_decisions_recorded")
case["comparative_reconciliation_decisions"] = [*retained, *normalized]
case["validation"] = None
_record_event(
case,
"comparative_reconciliation_decisions_recorded",
actor,
{"xbrl_concepts": sorted(seen)},
)
return case
def determine_forms(
case: dict[str, Any],
metrics: Sequence[Mapping[str, Any]],
rule_pack: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Calculate effective-dated form eligibility without selecting the form."""
_ensure_revision(case, expected_revision)
if case["state"] == CaseState.UNSUPPORTED:
raise ValueError("Unsupported cases have no eligible MVP form")
if (
str(rule_pack.get("id"))
!= str(case["rule_pack_versions"]["statutory_rule_pack"])
or _sha256_bytes(_canonical_json(rule_pack)) != case["rule_pack_checksum"]
):
raise ValueError(
"Statutory rule pack differs from the locked case; use explicit migration"
)
period_start = date.fromisoformat(case["period"]["start"])
_validate_effective_rule_pack(rule_pack, "STATUTORY_FORM_RULES", period_start)
first_year = bool(case["entity"].get("first_financial_year"))
required_years = 1 if first_year else 2
if isinstance(metrics, (str, bytes)) or not isinstance(metrics, Sequence):
raise ValueError("Form metrics must be a list")
if any(not isinstance(item, Mapping) or "year" not in item for item in metrics):
raise ValueError("Every form metric record requires a reporting year")
metric_years = [int(item["year"]) for item in metrics]
if len(metric_years) != len(set(metric_years)):
raise ValueError("Form metrics require one unique record per reporting year")
reporting_year = date.fromisoformat(case["period"]["end"]).year
expected_years = [reporting_year]
if not first_year:
expected_years.append(reporting_year - 1)
unexpected_years = sorted(set(metric_years) - set(expected_years))
if unexpected_years:
raise ValueError(
"Form metric years do not align with the reporting period: "
+ ", ".join(str(year) for year in unexpected_years)
)
metrics_by_year = {int(item["year"]): item for item in metrics}
missing: list[str] = []
missing_years = [year for year in expected_years if year not in metrics_by_year]
missing.extend(f"threshold_metrics_for_{year}" for year in missing_years)
ordered: list[Mapping[str, Any]] = []
for year in expected_years:
item = metrics_by_year.get(year)
if item is None:
continue
missing_metrics = [
key
for key in ("assets", "revenue", "employees")
if item.get(key) is None or not str(item.get(key)).strip()
]
missing.extend(f"threshold_{key}_for_{year}" for key in missing_metrics)
if not missing_metrics:
ordered.append(item)
prior_form = str(case["entity"].get("prior_year_form") or "").upper() or None
detail_rank = {"MICRO": 0, "ABBREVIATED": 1, "ORDINARY": 2}
form_components = rule_pack.get("form_components")
transition_consequences = rule_pack.get("form_change_consequences")
if not isinstance(form_components, Mapping) or set(form_components) != set(
detail_rank
):
raise ValueError("Statutory rule pack requires components for every form")
if not isinstance(transition_consequences, Mapping):
raise ValueError("Statutory rule pack requires form-change consequences")
analysis: dict[str, Any] = {}
for form in ("ABBREVIATED", "MICRO"):
thresholds = rule_pack["forms"][form]
years = [
_threshold_result(item, thresholds, actor)
for item in ordered[:required_years]
]
complete = len(years) == required_years
within = [bool(item["within_thresholds"]) for item in years]
continuing_or_moving_to_more_detail = bool(
prior_form and detail_rank[prior_form] <= detail_rank[form]
)
if first_year:
eligible_for_thresholds = complete and within == [True]
eligibility_basis = "FIRST_FINANCIAL_YEAR_CURRENT_THRESHOLDS"
elif continuing_or_moving_to_more_detail:
# Articles 2435-bis/ter require two consecutive exceedances before
# an entity already in the simplified regime must leave it.
eligible_for_thresholds = complete and not all(not item for item in within)
eligibility_basis = "CONTINUATION_UNTIL_TWO_CONSECUTIVE_EXCEEDANCES"
else:
# Entry from a more detailed form requires two consecutive years
# within the selected simplified-form thresholds.
eligible_for_thresholds = complete and all(within)
eligibility_basis = "ENTRY_REQUIRES_TWO_CONSECUTIVE_YEARS_WITHIN"
reasons: list[str] = []
if complete and not eligible_for_thresholds:
reasons.append(
"CURRENT_YEAR_THRESHOLDS_NOT_MET"
if first_year
else (
"TWO_CONSECUTIVE_THRESHOLD_EXCEEDANCES"
if continuing_or_moving_to_more_detail
else "TWO_CONSECUTIVE_YEARS_WITHIN_NOT_MET"
)
)
analysis[form] = {
"thresholds": {key: str(value) for key, value in thresholds.items()},
"years": years,
"eligible": eligible_for_thresholds,
"eligibility_basis": eligibility_basis,
"continuation_rule_applied": continuing_or_moving_to_more_detail,
"reasons": reasons,
}
micro_exclusions = list(case["entity"].get("micro_exclusion_flags", []))
if micro_exclusions:
analysis["MICRO"]["eligible"] = False
analysis["MICRO"]["reasons"].extend(micro_exclusions)
eligible = ["ORDINARY"]
if analysis["ABBREVIATED"]["eligible"]:
eligible.insert(0, "ABBREVIATED")
if analysis["MICRO"]["eligible"]:
eligible.insert(0, "MICRO")
recommended = eligible[0] if not missing else None
consequences: dict[str, Any] = {}
for form in ("MICRO", "ABBREVIATED", "ORDINARY"):
transition_key = f"{prior_form}_TO_{form}" if prior_form else None
consequences[form] = {
"change_type": (
"FIRST_YEAR_SELECTION"
if prior_form is None
else (
"UNCHANGED"
if prior_form == form
else (
"MORE_DETAILED"
if detail_rank[form] > detail_rank[prior_form]
else "LESS_DETAILED"
)
)
),
"required_components": list(form_components[form]),
"effects": (
[]
if transition_key is None or prior_form == form
else list(transition_consequences.get(transition_key, []))
),
"eligible": form in eligible and not missing,
}
_mutate(case, actor, "form_eligibility_calculated")
_clear_accounting_dependent_reviews(case)
case["form_analysis"] = {
"rule_pack": rule_pack["id"],
"eligible_forms": eligible if not missing else [],
"ineligible_forms": [
form for form in ("MICRO", "ABBREVIATED") if form not in eligible
],
"recommended_form": recommended,
"least_burdensome_only": True,
"prior_year_form": prior_form,
"first_year": first_year,
"consequences_of_changing_form": consequences,
"missing_fields": missing,
"calculations": analysis,
"computation_context": _computation_context(
case, "statutory-form-eligibility-v1"
),
}
case["selected_form"] = None
case["mappings"] = []
case["mapping_candidates"] = []
case["taxonomy_mapping_index"] = None
case["canonical_facts"] = []
case["statements"] = None
case["statutory_presentation"] = None
case["schedules"] = []
case["validation"] = None
_record_event(case, "form_eligibility_calculated", actor)
return case
def select_form(
case: dict[str, Any], selected_form: str, actor: str, expected_revision: str
) -> dict[str, Any]:
"""Record the user's form choice after deterministic eligibility analysis."""
_ensure_revision(case, expected_revision)
analysis = case.get("form_analysis") or {}
selected = selected_form.upper()
if selected not in analysis.get("eligible_forms", []):
raise ValueError("Selected form is not eligible under the locked rule pack")
_mutate(case, actor, "form_selected")
_clear_accounting_dependent_reviews(case)
case["selected_form"] = selected
case["mappings"] = []
case["mapping_candidates"] = []
case["taxonomy_mapping_index"] = None
case["canonical_facts"] = []
case["statements"] = None
case["statutory_presentation"] = None
case["schedules"] = []
case["state"] = CaseState.MAPPING_REVIEW
case["validation"] = None
_record_event(case, "form_selected", actor, {"form": selected})
return case
def generate_mapping_candidates(
case: dict[str, Any],
memory_path: Path,
source_system_template: str,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Attach exact tenant-isolated approved mapping candidates for review."""
_ensure_revision(case, expected_revision)
trial_balance = case.get("trial_balance")
if not trial_balance or not trial_balance.get("confirmed_convention"):
raise ValueError(
"Parser convention must be confirmed before mapping candidates"
)
if not case.get("selected_form"):
raise ValueError("A statutory form must be selected before mapping candidates")
if case.get("statutory_presentation_required", True) and not case.get(
"taxonomy_mapping_index"
):
raise ValueError(
"Official taxonomy mapping index must be recorded before candidates"
)
candidates = mapping_candidates(case, memory_path, source_system_template)
allowed_concepts = {
str(item["xbrl_concept"])
for item in (case.get("taxonomy_mapping_index") or {}).get("concepts", [])
if item.get("mapping_allowed") is True
}
if allowed_concepts:
candidates = [
candidate
for candidate in candidates
if all(
not allocation.get("xbrl_concept")
or str(allocation["xbrl_concept"]) in allowed_concepts
for allocation in candidate.get("allocations", [])
)
]
_mutate(case, actor, "mapping_candidates_generated")
case["mapping_candidates"] = candidates
case["mapping_candidate_context"] = {
"source_system_template": source_system_template,
"memory_file_name": memory_path.name,
"generated_at": _now(),
"computation_context": _computation_context(
case, "mapping-candidate-memory-v1"
),
}
case["validation"] = None
_record_event(
case,
"mapping_suggested",
actor,
{
"candidate_count": len(candidates),
"account_ids_sample": [item["account_id"] for item in candidates[:50]],
"account_ids_sha256": _sha256_bytes(
_canonical_json([item["account_id"] for item in candidates])
),
"suggestion_source": "APPROVED_MAPPING_MEMORY",
},
)
_record_event(
case,
"mapping_candidates_generated",
actor,
{"candidate_count": len(candidates)},
)
return case
def load_client_history(
case: dict[str, Any],
history_path: Path,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Load prior approved client knowledge as unconfirmed suggestions only."""
_ensure_revision(case, expected_revision)
suggestions = client_history_suggestions(case, history_path)
_mutate(case, actor, "client_history_loaded")
case["client_history_suggestions"] = suggestions
case["validation"] = None
_record_event(
case,
"client_history_loaded",
actor,
{
"prior_period_end": suggestions["prior_period_end"],
"answer_suggestion_count": len(suggestions["answer_suggestions"]),
"narrative_suggestion_count": len(suggestions["narrative_suggestions"]),
},
)
return case
def apply_mapping_decisions(
case: dict[str, Any],
decisions: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Apply reviewed, balancing account mappings and explicit exclusions."""
_ensure_revision(case, expected_revision)
trial_balance = case.get("trial_balance")
if not trial_balance or not trial_balance.get("confirmed_convention"):
raise ValueError("Parser convention must be confirmed before mapping")
if not case.get("selected_form"):
raise ValueError("A statutory form must be selected before mapping")
if not decisions:
raise ValueError("At least one mapping decision is required")
taxonomy_mapping_index = case.get("taxonomy_mapping_index")
if case.get("statutory_presentation_required", True) and not taxonomy_mapping_index:
raise ValueError(
"Official taxonomy mapping index must be recorded before mapping"
)
indexed_concepts = {
str(item["xbrl_concept"])
for item in (taxonomy_mapping_index or {}).get("concepts", [])
if item.get("mapping_allowed") is True
}
accounts = {row["account_id"]: row for row in trial_balance["entries"]}
comparative_required = case["entity"].get("first_financial_year") is not True
prior_mappings = {
str(item["account_id"]): item for item in case.get("mappings", [])
}
seen: set[str] = set()
normalized: list[dict[str, Any]] = []
for decision in decisions:
account_id = str(decision["account_id"])
if account_id in seen or account_id not in accounts:
raise ValueError(f"Unknown or duplicate mapping account: {account_id}")
seen.add(account_id)
account = accounts[account_id]
status = str(decision["decision"]).upper()
memory_scope = str(decision.get("memory_scope", "CLIENT")).upper()
if memory_scope not in {"CLIENT", "TENANT"}:
raise ValueError("Mapping memory scope must be CLIENT or TENANT")
if (
memory_scope == "TENANT"
and decision.get("tenant_reuse_approved") is not True
):
raise ValueError("Tenant-wide mapping reuse requires explicit approval")
if status == "EXCLUDED":
if not str(decision.get("reason", "")).strip():
raise ValueError(f"Excluded account {account_id} requires a reason")
current_balance = Decimal(str(account["closing_signed"]))
prior_balance = account.get("prior_closing_signed")
if current_balance != 0 or (
comparative_required
and (prior_balance is None or Decimal(str(prior_balance)) != 0)
):
raise ValueError(
f"Non-zero account {account_id} cannot be excluded from the statements"
)
allocations: list[dict[str, Any]] = []
elif status == "ACCEPTED":
allocations = []
raw_allocations = list(decision.get("allocations", []))
if not raw_allocations:
raise ValueError(f"Accepted account {account_id} requires allocations")
current_total = Decimal("0")
prior_total = Decimal("0")
for position, allocation in enumerate(raw_allocations, start=1):
evidence_status = EvidenceStatus(
str(allocation["evidence_status"]).upper()
)
if evidence_status not in {
EvidenceStatus.OBSERVED,
EvidenceStatus.USER_CONFIRMED,
}:
raise ValueError(
"Reviewed mapping allocations must be observed or user-confirmed"
)
current = normalize_decimal(allocation["current_amount"])
raw_prior = allocation.get("prior_amount")
if comparative_required:
prior = normalize_decimal(raw_prior)
else:
if raw_prior not in {None, ""}:
raise ValueError(
"First-financial-year mappings cannot contain comparative amounts"
)
prior = None
current_total += current
if prior is not None:
prior_total += prior
qname = allocation.get("xbrl_concept")
if qname is not None and not SAFE_QNAME.fullmatch(str(qname)):
raise ValueError(f"Invalid XBRL QName: {qname}")
if (
case.get("statutory_presentation_required", True)
and str(qname or "") not in indexed_concepts
):
raise ValueError(
f"Mapping concept is outside the selected-form primary network: {qname}"
)
xbrl_sign_multiplier = str(allocation.get("xbrl_sign_multiplier", ""))
if qname is not None and xbrl_sign_multiplier not in {"1", "-1"}:
raise ValueError(
"Mapped XBRL concepts require an explicit sign multiplier"
)
if qname is None:
xbrl_sign_multiplier = "1"
schedule_triggers = sorted(
{
str(value).upper()
for value in allocation.get("schedule_triggers", [])
}
)
unknown_triggers = set(schedule_triggers) - SCHEDULE_TYPES
if unknown_triggers:
raise ValueError(
f"Unknown schedule triggers: {sorted(unknown_triggers)}"
)
allocations.append(
{
"allocation_id": f"{account_id}_{position}",
"canonical_line": str(allocation["canonical_line"]),
"statement_section": str(
allocation["statement_section"]
).upper(),
"xbrl_concept": qname,
"xbrl_sign_multiplier": xbrl_sign_multiplier,
"current_amount": _decimal_text(current),
"prior_amount": (
None if prior is None else _decimal_text(prior)
),
"evidence_status": evidence_status.value,
"review_reason": str(allocation.get("review_reason", "")),
"schedule_triggers": schedule_triggers,
"source_refs": list(account["source_refs"]),
}
)
if current_total != Decimal(account["closing_signed"]):
raise ValueError(f"Current split for {account_id} does not balance")
if comparative_required and prior_total != Decimal(
account["prior_closing_signed"]
):
raise ValueError(f"Prior split for {account_id} does not balance")
else:
raise ValueError(f"Unsupported mapping decision: {status}")
normalized.append(
{
"account_id": account_id,
"decision": status,
"reason": str(decision.get("reason", "")),
"candidate_source": str(decision.get("candidate_source", "MANUAL")),
"memory_scope": memory_scope,
"approved_by": actor,
"approved_at": _now(),
"allocations": allocations,
}
)
replacements = {str(item["account_id"]): item for item in normalized}
merged = [
replacements.get(
str(row["account_id"]), prior_mappings.get(str(row["account_id"]))
)
for row in trial_balance["entries"]
if str(row["account_id"]) in replacements
or str(row["account_id"]) in prior_mappings
]
_mutate(case, actor, "mapping_decisions_applied")
_clear_accounting_dependent_reviews(case)
case["mappings"] = merged
case["adjustments"] = []
case["canonical_facts"] = []
case["statements"] = None
case["statutory_presentation"] = None
case["schedules"] = []
case["validation"] = None
case["state"] = CaseState.MAPPING_REVIEW
mapping_after_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
for mapping in normalized:
previous = prior_mappings.get(str(mapping["account_id"]))
previous_signature = (
{
key: previous.get(key)
for key in (
"decision",
"reason",
"candidate_source",
"memory_scope",
"allocations",
)
}
if previous
else None
)
current_signature = {
key: mapping.get(key)
for key in (
"decision",
"reason",
"candidate_source",
"memory_scope",
"allocations",
)
}
if previous_signature is not None and previous_signature != current_signature:
_record_event(
case,
"mapping_changed",
actor,
{
"account_id": mapping["account_id"],
"previous": _mapping_audit_summary(previous_signature),
"current": _mapping_audit_summary(current_signature),
},
after_hash=mapping_after_hash,
)
_record_event(
case,
(
"mapping_accepted"
if mapping["decision"] == "ACCEPTED"
else "mapping_excluded"
),
actor,
{"account_id": mapping["account_id"]},
after_hash=mapping_after_hash,
)
if len(mapping["allocations"]) > 1:
_record_event(
case,
"account_split_created",
actor,
{
"account_id": mapping["account_id"],
"allocation_ids": [
item["allocation_id"] for item in mapping["allocations"]
],
},
after_hash=mapping_after_hash,
)
return case
def record_adjustments(
case: dict[str, Any],
adjustments: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record explicit, balancing presentation reclassifications."""
_ensure_revision(case, expected_revision)
if not case.get("mappings"):
raise ValueError("Reviewed mappings are required before adjustments")
normalized: list[dict[str, Any]] = []
comparative_required = case["entity"].get("first_financial_year") is not True
seen: set[str] = set()
for raw in adjustments:
adjustment_id = str(raw["adjustment_id"])
if not SAFE_ID.fullmatch(adjustment_id) or adjustment_id in seen:
raise ValueError("Adjustment identifiers must be unique safe IDs")
seen.add(adjustment_id)
reason = str(raw.get("reason", "")).strip()
if not reason:
raise ValueError("Every presentation adjustment requires a reason")
raw_lines = list(raw.get("lines", []))
if len(raw_lines) < 2:
raise ValueError("A presentation adjustment requires at least two lines")
lines: list[dict[str, Any]] = []
current_total = Decimal("0")
prior_total = Decimal("0")
for position, raw_line in enumerate(raw_lines, start=1):
current = normalize_decimal(raw_line["current_amount"])
raw_prior = raw_line.get("prior_amount")
if comparative_required:
prior = normalize_decimal(raw_prior)
else:
if raw_prior not in {None, ""}:
raise ValueError(
"First-financial-year adjustments cannot contain comparative amounts"
)
prior = None
current_total += current
if prior is not None:
prior_total += prior
qname = raw_line.get("xbrl_concept")
if qname is not None and not SAFE_QNAME.fullmatch(str(qname)):
raise ValueError(f"Invalid XBRL QName: {qname}")
xbrl_sign_multiplier = str(raw_line.get("xbrl_sign_multiplier", ""))
if qname is not None and xbrl_sign_multiplier not in {"1", "-1"}:
raise ValueError(
"Adjusted XBRL concepts require an explicit sign multiplier"
)
if qname is None:
xbrl_sign_multiplier = "1"
lines.append(
{
"line_id": f"{adjustment_id}_{position}",
"canonical_line": str(raw_line["canonical_line"]),
"statement_section": str(raw_line["statement_section"]).upper(),
"xbrl_concept": qname,
"xbrl_sign_multiplier": xbrl_sign_multiplier,
"current_amount": _decimal_text(current),
"prior_amount": None if prior is None else _decimal_text(prior),
"source_refs": sorted(
{str(item) for item in raw_line.get("source_refs", [])}
),
}
)
if lines[-1]["source_refs"] and not set(
lines[-1]["source_refs"]
) <= _available_evidence_refs(case):
raise ValueError("Adjustment lines reference evidence outside the case")
if current_total or (comparative_required and prior_total):
raise ValueError(
"Presentation adjustment lines must balance in both periods"
)
normalized.append(
{
"adjustment_id": adjustment_id,
"kind": "PRESENTATION_RECLASSIFICATION",
"reason": reason,
"lines": lines,
"approved_by": actor,
"approved_at": _now(),
}
)
_mutate(case, actor, "adjustments_recorded")
_clear_accounting_dependent_reviews(case)
case["adjustments"] = normalized
case["canonical_facts"] = []
case["statements"] = None
case["statutory_presentation"] = None
case["schedules"] = []
case["validation"] = None
case["state"] = CaseState.MAPPING_REVIEW
_record_event(
case,
"adjustments_recorded",
actor,
{"adjustment_ids": [item["adjustment_id"] for item in normalized]},
)
return case
def record_taxonomy_facts(
case: dict[str, Any],
facts: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record reviewed text, nil, or dimensional facts for the taxonomy adapter."""
_ensure_revision(case, expected_revision)
if not case.get("selected_form") or not case.get("statements"):
raise ValueError("Statements and form are required before taxonomy facts")
normalized: list[dict[str, Any]] = []
seen_ids: set[str] = set()
seen_keys: set[tuple[str, str, tuple[tuple[str, str], ...]]] = set()
allowed_periods = {
"current_instant",
"current_duration",
}
if case["entity"].get("first_financial_year") is not True:
allowed_periods.update({"prior_instant", "prior_duration"})
for raw in facts:
fact_id = str(raw["fact_id"])
qname = str(raw["xbrl_concept"])
period = str(raw["period"])
fact_type = str(raw["fact_type"]).upper()
status = EvidenceStatus(str(raw["status"]).upper())
if not SAFE_ID.fullmatch(fact_id) or fact_id in seen_ids:
raise ValueError("Taxonomy fact IDs must be unique safe IDs")
seen_ids.add(fact_id)
if not SAFE_QNAME.fullmatch(qname):
raise ValueError(f"Invalid taxonomy fact QName: {qname}")
if period not in allowed_periods:
raise ValueError(f"Unsupported taxonomy fact period: {period}")
if fact_type not in {"MONETARY", "TEXT", "NIL"}:
raise ValueError(f"Unsupported taxonomy fact type: {fact_type}")
if status not in {
EvidenceStatus.OBSERVED,
EvidenceStatus.DERIVED,
EvidenceStatus.USER_CONFIRMED,
}:
raise ValueError("Taxonomy facts must be reviewed and exportable")
dimensions = {
str(axis): str(member)
for axis, member in sorted(dict(raw.get("dimensions", {})).items())
}
if any(
not SAFE_QNAME.fullmatch(axis) or not SAFE_QNAME.fullmatch(member)
for axis, member in dimensions.items()
):
raise ValueError("Dimension axes and members must be valid QNames")
duplicate_key = (qname, period, tuple(dimensions.items()))
if duplicate_key in seen_keys:
raise ValueError("Conflicting duplicate taxonomy fact")
seen_keys.add(duplicate_key)
source_refs = sorted({str(item) for item in raw.get("source_refs", [])})
derivation = raw.get("derivation")
value: str | None
language: str | None = None
nil_reason: str | None = None
if fact_type == "MONETARY":
if str(raw.get("currency", "EUR")).upper() != "EUR":
raise ValueError("MVP taxonomy monetary facts must use EUR")
value = _decimal_text(normalize_decimal(raw["value"]))
elif fact_type == "TEXT":
value = str(raw.get("value", "")).strip()
if not value:
raise ValueError("Text taxonomy facts require a value")
language = str(
raw.get("language", case.get("output_language", "it"))
).lower()
if language != case.get("output_language", "it"):
raise ValueError(
"Text taxonomy facts cannot mix languages in one output"
)
else:
value = None
nil_reason = str(raw.get("nil_reason", "")).strip()
if raw.get("nil_allowed_by_review") is not True or not nil_reason:
raise ValueError("Nil facts require an explicit reviewed reason")
if fact_type != "NIL" and not source_refs and not derivation:
raise ValueError("Taxonomy facts require source refs or a derivation")
if source_refs and not set(source_refs) <= _available_evidence_refs(case):
raise ValueError("Taxonomy fact references evidence outside the case")
normalized.append(
{
"fact_id": fact_id,
"xbrl_concept": qname,
"period": period,
"fact_type": fact_type,
"value": value,
"currency": "EUR" if fact_type == "MONETARY" else None,
"language": language if fact_type == "TEXT" else None,
"status": status.value,
"source_refs": source_refs,
"derivation": derivation,
"dimensions": dimensions,
"nil_reason": nil_reason,
"confirmed_by": actor,
"confirmed_at": _now(),
}
)
_mutate(case, actor, "taxonomy_facts_recorded")
_clear_schedule_taxonomy_adapter(case)
_clear_narrative_reviews(case)
case["taxonomy_facts"] = normalized
case["taxonomy_fact_context"] = _computation_context(
case, "taxonomy-fact-adapter-v1"
)
case["statutory_presentation"] = None
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"taxonomy_facts_recorded",
actor,
{"fact_ids": [item["fact_id"] for item in normalized]},
)
return case
def record_schedule_taxonomy_adapter(
case: dict[str, Any],
catalogue_path: Path,
rule_pack: Mapping[str, Any],
decisions: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Compile professionally reviewed schedules into official note-table facts."""
_ensure_revision(case, expected_revision)
period_start = date.fromisoformat(case["period"]["start"])
if not (
date.fromisoformat(str(rule_pack["effective_from"]))
<= period_start
<= date.fromisoformat(str(rule_pack["effective_to"]))
):
raise ValueError("Schedule taxonomy rule pack is not effective for the case")
catalogue = _taxonomy_catalogue(
catalogue_path,
case["rule_pack_versions"]["taxonomy_id"],
case.get("taxonomy_checksum"),
)
adapter = compile_schedule_taxonomy_adapter(
case, catalogue, rule_pack, decisions, actor
)
checksum = _sha256_bytes(_canonical_json(rule_pack))
_mutate(case, actor, "schedule_taxonomy_adapter_recorded")
_clear_narrative_reviews(case)
recorded_at = _now()
for fact in adapter["generated_facts"]:
fact["confirmed_at"] = recorded_at
adapter["recorded_at"] = recorded_at
adapter["computation_context"] = _computation_context(case, str(rule_pack["id"]))
case["rule_pack_versions"]["schedule_taxonomy_adapter"] = str(rule_pack["id"])
case["schedule_taxonomy_adapter_rule_pack_checksum"] = checksum
case["schedule_taxonomy_adapter"] = adapter
case["schedule_taxonomy_facts"] = list(adapter["generated_facts"])
case["validation"] = None
_record_event(
case,
"schedule_taxonomy_adapter_recorded",
actor,
{
"rule_pack_id": str(rule_pack["id"]),
"inventory_sha256": adapter["inventory"]["inventory_sha256"],
"adapter_sha256": adapter["adapter_sha256"],
"schedule_count": len(adapter["coverage"]),
"generated_fact_count": len(adapter["generated_facts"]),
},
)
return case
def record_taxonomy_mapping_index(
case: dict[str, Any],
catalogue_path: Path,
rule_pack: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Expose the selected form's official concepts to semantic mapping review."""
_ensure_revision(case, expected_revision)
if not case.get("selected_form"):
raise ValueError("A statutory form must be selected before taxonomy indexing")
period_start = date.fromisoformat(case["period"]["start"])
if not (
date.fromisoformat(str(rule_pack["effective_from"]))
<= period_start
<= date.fromisoformat(str(rule_pack["effective_to"]))
):
raise ValueError(
"Statutory presentation rule pack is not effective for the case period"
)
presentation_checksum = _sha256_bytes(_canonical_json(rule_pack))
locked_presentation_id = case["rule_pack_versions"].get(
"statutory_presentation_rule_pack"
)
locked_presentation_checksum = case.get("statutory_presentation_rule_pack_checksum")
if locked_presentation_id is not None and (
str(locked_presentation_id) != str(rule_pack.get("id"))
or locked_presentation_checksum != presentation_checksum
):
raise ValueError(
"Statutory presentation rule pack differs from the locked case"
)
catalogue = _taxonomy_catalogue(
catalogue_path,
case["rule_pack_versions"]["taxonomy_id"],
case.get("taxonomy_checksum"),
)
inventory = build_primary_presentation_inventory(
catalogue, rule_pack, str(case["selected_form"])
)
schedule_lookup: dict[str, list[str]] = {}
for schedule_type, concepts in inventory["schedule_trigger_concepts"].items():
for qname in concepts:
schedule_lookup.setdefault(str(qname), []).append(str(schedule_type))
concepts = []
for kind, items in (
("LEAF", inventory["requirements"]),
("TOTAL", inventory["totals"]),
):
for item in items:
qname = str(item["xbrl_concept"])
concepts.append(
{
"xbrl_concept": qname,
"label_it": item["label_it"],
"period_type": item["period_type"],
"balance": item["balance"],
"role_kinds": list(item["role_kinds"]),
"presentation_kind": kind,
"mapping_allowed": kind == "LEAF",
"schedule_types": sorted(schedule_lookup.get(qname, [])),
}
)
_mutate(case, actor, "taxonomy_mapping_index_recorded")
case["rule_pack_versions"]["statutory_presentation_rule_pack"] = str(
rule_pack["id"]
)
case["statutory_presentation_rule_pack_checksum"] = presentation_checksum
case["taxonomy_mapping_index"] = {
"selected_form": case["selected_form"],
"inventory_sha256": inventory["inventory_sha256"],
"taxonomy_package_sha256": inventory["taxonomy_package_sha256"],
"concepts": sorted(concepts, key=lambda item: item["xbrl_concept"]),
"recorded_by": actor,
"recorded_at": _now(),
"computation_context": _computation_context(
case, f"{rule_pack['id']}-mapping-index"
),
}
case["mapping_candidates"] = []
case["validation"] = None
_record_event(
case,
"taxonomy_mapping_index_recorded",
actor,
{
"selected_form": case["selected_form"],
"concept_count": len(concepts),
"inventory_sha256": inventory["inventory_sha256"],
},
)
return case
def record_statutory_presentation(
case: dict[str, Any],
catalogue_path: Path,
rule_pack: Mapping[str, Any],
decisions: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record complete primary-statement coverage against official networks."""
_ensure_revision(case, expected_revision)
if not case.get("taxonomy_mapping_index"):
raise ValueError(
"The selected-form taxonomy mapping index is required before presentation review"
)
if case["rule_pack_versions"].get("statutory_presentation_rule_pack") != str(
rule_pack.get("id")
) or case.get("statutory_presentation_rule_pack_checksum") != _sha256_bytes(
_canonical_json(rule_pack)
):
raise ValueError(
"Statutory presentation rule pack differs from the locked mapping index"
)
catalogue = _taxonomy_catalogue(
catalogue_path,
case["rule_pack_versions"]["taxonomy_id"],
case.get("taxonomy_checksum"),
)
coverage = build_statutory_presentation_coverage(
case, catalogue, rule_pack, decisions, actor
)
if (
coverage["inventory"]["inventory_sha256"]
!= case["taxonomy_mapping_index"]["inventory_sha256"]
):
raise ValueError("Statutory presentation inventory differs from mapping review")
_mutate(case, actor, "statutory_presentation_recorded")
_clear_schedule_taxonomy_adapter(case)
coverage["recorded_by"] = actor
coverage["recorded_at"] = _now()
coverage["computation_context"] = _computation_context(case, str(rule_pack["id"]))
case["statutory_presentation"] = coverage
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"statutory_presentation_recorded",
actor,
{
"status": coverage["status"],
"inventory_sha256": coverage["inventory"]["inventory_sha256"],
**coverage["summary"],
},
)
return case
def record_taxonomy_representation(
case: dict[str, Any],
payload: Mapping[str, Any],
reviewer: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record the professional treatment of substantive taxonomy differences."""
_ensure_revision(case, expected_revision)
mismatch_present = payload.get("mismatch_present") is True
allowed_treatments = {
"NO_SUBSTANTIVE_MISMATCH",
"XBRL_REPRESENTATION_ACCEPTED_WITH_DISCLOSED_DIFFERENCES",
"ACCOUNTS_REVISED_TO_TAXONOMY",
"DOUBLE_FORMAT_ROUTE_REFERRED_FOR_PROFESSIONAL_FILING",
}
treatment = str(payload.get("chosen_treatment", "")).upper()
if treatment not in allowed_treatments:
raise ValueError("A supported taxonomy representation treatment is required")
if mismatch_present and treatment == "NO_SUBSTANTIVE_MISMATCH":
raise ValueError("A substantive mismatch requires a professional treatment")
if not mismatch_present and treatment != "NO_SUBSTANTIVE_MISMATCH":
raise ValueError("No-mismatch cases must use NO_SUBSTANTIVE_MISMATCH")
reason = str(payload.get("reviewer_reason", "")).strip()
affected_sections = sorted(
{str(item) for item in payload.get("affected_sections", []) if str(item)}
)
differences = [
{
"difference_id": str(item["difference_id"]),
"description": str(item["description"]).strip(),
"affected_facts": sorted(
{str(ref) for ref in item.get("affected_facts", [])}
),
"source_refs": sorted({str(ref) for ref in item.get("source_refs", [])}),
}
for item in payload.get("differences", [])
]
if mismatch_present and (not reason or not affected_sections or not differences):
raise ValueError(
"A substantive taxonomy mismatch requires sections, differences, and a reviewer reason"
)
if any(
not SAFE_ID.fullmatch(item["difference_id"]) or not item["description"]
for item in differences
):
raise ValueError("Taxonomy differences require safe IDs and descriptions")
if len({item["difference_id"] for item in differences}) != len(differences):
raise ValueError("Taxonomy difference IDs must be unique")
if any(
item["source_refs"]
and not set(item["source_refs"]) <= _available_evidence_refs(case)
for item in differences
):
raise ValueError("Taxonomy difference references evidence outside the case")
_mutate(case, reviewer, "taxonomy_representation_recorded")
case["taxonomy_representation"] = {
"mismatch_present": mismatch_present,
"affected_sections": affected_sections,
"differences": differences,
"chosen_treatment": treatment,
"reviewer_reason": reason,
"reviewed_by": reviewer,
"reviewed_at": _now(),
"vera_did_not_select_filing_route": True,
}
case["validation"] = None
_record_event(
case,
"taxonomy_representation_recorded",
reviewer,
{
"mismatch_present": mismatch_present,
"chosen_treatment": treatment,
},
)
return case
def record_micro_reporting(
case: dict[str, Any],
payload: Mapping[str, Any],
reviewer: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record whether a micro case uses reviewed footer disclosures or notes."""
_ensure_revision(case, expected_revision)
if case.get("selected_form") != "MICRO":
raise ValueError("Micro reporting choices apply only to the MICRO form")
mode = str(payload["mode"]).upper()
if mode not in {"FOOTER_ONLY", "NOTES"}:
raise ValueError("Micro reporting mode must be FOOTER_ONLY or NOTES")
required_footer_keys = {
"guarantees_commitments_contingencies",
"director_auditor_compensation",
"own_and_parent_shares",
}
footer_items: list[dict[str, Any]] = []
if mode == "FOOTER_ONLY":
supplied = {str(item["key"]): item for item in payload.get("footer_items", [])}
if set(supplied) != required_footer_keys:
raise ValueError(
"Footer-only micro accounts require every statutory footer item"
)
for key in sorted(required_footer_keys):
raw = supplied[key]
status = str(raw["status"]).upper()
value = raw.get("value")
reason = str(raw.get("reason", "")).strip()
if status not in {"PRESENT", "NOT_APPLICABLE_CONFIRMED"}:
raise ValueError("Micro footer items require a reviewed status")
if status == "PRESENT" and value in {None, ""}:
raise ValueError("Present micro footer items require reviewed content")
if status == "NOT_APPLICABLE_CONFIRMED" and not reason:
raise ValueError("Negative micro footer items require a reason")
footer_items.append(
{
"key": key,
"status": status,
"value": value,
"reason": reason,
"source_refs": sorted(
{str(ref) for ref in raw.get("source_refs", [])}
),
}
)
if footer_items[-1]["source_refs"] and not set(
footer_items[-1]["source_refs"]
) <= _available_evidence_refs(case):
raise ValueError("Micro footer references evidence outside the case")
output_contract = (case.get("taxonomy_output_contracts") or {}).get(
"MICRO_FOOTER_TEXT"
)
if not isinstance(output_contract, Mapping):
raise ValueError("The case has no versioned micro-footer output contract")
output_language = str(case.get("output_language", "it"))
labels = output_contract["labels"][output_language]
negative_text = str(output_contract["not_applicable_text"][output_language])
rendered_text = None
output_source_refs: list[str] = []
if mode == "FOOTER_ONLY":
rendered_lines = []
for item in footer_items:
content = (
str(item["value"]).strip()
if item["status"] == "PRESENT"
else f"{negative_text} ({item['reason']})"
)
rendered_lines.append(f"{labels[item['key']]}: {content}.")
output_source_refs.extend(item["source_refs"])
rendered_text = " ".join(rendered_lines)
_mutate(case, reviewer, "micro_reporting_recorded")
_clear_narrative_reviews(case)
case["micro_reporting"] = {
"mode": mode,
"status": "CONFIRMED",
"footer_items": footer_items,
"xbrl_concept": (
str(output_contract["xbrl_concept"]) if mode == "FOOTER_ONLY" else None
),
"rendered_text": rendered_text,
"source_refs": sorted(set(output_source_refs)),
"reviewed_by": reviewer,
"reviewed_at": _now(),
}
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"micro_reporting_recorded",
reviewer,
{"mode": mode},
)
return case
def build_statements(
case: dict[str, Any], actor: str, expected_revision: str
) -> dict[str, Any]:
"""Aggregate reviewed mappings into exact current/comparative statement facts."""
_ensure_revision(case, expected_revision)
entries = (case.get("trial_balance") or {}).get("entries", [])
mapped_ids = {item["account_id"] for item in case.get("mappings", [])}
if {row["account_id"] for row in entries} != mapped_ids:
raise ValueError("Every account must be mapped or explicitly excluded")
comparative_required = case["entity"].get("first_financial_year") is not True
grouped: dict[tuple[str, str, str | None, str], dict[str, Any]] = {}
for mapping in case["mappings"]:
for allocation in mapping["allocations"]:
key = (
allocation["statement_section"],
allocation["canonical_line"],
allocation["xbrl_concept"],
allocation["xbrl_sign_multiplier"],
)
item = grouped.setdefault(
key,
{
"statement_section": key[0],
"canonical_line": key[1],
"xbrl_concept": key[2],
"xbrl_sign_multiplier": key[3],
"current_amount": Decimal("0"),
"prior_amount": Decimal("0") if comparative_required else None,
"source_refs": [],
"allocation_refs": [],
"adjustment_refs": [],
},
)
item["current_amount"] += Decimal(allocation["current_amount"])
if comparative_required:
item["prior_amount"] += Decimal(allocation["prior_amount"])
item["source_refs"].extend(allocation["source_refs"])
item["allocation_refs"].append(allocation["allocation_id"])
for adjustment in case.get("adjustments", []):
for line in adjustment["lines"]:
key = (
line["statement_section"],
line["canonical_line"],
line["xbrl_concept"],
line["xbrl_sign_multiplier"],
)
item = grouped.setdefault(
key,
{
"statement_section": key[0],
"canonical_line": key[1],
"xbrl_concept": key[2],
"xbrl_sign_multiplier": key[3],
"current_amount": Decimal("0"),
"prior_amount": Decimal("0") if comparative_required else None,
"source_refs": [],
"allocation_refs": [],
"adjustment_refs": [],
},
)
item["current_amount"] += Decimal(line["current_amount"])
if comparative_required:
item["prior_amount"] += Decimal(line["prior_amount"])
item["source_refs"].extend(line["source_refs"])
item["adjustment_refs"].append(line["line_id"])
facts: list[dict[str, Any]] = []
for position, item in enumerate(grouped.values(), start=1):
facts.append(
{
"fact_id": f"fact_{position:06d}",
"domain": "statements",
"key": item["canonical_line"],
"statement_section": item["statement_section"],
"xbrl_concept": item["xbrl_concept"],
"xbrl_sign_multiplier": item["xbrl_sign_multiplier"],
"period": case["period"]["end"],
"current_value": _decimal_text(item["current_amount"]),
"prior_value": (
_decimal_text(item["prior_amount"])
if comparative_required
else None
),
"currency": "EUR",
"status": EvidenceStatus.DERIVED,
"source_refs": sorted(set(item["source_refs"])),
"derivation": {
"operation": "SUM_REVIEWED_ALLOCATIONS_AND_ADJUSTMENTS",
"allocation_refs": item["allocation_refs"],
"adjustment_refs": item["adjustment_refs"],
},
"confirmed_by": None,
"confidence": "HIGH",
"materiality": "UNASSESSED",
"rule_pack": case["rule_pack_versions"]["statutory_rule_pack"],
}
)
# EQUITY_RESULT is a reviewed component of liabilities/equity. Include its
# exact amount in the parent total without copying facts or deriving a
# profit decision. Reuse the same membership for presentation rounding.
section_groups: dict[str, list[dict[str, Any]]] = {}
for fact in facts:
section_groups.setdefault(fact["statement_section"], []).append(fact)
if fact["statement_section"] == "EQUITY_RESULT":
section_groups.setdefault("LIABILITIES_EQUITY", []).append(fact)
totals: dict[str, dict[str, str]] = {}
for section, selected in sorted(section_groups.items()):
totals[section] = {
"current": _decimal_text(
sum(Decimal(fact["current_value"]) for fact in selected)
),
"prior": (
_decimal_text(sum(Decimal(fact["prior_value"]) for fact in selected))
if comparative_required
else None
),
}
precision = int(case.get("reporting_precision", 0))
presentation_facts = [
{
"fact_id": fact["fact_id"],
"current_value": _decimal_text(
_reported_decimal(Decimal(fact["current_value"]), precision)
),
"prior_value": (
_decimal_text(
_reported_decimal(Decimal(fact["prior_value"]), precision)
)
if comparative_required
else None
),
}
for fact in facts
]
presentation_lookup = {item["fact_id"]: item for item in presentation_facts}
rounding_adjustments: list[dict[str, Any]] = []
for section, exact in totals.items():
section_facts = section_groups[section]
period_fields = [("current", "current_value")]
if comparative_required:
period_fields.append(("prior", "prior_value"))
for period_key, value_key in period_fields:
rounded_total = _reported_decimal(Decimal(exact[period_key]), precision)
displayed_sum = sum(
Decimal(presentation_lookup[fact["fact_id"]][value_key])
for fact in section_facts
)
difference = rounded_total - displayed_sum
if difference:
rounding_adjustments.append(
{
"adjustment_id": (f"rounding_{section.lower()}_{period_key}"),
"kind": "PRESENTATION_ROUNDING",
"statement_section": section,
"period": period_key,
"amount": _decimal_text(difference),
"precision": precision,
"derivation": {
"operation": "ROUNDED_EXACT_TOTAL_MINUS_SUM_ROUNDED_LINES",
"fact_refs": [fact["fact_id"] for fact in section_facts],
},
"repairs_substantive_imbalance": False,
}
)
_mutate(case, actor, "statements_computed")
_clear_accounting_dependent_reviews(case, preserve_adjustments=True)
case["canonical_facts"] = facts
case["statements"] = {
"facts": facts,
"section_totals": totals,
"presentation_facts": presentation_facts,
"rounding_adjustments": rounding_adjustments,
"reporting_precision": precision,
"comparative_status": (
"PRESENT" if comparative_required else "NOT_APPLICABLE_FIRST_FINANCIAL_YEAR"
),
"computed_at": _now(),
"computation_context": _computation_context(case, "statement-engine-v2"),
}
case["statutory_presentation"] = None
case["state"] = CaseState.STATEMENT_REVIEW
case["schedules"] = []
case["validation"] = None
_record_event(case, "statements_computed", actor, {"fact_count": len(facts)})
_record_event(
case,
"fact_derived",
actor,
{"fact_ids": [fact["fact_id"] for fact in facts]},
)
return case
def record_schedule(
case: dict[str, Any],
payload: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record a source-evidenced schedule and its exact reconciliation results."""
_ensure_revision(case, expected_revision)
statements = case.get("statements")
if not statements:
raise ValueError("Statements must be computed before schedules")
reviewed_payload = deepcopy(dict(payload))
schedule_type = str(reviewed_payload.get("schedule_type", "")).upper()
if schedule_type == "INVENTORIES":
statement_line = str(reviewed_payload.get("statement_line", ""))
inventory_concepts = {
str(item["xbrl_concept"])
for item in (case.get("taxonomy_mapping_index") or {}).get("concepts", [])
if schedule_type
in {str(value).upper() for value in item.get("schedule_types", [])}
}
line_classifications: dict[str, list[bool]] = {}
for mapping in case.get("mappings", []):
for allocation in mapping.get("allocations", []):
line = str(allocation["canonical_line"])
is_inventory = (
str(allocation.get("xbrl_concept") or "") in inventory_concepts
)
if case.get("statutory_presentation_required") is False:
is_inventory = is_inventory or schedule_type in {
str(value).upper()
for value in allocation.get("schedule_triggers", [])
}
line_classifications.setdefault(line, []).append(is_inventory)
for adjustment in case.get("adjustments", []):
for line_item in adjustment.get("lines", []):
line = str(line_item["canonical_line"])
line_classifications.setdefault(line, []).append(
str(line_item.get("xbrl_concept") or "") in inventory_concepts
)
allowed_lines = {
line
for line, classifications in line_classifications.items()
if classifications and all(classifications)
}
if statement_line not in allowed_lines:
raise ValueError(
"Inventory schedules require a statement line backed only by "
"reviewed inventory mappings"
)
valid_evidence_refs = _available_evidence_refs(case)
schedule_id = str(reviewed_payload.get("schedule_id", "schedule"))
collections = [
("rows", "row_id"),
("items", "item_id"),
]
for collection_name, identifier_key in collections:
normalized_items = []
for position, raw_item in enumerate(
reviewed_payload.get(collection_name, []), start=1
):
item = dict(raw_item)
item_id = str(item.get(identifier_key) or position)
status = str(item.get("evidence_status", "")).upper()
refs = sorted(
{
str(ref).strip()
for ref in item.get("source_refs", [])
if str(ref).strip()
}
)
if status == EvidenceStatus.OBSERVED:
if not refs or not set(refs) <= valid_evidence_refs:
raise ValueError(
"Observed schedule rows must cite evidence anchored in the case"
)
elif status == EvidenceStatus.USER_CONFIRMED:
anchored = sorted(set(refs) & valid_evidence_refs)
item["source_refs"] = anchored or [
f"professional_input:{actor}:{schedule_id}:{item_id}"
]
normalized_items.append(item)
if collection_name in reviewed_payload:
reviewed_payload[collection_name] = normalized_items
exception = reviewed_payload.get("amortisation_reconciliation_exception")
if isinstance(exception, Mapping):
normalized_exception = dict(exception)
anchored = sorted(
set(str(ref) for ref in exception.get("source_refs", []))
& valid_evidence_refs
)
normalized_exception["source_refs"] = anchored or [
f"professional_input:{actor}:{schedule_id}:amortisation_exception"
]
reviewed_payload["amortisation_reconciliation_exception"] = normalized_exception
schedule = normalize_schedule(
reviewed_payload,
statements["facts"],
comparative_required=(case["entity"].get("first_financial_year") is not True),
)
if (
schedule["schedule_type"] == "CASH_FLOW"
and case.get("selected_form") != "ORDINARY"
):
raise ValueError("A cash-flow statement belongs to ordinary accounts in MVP")
existing = [
item
for item in case.get("schedules", [])
if item["schedule_id"] != schedule["schedule_id"]
]
_mutate(case, actor, "schedule_recorded")
_clear_schedule_taxonomy_adapter(case)
_clear_narrative_reviews(case)
schedule["computation_context"] = _computation_context(
case, f"schedule-{str(schedule['schedule_type']).lower()}-v1"
)
case["schedules"] = [*existing, schedule]
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"schedule_recorded",
actor,
{
"schedule_id": schedule["schedule_id"],
"schedule_type": schedule["schedule_type"],
"status": schedule["status"],
},
)
return case
def ingest_schedule_file(
case: dict[str, Any],
source_path: Path,
schedule_type: str,
schedule_id: str,
statement_line: str | None,
options: Mapping[str, Any],
actor: str,
expected_revision: str,
sheet: str | None = None,
) -> dict[str, Any]:
"""Import a documented CSV/XLSX schedule template with per-cell anchors."""
_ensure_revision(case, expected_revision)
if not case.get("statements"):
raise ValueError("Statements must be computed before schedule ingestion")
if source_path.is_symlink() or not source_path.is_file():
raise ValueError("Schedule source must be a regular local file")
normalized_type = schedule_type.upper()
required_fields = schedule_template_fields(normalized_type)
required_text_fields = schedule_template_text_fields(normalized_type)
headers, raw_rows, sheet_name = _load_table(source_path.resolve(), sheet)
header_columns = _normalized_header_columns(headers)
normalized_headers = [item["normalized_column"] for item in header_columns]
required_headers = {"row_id", *required_fields, *required_text_fields}
missing_headers = sorted(required_headers - set(normalized_headers))
if missing_headers:
raise ValueError(
f"Schedule template is missing columns: {', '.join(missing_headers)}"
)
document_id = _next_document_id(case)
anchors: list[dict[str, Any]] = []
rows: list[dict[str, Any]] = []
seen: set[str] = set()
for row_number, values in enumerate(raw_rows, start=2):
row = {
name: values[position] if position < len(values) else None
for position, name in enumerate(normalized_headers)
}
row_id = str(row.get("row_id") or "").strip()
if not row_id and all(value in {None, ""} for value in row.values()):
continue
if not row_id or row_id in seen:
raise ValueError("Schedule row IDs must be present and unique")
seen.add(row_id)
source_refs: list[str] = []
normalized_row: dict[str, Any] = {
"row_id": row_id,
"label": str(row.get("label") or row_id),
"evidence_status": EvidenceStatus.OBSERVED,
}
for column in header_columns:
position = int(column["column_index"]) - 1
column_name = str(column["normalized_column"])
anchor_id = f"src_{document_id}_{len(anchors) + 1:07d}"
raw_value = values[position] if position < len(values) else None
anchor: dict[str, Any] = {
"source_ref": anchor_id,
"document_id": document_id,
"sheet": sheet_name,
"row": row_number,
"column": column["column"],
"column_header": column["column_header"],
"normalized_column": column_name,
"raw_value": "" if raw_value is None else str(raw_value),
"parser_profile": f"{normalized_type.lower()}_template_v1",
"confidence": "HIGH",
}
if column_name in required_fields:
value = normalize_decimal(raw_value)
normalized_row[column_name] = _decimal_text(value)
anchor["normalized_value"] = _decimal_text(value)
elif column_name in required_text_fields:
text_value = str(raw_value or "").strip()
normalized_row[column_name] = text_value or "UNKNOWN"
anchor["normalized_value"] = text_value or "UNKNOWN"
else:
anchor["normalized_value"] = str(raw_value or "").strip()
anchors.append(anchor)
source_refs.append(anchor_id)
normalized_row["source_refs"] = source_refs
rows.append(normalized_row)
payload = {
"schedule_id": schedule_id,
"schedule_type": normalized_type,
"statement_line": statement_line,
"rows": rows,
**dict(options),
}
schedule = normalize_schedule(payload, case["statements"]["facts"])
document = {
"document_id": document_id,
"purpose": f"{normalized_type}_SCHEDULE",
"file_name": source_path.name,
"media_type": (
"text/csv"
if source_path.suffix.lower() == ".csv"
else "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
),
"sha256": _sha256_file(source_path.resolve()),
"size_bytes": source_path.resolve().stat().st_size,
"sheet": sheet_name,
"columns": header_columns,
"parser_profile": f"{normalized_type.lower()}_template_v1",
"source_anchors": anchors,
"parsed_at": _now(),
}
existing = [
item for item in case.get("schedules", []) if item["schedule_id"] != schedule_id
]
_mutate(case, actor, "schedule_document_parsed")
_clear_schedule_taxonomy_adapter(case)
_clear_narrative_reviews(case)
case["source_documents"].append(document)
schedule["computation_context"] = _computation_context(
case, f"schedule-{normalized_type.lower()}-template-v1"
)
case["schedules"] = [*existing, schedule]
case["validation"] = None
_refresh_disclosures(case)
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
_record_event(
case,
"document_parsed",
actor,
{"document_id": document_id, "schedule_type": normalized_type},
)
_record_event(
case,
"schedule_recorded",
actor,
{"schedule_id": schedule_id, "status": schedule["status"]},
)
return case
def _refresh_disclosures(case: dict[str, Any]) -> None:
rule_pack = case.get("disclosure_rule_pack")
if not rule_pack or not case.get("selected_form") or not case.get("statements"):
return
coverage = build_disclosure_coverage(
case, _effective_disclosure_rule_pack(case, rule_pack)
)
coverage["computation_context"] = _computation_context(
case, "disclosure-coverage-v1"
)
case["disclosure_coverage"] = coverage
case["questionnaire"] = coverage["questions"]
outline = note_outline(case)
accepted_sections = {
block["section_id"]
for block in case.get("narrative_blocks", [])
if block.get("status") == "ACCEPTED"
}
for section in outline:
if section["section_id"] in accepted_sections:
section["status"] = "ACCEPTED"
case["note_outline"] = outline
case["note_outline_context"] = _computation_context(case, "note-outline-v1")
_update_prevalidation_state(case)
def record_disclosure_trigger_decisions(
case: dict[str, Any],
decisions: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record professional decisions for semantically ambiguous applicability."""
_ensure_revision(case, expected_revision)
rule_pack = case.get("disclosure_rule_pack")
if not rule_pack:
raise ValueError("Disclosure rules must be activated before trigger review")
valid_flags = manual_disclosure_flags(rule_pack)
if not decisions:
raise ValueError("At least one disclosure trigger decision is required")
normalized: list[dict[str, Any]] = []
seen: set[str] = set()
for raw in decisions:
if set(raw) != {"flag", "status", "reason", "source_refs"}:
raise ValueError(
"Disclosure trigger decisions require flag, status, reason, and source_refs"
)
flag = str(raw["flag"])
status = str(raw["status"]).upper()
reason = str(raw["reason"]).strip()
raw_source_refs = raw["source_refs"]
if not isinstance(raw_source_refs, list):
raise ValueError("Disclosure trigger source_refs must be a list")
source_refs = sorted(
{str(item) for item in raw_source_refs if str(item).strip()}
)
if flag not in valid_flags or flag in seen:
raise ValueError(f"Unknown or duplicate disclosure trigger flag: {flag}")
if status not in {"TRIGGERED", "NOT_APPLICABLE_CONFIRMED"}:
raise ValueError("Disclosure trigger status is not reviewable")
if not reason or not source_refs:
raise ValueError(
"Disclosure trigger decisions require a reason and evidence references"
)
if not set(source_refs) <= _available_evidence_refs(case):
raise ValueError(
"Disclosure trigger decision references evidence outside the case"
)
seen.add(flag)
normalized.append(
{
"flag": flag,
"status": status,
"reason": reason,
"source_refs": source_refs,
"reviewed_by": actor,
"reviewed_at": _now(),
}
)
retained = [
item
for item in case.get("disclosure_trigger_decisions", [])
if item["flag"] not in seen
]
_mutate(case, actor, "disclosure_trigger_decisions_recorded")
_clear_narrative_reviews(case)
case["disclosure_trigger_decisions"] = [*retained, *normalized]
case["disclosure_trigger_flags"] = sorted(
item["flag"]
for item in case["disclosure_trigger_decisions"]
if item["status"] == "TRIGGERED"
)
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"disclosure_trigger_decisions_recorded",
actor,
{"flags": sorted(seen)},
)
return case
def _update_prevalidation_state(case: dict[str, Any]) -> None:
"""Distinguish source-backed data gaps from narrative drafting work."""
if case.get("state") in {
CaseState.APPROVED,
CaseState.EXPORTED,
CaseState.ARCHIVED,
CaseState.UNSUPPORTED,
}:
return
required_schedules = required_schedule_types(case)
schedules = case.get("schedules", [])
present_schedules = {str(item["schedule_type"]) for item in schedules}
schedule_gaps = bool(required_schedules - present_schedules) or any(
item.get("status") != "COMPLETE" for item in schedules
)
coverage = (case.get("disclosure_coverage") or {}).get("coverage", [])
structured_gaps = any(
item.get("triggered")
and any(
not requirement.get("complete")
and requirement.get("kind") != "NARRATIVE_SECTION"
for requirement in item.get("requirements", [])
)
for item in coverage
)
case["state"] = (
CaseState.DATA_GAPS
if schedule_gaps or structured_gaps
else CaseState.NOTE_DRAFT
)
def activate_disclosures(
case: dict[str, Any],
rule_pack: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Lock an effective-dated disclosure pack and build relevant questions."""
_ensure_revision(case, expected_revision)
if not case.get("selected_form") or not case.get("statements"):
raise ValueError("Statements and statutory form are required for disclosures")
_validate_disclosure_rule_pack(
rule_pack, date.fromisoformat(case["period"]["start"])
)
incoming_checksum = disclosure_rule_pack_hash(rule_pack)
locked_id = case["rule_pack_versions"].get("disclosure_rule_pack")
locked_checksum = case.get("disclosure_rule_pack_checksum")
if locked_id is not None and (
str(rule_pack.get("id")) != str(locked_id)
or incoming_checksum != locked_checksum
):
raise ValueError(
"Disclosure rule pack differs from the locked case; use explicit migration"
)
_mutate(case, actor, "disclosure_rules_activated")
case["disclosure_rule_pack"] = deepcopy(dict(rule_pack))
case["rule_pack_versions"]["disclosure_rule_pack"] = str(rule_pack["id"])
case["disclosure_rule_pack_checksum"] = incoming_checksum
history_narratives = list(
(case.get("client_history_suggestions") or {}).get("narrative_suggestions", [])
)
case["prior_narrative_suggestions"] = [
*prior_narrative_suggestions(case),
*history_narratives,
]
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"disclosure_rules_activated",
actor,
{
"rule_pack": rule_pack["id"],
"triggered_count": case["disclosure_coverage"]["triggered_count"],
},
)
return case
def record_narrative_blocks(
case: dict[str, Any],
blocks: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record sentence-level provenance-bound note blocks for reviewer acceptance."""
_ensure_revision(case, expected_revision)
if not case.get("disclosure_rule_pack"):
raise ValueError("Disclosure rules must be activated before note blocks")
normalized = normalize_narrative_blocks(
blocks, actor, str(case.get("output_language", "it"))
)
accepted_answers = {
str(answer["key"])
for answer in case.get("disclosure_answers", [])
if disclosure_answer_complete(answer)
}
valid_refs = {
str(fact["fact_id"]) for fact in case.get("canonical_facts", [])
} | accepted_answers
valid_refs.update(
str(ref)
for answer in case.get("disclosure_answers", [])
for ref in answer.get("source_refs", [])
)
valid_refs.update(
str(ref)
for schedule in case.get("schedules", [])
for row in schedule.get("rows", [])
for ref in row.get("source_refs", [])
)
structured_facts = {
str(fact["fact_id"]): fact
for fact in [
*case.get("canonical_facts", []),
*case.get("taxonomy_facts", []),
*case.get("schedule_taxonomy_facts", []),
*[
schedule_fact
for schedule in case.get("schedules", [])
if schedule.get("status") == "COMPLETE"
for schedule_fact in schedule_fact_records(schedule)
],
]
}
valid_refs.update(structured_facts)
accepted_qnames: set[str] = set()
for block in normalized:
qname = block.get("xbrl_concept")
if qname is not None and not SAFE_QNAME.fullmatch(str(qname)):
raise ValueError(f"Invalid narrative XBRL QName: {qname}")
if qname and block["status"] != "ACCEPTED":
raise ValueError("Only accepted narrative may be bound to an XBRL concept")
if block["status"] == "ACCEPTED" and not qname:
raise ValueError("Accepted narrative requires an XBRL concept")
if qname in accepted_qnames:
raise ValueError("Accepted narrative XBRL concepts must be unique")
if qname:
accepted_qnames.add(str(qname))
for claim in block["claims"]:
if (
claim["kind"] == "FACTUAL"
and not set(claim["source_refs"]) <= valid_refs
):
raise ValueError(
"Narrative factual claim references unaccepted evidence"
)
if claim["kind"] != "FACTUAL" or block["status"] != "ACCEPTED":
continue
support = claim.get("semantic_support")
if (
not isinstance(support, Mapping)
or str(support.get("status", "")).upper() != "SUPPORTED"
or not str(support.get("reason", "")).strip()
):
raise ValueError(
"Accepted factual narrative requires explicit semantic support"
)
assertions: list[Decimal] = []
for assertion in claim.get("fact_assertions", []):
if set(assertion) != {"fact_ref", "value_field", "value"}:
raise ValueError(
"Narrative fact assertions require fact_ref, value_field, and value"
)
fact_ref = str(assertion["fact_ref"])
value_field = str(assertion["value_field"])
if (
fact_ref not in claim["source_refs"]
or fact_ref not in structured_facts
):
raise ValueError(
"Narrative fact assertion must reference cited structured evidence"
)
fact = structured_facts[fact_ref]
if value_field not in {"current_value", "prior_value", "value"} or (
value_field not in fact
):
raise ValueError(
"Narrative fact assertion uses an invalid value field"
)
asserted_value = normalize_decimal(assertion["value"])
if asserted_value != normalize_decimal(fact[value_field]):
raise ValueError(
"Narrative fact assertion does not match structured evidence"
)
assertions.append(asserted_value)
literal_values = [
_normalize_narrative_money_literal(
match, str(case.get("output_language", "it"))
)
for match in MONEY_LITERAL.findall(str(claim["sentence"]))
]
if sorted(literal_values) != sorted(assertions):
raise ValueError(
"Narrative monetary statements must exactly match fact assertions"
)
prior_id = block.get("prior_suggestion_id")
if prior_id:
prior = next(
(
item
for item in case.get("prior_narrative_suggestions", [])
if item["suggestion_id"] == prior_id
),
None,
)
if prior is None:
raise ValueError(
"Narrative block references an unknown prior suggestion"
)
if prior.get("language") not in {
None,
"",
case.get("output_language", "it"),
}:
raise ValueError(
"Prior narrative language differs from the current output"
)
block["redline"] = narrative_redline(prior["text"], block["text"])
block["prior_source_refs"] = list(prior["source_refs"])
_mutate(case, actor, "narrative_blocks_recorded")
case["narrative_blocks"] = normalized
case["narrative_context"] = _computation_context(case, "reviewed-narrative-v1")
case["validation"] = None
_refresh_disclosures(case)
_record_event(
case,
"narrative_edited",
actor,
{
"block_count": len(normalized),
"accepted_count": sum(
block["status"] == "ACCEPTED" for block in normalized
),
},
)
return case
def record_intelligence_suggestion(
case: dict[str, Any],
task: IntelligenceTask | str,
subject_ids: Sequence[str],
output: Mapping[str, Any],
model_metadata: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record a validated, non-authoritative semantic suggestion for review."""
_ensure_revision(case, expected_revision)
if case.get("state") in {
CaseState.APPROVED,
CaseState.EXPORTED,
CaseState.ARCHIVED,
}:
raise ValueError("Intelligence suggestions cannot mutate a locked case")
required_metadata = {"provider", "model", "prompt_template_version"}
if set(model_metadata) != required_metadata or any(
not str(model_metadata[key]).strip() for key in required_metadata
):
raise ValueError(
"Model metadata requires provider, model, and prompt_template_version"
)
packet = build_intelligence_packet(case, task, subject_ids)
normalized = validate_intelligence_output(packet, output)
selected_task = IntelligenceTask(str(task))
_mutate(case, actor, "intelligence_suggestion_recorded")
run = {
"run_id": f"intel_{len(case.get('intelligence_runs', [])) + 1:06d}",
"task": selected_task.value,
"subject_ids": list(subject_ids),
"status": EvidenceStatus.MODEL_SUGGESTED,
"input_packet_hash": intelligence_packet_hash(packet),
"input_context_receipt": dict(packet["context_receipt"]),
"input_revision_id": expected_revision,
"model_metadata": dict(model_metadata),
"output": normalized,
"recorded_by": actor,
"recorded_at": _now(),
"requires_review": True,
"computation_context": _computation_context(
case,
str(model_metadata["prompt_template_version"]),
model_version=str(model_metadata["model"]),
),
}
case.setdefault("intelligence_runs", []).append(run)
if selected_task is IntelligenceTask.WORKFLOW_GUIDANCE:
case["latest_workflow_guidance"] = {
"run_id": run["run_id"],
**normalized,
}
elif selected_task is IntelligenceTask.ACCOUNT_MAPPING:
case["model_mapping_suggestions"] = list(normalized["suggestions"])
elif selected_task is IntelligenceTask.DISCLOSURE_ACTIVATION:
case["disclosure_activation_suggestions"] = list(normalized["suggestions"])
elif selected_task is IntelligenceTask.NARRATIVE_DRAFT:
case["narrative_suggestions"] = list(normalized["blocks"])
case["validation"] = None
if selected_task is IntelligenceTask.ACCOUNT_MAPPING:
_record_event(
case,
"mapping_suggested",
actor,
{
"run_id": run["run_id"],
"candidate_count": len(normalized["suggestions"]),
"suggestion_source": "MODEL",
},
)
elif selected_task is IntelligenceTask.NARRATIVE_DRAFT:
_record_event(
case,
"narrative_generated",
actor,
{
"run_id": run["run_id"],
"block_count": len(normalized["blocks"]),
"status": "MODEL_SUGGESTED",
},
)
_record_event(
case,
"model_suggestion_recorded",
actor,
{"run_id": run["run_id"], "task": selected_task.value},
)
return case
def render_preview_html(case: Mapping[str, Any]) -> bytes:
"""Render native accounts using language-matched review navigation."""
return render_accounts_preview(case)
def create_preview(
case: dict[str, Any],
output_path: Path,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Write and hash the review preview before the final validation run."""
_ensure_revision(case, expected_revision)
if not case.get("statements"):
raise ValueError("Statements are required before preview rendering")
if output_path.suffix.lower() != ".html":
raise ValueError("Preview output must use the .html extension")
if output_path.is_symlink():
raise ValueError("Preview must not be written through a symbolic link")
if output_path.parent.is_symlink():
raise ValueError("Preview directory must not be a symbolic link")
_mutate(case, actor, "preview_rendered")
# Recompute content checks on a copy rather than embedding stale validation.
# Preview-integrity checks belong to the subsequent final validation: the
# previous preview is being replaced by the exact bytes rendered here.
preview_case = deepcopy(case)
checks = validate_case(preview_case)
checks["issues"] = [
issue
for issue in checks["issues"]
if issue["rule_id"]
not in {
"REVIEW.PREVIEW_REQUIRED",
"REVIEW.PREVIEW_STALE",
"REVIEW.PREVIEW_INTEGRITY",
}
]
preview_case["validation"] = checks
preview_case["preview_checks_recalculated"] = True
preview = render_preview_html(preview_case)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(preview)
case["preview"] = {
"file_name": output_path.name,
"sha256": _sha256_bytes(preview),
"size_bytes": len(preview),
"content_base64": base64.b64encode(preview).decode("ascii"),
"rendered_revision_id": case["revision_id"],
"rendered_at": _now(),
"review_content_hash": _review_content_hash(case),
"computation_context": _computation_context(case, "bilancio-preview-v1"),
}
case["validation"] = None
_record_event(
case,
"preview_rendered",
actor,
{
key: value
for key, value in case["preview"].items()
if key != "content_base64"
},
)
return case
def record_external_validation(
case: dict[str, Any],
report_path: Path,
result: str,
reported_issues: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Attach a user-supplied TEBENI report without automating transmission."""
_ensure_revision(case, expected_revision)
if case.get("state") not in {CaseState.APPROVED, CaseState.EXPORTED}:
raise ValueError(
"External validation results attach only to an approved or exported snapshot"
)
before_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
record = record_external_validation_result(
case, report_path, result, reported_issues, actor
)
document_id = _next_document_id(case)
document = {
"document_id": document_id,
"purpose": "EXTERNAL_VALIDATION_REPORT",
**record["report"],
"parser_profile": "user_declared_tebeni_result_v1",
"parsed_at": _now(),
}
record["document_id"] = document_id
case["source_documents"] = [
item
for item in case.get("source_documents", [])
if item.get("purpose") != "EXTERNAL_VALIDATION_REPORT"
] + [document]
case["external_validation"] = record
case["updated_at"] = _now()
case["_pending_before_hash"] = before_hash
_record_event(case, "document_uploaded", actor, {"document_id": document_id})
_record_evidence_attached(case, actor, document)
_record_event(
case,
"external_validation_recorded",
actor,
{"document_id": document_id, "result": record["result"]},
)
return case
def record_file_security_scan(
case: dict[str, Any], receipt: Mapping[str, Any], actor: str
) -> dict[str, Any]:
"""Bind one host scanner clean verdict to an imported document checksum."""
if str(receipt.get("status")) != "CLEAN":
raise ValueError("Only a clean malware-scan verdict may be recorded")
required = {
"sha256",
"size_bytes",
"engine",
"signature_version",
"scanned_at",
}
missing = sorted(key for key in required if receipt.get(key) in {None, ""})
if missing:
raise ValueError(
f"Malware-scan receipt is missing fields: {', '.join(missing)}"
)
source_sha256 = str(receipt["sha256"])
documents = [
document
for document in case.get("source_documents", [])
if document.get("sha256") == source_sha256
and int(document.get("size_bytes", -1)) == int(receipt["size_bytes"])
]
if not documents:
raise ValueError("Malware-scan receipt does not match an imported document")
document = documents[-1]
scan_id = f"scan_{len(case.get('file_security_scans', [])) + 1:06d}"
normalized = {
"scan_id": scan_id,
"document_id": document["document_id"],
"status": "CLEAN",
"sha256": source_sha256,
"size_bytes": int(receipt["size_bytes"]),
"engine": str(receipt["engine"]),
"signature_version": str(receipt["signature_version"]),
"scanned_at": str(receipt["scanned_at"]),
}
document["security_scan_id"] = scan_id
case.setdefault("file_security_scans", []).append(normalized)
_record_event(
case,
"document_malware_scanned",
actor,
{
"scan_id": scan_id,
"document_id": document["document_id"],
"engine": normalized["engine"],
"signature_version": normalized["signature_version"],
},
)
return case
def record_artifact_access(
case: dict[str, Any], action: str, actor: str, details: Mapping[str, Any]
) -> dict[str, Any]:
"""Append an artifact grant or redemption event without changing revision."""
if action not in {"artifact_download_grant_issued", "artifact_downloaded"}:
raise ValueError("Unsupported artifact access event")
required = {"grant_id", "artifact_id", "sha256"}
missing = sorted(key for key in required if not str(details.get(key, "")))
if missing:
raise ValueError(f"Artifact access event is missing: {', '.join(missing)}")
_record_event(
case,
action,
actor,
{key: str(details[key]) for key in sorted(required)},
)
return case
def record_taxonomy_catalogue_build(
case: dict[str, Any],
receipt: Mapping[str, Any],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Attach a deployment-controlled catalogue build receipt to one case."""
_ensure_revision(case, expected_revision)
required = {
"file_name",
"sha256",
"size_bytes",
"taxonomy_id",
"taxonomy_package_sha256",
"registry_sha256",
"concept_count",
"built_at",
}
missing = sorted(key for key in required if receipt.get(key) in {None, ""})
if missing:
raise ValueError(
f"Taxonomy catalogue build receipt is missing: {', '.join(missing)}"
)
if str(receipt["taxonomy_id"]) != str(case["rule_pack_versions"]["taxonomy_id"]):
raise ValueError("Built taxonomy catalogue identifier differs from the case")
if str(receipt["taxonomy_package_sha256"]) != str(case.get("taxonomy_checksum")):
raise ValueError("Built taxonomy package checksum differs from the case")
_mutate(case, actor, "taxonomy_catalogue_built")
case["taxonomy_catalogue_build"] = dict(receipt)
case["statutory_presentation"] = None
case["validation"] = None
_record_event(
case,
"taxonomy_catalogue_built",
actor,
{
"sha256": str(receipt["sha256"]),
"concept_count": int(receipt["concept_count"]),
"registry_sha256": str(receipt["registry_sha256"]),
},
)
return case
def archive_case(
case: dict[str, Any],
actor: str,
expected_revision: str,
*,
retention_days: int,
reason: str,
) -> dict[str, Any]:
"""Archive one case under an explicit host retention policy."""
_ensure_revision(case, expected_revision)
if case.get("state") == CaseState.ARCHIVED:
raise ValueError("Case is already archived")
if isinstance(retention_days, bool) or not 1 <= retention_days <= 3650:
raise ValueError("Retention days must be from 1 to 3650")
normalized_reason = reason.strip()
if not normalized_reason:
raise ValueError("Case archive reason is required")
archived_at = datetime.now(tz=timezone.utc)
before_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
case["state"] = CaseState.ARCHIVED
case["archive"] = {
"archived_at": archived_at.isoformat(timespec="seconds"),
"archived_by": actor,
"reason": normalized_reason,
"retention_days": retention_days,
"retain_until": (archived_at + timedelta(days=retention_days)).isoformat(
timespec="seconds"
),
}
case["updated_at"] = _now()
case["_pending_before_hash"] = before_hash
_record_event(
case,
"case_archived",
actor,
{
"reason": normalized_reason,
"retain_until": case["archive"]["retain_until"],
},
)
return case
def remember_mappings(
case: dict[str, Any],
memory_path: Path,
source_system_template: str,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Write approved classifications into an explicitly tenant-scoped store."""
_ensure_revision(case, expected_revision)
result = remember_approved_mappings(
case, memory_path, source_system_template, actor
)
_record_event(case, "approved_mappings_remembered", actor, result)
return case
def remember_client_history(
case: dict[str, Any],
history_path: Path,
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Persist approved client knowledge for unconfirmed future suggestions."""
_ensure_revision(case, expected_revision)
result = remember_approved_client_history(case, history_path, actor)
_record_event(case, "approved_client_history_remembered", actor, result)
return case
def _issue(rule_id: str, severity: str, message: str, index: int) -> ValidationIssue:
return ValidationIssue(
issue_id=f"iss_{index:04d}",
severity=severity,
rule_id=rule_id,
message=message,
override_allowed=severity not in {"BLOCKER"},
)
def _current_review_decision(
case: Mapping[str, Any], issue: Mapping[str, Any]
) -> Mapping[str, Any] | None:
fingerprint = str(issue.get("fingerprint") or _issue_fingerprint(issue))
for decision in reversed(case.get("review_decisions", [])):
if decision.get("issue_fingerprint") == fingerprint:
return decision
return None
def validate_case(case: dict[str, Any]) -> dict[str, Any]:
"""Run deterministic input, accounting, disclosure, XBRL, and review gates."""
issues: list[ValidationIssue] = []
def add(rule_id: str, severity: str, message: str) -> None:
issues.append(_issue(rule_id, severity, message, len(issues) + 1))
if case["state"] == CaseState.UNSUPPORTED or case.get("unsupported_reasons"):
add("INPUT.UNSUPPORTED_CASE", "BLOCKER", "The entity is outside the MVP scope")
trial_balance = case.get("trial_balance")
if not trial_balance:
add("INPUT.TRIAL_BALANCE_REQUIRED", "BLOCKER", "A trial balance is required")
elif not trial_balance.get("confirmed_convention"):
add(
"INPUT.PARSER_CONFIRMATION_REQUIRED",
"BLOCKER",
"The debit/credit convention is not confirmed",
)
if case["entity"].get("first_financial_year") is not True and not case.get(
"prior_xbrl"
):
add(
"INPUT.PRIOR_XBRL_RECOMMENDED",
"MEDIUM",
"The prior filed XBRL is not attached for comparative and opening checks",
)
prior_reconciliation = _prior_xbrl_reconciliation(case)
for item in prior_reconciliation["issues"]:
add(
f"INPUT.{item['code']}",
"BLOCKER",
"Current comparative facts do not reconcile to the attached prior XBRL: "
+ str(item.get("xbrl_concept", "unknown concept")),
)
if any(
item.get("status") == "RESTATEMENT_CONFIRMED"
for item in prior_reconciliation["checks"]
):
add(
"INPUT.PRIOR_XBRL_RESTATEMENT_REVIEWED",
"MEDIUM",
"Comparative amounts differ from the prior filing under an evidenced restatement decision",
)
if not case.get("selected_form"):
add("FORM.SELECTION_REQUIRED", "BLOCKER", "The statutory form is not confirmed")
if case.get("selected_form") == "MICRO":
micro_reporting = case.get("micro_reporting") or {}
if micro_reporting.get("status") != "CONFIRMED":
add(
"MICRO.REPORTING_CHOICE_REQUIRED",
"BLOCKER",
"The micro footer-versus-notes treatment is not confirmed",
)
elif micro_reporting.get("mode") == "FOOTER_ONLY":
if (
not SAFE_QNAME.fullmatch(str(micro_reporting.get("xbrl_concept", "")))
or not str(micro_reporting.get("rendered_text", "")).strip()
or not str(micro_reporting.get("reviewed_by", "")).strip()
):
add(
"MICRO.FOOTER_XBRL_FACT_REQUIRED",
"BLOCKER",
"Footer-only micro accounts require a reviewed renderable XBRL fact",
)
if any(
block.get("status") == "ACCEPTED"
for block in case.get("narrative_blocks", [])
):
add(
"MICRO.FOOTER_ONLY_HAS_NOTES",
"BLOCKER",
"Footer-only micro accounts cannot also export a note draft",
)
positive_answers = []
for answer in case.get("disclosure_answers", []):
value = answer.get("value")
if value is True or (
isinstance(value, Mapping)
and any(item is True for item in value.values())
):
positive_answers.append(str(answer["key"]))
prohibited = sorted(
set(positive_answers)
& {
"accounting_policy_changes",
"contingent_liabilities",
"derivatives",
"double_format_events",
"going_concern_uncertainties",
"guarantees_and_commitments",
"non_market_transactions",
"off_balance_sheet_arrangements",
"post_closing_events",
"prior_period_errors",
"related_party_transactions",
}
)
if prohibited:
add(
"MICRO.FOOTER_ONLY_DISCLOSURE_CONFLICT",
"BLOCKER",
"Footer-only treatment is incompatible with positive disclosures: "
+ ", ".join(prohibited),
)
entries = (trial_balance or {}).get("entries", [])
mapping_ids = {item["account_id"] for item in case.get("mappings", [])}
missing_mappings = [
row["account_id"] for row in entries if row["account_id"] not in mapping_ids
]
if missing_mappings:
add(
"MAPPING.COVERAGE",
"BLOCKER",
f"{len(missing_mappings)} accounts are not mapped or excluded",
)
entries_by_id = {str(row["account_id"]): row for row in entries}
invalid_exclusions = []
comparative_required = case["entity"].get("first_financial_year") is not True
for mapping in case.get("mappings", []):
if mapping.get("decision") != "EXCLUDED":
continue
account = entries_by_id.get(str(mapping.get("account_id")))
if account is None:
continue
current_nonzero = Decimal(str(account["closing_signed"])) != 0
prior_value = account.get("prior_closing_signed")
prior_nonzero = comparative_required and (
prior_value is None or Decimal(str(prior_value)) != 0
)
if current_nonzero or prior_nonzero:
invalid_exclusions.append(str(mapping["account_id"]))
if invalid_exclusions:
add(
"MAPPING.NONZERO_EXCLUSION",
"BLOCKER",
"Non-zero accounts cannot be excluded: "
+ ", ".join(sorted(invalid_exclusions)),
)
for adjustment in case.get("adjustments", []):
for amount_key in ("current_amount", "prior_amount"):
total = sum(
Decimal(line[amount_key])
for line in adjustment.get("lines", [])
if line.get(amount_key) is not None
)
if total:
add(
"ADJUSTMENT.UNBALANCED",
"BLOCKER",
f"Adjustment {adjustment['adjustment_id']} is not presentation-neutral",
)
statements = case.get("statements")
if not statements:
add("STATEMENT.NOT_COMPUTED", "BLOCKER", "Statements have not been computed")
else:
totals = statements["section_totals"]
if "ASSETS" in totals and "LIABILITIES_EQUITY" in totals:
balance_periods = [("current", "STATEMENT.BALANCE_SHEET", "Current")]
if case["entity"].get("first_financial_year") is not True:
balance_periods.append(
(
"prior",
"STATEMENT.COMPARATIVE_BALANCE_SHEET",
"Comparative",
)
)
for period, rule_id, period_label in balance_periods:
asset_value = totals["ASSETS"].get(period)
liability_value = totals["LIABILITIES_EQUITY"].get(period)
if asset_value is None or liability_value is None:
add(
rule_id,
"BLOCKER",
f"{period_label} balance-sheet totals are missing",
)
elif Decimal(asset_value) + Decimal(liability_value) != 0:
add(
rule_id,
"BLOCKER",
f"{period_label} assets do not equal liabilities and equity",
)
result_facts = [
fact
for fact in statements["facts"]
if fact["statement_section"] in {"INCOME_RESULT", "EQUITY_RESULT"}
]
result_periods = [("current_value", "STATEMENT.RESULT_TIE_OUT", "Current")]
if case["entity"].get("first_financial_year") is not True:
result_periods.append(
(
"prior_value",
"STATEMENT.COMPARATIVE_RESULT_TIE_OUT",
"Comparative",
)
)
for value_field, rule_id, period_label in result_periods:
if any(fact.get(value_field) is None for fact in result_facts):
add(
rule_id,
"BLOCKER",
f"{period_label} result reconciliation values are missing",
)
continue
result_income = sum(
Decimal(fact[value_field])
for fact in result_facts
if fact["statement_section"] == "INCOME_RESULT"
)
result_equity = sum(
Decimal(fact[value_field])
for fact in result_facts
if fact["statement_section"] == "EQUITY_RESULT"
)
if (result_income or result_equity) and result_income + result_equity != 0:
add(
rule_id,
"BLOCKER",
f"{period_label} income-statement result does not reconcile to equity",
)
if case.get("statutory_presentation_required", True):
mapping_index = case.get("taxonomy_mapping_index")
if not mapping_index:
add(
"MAPPING.TAXONOMY_INDEX_REQUIRED",
"BLOCKER",
"The selected-form official taxonomy mapping index is missing",
)
elif mapping_index.get("selected_form") != case.get(
"selected_form"
) or mapping_index.get("taxonomy_package_sha256") != case.get(
"taxonomy_checksum"
):
add(
"MAPPING.TAXONOMY_INDEX_STALE",
"BLOCKER",
"The mapping index does not match the selected form and taxonomy package",
)
presentation = case.get("statutory_presentation")
if not presentation:
add(
"STATEMENT.STATUTORY_PRESENTATION_REQUIRED",
"BLOCKER",
"Primary statutory taxonomy presentation coverage has not been reviewed",
)
elif presentation.get("status") != "COMPLETE":
summary = presentation.get("summary") or {}
add(
"STATEMENT.STATUTORY_PRESENTATION_INCOMPLETE",
"BLOCKER",
"Primary statutory presentation has "
f"{summary.get('missing_period_decisions', 0)} missing period decisions "
f"and {summary.get('issues', 0)} arithmetic issues",
)
elif (presentation.get("semantic_reconciliation") or {}).get("status") != (
"PASS"
):
add(
"STATEMENT.XBRL_RECONCILIATION_REQUIRED",
"BLOCKER",
"Canonical statement totals do not reconcile to statutory XBRL roots",
)
elif str(
(presentation.get("inventory") or {}).get("taxonomy_package_sha256", "")
) != str(case.get("taxonomy_checksum")):
add(
"STATEMENT.STATUTORY_PRESENTATION_TAXONOMY_MISMATCH",
"BLOCKER",
"Primary statutory presentation uses a different taxonomy package",
)
elif mapping_index and (presentation.get("inventory") or {}).get(
"inventory_sha256"
) != mapping_index.get("inventory_sha256"):
add(
"STATEMENT.STATUTORY_PRESENTATION_MAPPING_INDEX_MISMATCH",
"BLOCKER",
"Statutory presentation differs from the inventory used for account mapping",
)
else:
for fact in presentation.get("output_facts", []):
if fact.get("status") not in EXPORTABLE_STATUSES:
add(
"STATEMENT.STATUTORY_PRESENTATION_NON_EXPORTABLE",
"BLOCKER",
f"Presentation fact {fact.get('fact_id')} is not exportable",
)
if not fact.get("derivation") and not fact.get("confirmed_by"):
add(
"STATEMENT.STATUTORY_PRESENTATION_PROVENANCE",
"BLOCKER",
f"Presentation fact {fact.get('fact_id')} has no provenance",
)
required_schedules = required_schedule_types(case)
schedules = case.get("schedules", [])
present_schedule_types = {item["schedule_type"] for item in schedules}
duplicate_schedule_types = sorted(
schedule_type
for schedule_type in present_schedule_types
if sum(item["schedule_type"] == schedule_type for item in schedules) > 1
)
if duplicate_schedule_types:
add(
"SCHEDULE.DUPLICATE_TYPE",
"BLOCKER",
"Only one reviewed schedule is allowed per type: "
+ ", ".join(duplicate_schedule_types),
)
missing_schedules = sorted(required_schedules - present_schedule_types)
if missing_schedules:
add(
"SCHEDULE.REQUIRED",
"BLOCKER",
f"Required schedules are missing: {', '.join(missing_schedules)}",
)
incomplete_schedules = [
item["schedule_id"] for item in schedules if item.get("status") != "COMPLETE"
]
if incomplete_schedules:
add(
"SCHEDULE.RECONCILIATION",
"BLOCKER",
f"Schedules have unresolved arithmetic issues: {', '.join(incomplete_schedules)}",
)
non_cash_schedules = [
item for item in schedules if item.get("schedule_type") != "CASH_FLOW"
]
if non_cash_schedules and case.get("taxonomy_mapping_index") is not None:
adapter = case.get("schedule_taxonomy_adapter")
expected_types = sorted(
str(item["schedule_type"]) for item in non_cash_schedules
)
covered_types = sorted(
str(item.get("schedule_type"))
for item in (adapter or {}).get("coverage", [])
if item.get("status") == "COMPLETE"
)
if not adapter or adapter.get("status") != "COMPLETE":
add(
"SCHEDULE.TAXONOMY_ADAPTER_REQUIRED",
"BLOCKER",
"Every reviewed non-cash schedule requires official taxonomy-table coverage",
)
elif covered_types != expected_types:
add(
"SCHEDULE.TAXONOMY_ADAPTER_INCOMPLETE",
"BLOCKER",
"Schedule taxonomy coverage differs from the current schedules",
)
elif str(
(adapter.get("inventory") or {}).get("taxonomy_package_sha256")
) != str(case.get("taxonomy_checksum")):
add(
"SCHEDULE.TAXONOMY_ADAPTER_TAXONOMY_MISMATCH",
"BLOCKER",
"Schedule taxonomy coverage uses a different taxonomy package",
)
elif case.get("schedule_taxonomy_facts") != adapter.get("generated_facts"):
add(
"SCHEDULE.TAXONOMY_ADAPTER_FACT_MISMATCH",
"BLOCKER",
"Schedule taxonomy facts differ from their reviewed adapter result",
)
if case.get("selected_form") == "ORDINARY" and case.get(
"statutory_presentation_required", True
):
cash_schedules = [
item for item in schedules if item.get("schedule_type") == "CASH_FLOW"
]
cash_flow_values = (case.get("statutory_presentation") or {}).get(
"cash_flow_values"
)
if len(cash_schedules) == 1 and cash_schedules[0].get("status") == "COMPLETE":
schedule = cash_schedules[0]
schedule_change = Decimal(str(schedule["closing_cash"])) - Decimal(
str(schedule["opening_cash"])
)
xbrl_change = (
None
if not isinstance(cash_flow_values, Mapping)
or cash_flow_values.get("current_value") is None
else Decimal(str(cash_flow_values["current_value"]))
)
if xbrl_change is None or xbrl_change != schedule_change:
add(
"CASH_FLOW.XBRL_NET_CHANGE_RECONCILIATION",
"BLOCKER",
"The reviewed cash-flow net change does not reconcile to the statutory XBRL cash-flow root",
)
for fact in case.get("canonical_facts", []):
if fact["status"] not in EXPORTABLE_STATUSES:
add(
"FACT.NON_EXPORTABLE_STATUS",
"BLOCKER",
f"Fact {fact['fact_id']} has non-exportable status {fact['status']}",
)
if not fact.get("source_refs") and not fact.get("derivation"):
add(
"FACT.PROVENANCE_REQUIRED",
"BLOCKER",
f"Fact {fact['fact_id']} has no provenance",
)
if fact.get("xbrl_concept") and str(
fact.get("xbrl_sign_multiplier", "")
) not in {"1", "-1"}:
add(
"FACT.XBRL_SIGN_REQUIRED",
"BLOCKER",
f"Fact {fact['fact_id']} has no reviewed XBRL sign convention",
)
for fact in [
*case.get("taxonomy_facts", []),
*case.get("schedule_taxonomy_facts", []),
]:
if fact.get("status") not in EXPORTABLE_STATUSES:
add(
"FACT.TAXONOMY_NON_EXPORTABLE_STATUS",
"BLOCKER",
f"Taxonomy fact {fact['fact_id']} is not reviewed",
)
if (
fact.get("fact_type") != "NIL"
and not fact.get("source_refs")
and not fact.get("derivation")
):
add(
"FACT.TAXONOMY_PROVENANCE_REQUIRED",
"BLOCKER",
f"Taxonomy fact {fact['fact_id']} has no provenance",
)
if fact.get("fact_type") == "NIL" and not fact.get("nil_reason"):
add(
"FACT.NIL_REASON_REQUIRED",
"BLOCKER",
f"Nil taxonomy fact {fact['fact_id']} has no reviewed reason",
)
narrative_qnames: set[str] = set()
for block in case.get("narrative_blocks", []):
if block.get("status") != "ACCEPTED":
continue
qname = str(block.get("xbrl_concept") or "")
if not qname:
add(
"NARRATIVE.XBRL_CONCEPT_REQUIRED",
"BLOCKER",
f"Accepted narrative block {block.get('block_id')} has no XBRL concept",
)
elif qname in narrative_qnames:
add(
"NARRATIVE.XBRL_CONCEPT_DUPLICATE",
"BLOCKER",
f"Accepted narrative concept is duplicated: {qname}",
)
else:
narrative_qnames.add(qname)
accepted_answers = {
answer["key"]
for answer in case.get("disclosure_answers", [])
if disclosure_answer_complete(answer)
}
missing_confirmations = sorted(ANNUAL_NEGATIVE_CONFIRMATION_KEYS - accepted_answers)
if missing_confirmations:
add(
"DISCLOSURE.NEGATIVE_CONFIRMATIONS",
"BLOCKER",
f"{len(missing_confirmations)} annual negative confirmations are missing",
)
double_format_answer = next(
(
answer
for answer in case.get("disclosure_answers", [])
if answer.get("key") == "double_format_events"
and disclosure_answer_complete(answer)
),
None,
)
double_value = (double_format_answer or {}).get("value")
mismatch_declared = double_value is True or (
isinstance(double_value, Mapping)
and double_value.get("differences_present") is True
)
representation = case.get("taxonomy_representation")
if mismatch_declared and not representation:
add(
"XBRL.SUBSTANTIVE_TAXONOMY_MISMATCH",
"BLOCKER",
"Substantive taxonomy differences require a reviewer-owned differences report and treatment",
)
elif mismatch_declared and representation.get("mismatch_present") is not True:
add(
"XBRL.TAXONOMY_MISMATCH_CONTRADICTION",
"BLOCKER",
"The disclosure answer and taxonomy representation review disagree",
)
elif representation and representation.get("mismatch_present") is True:
if not mismatch_declared:
add(
"XBRL.TAXONOMY_MISMATCH_CONTRADICTION",
"BLOCKER",
"The taxonomy differences report conflicts with the annual confirmation",
)
else:
add(
"XBRL.SUBSTANTIVE_TAXONOMY_MISMATCH_REVIEWED",
"MEDIUM",
"Substantive taxonomy differences remain and the reviewer-selected treatment must be understood before approval",
)
disclosure_coverage = case.get("disclosure_coverage")
if not case.get("disclosure_rule_pack") or not disclosure_coverage:
add(
"DISCLOSURE.RULE_PACK_REQUIRED",
"BLOCKER",
"An effective-dated disclosure rule pack has not been activated",
)
else:
reviewed_manual_flags = {
str(item["flag"]) for item in case.get("disclosure_trigger_decisions", [])
}
missing_manual_flags = sorted(
manual_disclosure_flags(case["disclosure_rule_pack"])
- reviewed_manual_flags
)
if missing_manual_flags:
add(
"DISCLOSURE.MANUAL_TRIGGER_REVIEW_REQUIRED",
"BLOCKER",
"Disclosure applicability requires professional decisions for: "
+ ", ".join(missing_manual_flags),
)
for item in disclosure_coverage["coverage"]:
if item["triggered"] and not item["complete"]:
missing = [
f"{requirement['kind']}:{requirement['key']}"
for requirement in item["requirements"]
if not requirement["complete"]
]
add(
f"DISCLOSURE.{item['rule_id']}",
item["severity_if_missing"],
f"Triggered disclosure is incomplete: {', '.join(missing)}",
)
preview = case.get("preview")
if not preview:
add("REVIEW.PREVIEW_REQUIRED", "BLOCKER", "A review preview is required")
elif preview.get("review_content_hash") != _review_content_hash(case):
add(
"REVIEW.PREVIEW_STALE",
"BLOCKER",
"The review preview does not match the current substantive case content",
)
else:
try:
_reviewed_preview_bytes(preview)
except ValueError as exc:
add(
"REVIEW.PREVIEW_INTEGRITY",
"BLOCKER",
str(exc),
)
xbrl_review = case.get("xbrl_review")
if xbrl_review:
if xbrl_review.get("review_content_hash") != _review_content_hash(case):
add(
"XBRL.LOCAL_REVIEW_STALE",
"BLOCKER",
"The locally validated XBRL does not match the current case content",
)
elif xbrl_review.get("status") != "PASS":
add(
"XBRL.LOCAL_VALIDATION_FAILED",
"BLOCKER",
"The pre-approval XBRL instance failed local processor validation",
)
issue_payloads = [issue.as_dict() for issue in issues]
unresolved_high = 0
review_required = 0
overrides = 0
for issue in issue_payloads:
decision = _current_review_decision(case, issue)
action = str((decision or {}).get("action", ""))
if decision:
issue["review_decision_id"] = decision["decision_id"]
issue["review_status"] = action
issue["reviewed_by"] = decision["reviewed_by"]
else:
issue["review_status"] = "UNREVIEWED"
if issue["severity"] == "HIGH":
if action == "OVERRIDDEN":
overrides += 1
else:
unresolved_high += 1
elif issue["severity"] in {"MEDIUM", "LOW", "INFO"}:
if action != "ACKNOWLEDGED":
review_required += 1
blockers = sum(issue.severity == "BLOCKER" for issue in issues)
highs = sum(issue.severity == "HIGH" for issue in issues)
status = "PASS" if blockers == 0 and unresolved_high == 0 else "FAIL"
return {
"status": status,
"layers": ["INPUT", "ACCOUNTING", "DISCLOSURE", "XBRL", "REVIEW_POLICY"],
"blockers": blockers,
"high": highs,
"unresolved_high": unresolved_high,
"review_required": review_required,
"approved_overrides": overrides,
"issues": issue_payloads,
"prior_xbrl_reconciliation": prior_reconciliation,
"validated_revision_id": case["revision_id"],
"input_manifest_hash": _manifest_hash(case),
"validated_at": _now(),
"computation_context": _computation_context(case, "bilancio-validation-v1"),
}
def record_issue_reviews(
case: dict[str, Any],
decisions: Sequence[Mapping[str, Any]],
reviewer: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record reviewer-owned acknowledgements and permitted HIGH overrides."""
_ensure_revision(case, expected_revision)
validation = case.get("validation") or {}
if validation.get("validated_revision_id") != case["revision_id"]:
raise ValueError("Issue reviews require validation of the current revision")
current_issues = {
str(issue["issue_id"]): issue for issue in validation.get("issues", [])
}
if not decisions:
raise ValueError("At least one issue review decision is required")
normalized: list[dict[str, Any]] = []
seen: set[str] = set()
for raw in decisions:
issue_id = str(raw["issue_id"])
if issue_id in seen:
raise ValueError("Each issue may be reviewed only once per request")
seen.add(issue_id)
issue = current_issues.get(issue_id)
if issue is None:
raise ValueError(f"Unknown current validation issue: {issue_id}")
action = str(raw["action"]).upper()
severity = str(issue["severity"])
reason = str(raw.get("reason", "")).strip()
if severity == "BLOCKER" or not issue.get("override_allowed"):
raise ValueError("Structural and blocking issues cannot be overridden")
if action == "OVERRIDDEN":
if severity != "HIGH":
raise ValueError("Only HIGH issues may be professionally overridden")
if len(reason) < 10:
raise ValueError("A professional override requires a specific reason")
elif action == "ACKNOWLEDGED":
if severity not in {"MEDIUM", "LOW", "INFO"}:
raise ValueError("Only non-blocking warnings may be acknowledged")
if not reason:
raise ValueError("A warning acknowledgement requires a review note")
else:
raise ValueError(f"Unsupported issue review action: {action}")
normalized.append(
{
"issue_id": issue_id,
"issue_fingerprint": str(
issue.get("fingerprint") or _issue_fingerprint(issue)
),
"rule_id": issue["rule_id"],
"severity": severity,
"action": action,
"reason": reason,
"reviewed_by": reviewer,
"reviewed_at": _now(),
}
)
_mutate(case, reviewer, "issue_reviews_recorded")
for decision in normalized:
decision["decision_id"] = (
f"decision_{len(case.get('review_decisions', [])) + 1:06d}"
)
case.setdefault("review_decisions", []).append(decision)
current_issues[decision["issue_id"]]["review_decision_id"] = decision[
"decision_id"
]
current_issues[decision["issue_id"]]["review_status"] = decision["action"]
current_issues[decision["issue_id"]]["reviewed_by"] = reviewer
for decision in normalized:
if decision["action"] == "OVERRIDDEN":
_record_event(case, "override_created", reviewer, decision)
_record_event(case, "override_approved", reviewer, decision)
else:
_record_event(case, "issue_acknowledged", reviewer, decision)
_record_event(
case,
"issue_resolved",
reviewer,
{
"issue_id": decision["issue_id"],
"decision_id": decision["decision_id"],
"resolution": decision["action"],
},
)
_record_event(
case,
"issue_reviews_recorded",
reviewer,
{"decision_ids": [item["decision_id"] for item in normalized]},
)
return case
def record_disclosure_answers(
case: dict[str, Any],
answers: Sequence[Mapping[str, Any]],
actor: str,
expected_revision: str,
) -> dict[str, Any]:
"""Record structured, evidenced questionnaire answers without inventing zeros."""
_ensure_revision(case, expected_revision)
normalized: list[dict[str, Any]] = []
seen: set[str] = set()
allowed_statuses = {
"OPEN",
"ASSIGNED",
"ANSWERED_UNREVIEWED",
"ACCEPTED",
"REJECTED",
"NOT_APPLICABLE_CONFIRMED",
}
allowed_keys = {
str(question["answer_key"]) for question in case.get("questionnaire", [])
} | ANNUAL_NEGATIVE_CONFIRMATION_KEYS
for answer in answers:
key = str(answer["key"])
if not key or key in seen:
raise ValueError("Disclosure answer keys must be present and unique")
if key not in allowed_keys:
raise ValueError(f"Disclosure answer key is not active or annual: {key}")
seen.add(key)
status = str(answer["status"]).upper()
if status not in allowed_statuses:
raise ValueError(f"Unsupported disclosure answer status: {status}")
if status in {"ACCEPTED", "NOT_APPLICABLE_CONFIRMED"}:
if not actor.strip():
raise ValueError(
"Accepted disclosure answers require an authenticated actor"
)
claimed_actor = answer.get("confirmed_by")
if claimed_actor is not None and str(claimed_actor) != actor:
raise ValueError(
"Disclosure confirmation identity must match the authenticated actor"
)
terminal_answer = {
**dict(answer),
"status": status,
"confirmed_by": actor,
}
if status == "ACCEPTED" and not disclosure_answer_complete(terminal_answer):
raise ValueError(
"Accepted disclosure answers require a reviewed structured value"
)
if status == "NOT_APPLICABLE_CONFIRMED" and not disclosure_answer_complete(
terminal_answer
):
raise ValueError(
"Not-applicable disclosure confirmations require a specific reason"
)
if status == "ASSIGNED" and not str(answer.get("owner", "")).strip():
raise ValueError("Assigned disclosure answers require an owner")
source_refs = sorted(
{
str(ref).strip()
for ref in answer.get("source_refs", [])
if str(ref).strip()
}
)
if source_refs and not set(source_refs) <= _available_evidence_refs(case):
raise ValueError("Disclosure answer references evidence outside the case")
normalized.append(
{
"key": key,
"status": status,
"value": answer.get("value"),
"source_refs": source_refs,
"reason": str(answer.get("reason", "")),
"owner": str(answer.get("owner", "")) or None,
"confirmed_by": (
actor
if status in {"ACCEPTED", "NOT_APPLICABLE_CONFIRMED"}
else None
),
"confirmed_at": (
_now()
if status in {"ACCEPTED", "NOT_APPLICABLE_CONFIRMED"}
else None
),
}
)
existing = {
str(answer["key"]): answer for answer in case.get("disclosure_answers", [])
}
for answer in normalized:
existing[answer["key"]] = answer
_mutate(case, actor, "questionnaire_answers_recorded")
_clear_narrative_reviews(case)
case["disclosure_answers"] = list(existing.values())
case["validation"] = None
_refresh_disclosures(case)
_record_event(case, "question_answered", actor, {"answer_count": len(normalized)})
return case
def run_validation(
case: dict[str, Any], actor: str, expected_revision: str
) -> dict[str, Any]:
"""Persist a validation run and move the case to the corresponding gate."""
_ensure_revision(case, expected_revision)
_mutate(case, actor, "validation_run")
result = validate_case(case)
case["validation"] = result
case["state"] = (
CaseState.READY_FOR_REVIEW
if result["status"] == "PASS"
else CaseState.VALIDATION_FAILED
)
migrations = case.get("regulatory_migrations", [])
if migrations:
latest_migration = migrations[-1]
latest_migration.setdefault("revalidation_runs", []).append(
{
"revision_id": case["revision_id"],
"result": result["status"],
"completed_at": _now(),
}
)
latest_migration["revalidation_status"] = (
"PASSED" if result["status"] == "PASS" else "FAILED"
)
_record_event(case, "validation_run", actor, {"status": result["status"]})
return case
def prepare_xbrl_review(
case: dict[str, Any],
catalogue_path: Path,
taxonomy_package: Path,
output_dir: Path,
actor: str,
expected_revision: str,
validator: (
Callable[[Path, Path, Path | None, str | None], dict[str, object]] | None
) = None,
) -> dict[str, Any]:
"""Render and locally validate the current case before reviewer approval."""
_ensure_revision(case, expected_revision)
validation = case.get("validation") or {}
if (
case.get("state") != CaseState.READY_FOR_REVIEW
or validation.get("status") != "PASS"
or validation.get("validated_revision_id") != case["revision_id"]
):
raise ValueError(
"Pre-approval XBRL review requires current passing case validation"
)
output = _controlled_output_path(output_dir, "XBRL review output directory")
_reject_symlink_path_components(catalogue_path, "Taxonomy catalogue path")
if not catalogue_path.is_file():
raise ValueError("Taxonomy catalogue must be a regular local file")
if taxonomy_package is not None:
_reject_symlink_path_components(taxonomy_package, "Taxonomy package path")
if not taxonomy_package.is_file():
raise ValueError("Taxonomy package must be a regular local file")
catalogue_bytes = catalogue_path.read_bytes()
content_hash = _review_content_hash(case)
snapshot = _case_payload_for_hash(case)
snapshot_hash = _sha256_bytes(_canonical_json(snapshot))
candidate = {
"state": CaseState.APPROVED,
"approval": {"snapshot": snapshot, "snapshot_hash": snapshot_hash},
}
xml = render_xbrl(candidate, catalogue_path, catalogue_bytes=catalogue_bytes)
candidate_sha256 = _sha256_bytes(xml)
selected_validator = validator or validate_instance
with tempfile.TemporaryDirectory(
dir=output.parent, prefix=f".{output.name}.staging-"
) as temporary:
staging = Path(temporary)
candidate_path = staging / "review-candidate.xbrl"
candidate_path.write_bytes(xml)
report_path = staging / "local-xbrl-validation.json"
result = selected_validator(
candidate_path,
report_path,
taxonomy_package,
case.get("taxonomy_checksum"),
)
if not report_path.is_file():
report_path.write_text(
json.dumps(result, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
if _sha256_file(candidate_path) != candidate_sha256:
raise ValueError(
"Local validation modified the rendered XBRL review candidate"
)
status = str(result.get("status", "FAIL"))
if status not in {"PASS", "FAIL"}:
raise ValueError("Local XBRL validator returned an unsupported status")
_publish_output_directory(staging, output, "XBRL review output directory")
candidate_path = output / "review-candidate.xbrl"
report_path = output / "local-xbrl-validation.json"
_mutate(case, actor, "xbrl_review_prepared")
case["xbrl_review"] = {
"review_content_hash": content_hash,
"candidate_snapshot_hash": snapshot_hash,
"candidate_file_name": candidate_path.name,
"candidate_sha256": candidate_sha256,
"validation_report_file_name": report_path.name,
"validation_report_sha256": _sha256_file(report_path),
"catalogue_sha256": _sha256_bytes(catalogue_bytes),
"taxonomy_package_sha256": result.get("taxonomy_package_sha256"),
"processor": result.get("processor", "injected-validator"),
"status": status,
"prepared_by": actor,
"prepared_at": _now(),
"computation_context": _computation_context(case, "local-xbrl-review-v1"),
}
current_validation = validate_case(case)
case["validation"] = current_validation
case["state"] = (
CaseState.READY_FOR_REVIEW
if status == "PASS" and current_validation["status"] == "PASS"
else CaseState.VALIDATION_FAILED
)
_record_event(
case,
"xbrl_review_prepared",
actor,
{
"status": status,
"candidate_sha256": case["xbrl_review"]["candidate_sha256"],
"validation_report_sha256": case["xbrl_review"]["validation_report_sha256"],
},
)
return case
def approve_case(
case: dict[str, Any],
reviewer: str,
expected_revision: str,
declaration: Mapping[str, Any],
) -> dict[str, Any]:
"""Create an immutable approval snapshot only from the validated current revision."""
_ensure_revision(case, expected_revision)
before_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
validation = case.get("validation") or {}
if (
case["state"] != CaseState.READY_FOR_REVIEW
or validation.get("status") != "PASS"
or validation.get("validated_revision_id") != case["revision_id"]
):
raise ValueError("Only a fully validated READY_FOR_REVIEW case can be approved")
migrations = case.get("regulatory_migrations", [])
if migrations and migrations[-1].get("revalidation_status") != "PASSED":
raise ValueError("Regulatory migration requires a passing full revalidation")
xbrl_review = case.get("xbrl_review") or {}
if xbrl_review.get("status") != "PASS" or xbrl_review.get(
"review_content_hash"
) != _review_content_hash(case):
raise ValueError(
"Approval requires a passing local XBRL review of the current content"
)
required = {
"entity_period_confirmed",
"form_confirmed",
"evidence_reviewed",
"preview_reviewed",
"filing_boundary_understood",
"rendered_output_confirmed",
"outstanding_warnings_understood",
}
if any(declaration.get(key) is not True for key in required):
raise ValueError("The reviewer declaration is incomplete")
if int(validation.get("review_required", 0)):
raise ValueError("All current warnings must be reviewed before approval")
if (
int(validation.get("approved_overrides", 0))
and declaration.get("overrides_reviewed") is not True
):
raise ValueError("The approval must explicitly confirm professional overrides")
snapshot_payload = _case_payload_for_hash(case)
snapshot_hash = _sha256_bytes(_canonical_json(snapshot_payload))
case["approval"] = {
"snapshot_id": f"snap_{len(case.get('approval_snapshots', [])) + 1:04d}",
"revision_id": case["revision_id"],
"snapshot_hash": snapshot_hash,
"input_manifest_hash": _manifest_hash(case),
"approved_by": reviewer,
"approved_at": _now(),
"declaration": dict(declaration),
"snapshot": snapshot_payload,
"audit_events": deepcopy(case.get("audit_events", [])),
}
case["state"] = CaseState.APPROVED
case["updated_at"] = _now()
case["_pending_before_hash"] = before_hash
_record_event(case, "snapshot_approved", reviewer, {"snapshot_hash": snapshot_hash})
return case
def _taxonomy_catalogue(
path: Path,
expected_id: str,
expected_checksum: str | None,
*,
content: bytes | None = None,
) -> dict[str, Any]:
if content is None:
if path.is_symlink() or not path.is_file():
raise ValueError("Taxonomy catalogue must be a regular local file")
content = path.read_bytes()
try:
catalogue = json.loads(content.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError("Taxonomy catalogue must be valid UTF-8 JSON") from exc
if catalogue.get("schema_version") != 2:
raise ValueError("Taxonomy catalogue schema version 2 is required")
if catalogue.get("taxonomy_id") != expected_id:
raise ValueError("Taxonomy catalogue identifier does not match the locked case")
package_checksum = catalogue.get("taxonomy_package_sha256")
if not package_checksum or package_checksum == "UNVERIFIED":
raise ValueError(
"Taxonomy catalogue is not bound to a verified official package checksum"
)
if expected_checksum and package_checksum != expected_checksum:
raise ValueError("Taxonomy package checksum does not match the locked case")
concepts = catalogue.get("concepts")
if not isinstance(concepts, list) or not concepts:
raise ValueError("Taxonomy catalogue has no concepts")
qnames: set[str] = set()
for concept in concepts:
if not isinstance(concept, Mapping) or not str(concept.get("qname", "")):
raise ValueError("Taxonomy catalogue contains an invalid concept record")
qname = str(concept["qname"])
if qname in qnames:
raise ValueError(f"Taxonomy catalogue repeats concept {qname}")
qnames.add(qname)
if not isinstance(concept.get("is_item"), bool) or not isinstance(
concept.get("is_tuple"), bool
):
raise ValueError(
f"Taxonomy concept lacks item-versus-tuple metadata: {qname}"
)
if concept["is_item"] is True and concept["is_tuple"] is True:
raise ValueError(f"Taxonomy concept cannot be both item and tuple: {qname}")
return catalogue
def _is_reportable_item(concept: Mapping[str, Any] | None) -> bool:
"""Return whether one taxonomy concept can be emitted as an item fact."""
return bool(
concept
and concept.get("abstract") is not True
and concept.get("is_item") is True
and concept.get("is_tuple") is False
and concept.get("period_type") in {"instant", "duration"}
)
def _taxonomy_presentation_order(
catalogue: Mapping[str, Any],
selected_form: str,
presentation: Mapping[str, Any],
) -> dict[str, int]:
"""Return stable concept ordinals from the selected official presentation roles."""
role_order = [
str(item["role"])
for item in (presentation.get("inventory") or {}).get("roles", [])
]
rows = list((catalogue.get("relationships") or {}).get("presentation", []))
order: dict[str, int] = {}
ordinal = 0
for role in role_order:
role_rows = [
row
for row in rows
if str(row.get("form")) == selected_form and str(row.get("role")) == role
]
children: dict[str, list[Mapping[str, Any]]] = {}
parents: set[str] = set()
targets: set[str] = set()
for row in role_rows:
parent = str(row["from"])
parents.add(parent)
targets.add(str(row["to"]))
children.setdefault(parent, []).append(row)
roots = sorted(parents - targets)
visited: set[str] = set()
def visit(qname: str) -> None:
nonlocal ordinal
if qname in visited:
return
visited.add(qname)
if qname not in order:
order[qname] = ordinal
ordinal += 1
for row in sorted(
children.get(qname, []),
key=lambda item: (
Decimal(str(item.get("order") or "0")),
str(item["to"]),
),
):
visit(str(row["to"]))
for root_qname in roots:
visit(root_qname)
return order
def render_xbrl(
case: Mapping[str, Any],
catalogue_path: Path,
*,
catalogue_bytes: bytes | None = None,
) -> bytes:
"""Render an approved snapshot into deterministic XBRL XML."""
approval = case.get("approval")
if not approval or case.get("state") not in {
CaseState.APPROVED,
CaseState.EXPORTED,
}:
raise ValueError("XBRL rendering requires an approved snapshot")
snapshot = approval["snapshot"]
expected_hash = _sha256_bytes(_canonical_json(snapshot))
if expected_hash != approval["snapshot_hash"]:
raise ValueError("Approval snapshot hash is invalid")
catalogue = _taxonomy_catalogue(
catalogue_path,
snapshot["rule_pack_versions"]["taxonomy_id"],
snapshot.get("taxonomy_checksum"),
content=catalogue_bytes,
)
concept_lookup = {item["qname"]: item for item in catalogue["concepts"]}
selected_form = snapshot.get("selected_form")
presentation = snapshot.get("statutory_presentation") or {}
presentation_order = _taxonomy_presentation_order(
catalogue, str(selected_form), presentation
)
def fact_sort_key(fact: Mapping[str, Any]) -> tuple[int, str, str]:
qname = str(fact.get("xbrl_concept") or "")
return (
presentation_order.get(qname, len(presentation_order)),
qname,
str(fact.get("fact_id") or fact.get("block_id") or ""),
)
namespaces = {
"xbrli": XBRLI_NS,
"xbrldi": XBRLDI_NS,
"xsi": XSI_NS,
"link": LINK_NS,
"xlink": XLINK_NS,
"iso4217": ISO4217_NS,
**catalogue["namespaces"],
}
root = etree.Element(etree.QName(XBRLI_NS, "xbrl"), nsmap=namespaces)
schema_ref = etree.SubElement(root, etree.QName(LINK_NS, "schemaRef"))
schema_ref.set(etree.QName(XLINK_NS, "type"), "simple")
entry_point = catalogue.get("entry_points", {}).get(selected_form)
if not entry_point:
raise ValueError(f"Taxonomy catalogue has no entry point for {selected_form}")
schema_ref.set(etree.QName(XLINK_NS, "href"), entry_point)
entity_identifier = str(snapshot["entity"]["tax_identifier"])
period_end = snapshot["period"]["end"]
period_start = snapshot["period"]["start"]
contexts = {
"current_instant": (None, period_end),
"current_duration": (period_start, period_end),
}
first_year = snapshot["entity"].get("first_financial_year") is True
if not first_year:
prior_end = str(snapshot["entity"].get("prior_period_end") or "")
prior_start = str(snapshot["entity"].get("prior_period_start") or "")
if not prior_start or not prior_end:
raise ValueError(
"Approved comparative contexts require explicit start and end dates"
)
if date.fromisoformat(prior_start) > date.fromisoformat(prior_end):
raise ValueError("Approved comparative period start is after its end")
contexts.update(
{
"prior_instant": (None, prior_end),
"prior_duration": (prior_start, prior_end),
}
)
for context_id, (start, end) in contexts.items():
context = etree.SubElement(
root, etree.QName(XBRLI_NS, "context"), id=context_id
)
entity = etree.SubElement(context, etree.QName(XBRLI_NS, "entity"))
identifier = etree.SubElement(
entity,
etree.QName(XBRLI_NS, "identifier"),
scheme="http://www.registroimprese.it",
)
identifier.text = entity_identifier
period = etree.SubElement(context, etree.QName(XBRLI_NS, "period"))
if start is None:
etree.SubElement(period, etree.QName(XBRLI_NS, "instant")).text = end
else:
etree.SubElement(period, etree.QName(XBRLI_NS, "startDate")).text = start
etree.SubElement(period, etree.QName(XBRLI_NS, "endDate")).text = end
dimension_contexts: dict[tuple[str, tuple[tuple[str, str], ...]], str] = {}
for fact in [
*snapshot.get("taxonomy_facts", []),
*snapshot.get("schedule_taxonomy_facts", []),
]:
dimensions = tuple(sorted(dict(fact.get("dimensions", {})).items()))
if not dimensions:
continue
period_key = str(fact["period"])
if period_key not in contexts:
raise ValueError(f"Unsupported dimensional fact period: {period_key}")
key = (period_key, dimensions)
if key in dimension_contexts:
continue
for axis, member in dimensions:
if axis not in concept_lookup or member not in concept_lookup:
raise ValueError(f"Unknown dimension axis or member: {axis}={member}")
axis_concept = concept_lookup[axis]
member_concept = concept_lookup[member]
if axis_concept.get("is_dimension_item") is not True:
raise ValueError(f"Dimension axis is not a dimension item: {axis}")
if (
member_concept.get("is_item") is not True
or member_concept.get("is_tuple") is not False
or member_concept.get("is_dimension_item") is True
or member_concept.get("is_hypercube_item") is True
):
raise ValueError(f"Dimension member is not a domain item: {member}")
if axis.split(":", 1)[0] not in namespaces:
raise ValueError(f"Missing namespace for dimension axis: {axis}")
if member.split(":", 1)[0] not in namespaces:
raise ValueError(f"Missing namespace for dimension member: {member}")
context_id = f"ctx_dim_{len(dimension_contexts) + 1:04d}"
dimension_contexts[key] = context_id
start, end = contexts[period_key]
context = etree.SubElement(
root, etree.QName(XBRLI_NS, "context"), id=context_id
)
entity = etree.SubElement(context, etree.QName(XBRLI_NS, "entity"))
identifier = etree.SubElement(
entity,
etree.QName(XBRLI_NS, "identifier"),
scheme="http://www.registroimprese.it",
)
identifier.text = entity_identifier
period = etree.SubElement(context, etree.QName(XBRLI_NS, "period"))
if start is None:
etree.SubElement(period, etree.QName(XBRLI_NS, "instant")).text = end
else:
etree.SubElement(period, etree.QName(XBRLI_NS, "startDate")).text = start
etree.SubElement(period, etree.QName(XBRLI_NS, "endDate")).text = end
scenario = etree.SubElement(context, etree.QName(XBRLI_NS, "scenario"))
for axis, member in dimensions:
explicit = etree.SubElement(
scenario, etree.QName(XBRLDI_NS, "explicitMember")
)
explicit.set("dimension", axis)
explicit.text = member
unit = etree.SubElement(root, etree.QName(XBRLI_NS, "unit"), id="EUR")
etree.SubElement(unit, etree.QName(XBRLI_NS, "measure")).text = "iso4217:EUR"
precision = int(snapshot.get("reporting_precision", 0))
decimals = str(precision)
output_language = str(snapshot.get("output_language", "it"))
if output_language not in {"it", "en"}:
raise ValueError("Approved snapshot output language must be it or en")
seen: set[tuple[str, str, str]] = set()
tuple_containers: dict[tuple[tuple[str, ...], str], etree._Element] = {}
def tuple_parent(tuple_path: list[str], instance_id: str) -> etree._Element:
"""Create or reuse one deterministic nested tuple occurrence."""
parent = root
for depth, tuple_qname in enumerate(tuple_path, start=1):
path_key = tuple(tuple_path[:depth])
key = (path_key, instance_id)
if key in tuple_containers:
parent = tuple_containers[key]
continue
tuple_concept = concept_lookup.get(tuple_qname)
if (
not tuple_concept
or tuple_concept.get("is_tuple") is not True
or tuple_concept.get("is_item") is not False
or tuple_concept.get("abstract") is True
):
raise ValueError(
f"Tuple path contains a non-tuple concept: {tuple_qname}"
)
forms = tuple_concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Tuple concept {tuple_qname} is not available for {selected_form}"
)
prefix, local_name = tuple_qname.split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for tuple concept: {tuple_qname}")
container = etree.SubElement(parent, etree.QName(namespace, local_name))
tuple_digest = _sha256_bytes(
_canonical_json({"path": path_key, "instance": instance_id})
)[:20]
container.set("id", _xbrl_fact_id("tuple", tuple_digest, depth))
tuple_containers[key] = container
parent = container
return parent
for fact in sorted(snapshot.get("canonical_facts", []), key=fact_sort_key):
qname = fact.get("xbrl_concept")
if not qname:
continue
if fact["status"] not in EXPORTABLE_STATUSES:
raise ValueError(f"Fact {fact['fact_id']} has a non-exportable status")
if first_year and fact.get("prior_value") is not None:
raise ValueError(
"First-financial-year canonical facts cannot contain comparative values"
)
sign_multiplier = str(fact.get("xbrl_sign_multiplier", ""))
if sign_multiplier not in {"1", "-1"}:
raise ValueError(
f"Fact {fact['fact_id']} has no reviewed XBRL sign convention"
)
concept = concept_lookup.get(qname)
if not _is_reportable_item(concept):
raise ValueError(
f"Unknown, abstract, or non-item taxonomy concept: {qname}"
)
if "monetaryItemType" not in str(concept.get("type", "")):
raise ValueError(f"Statement fact concept is not monetary: {qname}")
forms = concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Taxonomy concept {qname} is not available for {selected_form}"
)
prefix, local_name = qname.split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for concept: {qname}")
current_context = (
"current_instant"
if concept["period_type"] == "instant"
else "current_duration"
)
prior_context = (
"prior_instant" if concept["period_type"] == "instant" else "prior_duration"
)
period_values = [(current_context, fact["current_value"])]
if not first_year:
period_values.append((prior_context, fact["prior_value"]))
for context_ref, value in period_values:
duplicate_key = (qname, context_ref, "")
if duplicate_key in seen:
raise ValueError(
f"Conflicting duplicate fact: {qname} in {context_ref}"
)
seen.add(duplicate_key)
element = etree.SubElement(root, etree.QName(namespace, local_name))
element.set(
"id",
_xbrl_fact_id("canonical", fact["fact_id"], context_ref),
)
element.set("contextRef", context_ref)
element.set("unitRef", "EUR")
element.set("decimals", decimals)
element.text = _decimal_text(
_reported_decimal(
Decimal(str(value)) * Decimal(sign_multiplier), precision
)
)
if (
snapshot.get("statutory_presentation_required", True)
and presentation.get("status") != "COMPLETE"
):
raise ValueError(
"XBRL rendering requires complete primary statutory presentation coverage"
)
if (
snapshot.get("statutory_presentation_required", True)
and (presentation.get("semantic_reconciliation") or {}).get("status") != "PASS"
):
raise ValueError(
"XBRL rendering requires canonical statements to reconcile to statutory roots"
)
for presentation_index, fact in enumerate(
sorted(presentation.get("output_facts", []), key=fact_sort_key), start=1
):
if fact.get("status") not in EXPORTABLE_STATUSES:
raise ValueError("Statutory presentation fact is not exportable")
if not fact.get("derivation") and not fact.get("confirmed_by"):
raise ValueError("Statutory presentation fact has no provenance")
if first_year and fact.get("prior_value") is not None:
raise ValueError(
"First-financial-year presentation cannot contain a comparative fact"
)
qname = str(fact["xbrl_concept"])
concept = concept_lookup.get(qname)
if not _is_reportable_item(concept):
raise ValueError(
f"Unknown, abstract, or non-item presentation concept: {qname}"
)
if "monetaryItemType" not in str(concept.get("type", "")):
raise ValueError(f"Presentation fact concept is not monetary: {qname}")
forms = concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Presentation concept {qname} is not available for {selected_form}"
)
prefix, local_name = qname.split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for presentation concept: {qname}")
current_context = (
"current_instant"
if concept["period_type"] == "instant"
else "current_duration"
)
prior_context = (
"prior_instant" if concept["period_type"] == "instant" else "prior_duration"
)
for context_ref, value in (
(current_context, fact.get("current_value")),
(prior_context, fact.get("prior_value")),
):
if value is None:
continue
duplicate_key = (qname, context_ref, "")
if duplicate_key in seen:
raise ValueError(
f"Conflicting duplicate fact: {qname} in {context_ref}"
)
seen.add(duplicate_key)
element = etree.SubElement(root, etree.QName(namespace, local_name))
element.set(
"id",
_xbrl_fact_id(
"presentation",
fact.get("fact_id") or presentation_index,
context_ref,
),
)
element.set("contextRef", context_ref)
element.set("unitRef", "EUR")
element.set("decimals", decimals)
element.text = _decimal_text(
_reported_decimal(Decimal(str(value)), precision)
)
for taxonomy_index, fact in enumerate(
sorted(
[
*snapshot.get("taxonomy_facts", []),
*snapshot.get("schedule_taxonomy_facts", []),
],
key=fact_sort_key,
),
start=1,
):
qname = str(fact["xbrl_concept"])
concept = concept_lookup.get(qname)
if not _is_reportable_item(concept):
raise ValueError(
f"Unknown, abstract, or non-item taxonomy concept: {qname}"
)
forms = concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Taxonomy concept {qname} is not available for {selected_form}"
)
dimensions = tuple(sorted(dict(fact.get("dimensions", {})).items()))
period_key = str(fact["period"])
period_type = str(concept.get("period_type", ""))
if period_type not in {"instant", "duration"} or not period_key.endswith(
f"_{period_type}"
):
raise ValueError(
f"Taxonomy fact period does not match concept period type: {qname}"
)
if period_key not in contexts:
raise ValueError(
f"Taxonomy fact uses a period unavailable in the approved case: {period_key}"
)
context_ref = (
dimension_contexts[(period_key, dimensions)] if dimensions else period_key
)
tuple_path = [str(item) for item in fact.get("tuple_path", [])]
tuple_instance_id = fact.get("tuple_instance_id")
if tuple_path and not str(tuple_instance_id or "").strip():
raise ValueError("Tuple facts require a stable tuple instance identifier")
if not tuple_path and tuple_instance_id is not None:
raise ValueError("Flat facts cannot carry a tuple instance identifier")
occurrence = f"{'/'.join(tuple_path)}#{tuple_instance_id}" if tuple_path else ""
duplicate_key = (qname, context_ref, occurrence)
if duplicate_key in seen:
raise ValueError(f"Conflicting duplicate fact: {qname} in {context_ref}")
seen.add(duplicate_key)
prefix, local_name = qname.split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for concept: {qname}")
parent = (
tuple_parent(tuple_path, str(tuple_instance_id)) if tuple_path else root
)
element = etree.SubElement(parent, etree.QName(namespace, local_name))
element.set(
"id",
_xbrl_fact_id(
"taxonomy", fact.get("fact_id") or taxonomy_index, context_ref
),
)
element.set("contextRef", context_ref)
fact_type = str(fact["fact_type"])
if fact_type == "MONETARY":
if "monetaryItemType" not in str(concept.get("type", "")):
raise ValueError(f"Monetary fact uses a non-monetary concept: {qname}")
element.set("unitRef", "EUR")
element.set("decimals", decimals)
element.text = _decimal_text(
_reported_decimal(Decimal(str(fact["value"])), precision)
)
elif fact_type == "TEXT":
if "monetaryItemType" in str(concept.get("type", "")):
raise ValueError(f"Text fact uses a monetary concept: {qname}")
language = str(fact.get("language") or output_language)
if language != output_language:
raise ValueError("Text fact language differs from approved output")
element.set(etree.QName(XML_NS, "lang"), language)
element.text = str(fact["value"])
elif fact_type == "NIL":
if concept.get("nillable") is not True:
raise ValueError(f"Taxonomy concept does not permit nil: {qname}")
element.set(etree.QName(XSI_NS, "nil"), "true")
if "monetaryItemType" in str(concept.get("type", "")):
element.set("unitRef", "EUR")
element.set("decimals", decimals)
else:
raise ValueError(f"Unsupported taxonomy fact type: {fact_type}")
for narrative_index, block in enumerate(
sorted(snapshot.get("narrative_blocks", []), key=fact_sort_key), start=1
):
if block.get("status") != "ACCEPTED":
continue
qname = block.get("xbrl_concept")
if not qname:
raise ValueError("Accepted narrative requires an XBRL concept")
language = str(block.get("language") or output_language)
if language != output_language:
raise ValueError("Narrative language differs from approved output")
concept = concept_lookup.get(qname)
if not _is_reportable_item(concept):
raise ValueError(
f"Unknown, abstract, or non-item narrative concept: {qname}"
)
if "monetaryItemType" in str(concept.get("type", "")):
raise ValueError(f"Narrative block uses a monetary concept: {qname}")
forms = concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Narrative concept {qname} is not available for {selected_form}"
)
prefix, local_name = str(qname).split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for narrative concept: {qname}")
context_ref = (
"current_instant"
if concept["period_type"] == "instant"
else "current_duration"
)
duplicate_key = (str(qname), context_ref, "")
if duplicate_key in seen:
raise ValueError(f"Conflicting duplicate fact: {qname} in {context_ref}")
seen.add(duplicate_key)
element = etree.SubElement(root, etree.QName(namespace, local_name))
element.set(
"id",
_xbrl_fact_id(
"narrative", block.get("block_id") or narrative_index, context_ref
),
)
element.set("contextRef", context_ref)
element.set(etree.QName(XML_NS, "lang"), language)
element.text = str(block["text"])
micro_reporting = snapshot.get("micro_reporting") or {}
if micro_reporting.get("mode") == "FOOTER_ONLY":
qname = str(micro_reporting.get("xbrl_concept") or "")
text = str(micro_reporting.get("rendered_text") or "").strip()
if not qname or not text or not micro_reporting.get("reviewed_by"):
raise ValueError("Micro footer is not a reviewed renderable XBRL fact")
concept = concept_lookup.get(qname)
if not _is_reportable_item(concept):
raise ValueError(
f"Unknown, abstract, or non-item micro-footer concept: {qname}"
)
if "monetaryItemType" in str(concept.get("type", "")):
raise ValueError(f"Micro footer uses a monetary concept: {qname}")
forms = concept.get("forms")
if forms and selected_form not in forms:
raise ValueError(
f"Micro-footer concept {qname} is not available for {selected_form}"
)
prefix, local_name = qname.split(":", 1)
namespace = namespaces.get(prefix)
if not namespace:
raise ValueError(f"Missing namespace for micro-footer concept: {qname}")
context_ref = (
"current_instant"
if concept["period_type"] == "instant"
else "current_duration"
)
duplicate_key = (qname, context_ref, "")
if duplicate_key in seen:
raise ValueError(f"Conflicting duplicate fact: {qname} in {context_ref}")
seen.add(duplicate_key)
element = etree.SubElement(root, etree.QName(namespace, local_name))
element.set("id", _xbrl_fact_id("micro_footer", "reviewed", context_ref))
element.set("contextRef", context_ref)
element.set(etree.QName(XML_NS, "lang"), output_language)
element.text = text
xml = etree.tostring(
root, xml_declaration=True, encoding="UTF-8", pretty_print=True
)
fromstring(xml)
return xml
def _safe_slug(value: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
return slug[:64] or "societa"
def _artifact_record(path: Path) -> dict[str, Any]:
"""Return the immutable checksum record for one written export artifact."""
return {
"file_name": path.name,
"sha256": _sha256_file(path),
"size_bytes": path.stat().st_size,
}
def _mapping_report(
case: Mapping[str, Any], snapshot: Mapping[str, Any]
) -> dict[str, Any]:
"""Build the standalone approved account-mapping review report."""
entries = (snapshot.get("trial_balance") or {}).get("entries", [])
decisions = {str(item["account_id"]): item for item in snapshot.get("mappings", [])}
candidates = {
str(item["account_id"]): item for item in snapshot.get("mapping_candidates", [])
}
rows = [
{
"account_id": entry["account_id"],
"account_code": entry["account_code"],
"account_description": entry["account_description"],
"current_balance": entry["closing_signed"],
"prior_balance": entry["prior_closing_signed"],
"source_refs": list(entry.get("source_refs", [])),
"candidate": candidates.get(str(entry["account_id"])),
"decision": decisions.get(str(entry["account_id"])),
}
for entry in entries
]
mapped = [item for item in decisions.values() if item.get("decision") == "ACCEPTED"]
excluded = [
item for item in decisions.values() if item.get("decision") == "EXCLUDED"
]
return {
"schema_version": 1,
"report_type": "ACCOUNT_MAPPING",
"case_id": case["case_id"],
"snapshot_id": case["approval"]["snapshot_id"],
"snapshot_hash": case["approval"]["snapshot_hash"],
"revision_id": case["approval"]["revision_id"],
"selected_form": snapshot.get("selected_form"),
"summary": {
"total_accounts": len(entries),
"accepted_accounts": len(mapped),
"excluded_accounts": len(excluded),
"split_accounts": sum(
len(item.get("allocations", [])) > 1 for item in mapped
),
"unresolved_accounts": len(entries) - len(decisions),
},
"rows": rows,
"manual_splits": [
item for item in mapped if len(item.get("allocations", [])) > 1
],
"exclusions": excluded,
"adjustments": list(snapshot.get("adjustments", [])),
"model_suggestions": list(snapshot.get("model_mapping_suggestions", [])),
"statutory_presentation": {
"status": (snapshot.get("statutory_presentation") or {}).get("status"),
"summary": (snapshot.get("statutory_presentation") or {}).get("summary"),
"inventory_sha256": (
(snapshot.get("statutory_presentation") or {}).get("inventory") or {}
).get("inventory_sha256"),
},
"taxonomy_mapping_index": snapshot.get("taxonomy_mapping_index"),
}
def _issue_report(
case: Mapping[str, Any], snapshot: Mapping[str, Any]
) -> dict[str, Any]:
"""Build the standalone approved issues, decisions, and overrides report."""
validation = snapshot.get("validation") or {}
decisions = list(snapshot.get("review_decisions", []))
return {
"schema_version": 1,
"report_type": "CASE_ISSUES",
"case_id": case["case_id"],
"snapshot_id": case["approval"]["snapshot_id"],
"snapshot_hash": case["approval"]["snapshot_hash"],
"revision_id": case["approval"]["revision_id"],
"validation_status": validation.get("status"),
"summary": {
key: validation.get(key, 0)
for key in (
"blockers",
"high",
"medium",
"low",
"info",
"review_required",
"approved_overrides",
)
},
"issues": list(validation.get("issues", [])),
"review_decisions": decisions,
"overrides": [item for item in decisions if item.get("action") == "OVERRIDDEN"],
"unresolved_non_blocking_warnings": [
issue
for issue in validation.get("issues", [])
if issue.get("severity") in {"MEDIUM", "LOW", "INFO"}
and issue.get("review_status") == "UNREVIEWED"
],
"local_xbrl_review": snapshot.get("xbrl_review"),
"taxonomy_representation": snapshot.get("taxonomy_representation"),
"external_validation_addendum": case.get("external_validation"),
}
def export_case(
case: dict[str, Any], output_dir: Path, catalogue_path: Path, actor: str
) -> dict[str, Any]:
"""Export reproducible XBRL, workpaper, validation, and checksum artifacts."""
if case.get("state") not in {CaseState.APPROVED, CaseState.EXPORTED}:
raise ValueError("Only an approved case can be exported")
approval = case.get("approval") or {}
snapshot = approval.get("snapshot")
if not isinstance(snapshot, Mapping):
raise ValueError("The approved snapshot is missing")
if _sha256_bytes(_canonical_json(snapshot)) != approval.get("snapshot_hash"):
raise ValueError("The approved snapshot checksum does not match its receipt")
xbrl_review = snapshot.get("xbrl_review") or {}
if xbrl_review.get("status") != "PASS":
raise ValueError("Export requires the approved passing local XBRL review")
_reject_symlink_path_components(catalogue_path, "Taxonomy catalogue path")
if not catalogue_path.is_file():
raise ValueError("Taxonomy catalogue must be a regular local file")
catalogue_bytes = catalogue_path.read_bytes()
if _sha256_bytes(catalogue_bytes) != xbrl_review.get("catalogue_sha256"):
raise ValueError(
"The export taxonomy catalogue differs from the approved review catalogue"
)
xml = render_xbrl(case, catalogue_path, catalogue_bytes=catalogue_bytes)
if _sha256_bytes(xml) != xbrl_review.get("candidate_sha256"):
raise ValueError(
"The final XBRL bytes differ from the approved review candidate"
)
preview_bytes = _reviewed_preview_bytes(snapshot.get("preview") or {})
before_hash = _sha256_bytes(_canonical_json(_case_payload_for_hash(case)))
final_output = _controlled_output_path(output_dir, "Export directory")
temporary_output = tempfile.TemporaryDirectory(
dir=final_output.parent, prefix=f".{final_output.name}.staging-"
)
output = Path(temporary_output.name)
year = date.fromisoformat(snapshot["period"]["end"]).year
taxonomy_slug = (
snapshot["rule_pack_versions"]["taxonomy_id"].replace("PCI_", "pci").lower()
)
name = f"bilancio_{_safe_slug(snapshot['entity']['legal_name'])}_{year}_{taxonomy_slug}.xbrl"
xbrl_path = output / name
xbrl_path.write_bytes(xml)
additional_artifact_paths: list[Path] = []
taxonomy_representation = snapshot.get("taxonomy_representation")
if taxonomy_representation and taxonomy_representation.get("mismatch_present"):
differences_path = output / "taxonomy_differences.json"
differences_path.write_bytes(_canonical_json(taxonomy_representation) + b"\n")
additional_artifact_paths.append(differences_path)
mapping_report_path = output / "mapping_report.json"
mapping_report_path.write_bytes(
_canonical_json(_mapping_report(case, snapshot)) + b"\n"
)
issue_report_path = output / "issue_report.json"
issue_report_path.write_bytes(
_canonical_json(_issue_report(case, snapshot)) + b"\n"
)
validation_report = {
"schema_version": 1,
"report_type": "CASE_VALIDATION",
"case_id": case["case_id"],
"snapshot_id": approval["snapshot_id"],
"snapshot_hash": approval["snapshot_hash"],
"revision_id": approval["revision_id"],
"taxonomy_id": snapshot["rule_pack_versions"]["taxonomy_id"],
"taxonomy_package_sha256": snapshot.get("taxonomy_checksum"),
"case_validation": snapshot["validation"],
"local_xbrl_review": snapshot.get("xbrl_review"),
"external_validation_addendum": case.get("external_validation"),
}
validation_path = output / "validation_report.json"
validation_path.write_bytes(_canonical_json(validation_report) + b"\n")
preview_path = output / "preview.html"
preview_path.write_bytes(preview_bytes)
peer_artifact_paths = [
xbrl_path,
mapping_report_path,
issue_report_path,
validation_path,
preview_path,
*additional_artifact_paths,
]
peer_artifacts = [_artifact_record(path) for path in peer_artifact_paths]
workpaper = {
"case_id": case["case_id"],
"entity": snapshot["entity"],
"period": snapshot["period"],
"approval": {
key: value
for key, value in approval.items()
if key not in {"snapshot", "audit_events"}
},
"rule_pack_versions": snapshot["rule_pack_versions"],
"oic_professional_review_rules": snapshot.get(
"oic_professional_review_rules", []
),
"regulatory_rule_pack_checksums": {
"statutory_rule_pack": snapshot.get("rule_pack_checksum"),
"oic_rule_pack": snapshot.get("oic_rule_pack_checksum"),
"filing_instruction_pack": snapshot.get("filing_instruction_pack_checksum"),
"disclosure_rule_pack": snapshot.get("disclosure_rule_pack_checksum"),
"statutory_presentation_rule_pack": snapshot.get(
"statutory_presentation_rule_pack_checksum"
),
"schedule_taxonomy_adapter_rule_pack": snapshot.get(
"schedule_taxonomy_adapter_rule_pack_checksum"
),
},
"filing_campaign_year": snapshot.get("filing_campaign_year"),
"taxonomy_checksum": snapshot.get("taxonomy_checksum"),
"source_documents": snapshot["source_documents"],
"file_security_scans": snapshot.get("file_security_scans", []),
"prior_xbrl": snapshot.get("prior_xbrl"),
"prior_xbrl_reconciliation": _prior_xbrl_reconciliation(snapshot),
"comparative_reconciliation_decisions": snapshot.get(
"comparative_reconciliation_decisions", []
),
"parser_calibration": (snapshot.get("trial_balance") or {}).get("calibration"),
"mapping_candidates": snapshot.get("mapping_candidates", []),
"taxonomy_mapping_index": snapshot.get("taxonomy_mapping_index"),
"mappings": snapshot["mappings"],
"adjustments": snapshot.get("adjustments", []),
"taxonomy_facts": snapshot.get("taxonomy_facts", []),
"schedule_taxonomy_adapter": snapshot.get("schedule_taxonomy_adapter"),
"schedule_taxonomy_facts": snapshot.get("schedule_taxonomy_facts", []),
"taxonomy_fact_context": snapshot.get("taxonomy_fact_context"),
"taxonomy_representation": taxonomy_representation,
"micro_reporting": snapshot.get("micro_reporting"),
"form_analysis": snapshot["form_analysis"],
"statements": snapshot["statements"],
"statutory_presentation": snapshot.get("statutory_presentation"),
"schedules": snapshot.get("schedules", []),
"disclosure_coverage": snapshot.get("disclosure_coverage"),
"disclosure_trigger_decisions": snapshot.get(
"disclosure_trigger_decisions", []
),
"disclosure_activation_suggestions": snapshot.get(
"disclosure_activation_suggestions", []
),
"questionnaire": snapshot.get("questionnaire", []),
"disclosure_answers": snapshot["disclosure_answers"],
"note_outline": snapshot.get("note_outline", []),
"note_outline_context": snapshot.get("note_outline_context"),
"narrative_blocks": snapshot.get("narrative_blocks", []),
"narrative_context": snapshot.get("narrative_context"),
"narrative_change_log": [
{
"block_id": block["block_id"],
"prior_suggestion_id": block.get("prior_suggestion_id"),
"redline": block.get("redline"),
"reviewed_by": block.get("reviewed_by"),
}
for block in snapshot.get("narrative_blocks", [])
],
"prior_narrative_suggestions": snapshot.get("prior_narrative_suggestions", []),
"intelligence_runs": snapshot.get("intelligence_runs", []),
"latest_workflow_guidance": snapshot.get("latest_workflow_guidance"),
"model_mapping_suggestions": snapshot.get("model_mapping_suggestions", []),
"narrative_suggestions": snapshot.get("narrative_suggestions", []),
"review_decisions": snapshot.get("review_decisions", []),
"regulatory_migrations": snapshot.get("regulatory_migrations", []),
"overrides": [
decision
for decision in snapshot.get("review_decisions", [])
if decision.get("action") == "OVERRIDDEN"
],
"unresolved_non_blocking_warnings": [
issue
for issue in snapshot["validation"].get("issues", [])
if issue.get("severity") in {"MEDIUM", "LOW", "INFO"}
and issue.get("review_status") == "UNREVIEWED"
],
"preview": {
key: value
for key, value in (snapshot.get("preview") or {}).items()
if key != "content_base64"
},
"xbrl_review": snapshot.get("xbrl_review"),
"validation": snapshot["validation"],
"assumptions": [],
"assumption_policy": (
"Missing or model-suggested facts are not assumptions and are not exportable."
),
"audit_events": approval.get("audit_events", []),
"filing_boundary": "Vera does not sign, approve corporate accounts, or submit the filing.",
"external_validation": case.get("external_validation"),
"external_validation_documents": [
document
for document in case.get("source_documents", [])
if document.get("purpose") == "EXTERNAL_VALIDATION_REPORT"
],
"external_validation_boundary": "Manual user-controlled TEBENI validation remains required.",
"artifact_manifest": {
"file_name": "artifact_manifest.json",
"snapshot_id": approval["snapshot_id"],
"snapshot_hash": approval["snapshot_hash"],
"peer_artifacts": peer_artifacts,
"integrity_note": (
"The workpaper and final manifest checksums are recorded in "
"artifact_manifest.json to avoid recursive self-hashing."
),
},
}
workpaper_path = output / "workpaper.json"
workpaper_path.write_bytes(_canonical_json(workpaper) + b"\n")
artifacts = [*peer_artifacts, _artifact_record(workpaper_path)]
manifest = {
"case_id": case["case_id"],
"snapshot_id": approval["snapshot_id"],
"snapshot_hash": approval["snapshot_hash"],
"export_basis": "IMMUTABLE_APPROVAL_SNAPSHOT",
"approved_by": approval["approved_by"],
"approved_at": approval["approved_at"],
"artifacts": artifacts,
}
manifest_path = output / "artifact_manifest.json"
manifest_path.write_bytes(_canonical_json(manifest) + b"\n")
manifest_artifact = _artifact_record(manifest_path)
_publish_output_directory(output, final_output, "Export directory")
temporary_output.cleanup()
case["state"] = CaseState.EXPORTED
case["updated_at"] = _now()
case["artifacts"] = [
*artifacts,
manifest_artifact,
]
case["_pending_before_hash"] = before_hash
_record_event(case, "artifact_exported", actor, {"manifest": manifest})
return case
def _read_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def _write_result(case_dir: Path, case: dict[str, Any]) -> None:
save_case(case_dir, case)
sys.stdout.write(
json.dumps(
{
"case_id": case["case_id"],
"revision_id": case["revision_id"],
"state": case["state"],
},
ensure_ascii=False,
)
+ "\n"
)
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command", required=True)
create = sub.add_parser("create")
create.add_argument("--case-dir", type=Path, required=True)
create.add_argument("--payload", type=Path, required=True)
create.add_argument("--rule-pack", type=Path, required=True)
create.add_argument("--actor", required=True)
ingest = sub.add_parser("ingest")
ingest.add_argument("--case-dir", type=Path, required=True)
ingest.add_argument("--source", type=Path, required=True)
ingest.add_argument("--sheet")
ingest.add_argument("--revision", required=True)
ingest.add_argument("--actor", required=True)
pdf_ingest = sub.add_parser("ingest-pdf")
pdf_ingest.add_argument("--case-dir", type=Path, required=True)
pdf_ingest.add_argument("--source", type=Path, required=True)
pdf_ingest.add_argument("--ocr-language", choices=["it", "en"], default="it")
pdf_ingest.add_argument("--no-ocr", action="store_true")
pdf_ingest.add_argument("--revision", required=True)
pdf_ingest.add_argument("--actor", required=True)
pdf_review = sub.add_parser("review-pdf-extraction")
pdf_review.add_argument("--case-dir", type=Path, required=True)
pdf_review.add_argument("--payload", type=Path, required=True)
pdf_review.add_argument("--revision", required=True)
pdf_review.add_argument("--reviewer", required=True)
prior = sub.add_parser("ingest-prior-xbrl")
prior.add_argument("--case-dir", type=Path, required=True)
prior.add_argument("--source", type=Path, required=True)
prior.add_argument("--revision", required=True)
prior.add_argument("--actor", required=True)
supporting = sub.add_parser("attach-supporting-document")
supporting.add_argument("--case-dir", type=Path, required=True)
supporting.add_argument("--source", type=Path, required=True)
supporting.add_argument("--purpose", required=True)
supporting.add_argument("--description", required=True)
supporting.add_argument("--revision", required=True)
supporting.add_argument("--actor", required=True)
confirm = sub.add_parser("confirm-parser")
confirm.add_argument("--case-dir", type=Path, required=True)
confirm.add_argument(
"--convention",
choices=[
item.value
for item in ParserConvention
if item is not ParserConvention.UNKNOWN
],
required=True,
)
confirm.add_argument("--revision", required=True)
confirm.add_argument("--actor", required=True)
forms = sub.add_parser("determine-forms")
forms.add_argument("--case-dir", type=Path, required=True)
forms.add_argument("--metrics", type=Path, required=True)
forms.add_argument("--rule-pack", type=Path, required=True)
forms.add_argument("--revision", required=True)
forms.add_argument("--actor", required=True)
select = sub.add_parser("select-form")
select.add_argument("--case-dir", type=Path, required=True)
select.add_argument(
"--form", choices=["MICRO", "ABBREVIATED", "ORDINARY"], required=True
)
select.add_argument("--revision", required=True)
select.add_argument("--actor", required=True)
taxonomy_index = sub.add_parser("taxonomy-mapping-index")
taxonomy_index.add_argument("--case-dir", type=Path, required=True)
taxonomy_index.add_argument("--catalogue", type=Path, required=True)
taxonomy_index.add_argument("--rule-pack", type=Path, required=True)
taxonomy_index.add_argument("--revision", required=True)
taxonomy_index.add_argument("--actor", required=True)
mapping = sub.add_parser("apply-mappings")
mapping.add_argument("--case-dir", type=Path, required=True)
mapping.add_argument("--decisions", type=Path, required=True)
mapping.add_argument("--revision", required=True)
mapping.add_argument("--actor", required=True)
adjustments = sub.add_parser("record-adjustments")
adjustments.add_argument("--case-dir", type=Path, required=True)
adjustments.add_argument("--adjustments", type=Path, required=True)
adjustments.add_argument("--revision", required=True)
adjustments.add_argument("--actor", required=True)
comparative = sub.add_parser("record-comparative-reconciliation")
comparative.add_argument("--case-dir", type=Path, required=True)
comparative.add_argument("--decisions", type=Path, required=True)
comparative.add_argument("--revision", required=True)
comparative.add_argument("--reviewer", required=True)
taxonomy_facts = sub.add_parser("record-taxonomy-facts")
taxonomy_facts.add_argument("--case-dir", type=Path, required=True)
taxonomy_facts.add_argument("--facts", type=Path, required=True)
taxonomy_facts.add_argument("--revision", required=True)
taxonomy_facts.add_argument("--actor", required=True)
statutory_presentation = sub.add_parser("record-statutory-presentation")
statutory_presentation.add_argument("--case-dir", type=Path, required=True)
statutory_presentation.add_argument("--catalogue", type=Path, required=True)
statutory_presentation.add_argument("--rule-pack", type=Path, required=True)
statutory_presentation.add_argument("--decisions", type=Path, required=True)
statutory_presentation.add_argument("--revision", required=True)
statutory_presentation.add_argument("--actor", required=True)
taxonomy_representation = sub.add_parser("record-taxonomy-representation")
taxonomy_representation.add_argument("--case-dir", type=Path, required=True)
taxonomy_representation.add_argument("--payload", type=Path, required=True)
taxonomy_representation.add_argument("--revision", required=True)
taxonomy_representation.add_argument("--reviewer", required=True)
micro_reporting = sub.add_parser("record-micro-reporting")
micro_reporting.add_argument("--case-dir", type=Path, required=True)
micro_reporting.add_argument("--payload", type=Path, required=True)
micro_reporting.add_argument("--revision", required=True)
micro_reporting.add_argument("--reviewer", required=True)
candidates = sub.add_parser("mapping-candidates")
candidates.add_argument("--case-dir", type=Path, required=True)
candidates.add_argument("--memory", type=Path, required=True)
candidates.add_argument("--source-system-template", required=True)
candidates.add_argument("--revision", required=True)
candidates.add_argument("--actor", required=True)
compute = sub.add_parser("compute-statements")
compute.add_argument("--case-dir", type=Path, required=True)
compute.add_argument("--revision", required=True)
compute.add_argument("--actor", required=True)
schedule = sub.add_parser("record-schedule")
schedule.add_argument("--case-dir", type=Path, required=True)
schedule.add_argument("--payload", type=Path, required=True)
schedule.add_argument("--revision", required=True)
schedule.add_argument("--actor", required=True)
schedule_taxonomy = sub.add_parser("record-schedule-taxonomy-adapter")
schedule_taxonomy.add_argument("--case-dir", type=Path, required=True)
schedule_taxonomy.add_argument("--catalogue", type=Path, required=True)
schedule_taxonomy.add_argument("--rule-pack", type=Path, required=True)
schedule_taxonomy.add_argument("--decisions", type=Path, required=True)
schedule_taxonomy.add_argument("--revision", required=True)
schedule_taxonomy.add_argument("--actor", required=True)
schedule_file = sub.add_parser("ingest-schedule")
schedule_file.add_argument("--case-dir", type=Path, required=True)
schedule_file.add_argument("--source", type=Path, required=True)
schedule_file.add_argument(
"--schedule-type",
choices=sorted(SCHEDULE_TYPES - {"CASH_FLOW"}),
required=True,
)
schedule_file.add_argument("--schedule-id", required=True)
schedule_file.add_argument("--statement-line")
schedule_file.add_argument("--options", type=Path, required=True)
schedule_file.add_argument("--sheet")
schedule_file.add_argument("--revision", required=True)
schedule_file.add_argument("--actor", required=True)
disclosures = sub.add_parser("activate-disclosures")
disclosures.add_argument("--case-dir", type=Path, required=True)
disclosures.add_argument("--rule-pack", type=Path, required=True)
disclosures.add_argument("--revision", required=True)
disclosures.add_argument("--actor", required=True)
disclosure_triggers = sub.add_parser("record-disclosure-triggers")
disclosure_triggers.add_argument("--case-dir", type=Path, required=True)
disclosure_triggers.add_argument("--decisions", type=Path, required=True)
disclosure_triggers.add_argument("--revision", required=True)
disclosure_triggers.add_argument("--reviewer", required=True)
answers = sub.add_parser("record-answers")
answers.add_argument("--case-dir", type=Path, required=True)
answers.add_argument("--answers", type=Path, required=True)
answers.add_argument("--revision", required=True)
answers.add_argument("--actor", required=True)
narratives = sub.add_parser("record-narratives")
narratives.add_argument("--case-dir", type=Path, required=True)
narratives.add_argument("--blocks", type=Path, required=True)
narratives.add_argument("--revision", required=True)
narratives.add_argument("--actor", required=True)
preview = sub.add_parser("preview")
preview.add_argument("--case-dir", type=Path, required=True)
preview.add_argument("--output", type=Path, required=True)
preview.add_argument("--revision", required=True)
preview.add_argument("--actor", required=True)
validate = sub.add_parser("validate")
validate.add_argument("--case-dir", type=Path, required=True)
validate.add_argument("--revision", required=True)
validate.add_argument("--actor", required=True)
xbrl_review = sub.add_parser("prepare-xbrl-review")
xbrl_review.add_argument("--case-dir", type=Path, required=True)
xbrl_review.add_argument("--catalogue", type=Path, required=True)
xbrl_review.add_argument("--taxonomy-package", type=Path, required=True)
xbrl_review.add_argument("--output-dir", type=Path, required=True)
xbrl_review.add_argument("--revision", required=True)
xbrl_review.add_argument("--actor", required=True)
issue_reviews = sub.add_parser("record-issue-reviews")
issue_reviews.add_argument("--case-dir", type=Path, required=True)
issue_reviews.add_argument("--decisions", type=Path, required=True)
issue_reviews.add_argument("--revision", required=True)
issue_reviews.add_argument("--reviewer", required=True)
approve = sub.add_parser("approve")
approve.add_argument("--case-dir", type=Path, required=True)
approve.add_argument("--declaration", type=Path, required=True)
approve.add_argument("--revision", required=True)
approve.add_argument("--reviewer", required=True)
export = sub.add_parser("export")
export.add_argument("--case-dir", type=Path, required=True)
export.add_argument("--output-dir", type=Path, required=True)
export.add_argument("--catalogue", type=Path, required=True)
export.add_argument("--actor", required=True)
external = sub.add_parser("record-external-validation")
external.add_argument("--case-dir", type=Path, required=True)
external.add_argument("--report", type=Path, required=True)
external.add_argument(
"--result", choices=["PASS", "FAIL", "WARNING"], required=True
)
external.add_argument("--issues", type=Path, required=True)
external.add_argument("--revision", required=True)
external.add_argument("--actor", required=True)
intel_packet = sub.add_parser("intelligence-packet")
intel_packet.add_argument("--case-dir", type=Path, required=True)
intel_packet.add_argument(
"--task", choices=[item.value for item in IntelligenceTask], required=True
)
intel_packet.add_argument("--subject-id", action="append", default=[])
intel = sub.add_parser("record-intelligence")
intel.add_argument("--case-dir", type=Path, required=True)
intel.add_argument("--payload", type=Path, required=True)
intel.add_argument("--revision", required=True)
intel.add_argument("--actor", required=True)
remember = sub.add_parser("remember-mappings")
remember.add_argument("--case-dir", type=Path, required=True)
remember.add_argument("--memory", type=Path, required=True)
remember.add_argument("--source-system-template", required=True)
remember.add_argument("--revision", required=True)
remember.add_argument("--actor", required=True)
history_load = sub.add_parser("load-client-history")
history_load.add_argument("--case-dir", type=Path, required=True)
history_load.add_argument("--history", type=Path, required=True)
history_load.add_argument("--revision", required=True)
history_load.add_argument("--actor", required=True)
history_remember = sub.add_parser("remember-client-history")
history_remember.add_argument("--case-dir", type=Path, required=True)
history_remember.add_argument("--history", type=Path, required=True)
history_remember.add_argument("--revision", required=True)
history_remember.add_argument("--actor", required=True)
status = sub.add_parser("status")
status.add_argument("--case-dir", type=Path, required=True)
return parser
def main(argv: list[str] | None = None) -> int:
"""Run the deterministic case command line interface."""
args = _parser().parse_args(argv)
try:
if args.command == "create":
case = create_case(
args.case_dir,
_read_json(args.payload),
_read_json(args.rule_pack),
args.actor,
)
else:
case = load_case(args.case_dir)
if args.command == "ingest":
case = ingest_trial_balance(
case, args.source, args.actor, args.revision, args.sheet
)
elif args.command == "ingest-pdf":
case = ingest_pdf_trial_balance(
case,
args.source,
args.actor,
args.revision,
ocr_enabled=not args.no_ocr,
ocr_language=args.ocr_language,
)
elif args.command == "review-pdf-extraction":
case = record_pdf_trial_balance_review(
case,
_read_json(args.payload),
args.reviewer,
args.revision,
)
elif args.command == "ingest-prior-xbrl":
case = ingest_prior_xbrl(case, args.source, args.actor, args.revision)
elif args.command == "attach-supporting-document":
case = attach_supporting_document(
case,
args.source,
args.purpose,
args.description,
args.actor,
args.revision,
)
elif args.command == "confirm-parser":
case = confirm_parser(case, args.convention, args.actor, args.revision)
elif args.command == "determine-forms":
case = determine_forms(
case,
_read_json(args.metrics),
_read_json(args.rule_pack),
args.actor,
args.revision,
)
elif args.command == "select-form":
case = select_form(case, args.form, args.actor, args.revision)
elif args.command == "taxonomy-mapping-index":
case = record_taxonomy_mapping_index(
case,
args.catalogue,
_read_json(args.rule_pack),
args.actor,
args.revision,
)
elif args.command == "apply-mappings":
case = apply_mapping_decisions(
case, _read_json(args.decisions), args.actor, args.revision
)
elif args.command == "record-adjustments":
case = record_adjustments(
case,
_read_json(args.adjustments),
args.actor,
args.revision,
)
elif args.command == "record-comparative-reconciliation":
case = record_comparative_reconciliation_decisions(
case,
_read_json(args.decisions),
args.reviewer,
args.revision,
)
elif args.command == "record-taxonomy-facts":
case = record_taxonomy_facts(
case,
_read_json(args.facts),
args.actor,
args.revision,
)
elif args.command == "record-statutory-presentation":
case = record_statutory_presentation(
case,
args.catalogue,
_read_json(args.rule_pack),
_read_json(args.decisions),
args.actor,
args.revision,
)
elif args.command == "record-taxonomy-representation":
case = record_taxonomy_representation(
case,
_read_json(args.payload),
args.reviewer,
args.revision,
)
elif args.command == "record-micro-reporting":
case = record_micro_reporting(
case,
_read_json(args.payload),
args.reviewer,
args.revision,
)
elif args.command == "mapping-candidates":
case = generate_mapping_candidates(
case,
args.memory,
args.source_system_template,
args.actor,
args.revision,
)
elif args.command == "compute-statements":
case = build_statements(case, args.actor, args.revision)
elif args.command == "record-schedule":
case = record_schedule(
case, _read_json(args.payload), args.actor, args.revision
)
elif args.command == "record-schedule-taxonomy-adapter":
case = record_schedule_taxonomy_adapter(
case,
args.catalogue,
_read_json(args.rule_pack),
_read_json(args.decisions),
args.actor,
args.revision,
)
elif args.command == "ingest-schedule":
case = ingest_schedule_file(
case,
args.source,
args.schedule_type,
args.schedule_id,
args.statement_line,
_read_json(args.options),
args.actor,
args.revision,
args.sheet,
)
elif args.command == "activate-disclosures":
case = activate_disclosures(
case, _read_json(args.rule_pack), args.actor, args.revision
)
elif args.command == "record-disclosure-triggers":
case = record_disclosure_trigger_decisions(
case,
_read_json(args.decisions),
args.reviewer,
args.revision,
)
elif args.command == "record-answers":
case = record_disclosure_answers(
case, _read_json(args.answers), args.actor, args.revision
)
elif args.command == "record-narratives":
case = record_narrative_blocks(
case, _read_json(args.blocks), args.actor, args.revision
)
elif args.command == "preview":
case = create_preview(case, args.output, args.actor, args.revision)
elif args.command == "validate":
case = run_validation(case, args.actor, args.revision)
elif args.command == "prepare-xbrl-review":
case = prepare_xbrl_review(
case,
args.catalogue,
args.taxonomy_package,
args.output_dir,
args.actor,
args.revision,
)
elif args.command == "record-issue-reviews":
case = record_issue_reviews(
case,
_read_json(args.decisions),
args.reviewer,
args.revision,
)
elif args.command == "approve":
case = approve_case(
case, args.reviewer, args.revision, _read_json(args.declaration)
)
elif args.command == "export":
case = export_case(case, args.output_dir, args.catalogue, args.actor)
elif args.command == "record-external-validation":
case = record_external_validation(
case,
args.report,
args.result,
_read_json(args.issues),
args.actor,
args.revision,
)
elif args.command == "intelligence-packet":
packet = build_intelligence_packet(case, args.task, args.subject_id)
sys.stdout.write(
json.dumps(packet, ensure_ascii=False, indent=2) + "\n"
)
return 0
elif args.command == "record-intelligence":
payload = _read_json(args.payload)
case = record_intelligence_suggestion(
case,
payload["task"],
payload.get("subject_ids", []),
payload["output"],
payload["model_metadata"],
args.actor,
args.revision,
)
elif args.command == "remember-mappings":
case = remember_mappings(
case,
args.memory,
args.source_system_template,
args.actor,
args.revision,
)
elif args.command == "load-client-history":
case = load_client_history(
case, args.history, args.actor, args.revision
)
elif args.command == "remember-client-history":
case = remember_client_history(
case, args.history, args.actor, args.revision
)
elif args.command == "status":
sys.stdout.write(json.dumps(case, ensure_ascii=False, indent=2) + "\n")
return 0
else:
raise AssertionError(args.command)
_write_result(args.case_dir, case)
return 0
except (KeyError, ValueError, OSError, json.JSONDecodeError) as exc:
LOGGER.error("%s", exc)
return 2
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(message)s")
raise SystemExit(main())
SHA-256: c05f03a0ee4647f67d3df3f250031bd1c3617b0e5050551bfb816a06402f3eb6