Spaces:
Sleeping
Sleeping
| """Gradio application for the FinRepair Seed60 formula audit.""" | |
| from __future__ import annotations | |
| import html | |
| import json | |
| import os | |
| import shutil | |
| from pathlib import Path | |
| from typing import Any | |
| os.environ.setdefault("GRADIO_TEMP_DIR", f"/tmp/finrepair-gradio-{os.getuid()}") | |
| import gradio as gr | |
| from audit_core import ( | |
| ANNOTATION_SCHEMA_VERSION, | |
| ANNOTATOR_SLOTS, | |
| CONFIDENCE_LEVELS, | |
| ISSUE_TYPES, | |
| AnnotationStore, | |
| AnnotationValidationError, | |
| AnnotatorSelectionError, | |
| CaseBundle, | |
| StorageError, | |
| build_annotation, | |
| current_annotations, | |
| resolve_annotator_slot, | |
| resolve_bundle_root, | |
| ) | |
| BUNDLE = CaseBundle.load( | |
| resolve_bundle_root(), | |
| verify_images=os.getenv("FINREPAIR_VERIFY_IMAGES", "1") != "0", | |
| ) | |
| STORE = AnnotationStore.from_environment() | |
| ANNOTATOR_LABELS = {key: label for label, key in ANNOTATOR_SLOTS.items()} | |
| CSS_PATH = Path(__file__).with_name("styles.css") | |
| GUIDE_MARKDOWN = Path(__file__).with_name("ANNOTATION_GUIDE.md").read_text(encoding="utf-8") | |
| IMAGE_EXPORT_DIR = Path(os.environ["GRADIO_TEMP_DIR"]) / "selected-images" | |
| IMAGE_EXPORT_DIR.mkdir(parents=True, exist_ok=True) | |
| ANNOTATION_EXPORT_DIR = Path(os.environ["GRADIO_TEMP_DIR"]) / "annotation-exports" | |
| ANNOTATION_EXPORT_DIR.mkdir(parents=True, exist_ok=True) | |
| MASKED_VIEW = "Masked" | |
| ORIGINAL_VIEW = "Original numbers" | |
| DECISION_CHOICES = [ | |
| ("Aligned", "Aligned"), | |
| ("Misaligned row mapping", "Misaligned"), | |
| ("Not verifiable / possible XBRL issue", "Not verifiable"), | |
| ] | |
| def export_image(case: dict[str, Any], original: bool = False) -> str: | |
| hash_field = "original_image_sha256" if original else "image_sha256" | |
| source = Path(BUNDLE.image_path(case, original=original)) | |
| destination = IMAGE_EXPORT_DIR / f"{case[hash_field]}.png" | |
| if not destination.is_file(): | |
| shutil.copy2(source, destination) | |
| return str(destination) | |
| def export_annotations(annotator_selection: str | None) -> str: | |
| try: | |
| annotator_key = resolve_annotator_slot(annotator_selection) | |
| document = STORE.load(annotator_key) | |
| except (AnnotatorSelectionError, StorageError) as exc: | |
| raise gr.Error(str(exc)) from exc | |
| destination = ANNOTATION_EXPORT_DIR / f"finrepair-{annotator_key}-annotations.json" | |
| temporary = destination.with_suffix(".json.tmp") | |
| temporary.write_text( | |
| json.dumps(document, indent=2, sort_keys=True, ensure_ascii=True) + "\n", | |
| encoding="utf-8", | |
| ) | |
| temporary.replace(destination) | |
| return str(destination) | |
| def operation_label(weight: int | float) -> str: | |
| numeric_weight = float(weight) | |
| verb = "Subtract" if numeric_weight < 0 else "Add" | |
| return f"{verb} ({numeric_weight:+g})" | |
| def reference_mapping_rows(case: dict[str, Any]) -> list[list[Any]]: | |
| proposal = case["proposal"] | |
| row_text = {str(row["row_id"]): str(row["text"]) for row in case["rows"]} | |
| parent = proposal["parent"] | |
| rows = [ | |
| [ | |
| "Parent", | |
| parent["label"], | |
| parent["row_id"], | |
| row_text.get(str(parent["row_id"]), ""), | |
| "Result", | |
| ] | |
| ] | |
| rows.extend( | |
| [ | |
| "Child", | |
| child["label"], | |
| child["row_id"], | |
| row_text.get(str(child["row_id"]), ""), | |
| operation_label(child["weight"]), | |
| ] | |
| for child in proposal["children"] | |
| ) | |
| return rows | |
| def formula_html(case: dict[str, Any]) -> str: | |
| proposal = case["proposal"] | |
| parent = html.escape(str(proposal["parent"]["label"])) | |
| terms: list[str] = [] | |
| for child in proposal["children"]: | |
| weight = float(child["weight"]) | |
| label = html.escape(str(child["label"])) | |
| operation = operation_label(weight) | |
| operation_class = "subtract" if weight < 0 else "add" | |
| terms.append( | |
| '<span class="formula-term">' | |
| f'<span class="operation-badge operation-{operation_class}">' | |
| f"{operation}</span>" | |
| f'<span class="term-label">{label}</span>' | |
| "</span>" | |
| ) | |
| return ( | |
| '<div class="formula-shell">' | |
| '<div class="formula-label">XBRL calculation-linkbase relation</div>' | |
| '<div class="formula-expression">' | |
| f'<strong class="formula-parent">{parent}</strong>' | |
| '<span class="formula-equals">=</span>' | |
| f'{"".join(terms)}</div>' | |
| '<div class="formula-sign-note"><strong>Keep the signs separate:</strong> ' | |
| 'red/green labels are fixed XBRL operations. Parentheses or minus signs ' | |
| 'printed in the statement are display signs. Never apply both signs twice.' | |
| "</div>" | |
| "</div>" | |
| ) | |
| def extracted_rows(case: dict[str, Any]) -> list[list[str]]: | |
| parent_id = case["proposal"]["parent"]["row_id"] | |
| child_ids = {child["row_id"] for child in case["proposal"]["children"]} | |
| result = [] | |
| for row in case["rows"]: | |
| role = ( | |
| "Reference parent" | |
| if row["row_id"] == parent_id | |
| else "Reference child" | |
| if row["row_id"] in child_ids | |
| else "" | |
| ) | |
| result.append([row["row_id"], row["text"], role]) | |
| return result | |
| def row_choices(case: dict[str, Any]) -> list[tuple[str, str]]: | |
| return [(f"{row['row_id']} | {row['text']}", row["row_id"]) for row in case["rows"]] | |
| def child_mapping_rows( | |
| case: dict[str, Any], | |
| annotation: dict[str, Any] | None = None, | |
| ) -> list[list[str]]: | |
| reference_children = case["proposal"]["children"] | |
| correction = (annotation or {}).get("correction") or {} | |
| corrected_children = correction.get("children") or [] | |
| use_correction = len(corrected_children) == len(reference_children) | |
| return [ | |
| [ | |
| str(reference["label"]), | |
| str(corrected_children[index]["row_id"]) | |
| if use_correction | |
| else str(reference["row_id"]), | |
| ] | |
| for index, reference in enumerate(reference_children) | |
| ] | |
| def progress_html(annotator_label: str | None, document: dict[str, Any] | None) -> str: | |
| completed = len(current_annotations(document)) | |
| total = len(BUNDLE.cases) | |
| percent = (100 * completed / total) if total else 0 | |
| annotator_text = html.escape(annotator_label) if annotator_label else "Select an annotator" | |
| return ( | |
| '<div class="progress-shell">' | |
| f'<div class="progress-copy"><strong>{completed} / {total}</strong> reviewed' | |
| f'<span>{percent:.1f}%</span><span class="annotator-name">{annotator_text}</span></div>' | |
| f'<progress max="{total}" value="{completed}"></progress></div>' | |
| ) | |
| def restore_values(case: dict[str, Any], annotation: dict[str, Any] | None) -> tuple[Any, ...]: | |
| if not annotation: | |
| return ( | |
| None, | |
| [], | |
| None, | |
| case["proposal"]["parent"]["row_id"], | |
| child_mapping_rows(case), | |
| "", | |
| ) | |
| correction = annotation.get("correction") or {} | |
| return ( | |
| annotation.get("decision"), | |
| annotation.get("issues") or [], | |
| annotation.get("confidence"), | |
| correction.get("parent_row_id") or case["proposal"]["parent"]["row_id"], | |
| child_mapping_rows(case, annotation), | |
| annotation.get("notes") or "", | |
| ) | |
| def case_view( | |
| index: int, | |
| annotator_key: str | None, | |
| document: dict[str, Any], | |
| status: str = "Ready", | |
| ) -> tuple[Any, ...]: | |
| index = max(0, min(int(index), len(BUNDLE.cases) - 1)) | |
| case = BUNDLE.cases[index] | |
| annotation = current_annotations(document).get(case["audit_id"]) | |
| decision, issues, confidence, parent, children, notes = restore_values( | |
| case, | |
| annotation, | |
| ) | |
| viewed_original = bool((annotation or {}).get("viewed_original", False)) | |
| reviewed = "Reviewed" if annotation else "Not reviewed" | |
| case_title = ( | |
| f"### Case {index + 1} of {len(BUNDLE.cases)}" | |
| f"\n`{case['audit_id']}` | {case['relation_size']} relation terms | {reviewed}" | |
| ) | |
| annotator_label = ANNOTATOR_LABELS.get(annotator_key) if annotator_key else None | |
| return ( | |
| index, | |
| progress_html(annotator_label, document), | |
| f"**{status}**", | |
| case_title, | |
| MASKED_VIEW, | |
| viewed_original, | |
| export_image(case, original=False), | |
| formula_html(case), | |
| reference_mapping_rows(case), | |
| extracted_rows(case), | |
| decision, | |
| issues, | |
| confidence, | |
| gr.update(choices=row_choices(case), value=parent), | |
| children, | |
| notes, | |
| ) | |
| def locked_view(message: str) -> tuple[Any, ...]: | |
| return ( | |
| 0, | |
| progress_html(None, None), | |
| f"**{message}**", | |
| "### Select an annotator", | |
| MASKED_VIEW, | |
| False, | |
| None, | |
| "", | |
| [], | |
| [], | |
| None, | |
| [], | |
| None, | |
| gr.update(choices=[], value=None), | |
| [["", ""]], | |
| "", | |
| ) | |
| def load_initial() -> tuple[Any, ...]: | |
| return case_view( | |
| 0, | |
| None, | |
| {"annotations": {}}, | |
| "Preview loaded; select Annotator 1 or Annotator 2 before saving", | |
| ) | |
| def load_annotator(annotator_selection: str | None) -> tuple[Any, ...]: | |
| try: | |
| annotator_key = resolve_annotator_slot(annotator_selection) | |
| document = STORE.load(annotator_key) | |
| except (AnnotatorSelectionError, StorageError) as exc: | |
| return locked_view(str(exc)) | |
| annotations = current_annotations(document) | |
| index = next( | |
| (i for i, case in enumerate(BUNDLE.cases) if case["audit_id"] not in annotations), | |
| 0, | |
| ) | |
| return case_view(index, annotator_key, document, "Progress loaded") | |
| def navigate( | |
| index: int, | |
| direction: str, | |
| annotator_selection: str | None, | |
| ) -> tuple[Any, ...]: | |
| try: | |
| annotator_key = resolve_annotator_slot(annotator_selection) | |
| document = STORE.load(annotator_key) | |
| except (AnnotatorSelectionError, StorageError) as exc: | |
| return locked_view(str(exc)) | |
| current = int(index or 0) | |
| if direction == "previous": | |
| target = max(0, current - 1) | |
| elif direction == "next": | |
| target = min(len(BUNDLE.cases) - 1, current + 1) | |
| else: | |
| annotations = current_annotations(document) | |
| scan = list(range(current + 1, len(BUNDLE.cases))) + list(range(0, current + 1)) | |
| target = next( | |
| (i for i in scan if BUNDLE.cases[i]["audit_id"] not in annotations), | |
| current, | |
| ) | |
| return case_view(target, annotator_key, document) | |
| def go_previous(index: int, annotator_selection: str | None) -> tuple[Any, ...]: | |
| return navigate(index, "previous", annotator_selection) | |
| def go_next(index: int, annotator_selection: str | None) -> tuple[Any, ...]: | |
| return navigate(index, "next", annotator_selection) | |
| def go_next_unreviewed(index: int, annotator_selection: str | None) -> tuple[Any, ...]: | |
| return navigate(index, "unreviewed", annotator_selection) | |
| def switch_image( | |
| index: int, | |
| view: str, | |
| viewed_original: bool, | |
| ) -> tuple[str, bool]: | |
| index = max(0, min(int(index or 0), len(BUNDLE.cases) - 1)) | |
| case = BUNDLE.cases[index] | |
| original = view == ORIGINAL_VIEW | |
| return export_image(case, original=original), bool(viewed_original or original) | |
| def save_and_next( | |
| index: int, | |
| annotator_selection: str | None, | |
| decision: str | None, | |
| issues: list[str] | None, | |
| confidence: str | None, | |
| mapped_parent: str | None, | |
| child_mappings: Any, | |
| notes: str, | |
| viewed_original: bool, | |
| ) -> tuple[Any, ...]: | |
| try: | |
| annotator_key = resolve_annotator_slot(annotator_selection) | |
| document = STORE.load(annotator_key) | |
| current = max(0, min(int(index or 0), len(BUNDLE.cases) - 1)) | |
| case = BUNDLE.cases[current] | |
| annotation = build_annotation( | |
| case=case, | |
| username=annotator_key, | |
| decision=decision, | |
| issues=issues, | |
| confidence=confidence, | |
| mapped_parent=mapped_parent, | |
| child_mappings=child_mappings, | |
| notes=notes, | |
| viewed_original=viewed_original, | |
| ) | |
| document["annotation_schema_version"] = ANNOTATION_SCHEMA_VERSION | |
| document["annotations"][case["audit_id"]] = annotation | |
| synced = STORE.save(annotator_key, document) | |
| except (AnnotatorSelectionError, AnnotationValidationError, StorageError) as exc: | |
| raise gr.Error(str(exc)) from exc | |
| annotations = current_annotations(document) | |
| scan = list(range(current + 1, len(BUNDLE.cases))) + list(range(0, current + 1)) | |
| target = next( | |
| (i for i in scan if BUNDLE.cases[i]["audit_id"] not in annotations), | |
| current, | |
| ) | |
| status = "Saved and synced" if synced else "Saved to local development storage" | |
| return case_view(target, annotator_key, document, status) | |
| def build_app() -> gr.Blocks: | |
| output_components: list[gr.Component] = [] | |
| with gr.Blocks( | |
| title="FinRepair PDF-XBRL Alignment Audit", | |
| fill_width=True, | |
| ) as demo: | |
| case_index = gr.State(0) | |
| viewed_original = gr.State(False) | |
| with gr.Row(elem_id="topbar"): | |
| gr.Markdown("# FinRepair PDF-XBRL Alignment Audit", elem_id="product-title") | |
| annotator_selector = gr.Dropdown( | |
| choices=list(ANNOTATOR_SLOTS), | |
| value=None, | |
| label="Annotator", | |
| filterable=False, | |
| elem_id="annotator-selector", | |
| min_width=220, | |
| ) | |
| download_button = gr.DownloadButton( | |
| "Download my annotations", | |
| elem_id="download-annotations", | |
| min_width=210, | |
| variant="secondary", | |
| ) | |
| with gr.Accordion("Annotation guide", open=True, elem_id="annotation-guide"): | |
| gr.Markdown(GUIDE_MARKDOWN) | |
| progress = gr.HTML(progress_html(None, None), elem_id="audit-progress") | |
| status = gr.Markdown("**Select an annotator to begin**", elem_id="save-status") | |
| case_title = gr.Markdown("### Select an annotator", elem_id="case-title") | |
| image_view = gr.Radio( | |
| choices=[MASKED_VIEW, ORIGINAL_VIEW], | |
| value=MASKED_VIEW, | |
| label="Table view", | |
| elem_id="image-view-selector", | |
| ) | |
| image = gr.Image( | |
| label="Financial statement", | |
| type="filepath", | |
| interactive=False, | |
| buttons=["fullscreen"], | |
| height="min(58vh, 680px)", | |
| elem_id="statement-image", | |
| ) | |
| formula = gr.HTML(formula_html(BUNDLE.cases[0]), elem_id="formula-display") | |
| with gr.Row(equal_height=False, elem_id="workbench"): | |
| with gr.Column(scale=6, min_width=420): | |
| reference_mapping = gr.Dataframe( | |
| headers=[ | |
| "Role", | |
| "XBRL label", | |
| "Mapped PDF row", | |
| "Extracted PDF label", | |
| "XBRL operation (fixed)", | |
| ], | |
| datatype=["str", "str", "str", "str", "str"], | |
| type="array", | |
| interactive=False, | |
| wrap=True, | |
| max_height=340, | |
| column_widths=[90, "28%", 120, "36%", 140], | |
| label="XBRL-to-PDF reference mapping", | |
| buttons=["fullscreen", "copy"], | |
| ) | |
| all_rows = gr.Dataframe( | |
| headers=["Row", "Extracted PDF label", "Reference role"], | |
| datatype=["str", "str", "str"], | |
| type="array", | |
| interactive=False, | |
| wrap=True, | |
| max_height=420, | |
| column_widths=[90, "70%", 140], | |
| show_search="search", | |
| label="All statement rows", | |
| buttons=["fullscreen", "copy"], | |
| ) | |
| with gr.Column(scale=4, min_width=340, elem_id="annotation-panel"): | |
| decision = gr.Radio( | |
| choices=DECISION_CHOICES, | |
| label="PDF-XBRL alignment", | |
| ) | |
| confidence = gr.Radio( | |
| choices=list(CONFIDENCE_LEVELS), | |
| label="Confidence", | |
| ) | |
| issues = gr.CheckboxGroup( | |
| choices=list(ISSUE_TYPES), | |
| label="Alignment issues", | |
| ) | |
| mapped_parent = gr.Dropdown( | |
| choices=[], | |
| label="Mapped parent PDF row (edit only if Misaligned)", | |
| filterable=True, | |
| ) | |
| child_mappings = gr.Dataframe( | |
| headers=["XBRL child (fixed)", "Mapped PDF row ID"], | |
| datatype=["str", "str"], | |
| type="array", | |
| interactive=True, | |
| column_widths=["70%", "30%"], | |
| label="Child row mappings (edit PDF row IDs only)", | |
| ) | |
| notes = gr.Textbox( | |
| label="Notes (required for Not verifiable / possible XBRL issue)", | |
| lines=3, | |
| max_lines=6, | |
| ) | |
| with gr.Row(elem_id="navigation"): | |
| previous_button = gr.Button("Previous", variant="secondary") | |
| next_button = gr.Button("Next", variant="secondary") | |
| unreviewed_button = gr.Button("Next unreviewed", variant="secondary") | |
| save_button = gr.Button("Save and next", variant="primary") | |
| output_components = [ | |
| case_index, | |
| progress, | |
| status, | |
| case_title, | |
| image_view, | |
| viewed_original, | |
| image, | |
| formula, | |
| reference_mapping, | |
| all_rows, | |
| decision, | |
| issues, | |
| confidence, | |
| mapped_parent, | |
| child_mappings, | |
| notes, | |
| ] | |
| image_view.change( | |
| switch_image, | |
| inputs=[case_index, image_view, viewed_original], | |
| outputs=[image, viewed_original], | |
| show_progress="hidden", | |
| api_visibility="private", | |
| ) | |
| annotator_load_event = annotator_selector.change( | |
| load_annotator, | |
| inputs=[annotator_selector], | |
| outputs=output_components, | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| annotator_load_event.then( | |
| export_annotations, | |
| inputs=[annotator_selector], | |
| outputs=[download_button], | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| demo.load( | |
| load_initial, | |
| inputs=None, | |
| outputs=output_components, | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| previous_button.click( | |
| go_previous, | |
| inputs=[case_index, annotator_selector], | |
| outputs=output_components, | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| next_button.click( | |
| go_next, | |
| inputs=[case_index, annotator_selector], | |
| outputs=output_components, | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| unreviewed_button.click( | |
| go_next_unreviewed, | |
| inputs=[case_index, annotator_selector], | |
| outputs=output_components, | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| save_event = save_button.click( | |
| save_and_next, | |
| inputs=[ | |
| case_index, | |
| annotator_selector, | |
| decision, | |
| issues, | |
| confidence, | |
| mapped_parent, | |
| child_mappings, | |
| notes, | |
| viewed_original, | |
| ], | |
| outputs=output_components, | |
| show_progress="full", | |
| api_visibility="private", | |
| ) | |
| save_event.then( | |
| export_annotations, | |
| inputs=[annotator_selector], | |
| outputs=[download_button], | |
| show_progress="minimal", | |
| api_visibility="private", | |
| ) | |
| return demo | |
| demo = build_app() | |
| if __name__ == "__main__": | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=int(os.getenv("PORT", "7860")), | |
| css_paths=[str(CSS_PATH)], | |
| footer_links=[], | |
| show_error=True, | |
| ssr_mode=False, | |
| ) | |