Parsing Inconsistent HTML Tables From Legacy State Portals #

from __future__ import annotations

import hashlib
import json
import logging
import re
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional

from bs4 import BeautifulSoup, Tag

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.table_parser")

# Header synonyms vary by portal; we match canonical fields against any known alias.
HEADER_ALIASES: dict[str, frozenset[str]] = {
    "status": frozenset({"status", "standing", "entity status", "status detail"}),
    "registered_agent": frozenset({"registered agent", "agent", "agent name", "ra"}),
    "next_filing_date": frozenset({"next report", "due date", "next filing date", "report due"}),
}


@dataclass(frozen=True)
class ComplianceRecord:
    entity_id: str
    jurisdiction: str
    status: str
    registered_agent: str
    next_filing_date: Optional[str]
    extraction_timestamp: str
    raw_html_hash: str          # hash of the exact bytes parsed — pins the record to its source
    audit_signature: str        # binds payload + raw hash + time into one tamper-evident digest


class PortalTableParser:
    """Deterministic extractor for structurally inconsistent legacy registry tables."""

    def __init__(self, jurisdiction: str) -> None:
        self.jurisdiction = jurisdiction

    @staticmethod
    def _resolve_span_matrix(table: Tag) -> list[list[str]]:
        """Flatten a table into a rectangular grid, carrying rowspan/colspan correctly.

        Legacy portals merge 'Active / Good Standing / Tax Paid' under one rowspan cell;
        if that value is not carried DOWN into later rows, every column after it shifts
        left and a delinquent entity reads as active. This is the core correctness step.
        """
        grid: list[list[str]] = []
        # col_index -> (value, rows_still_to_fill) for cells spilling down from above.
        carry: dict[int, tuple[str, int]] = {}

        for tr in table.find_all("tr"):
            row: list[str] = []
            col = 0
            for cell in tr.find_all(["td", "th"]):
                # Emit any cells still spilling down from a rowspan above this position.
                while col in carry:
                    value, remaining = carry[col]
                    row.append(value)
                    if remaining <= 1:
                        carry.pop(col, None)
                    else:
                        carry[col] = (value, remaining - 1)
                    col += 1

                text = re.sub(r"\s+", " ", cell.get_text(separator=" ", strip=True))
                rowspan = max(1, int(cell.get("rowspan", 1)))
                colspan = max(1, int(cell.get("colspan", 1)))
                for c in range(colspan):
                    row.append(text)
                    if rowspan > 1:
                        carry[col + c] = (text, rowspan - 1)
                col += colspan

            # Drain trailing rowspans that fall to the right of the last real cell.
            while col in carry:
                value, remaining = carry[col]
                row.append(value)
                if remaining <= 1:
                    carry.pop(col, None)
                else:
                    carry[col] = (value, remaining - 1)
                col += 1
            grid.append(row)
        return grid

    @staticmethod
    def _detect_header(grid: list[list[str]]) -> tuple[int, dict[str, int]]:
        """Find the header row by CONTENT, not by <th>: legacy portals rarely use <th>."""
        for idx, row in enumerate(grid):
            mapping: dict[str, int] = {}
            for col, cell in enumerate(row):
                key = cell.strip().lower()
                for field, aliases in HEADER_ALIASES.items():
                    if key in aliases and field not in mapping:
                        mapping[field] = col
            # A real header row resolves at least the status column.
            if "status" in mapping:
                return idx, mapping
        raise ValueError("no header row detected — schema drift or wrong table")

    def _audit(self, raw_html: str, payload: dict[str, object]) -> tuple[str, str]:
        raw_hash = hashlib.sha256(raw_html.encode("utf-8")).hexdigest()
        body = json.dumps({"raw": raw_hash, "payload": payload}, sort_keys=True)
        signature = hashlib.sha256(
            f"{body}:{datetime.now(timezone.utc).isoformat()}".encode("utf-8")
        ).hexdigest()
        return raw_hash, signature

    def parse(self, raw_html: str, entity_id: str) -> ComplianceRecord:
        # html5lib reconstructs unclosed </tr> tags the way a browser does, instead of
        # truncating the table the way the stdlib parser silently would.
        soup = BeautifulSoup(raw_html, "html5lib")
        table = soup.find("table")
        if not isinstance(table, Tag):
            raise ValueError("no <table> in response — likely a JS-rendered portal; fall back")

        grid = self._resolve_span_matrix(table)
        header_idx, cols = self._detect_header(grid)
        data_rows = grid[header_idx + 1:]
        if not data_rows:
            raise ValueError("header present but no data rows after span resolution")
        row = data_rows[0]

        # FAIL LOUD: a ragged row means a span resolved wrong; emitting it would shift columns.
        if any(idx >= len(row) for idx in cols.values()):
            raise ValueError(f"ragged row {len(row)} cols vs header {max(cols.values()) + 1}")

        def col(field: str) -> Optional[str]:
            return row[cols[field]] if field in cols else None

        payload = {
            "entity_id": entity_id,
            "jurisdiction": self.jurisdiction,
            "status": col("status") or "UNKNOWN",
            "registered_agent": col("registered_agent") or "UNKNOWN",
            "next_filing_date": col("next_filing_date"),
        }
        raw_hash, signature = self._audit(raw_html, payload)
        logger.info(json.dumps({
            "event": "table_extracted", "entity_id": entity_id,
            "jurisdiction": self.jurisdiction, "status": payload["status"],
            "audit_sig": signature[:12],
        }))
        return ComplianceRecord(
            **payload,
            extraction_timestamp=datetime.now(timezone.utc).isoformat(),
            raw_html_hash=raw_hash,
            audit_signature=signature,
        )