Spaces:
Sleeping
Sleeping
File size: 2,971 Bytes
f8bd24e b3c3675 f8bd24e b3c3675 f8bd24e b3c3675 f8bd24e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """Load extraction-stage table JSON using existing WaterScope media resolver.
Not a validated cell grid when structure_extracted=False. Preserve PDF word
coordinates, original strings and extraction quality flags; never silently
interpret position-only data as exact, verified spreadsheet cells.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Callable, Any
def load_table_evidence(folder: str, label: str, resolver: Callable[[str,str], Any]) -> dict | None:
manifest_path = resolver(folder, "manifest.json")
if manifest_path is None:
return None
manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8-sig"))
matches = [x for x in manifest.get("items", []) if x.get("label") == label and x.get("kind") == "table"]
if len(matches) != 1:
return None
filename = matches[0].get("data_file")
if not filename or Path(filename).name != filename or not filename.lower().endswith(".json"):
return None
path = resolver(folder, filename)
if path is None:
return None
raw = Path(path).read_bytes()
if len(raw) > 3_000_000:
raise ValueError("Table JSON exceeds 3MB limit")
table = json.loads(raw.decode("utf-8-sig"))
if table.get("kind") != "table" or table.get("label") != label:
raise ValueError("Table JSON does not match manifest label")
rows = table.get("text_rows") or []
cells = table.get("cells")
if not isinstance(rows, list) or len(rows) > 500:
raise ValueError("Invalid/oversized table text rows")
# Preserve every source row, but replace redundant per-word bounding boxes
# with a compact word@horizontal-position representation. A text row is NOT
# a verified table cell row, especially for merged/multi-line groups.
compact_rows = []
for row in rows:
words = []
for word in (row.get("words") or [])[:100]:
bbox = word.get("bbox_points") or []
token = str(word.get("text", ""))[:120]
if len(bbox) == 4:
words.append([token, round(float(bbox[0]), 1)])
else:
words.append([token, None])
compact_rows.append({"text": str(row.get("text", ""))[:1600], "positioned_words": words})
# Include all nonempty rows, including continuation rows and classifications.
# The original JSON remains in the dataset; this is only an inference view.
return {
"label": label, "data_file": filename,
"structure_extracted": bool(table.get("structure_extracted")),
"row_count": table.get("row_count"), "column_count": table.get("column_count"),
"cells": cells if table.get("structure_extracted") and isinstance(cells, list) else None,
"text_rows": compact_rows,
"source_quality": "structured_cells" if table.get("structure_extracted") and isinstance(cells,list)
else "positioned_text_unverified",
}
|