stack/dev/scripts/generate_cclf_models.py

"""Generate CCLF SQLTable models from the CCLF Information Packet PDF.

Parses Appendix B of the CCLF IP (cclf-information-packet.pdf) to
extract all 12 CCLF file layouts and generates a single Python module
at ``src/aco/table/cclf.py`` with one SQLTable subclass per file.

Usage::

    python dev/scripts/generate_cclf_models.py

Source: https://www.cms.gov/files/document/cclf-information-packet.pdf
"""

from __future__ import annotations

import re
from pathlib import Path

import pdfplumber


def extract_cclf_tables(pdf_path: str) -> dict:
    """Extract all CCLF file layouts from Appendix B."""
    pdf = pdfplumber.open(pdf_path)
    all_tables: dict[str, dict] = {}
    current_file: str | None = None

    for i in range(len(pdf.pages)):
        text = pdf.pages[i].extract_text() or ""

        # Detect CCLF file table headers
        for m in re.finditer(r"Table \d+: (.+?)\(CCLF(\w+)\)", text):
            table_title = m.group(1).strip()
            cclf_id = m.group(2)
            current_file = f"CCLF{cclf_id}"
            if current_file not in all_tables:
                all_tables[current_file] = {
                    "title": table_title,
                    "cclf_id": cclf_id,
                    "fields": [],
                }

        # Extract structured tables from this page
        tables = pdf.pages[i].extract_tables()
        for table in tables:
            if not table:
                continue
            header = table[0]
            if not any("Element" in str(h) for h in header):
                continue

            for row in table[1:]:
                if not row or not row[0]:
                    continue
                elem_num = str(row[0]).strip()
                if not elem_num.isdigit():
                    continue

                is_cclf0 = current_file == "CCLF0"
                field_label = _clean_field_name(str(row[1] or ""), is_cclf0=is_cclf0)
                field_name = str(row[2] or "").replace("\n", " ").strip()
                start_pos = str(row[3] or "").strip()
                end_pos = str(row[4] or "").strip()
                data_length = str(row[5] or "").strip()
                fmt = str(row[6] or "").replace("\n", "").strip()
                description = (
                    str(row[7] or "").replace("\n", " ").strip() if len(row) > 7 else ""
                )

                if current_file and current_file in all_tables:
                    all_tables[current_file]["fields"].append(
                        {
                            "num": int(elem_num),
                            "label": field_label,
                            "name": field_name,
                            "start": (int(start_pos) if start_pos.isdigit() else 0),
                            "end": (int(end_pos) if end_pos.isdigit() else 0),
                            "length": (
                                int(data_length) if data_length.isdigit() else 0
                            ),
                            "format": fmt,
                            "description": description,
                        }
                    )

    return all_tables


def _clean_field_name(raw: str, is_cclf0: bool = False) -> str:
    """Fix PDF line-break artifacts in field names.

    For CCLF1-CCLFB: field labels are UPPER_SNAKE_CASE system names
    that get split by PDF column boundaries.  Fix by removing all
    spaces (no legitimate spaces in system names).

    For CCLF0: field labels are human-readable ("File Type",
    "Number of records") — convert to snake_case.
    """
    name = raw.replace("\n", " ").strip()
    # Remove unicode artifacts (ellipsis, etc.)
    name = name.replace("…", "").replace("\u2026", "")

    if is_cclf0:
        # Human-readable labels → snake_case
        return re.sub(r"\s+", "_", name.strip()).lower()

    # System names: remove ALL spaces (PDF column-break artifacts)
    return name.replace(" ", "")


def _field_type(fmt: str) -> str:
    """Map CCLF format codes to Python types.

    X(nn) → str, 9(nn) → str (kept as str since these are
    fixed-width character data), YYYY-MM-DD → date,
    -S9(nn)V99 → float
    """
    if "YYYY" in fmt or "MM-DD" in fmt:
        return "date"
    if "V9" in fmt or "V99" in fmt:
        return "float"
    if fmt.startswith("-S9") or fmt.startswith("-9"):
        return "float"
    return "str"


def _python_name(label: str) -> str:
    """Convert CCLF field label to Python field name."""
    return label.lower()


def _class_name(cclf_id: str) -> str:
    """Convert CCLF ID to class name."""
    return f"Cclf{cclf_id}"


def _table_name(cclf_id: str) -> str:
    """Convert CCLF ID to table name."""
    return f"cclf{cclf_id.lower()}"


def generate_module(tables: dict) -> str:
    """Generate the cclf.py module source code."""
    lines = [
        '"""CCLF — Claim and Claim Line Feed file layouts.',
        "",
        "Auto-generated from the CCLF Information Packet (IP):",
        "https://www.cms.gov/files/document/cclf-information-packet.pdf",
        f"Version 41.0, 07/16/2025 — {sum(len(t['fields']) for t in tables.values())} "
        f"fields across {len(tables)} files.",
        "",
        "CCLF files are fixed-width text files delivered monthly to",
        "ACOs participating in the Medicare Shared Savings Program,",
        "ACO REACH, KCC, PCF, and IOTA models.",
        "",
        "Files::",
        "",
    ]

    for cclf_id in sorted(tables.keys()):
        info = tables[cclf_id]
        lines.append(f"    {cclf_id}: {info['title']}({len(info['fields'])} fields)")

    lines.extend(
        [
            '"""',
            "",
            "from __future__ import annotations",
            "",
            "from datetime import date  # noqa: F811",
            "",
            "from aco.table.base import SQLTable",
            "",
        ]
    )

    # Sort by CCLF ID (0, 1, 2, ..., 9, A, B)
    sort_order = [
        "CCLF0",
        "CCLF1",
        "CCLF2",
        "CCLF3",
        "CCLF4",
        "CCLF5",
        "CCLF6",
        "CCLF7",
        "CCLF8",
        "CCLF9",
        "CCLFA",
        "CCLFB",
    ]
    sorted_keys = [k for k in sort_order if k in tables]

    for cclf_key in sorted_keys:
        info = tables[cclf_key]
        cclf_id = info["cclf_id"]
        cls_name = _class_name(cclf_id)
        tbl_name = _table_name(cclf_id)
        title = info["title"]

        lines.append("")
        lines.append(f"class {cls_name}(SQLTable):")
        lines.append(f'    """CCLF{cclf_id}: {title}')
        lines.append("")
        lines.append(f"    {len(info['fields'])} fields, fixed-width layout.")
        lines.append('    """')
        lines.append("")
        lines.append('    __schema__ = "cclf"')
        lines.append(f'    __tablename__ = "{tbl_name}"')

        seen_names: set[str] = set()
        for field in info["fields"]:
            py_name = _python_name(field["label"])
            py_type = _field_type(field["format"])

            # Deduplicate (CCLF0 has repeated element numbers)
            if py_name in seen_names:
                continue
            # Skip delimiter/filler fields
            if py_name in ("delimiter", "filler", "|"):
                continue
            seen_names.add(py_name)

            # Clean description for docstring
            desc = field["description"]
            # Remove PII/PHI markers
            desc = re.sub(r"\s*[IH]\s*$", "", desc)
            desc = re.sub(r"\s+[IH]\s+", " ", desc)
            desc = desc.strip()
            # Truncate very long descriptions
            if len(desc) > 120:
                desc = desc[:117] + "..."
            # Escape quotes that would break triple-quote docstrings
            desc = desc.replace('"""', '\'""')
            # If desc starts with ", add a space to avoid """"
            if desc.startswith('"'):
                desc = " " + desc

            lines.append("")
            lines.append(f"    {py_name}: {py_type} | None = None")
            if desc:
                lines.append(f'    """{desc}"""')

        lines.append("")
        lines.append("")

    # Remove trailing blank lines
    while lines and lines[-1] == "":
        lines.pop()

    lines.append("")
    return "\n".join(lines)


def main() -> None:
    project_root = Path(__file__).resolve().parents[2]
    pdf_path = project_root / "dev" / "seeds" / "cclf-information-packet.pdf"
    output_path = project_root / "src" / "aco" / "table" / "cclf.py"

    print(f"Parsing {pdf_path.name}...")
    tables = extract_cclf_tables(str(pdf_path))

    total_fields = sum(len(t["fields"]) for t in tables.values())
    print(f"Extracted {len(tables)} CCLF files, {total_fields} fields")

    for cclf_id in sorted(tables.keys()):
        info = tables[cclf_id]
        print(f"  {cclf_id}: {info['title']}— {len(info['fields'])} fields")

    source = generate_module(tables)
    output_path.write_text(source)
    print(f"\nWritten to {output_path}")
    print(f"  {len(source.splitlines())} lines")


if __name__ == "__main__":
    main()