"""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(
''
f''
f"{operation}"
f'{label}'
""
)
return (
'"
)
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 (
''
f'
{completed} / {total} reviewed'
f'{percent:.1f}%{annotator_text}
'
f'