Map IRS 990 Part VII to JSON Schema

import logging
from typing import Generator
from lxml import etree

# Structured audit logger for ingestion telemetry.
# Python's stdlib has no JSONFormatter; use a plain Formatter with JSON-shaped output.
AUDIT_LOGGER = logging.getLogger("pipeline.ingestion")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(
    logging.Formatter(
        fmt='{"time":"%(asctime)s","level":"%(levelname)s","msg":"%(message)s"}',
        datefmt="%Y-%m-%dT%H:%M:%SZ",
    )
)
AUDIT_LOGGER.addHandler(_handler)


def stream_part_vii_chunks(
    xml_path: str,
    chunk_size: int = 50,
) -> Generator[list[dict], None, None]:
    """Stream IRS 990 Part VII records via lxml.iterparse with chunk boundaries.

    Yields lists of plain dicts; never retains DOM references between chunks,
    which keeps the heap footprint independent of filing size.
    """
    try:
        context = etree.iterparse(xml_path, events=("end",), tag="PartVII")
        buffer: list = []
        for _, elem in context:
            buffer.append(elem)
            if len(buffer) >= chunk_size:
                yield _serialize_chunk(buffer)
                AUDIT_LOGGER.info("chunk_yielded | records=%d", len(buffer))
                buffer.clear()
                elem.clear()  # release parsed element back to the allocator
        if buffer:
            yield _serialize_chunk(buffer)
            AUDIT_LOGGER.info("final_chunk_yielded | records=%d", len(buffer))
    except etree.XMLSyntaxError as exc:
        AUDIT_LOGGER.error("xml_parse_failure | error=%s", exc)
        raise RuntimeError(f"Ingestion aborted: malformed XML at {xml_path}") from exc


def _serialize_chunk(elements: list) -> list[dict]:
    """Convert lxml elements to native dicts without retaining DOM references."""
    return [
        {child.tag: child.text for child in elem if child.text}
        for elem in elements
    ]

from decimal import Decimal, InvalidOperation
from typing import Optional
from pydantic import BaseModel, Field, ValidationError, field_validator


class PartVIIPerson(BaseModel):
    """Strictly typed model for one Part VII compensation record."""

    model_config = {"extra": "forbid"}

    name: str = Field(..., min_length=1, max_length=150)
    position_title: str = Field(..., min_length=1, max_length=100)
    ein: Optional[str] = Field(None, pattern=r"^\d{2}-\d{7}$")
    reportable_comp_filing_org: Optional[Decimal] = Field(
        None, alias="ReportableCompensationFromOrgAmt"          # Sec A, col (D)
    )
    reportable_comp_related_org: Optional[Decimal] = Field(
        None, alias="ReportableCompensationFromRltdOrgAmt"      # Sec A, col (E)
    )
    other_compensation: Optional[Decimal] = Field(
        None, alias="OtherCompensationAmt"                      # Sec A, col (F)
    )

    @field_validator(
        "reportable_comp_filing_org",
        "reportable_comp_related_org",
        "other_compensation",
        mode="before",
    )
    @classmethod
    def coerce_to_decimal(cls, value: Optional[object]) -> Optional[Decimal]:
        """Coerce monetary strings to 2-place Decimal; never touch float."""
        if value is None:
            return None
        try:
            return Decimal(str(value)).quantize(Decimal("0.01"))
        except InvalidOperation as exc:
            raise ValueError(f"Invalid monetary value: {value!r}") from exc


def validate_batch(
    raw_records: list[dict],
) -> tuple[list[PartVIIPerson], list[dict]]:
    """Validate a chunk; return (valid_models, quarantined_exceptions)."""
    valid: list[PartVIIPerson] = []
    quarantined: list[dict] = []
    for idx, record in enumerate(raw_records):
        try:
            valid.append(PartVIIPerson.model_validate(record))
        except ValidationError as exc:
            quarantined.append({
                "record_index": idx,
                "error_type": "SchemaValidationError",
                "field_errors": exc.errors(),
                "raw_payload": record,
            })
    return valid, quarantined

import uuid
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any


def generate_compliance_metadata(
    valid_records: list[PartVIIPerson],
    filing_year: int,
    organization_ein: str,
) -> dict[str, Any]:
    """Stamp a validated batch with immutable audit lineage and tags."""
    total_comp = sum(
        (r.reportable_comp_filing_org or Decimal("0.00") for r in valid_records),
        start=Decimal("0.00"),
    )
    return {
        "audit_id": str(uuid.uuid4()),
        "schema_version": "irs_990_partvii_v1.0",
        "filing_year": filing_year,
        "organization_ein": organization_ein,
        "record_count": len(valid_records),
        "aggregate_compensation": str(total_comp),
        "compliance_tags": {
            "state_charity_registration": "pending_crosswalk",
            "grantor_rule_taxonomy": "threshold_check_applied",
            "metadata_standard": "v2.1.0",
        },
        "validation_timestamp": datetime.now(timezone.utc).isoformat(),
    }

import time
from functools import wraps
from typing import Any, Callable, TypeVar
import orjson

T = TypeVar("T")


def retry_with_backoff(max_attempts: int = 3, base_delay: float = 1.0):
    """Retry I/O with capped exponential backoff; re-raise on exhaustion."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> T:
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except (OSError, orjson.JSONEncodeError) as exc:
                    if attempt == max_attempts:
                        raise RuntimeError(
                            f"Retry exhausted after {max_attempts} attempts"
                        ) from exc
                    time.sleep(min(base_delay * 2 ** (attempt - 1), 10.0))
            raise RuntimeError("unreachable")
        return wrapper
    return decorator


@retry_with_backoff(max_attempts=3)
def serialize_and_dispatch(
    valid_records: list[PartVIIPerson],
    compliance_meta: dict[str, Any],
    output_path: str,
) -> str:
    """Write a deterministic, sorted-key JSON payload; return the path."""
    payload = {
        "schema_version": compliance_meta["schema_version"],
        "compliance_metadata": compliance_meta,
        "records": [r.model_dump(by_alias=True, mode="json") for r in valid_records],
    }
    serialized = orjson.dumps(
        payload, option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS
    )
    with open(output_path, "wb") as fh:
        fh.write(serialized)
    AUDIT_LOGGER.info(
        "payload_written | audit_id=%s | records=%d",
        compliance_meta["audit_id"],
        len(valid_records),
    )
    return output_path