stack/dev/scripts/generate_reach_participants.py

"""Generate ACO REACH Participants table model and rex parser.

Extracts the participant provider schema from the bulk upload template Excel file
and generates a SQLTable model plus rex parser.

Source: dev/ACO_REACH_bulk_upload_participants 5-19 (1).xlsx

Usage::

    uv run python dev/scripts/generate_reach_participants.py
"""

from __future__ import annotations

import zipfile
from pathlib import Path
from xml.etree import ElementTree as ET


def extract_headers_from_template(xlsx_path: str) -> list[str]:
    """Extract column headers from bulk upload template."""
    with zipfile.ZipFile(xlsx_path, "r") as zip_ref:
        # Read shared strings
        strings_xml = zip_ref.read("xl/sharedStrings.xml")
        root = ET.fromstring(strings_xml)
        ns_ss = {"ss": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
        shared_strings = [
            elem.text if elem.text else "" for elem in root.findall(".//ss:t", ns_ss)
        ]

        # Read ACO REACH Provider sheet (sheet3)
        sheet3_xml = zip_ref.read("xl/worksheets/sheet3.xml")
        sheet3_root = ET.fromstring(sheet3_xml)
        rows = sheet3_root.findall(
            ".//{http://schemas.openxmlformats.org/spreadsheetml/2006/main}row"
        )

        # Extract first row (header)
        header_row = rows[0]
        cells = header_row.findall(
            "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}c"
        )

        headers = []
        for cell in cells:
            cell_type = cell.get("t")
            value_elem = cell.find(
                "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}v"
            )

            if value_elem is not None:
                if cell_type == "s":
                    idx = int(value_elem.text)
                    if idx < len(shared_strings):
                        headers.append(shared_strings[idx])
                else:
                    headers.append(value_elem.text)
            else:
                headers.append(None)

        return headers


def normalize_field_name(header: str) -> str:
    """Convert header to Python snake_case field name."""
    import re

    # Remove newlines and extra spaces
    header = re.sub(r"\s+", " ", header.strip())

    # Convert to snake case
    # Replace spaces and special chars with underscore
    header = re.sub(r"[^\w\s]", "", header)
    header = re.sub(r"\s+", "_", header)
    header = header.lower()

    # Remove leading/trailing underscores
    header = header.strip("_")

    return header


def infer_field_type(header: str) -> str:
    """Infer field type from header name."""
    header_lower = header.lower()

    # Boolean attestations
    if "attest" in header_lower:
        return "bool | None"

    # Email
    if "email" in header_lower:
        return "str | None"

    # Identifiers (required)
    if any(k in header_lower for k in ["npi", "tin", "ccn"]):
        return "str | None"

    # Names and addresses
    if any(
        k in header_lower for k in ["name", "address", "city", "state", "county", "zip"]
    ):
        return "str | None"

    # Most other fields are optional strings
    return "str | None"


def generate_table_model(headers: list[str], output_path: Path) -> None:
    """Generate SQLTable model for REACH participants."""
    lines = []

    lines.append('"""ACO REACH Participant Provider table model.')
    lines.append("")
    lines.append("Generated from: dev/ACO_REACH_bulk_upload_participants 5-19 (1).xlsx")
    lines.append("")
    lines.append(
        "This table represents the participant provider records that ACOs submit"
    )
    lines.append("via bulk upload to add providers to their participant list.")
    lines.append('"""')
    lines.append("")
    lines.append("from __future__ import annotations")
    lines.append("")
    lines.append("from typing import ClassVar")
    lines.append("")
    lines.append("from aco.table.base import SQLTable")
    lines.append("")
    lines.append("")
    lines.append("class ReachParticipants(SQLTable):")
    lines.append('    """ACO REACH Participant Provider bulk upload records.')
    lines.append("")
    lines.append("    File: DCE_bulk_upload_participants_DXXXX.xlsx")
    lines.append("    Sheet: ACO REACH Provider")
    lines.append("    ")
    lines.append(
        "    Participant providers include individual practitioners, organizations,"
    )
    lines.append("    and institutional facilities that have agreements with the ACO.")
    lines.append('    """')
    lines.append("")

    # Add schema/tablename with ClassVar annotation
    lines.append('    schema__: ClassVar[str] = "reach"')
    lines.append('    tablename: ClassVar[str] = "participants"')
    lines.append("")

    # Process each header
    for header in headers:
        if not header or not header.strip():
            continue

        # Skip the instructions column
        if "This excel document" in header:
            continue

        field_name = normalize_field_name(header)
        if not field_name or field_name in ("", "_"):
            continue

        field_type = infer_field_type(header)

        # Add field
        lines.append(f"    {field_name}: {field_type}")
        lines.append(f'    """{header}"""')
        lines.append("")

    output_path.write_text("\n".join(lines))
    print(f"✓ Generated {output_path}")


def generate_rex_parser(headers: list[str], output_path: Path) -> None:
    """Generate rex parser for REACH participants."""
    lines = []

    lines.append('"""Rex parser for ACO REACH Participant Provider bulk upload files.')
    lines.append("")
    lines.append("Parses Excel bulk upload templates submitted by ACOs to add/update")
    lines.append("participant providers.")
    lines.append("")
    lines.append("File Pattern: DCE_bulk_upload_participants_DXXXX.xlsx")
    lines.append("Sheet: ACO REACH Provider")
    lines.append('"""')
    lines.append("")
    lines.append("from __future__ import annotations")
    lines.append("")
    lines.append("import zipfile")
    lines.append("from typing import Iterator")
    lines.append("from xml.etree import ElementTree as ET")
    lines.append("")
    lines.append("from aco.table.reach_participants import ReachParticipants")
    lines.append("")
    lines.append("")
    lines.append(
        "def parse_reach_participants(file_path: str) -> Iterator[ReachParticipants]:"
    )
    lines.append('    """Parse REACH participant provider bulk upload Excel file.')
    lines.append("")
    lines.append(
        "    Handles the corrupt stylesheet issue by reading XML directly from ZIP."
    )
    lines.append('    """')
    lines.append('    with zipfile.ZipFile(file_path, "r") as zip_ref:')
    lines.append("        # Read shared strings")
    lines.append('        strings_xml = zip_ref.read("xl/sharedStrings.xml")')
    lines.append("        root = ET.fromstring(strings_xml)")
    lines.append(
        '        ns_ss = {"ss": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}'
    )
    lines.append("        shared_strings = [")
    lines.append(
        '            elem.text if elem.text else "" for elem in root.findall(".//ss:t", ns_ss)'
    )
    lines.append("        ]")
    lines.append("")
    lines.append("        # Read ACO REACH Provider sheet (sheet3)")
    lines.append('        sheet3_xml = zip_ref.read("xl/worksheets/sheet3.xml")')
    lines.append("        sheet3_root = ET.fromstring(sheet3_xml)")
    lines.append("        rows = sheet3_root.findall(")
    lines.append(
        '            ".//{http://schemas.openxmlformats.org/spreadsheetml/2006/main}row"'
    )
    lines.append("        )")
    lines.append("")
    lines.append("        # Extract header from first row")
    lines.append("        header_row = rows[0]")
    lines.append("        header_cells = header_row.findall(")
    lines.append(
        '            "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}c"'
    )
    lines.append("        )")
    lines.append("")
    lines.append("        headers = []")
    lines.append("        for cell in header_cells:")
    lines.append('            cell_type = cell.get("t")')
    lines.append("            value_elem = cell.find(")
    lines.append(
        '                "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}v"'
    )
    lines.append("            )")
    lines.append('            if value_elem is not None and cell_type == "s":')
    lines.append("                idx = int(value_elem.text)")
    lines.append(
        "                headers.append(shared_strings[idx] if idx < len(shared_strings) else None)"
    )
    lines.append("            else:")
    lines.append("                headers.append(None)")
    lines.append("")
    lines.append("        # Map headers to field names")
    lines.append("        field_mapping = {}")
    lines.append("        for i, header in enumerate(headers):")
    lines.append(
        '            if header and header.strip() and "This excel document" not in header:'
    )
    lines.append("                field_name = normalize_field_name(header)")
    lines.append("                if field_name:")
    lines.append("                    field_mapping[i] = field_name")
    lines.append("")
    lines.append("        # Process data rows (starting from row 2)")
    lines.append("        for row in rows[1:]:")
    lines.append("            cells = row.findall(")
    lines.append(
        '                "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}c"'
    )
    lines.append("            )")
    lines.append("")
    lines.append("            # Extract cell values")
    lines.append("            row_data = {}")
    lines.append("            for i, cell in enumerate(cells):")
    lines.append("                if i not in field_mapping:")
    lines.append("                    continue")
    lines.append("")
    lines.append('                cell_type = cell.get("t")')
    lines.append("                value_elem = cell.find(")
    lines.append(
        '                    "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}v"'
    )
    lines.append("                )")
    lines.append("")
    lines.append("                if value_elem is not None:")
    lines.append('                    if cell_type == "s":')
    lines.append("                        idx = int(value_elem.text)")
    lines.append(
        "                        value = shared_strings[idx] if idx < len(shared_strings) else None"
    )
    lines.append("                    else:")
    lines.append("                        value = value_elem.text")
    lines.append("                    row_data[field_mapping[i]] = value")
    lines.append("                else:")
    lines.append("                    row_data[field_mapping[i]] = None")
    lines.append("")
    lines.append("            # Skip empty rows")
    lines.append(
        "            if not any(v for v in row_data.values() if v and str(v).strip()):"
    )
    lines.append("                continue")
    lines.append("")
    lines.append("            # Convert boolean fields")
    lines.append("            for field_name, value in row_data.items():")
    lines.append('                if field_name.startswith("i_attest"):')
    lines.append(
        '                    if value and str(value).strip().upper() in ("Y", "YES", "TRUE", "1"):'
    )
    lines.append("                        row_data[field_name] = True")
    lines.append(
        '                    elif value and str(value).strip().upper() in ("N", "NO", "FALSE", "0"):'
    )
    lines.append("                        row_data[field_name] = False")
    lines.append("                    else:")
    lines.append("                        row_data[field_name] = None")
    lines.append("")
    lines.append("            yield ReachParticipants(**row_data)")
    lines.append("")
    lines.append("")
    lines.append("def normalize_field_name(header: str) -> str:")
    lines.append('    """Convert header to Python snake_case field name."""')
    lines.append("    import re")
    lines.append('    header = re.sub(r"\\s+", " ", header.strip())')
    lines.append('    header = re.sub(r"[^\\w\\s]", "", header)')
    lines.append('    header = re.sub(r"\\s+", "_", header)')
    lines.append('    return header.lower().strip("_")')
    lines.append("")
    lines.append("")
    lines.append(
        "def load_reach_participants(file_path: str) -> list[ReachParticipants]:"
    )
    lines.append('    """Load all participant records from bulk upload file."""')
    lines.append("    return list(parse_reach_participants(file_path))")
    lines.append("")

    output_path.write_text("\n".join(lines))
    print(f"✓ Generated {output_path}")


def main():
    """Generate REACH participants table model and rex parser."""
    project_root = Path(__file__).resolve().parents[2]
    xlsx_path = (
        project_root
        / "dev"
        / "seeds"
        / "ACO_REACH_bulk_upload_participants 5-19 (1).xlsx"
    )

    print("Generating ACO REACH Participants integration...")
    print("=" * 70)

    # Extract headers
    print(f"\n✓ Extracting headers from {xlsx_path.name}")
    headers = extract_headers_from_template(str(xlsx_path))
    print(f"  Found {len(headers)} columns")

    # Generate table model
    table_path = project_root / "src" / "aco" / "table" / "reach_participants.py"
    generate_table_model(headers, table_path)

    # Generate rex parser
    rex_path = project_root / "src" / "aco" / "rex" / "reach_participants.py"
    generate_rex_parser(headers, rex_path)

    print("\n" + "=" * 70)
    print("✓ REACH Participants integration complete")
    print("=" * 70)


if __name__ == "__main__":
    main()