f66b35f5fe
Split dev/ flat directory into dev/scripts/ (29 .py files) and dev/seeds/ (PDFs, Excel, ZIPs, NDJSON, grafana, PY2023). Update all Path(__file__) references to use parents[2] for project root and dev/seeds/ for seed data. Fix path references in src/ too. Add PY2024 CMS quality measure value sets (HWR, UAMCC, ACR) to the REGISTRY and load into DuckDB — enables 2024->2025->2026 diffs. Add HWR tables to notebook DIFF_SPECS now that two years exist.
366 lines
13 KiB
Python
366 lines
13 KiB
Python
"""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()
|