VTdevelops/bond-text-extraction

1from __future__ import annotations2 3import contextlib4import math5import re6from pathlib import Path7from typing import Any, Iterable, Sequence8from uuid import uuid49 10import gradio as gr11from gradio import utils as gr_utils12 13from .models import BondRecord14from .pipeline import ExtractionPipeline15from .xml_builder import FIELD_GROUPS, build_bond_xml16 17 18def _split_camel_case(text: str) -> list[str]:19    parts = re.findall(r"[A-Z]+(?=[A-Z][a-z]|\b)|[A-Z]?[a-z]+|\d+", text)20    normalised: list[str] = []21    for part in parts:22        if part.isupper():23            normalised.append(part)24        else:25            normalised.append(part.capitalize())26    return normalised27 28 29def _make_field_specs() -> list[tuple[str, str]]:30    specs: list[tuple[str, str]] = []31    for _group_name, entries in FIELD_GROUPS:32        for tag, attr in entries:33            label = " ".join(_split_camel_case(tag))34            specs.append((label, attr))35    return specs36 37 38FIELD_SPECS: list[tuple[str, str]] = _make_field_specs()39 40 41def _format_table(records: Iterable[BondRecord]) -> tuple[list[list[str]], list[str]]:42    record_list = list(records)43    column_count = max(1, len(record_list))44    headers = ["Field"] + [f"Bond {index}" for index in range(1, column_count + 1)]45 46    rows: list[list[str]] = []47    for label, attr in FIELD_SPECS:48        row = [label]49        for record in record_list:50            row.append(getattr(record, attr))51        while len(row) < len(headers):52            row.append("")53        rows.append(row)54 55    return rows, headers56 57 58def _normalise_rows(rows: Sequence[Sequence[Any]] | Any | None) -> list[list[str]]:59    if rows is None:60        return []61 62    # Convert pandas / numpy objects into plain python lists63    if hasattr(rows, "to_numpy"):64        rows = rows.to_numpy().tolist()  # type: ignore[assignment]65    elif hasattr(rows, "tolist") and not isinstance(rows, list):66        rows = rows.tolist()  # type: ignore[assignment]67 68    if not isinstance(rows, list):69        rows = list(rows)  # type: ignore[assignment]70 71    target_length = max(72        (len(row) if isinstance(row, Sequence) else 1) for row in rows73    ) if rows else len(FIELD_SPECS) + 174    target_length = max(target_length, len(FIELD_SPECS) + 1)75 76    normalised: list[list[str]] = []77    for raw_row in rows:  # type: ignore[arg-type]78        if raw_row is None:79            values: list[Any] = []80        elif isinstance(raw_row, list):81            values = raw_row82        elif isinstance(raw_row, Sequence):83            values = list(raw_row)84        else:85            values = [raw_row]86 87        cleaned: list[str] = []88        for idx in range(target_length):89            cell = values[idx] if idx < len(values) else ""90            if cell is None:91                cleaned.append("")92                continue93            if isinstance(cell, float) and math.isnan(cell):94                cleaned.append("")95                continue96            cleaned.append(str(cell).strip())97        normalised.append(cleaned)98    return normalised99 100 101def _rows_to_records(rows: Sequence[Sequence[Any]] | None) -> list[BondRecord]:102    normalised = _normalise_rows(rows)103    if not normalised:104        return []105 106    field_map: dict[str, list[str]] = {}107    max_columns = 0108    for row in normalised:109        if not row:110            continue111        field_name = str(row[0]).strip().lower()112        values = row[1:] if len(row) > 1 else []113        field_map[field_name] = values114        max_columns = max(max_columns, len(values))115 116    records: list[BondRecord] = []117    for col_index in range(max_columns):118        payload: dict[str, str] = {}119        for row_index, (label, attr) in enumerate(FIELD_SPECS):120            key = label.lower()121            values = field_map.get(key)122            if values is None and row_index < len(normalised):123                row_values = normalised[row_index]124                values = row_values[1:] if len(row_values) > 1 else []125            value = values[col_index] if values and col_index < len(values) else ""126            payload[attr] = value.strip()127        if not any(payload.values()):128            continue129        records.append(BondRecord(**payload))130    return records131 132 133def _build_status_message(pdf_count: int, record_count: int) -> str:134    docs_fragment = "document" if pdf_count == 1 else "documents"135    bonds_fragment = "bond" if record_count == 1 else "bonds"136    return f"Processed {pdf_count} {docs_fragment}; extracted {record_count} {bonds_fragment}."137 138 139def build_interface(pipeline: ExtractionPipeline | None = None) -> gr.Blocks:140    extractor = pipeline or ExtractionPipeline()141    download_dir = Path(gr_utils.get_cache_folder()) / "bond_extraction_downloads"142    download_dir.mkdir(parents=True, exist_ok=True)143 144    def _prepare_download(xml_doc: str) -> str:145        for existing in download_dir.glob("*.xml"):146            with contextlib.suppress(Exception):147                existing.unlink()148        target = download_dir / f"bonds_{uuid4().hex}.xml"149        target.write_text(xml_doc, encoding="utf-8")150        return str(target)151 152    def extract(pdf_files: list[gr.File], instructions: str) -> tuple:153        if not pdf_files:154            raise gr.Error("Please upload at least one PDF document.")155 156        pdf_paths = [Path(temp_file.name) for temp_file in pdf_files]157        records, xml_doc = extractor.run(pdf_paths, extra_instructions=(instructions or None))158 159        table_rows, table_headers = _format_table(records)160        status_message = _build_status_message(len(pdf_files), len(records))161 162        return (163            gr.update(value=table_rows, headers=table_headers),164            xml_doc,165            gr.update(value=status_message, visible=True),166            len(pdf_files),167            gr.update(value=_prepare_download(xml_doc), visible=True),168        )169 170    def regenerate_xml(table_rows: list[list[str]] | None, pdf_count: int) -> tuple[str, Any, Any]:171        records = _rows_to_records(table_rows)172        xml_doc = build_bond_xml(records)173        status_message = _build_status_message(pdf_count, len(records)) + " (updated manually)"174        return (175            xml_doc,176            gr.update(value=status_message, visible=True),177            gr.update(value=_prepare_download(xml_doc), visible=True),178        )179 180    with gr.Blocks(title="Bond Extraction Review") as demo:181        gr.HTML(182            """183            <style>184              .gradio-container {max-width: 1200px !important; margin: auto;}185              .bond-header h1 {font-size: 1.75rem; font-weight: 600; margin-bottom: 4px; color: #0f172a;}186              .bond-header p {margin: 0; color: #475569;}187              .bond-subtitle {text-align:right; color:#64748b; font-size:0.9rem;}188              .bond-run-button {padding: 0.9rem 1.5rem !important; font-size: 1rem !important;}189              .bond-table table {font-size: 0.95rem;}190              .bond-card {background:#ffffff;border:1px solid #e2e8f0;border-radius:12px;padding:20px;margin-bottom:18px;box-shadow:0 4px 12px rgba(15,23,42,0.04);} 191              .bond-card h3 {margin-top:0;color:#0f172a;}192              .bond-xml-preview textarea {min-height:320px;max-height:65vh;overflow-y:auto !important;resize:vertical;}193            </style>194            <div style="display:flex;align-items:center;justify-content:space-between;padding:16px 0 8px;" class="bond-header">195              <div>196                <h1>Institutional Bond Extraction</h1>197                <p>Upload term sheets, validate detected fields, and produce XML ready for downstream systems.</p>198              </div>199              <div class="bond-subtitle">200                <div>LLM-assisted parsing • Human review ready</div>201              </div>202            </div>203            """204        )205 206        default_rows, default_headers = _format_table([])207 208        with gr.Group(elem_classes=["bond-card"]):209            gr.Markdown(210                """211                ### Step 1 – Provide documents212 213                Upload one or more PDF term sheets. Optionally include guidance for nuances such as covenants or bespoke coupon language.214                """215            )216            pdf_input = gr.File(217                label="PDF document upload",218                file_count="multiple",219                file_types=[".pdf"],220                height=180,221            )222            with gr.Accordion("Add analyst guidance", open=False):223                instructions_input = gr.Textbox(224                    label="Instructions for the LLM",225                    placeholder="Highlight any nuances for this batch (optional)",226                    lines=4,227                )228            submit_button = gr.Button(229                "Run extraction",230                variant="primary",231                size="lg",232                elem_classes=["bond-run-button"],233            )234            status_label = gr.Markdown(visible=False)235            pdf_count_state = gr.State(0)236 237        with gr.Group(elem_classes=["bond-card"]):238            gr.Markdown(239                """240                ### Step 2 – Review & curate241 242                Adjust the detected values directly in the grid. When satisfied, regenerate the authoritative XML and download it for your internal systems.243                """244            )245 246            with gr.Tabs():247                with gr.TabItem("Review bonds"):248                    records_df = gr.Dataframe(249                        value=default_rows,250                        headers=default_headers,251                        datatype="str",252                        row_count=(len(FIELD_SPECS), "fixed"),253                        col_count=(2, "dynamic"),254                        interactive=True,255                        type="array",256                        label="Detected bond fields",257                        elem_classes=["bond-table"],258                    )259                    regenerate_button = gr.Button(260                        "Regenerate XML from edited table",261                        variant="secondary",262                    )263 264                with gr.TabItem("XML output"):265                    xml_output = gr.Textbox(266                        label="XML preview",267                        lines=16,268                        interactive=True,269                        elem_classes=["bond-xml-preview"],270                    )271                    xml_download = gr.DownloadButton(272                        label="Download XML",273                        value=None,274                        visible=False,275                        variant="primary",276                        size="md",277                    )278 279        submit_button.click(280            fn=extract,281            inputs=[pdf_input, instructions_input],282            outputs=[283                records_df,284                xml_output,285                status_label,286                pdf_count_state,287                xml_download,288            ],289        )290 291        regenerate_button.click(292            fn=regenerate_xml,293            inputs=[records_df, pdf_count_state],294            outputs=[xml_output, status_label, xml_download],295        )296 297    return demo298 299 300def launch(pipeline: ExtractionPipeline | None = None, **launch_kwargs) -> None:301    interface = build_interface(pipeline=pipeline)302    interface.launch(**launch_kwargs)303 304 305if __name__ == "__main__":  # pragma: no cover - interactive entry point306    launch()307