"""Ingest quber *fusion* output into the pgvector playground schema.
The package's only ingester. It requires the source PDF and two artifacts the
quber fusion workflow writes:
.unified.json — the unified DoclingDocument (spine: reading-order
text + headings, each with page + PDF-point bbox)
.tables.json — the corrected ExtractedTable list (LLM-cleaned
markdown, title, units, page, normalized regions)
The PDF is hashed for the document row and rendered for the footnote lookup
agent. The fusion report and three figure-run artifacts are optional inputs,
described below.
Mapping to the RAG schema, by chunk_type:
- text <- unified.json `texts` (bbox normalized to top-left 0..1),
and in the grouped stream one record per binding group
- picture <- unified.json `pictures` that carry a description or values
and are not a logo, icon or watermark
- figure_values <- one record per page of figure values no picture claims
- table <- tables.json, plus SoM-missed docling tables. A tables.json
chunk leads with its section heading and header lines,
then the cell-id HTML grid (the markdown when the table has
no grid), then its footnote block. Its region is already
normalized top-left 0..1.
- line_item <- one record per printed line of a gridded table that
carries a header depth
Grounding is cell-level: each table's `corrected_grid` supplies one
GroundedCell per markdown cell — text, a box normalized 0..1 top-left, a
provenance status, and the status inspector's note where one was taken. The
table chunk's content is an HTML table with an id on every cell
(`| `), so the agent can cite individual cells and
the app resolves them to overlays. Every
cell is served regardless of status; a cell whose box could not be traced
falls back to a box read off its boxed neighbours, then to the table's region,
and keeps its status/note so the UI can flag it. A table without a
`corrected_grid` gets no cell groundings. Its chunk is its markdown with no
cell ids, so it can be cited only as a whole table.
Footnotes attach to the chunks that reference them. Each table's and each
figure's markers are resolved in tier order: in-crop pairs, then a
reading-order scan, then the demand-driven lookup agent. A marker the
correction agent judged a cross-reference to a named section resolves to a
pointer at that heading rather than to definition text. The resolved text is
appended as a footnote block — on the whole-table chunk (all notes, plus
unmarked general notes), on each line record whose rendered rows carry the
marker, and on each picture chunk. The block is kept out of the embedding. It
stays in the stored content, so keyword search matches it. Unresolved markers,
unreferenced definitions, and unplaced markers land as tableFlag groundings,
quoted verbatim.
The fusion report (`.fusion.json`), when given, matches each SoM table to
its docling table. That match gives the table chunk its fused caption lines and
section heading, places the table in the reading order so its footnote scan
starts after it and stops at the next table, and gives a footnote text that
directly follows the docling table a parent link to the SoM table's chunk. A
table with no match, which includes every scanned table, falls back to its own
caption, title, subtitle and units, takes no local section heading, gets no
footnote parent link, and scans every line from its page onward for footnotes.
The report also names any docling table the Set-of-Mark engine missed
(`som_miss`). Those tables exist only in the unified document, so they are
ingested from there as plain table chunks — docling's own cell text,
table-level grounding — rather than dropped.
The figure run (`quber.core.figures.orchestrator`) writes the three optional
artifacts. Its scanned
tables (`.scanned-tables.json`) are `ExtractedTable`s like any other and
join the table engine's own, so a table read off a page image is chunked,
grounded and footnoted by the same code. Its run record (`.figures.json`)
names the parse tables those replaced, and the parse's reading of a replaced
table is then not ingested beside the scan's.
Its figure values (`.figure-values.json`) carry the
reconciled per-value reading of each figure with per-value provenance. Each
value becomes a grounding of its own — ref id `fv-`, its reconciliation
status and note, and the value's box on the page (already normalized top-left
0..1; a value with no box of its own borrows its picture's). The values are
also listed as `[fv-]`-tagged plain text lines, on the picture chunk they
belong to or, for a value with no picture, on its page's figure_values
record. The answering model cites a plotted value by that id as precisely as
a table cell. A value whose picture is a logo, icon or watermark is listed on
no chunk and exists only as its grounding.
"""
from __future__ import annotations
import hashlib
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any, Optional, Sequence
from loguru import logger
# Run at INFO: loguru's unconfigured default is DEBUG, which floods the
# upload job's log view with per-batch embedding lines.
logger.remove()
logger.add(sys.stderr, level="INFO")
from quber.agents.llm_client import FootnoteDef
from quber.core.figures.dpt3.blocks import SCAN_BLOCK
from quber.core.figures.graft import FOOTNOTE_MARKS_FIELD, FOOTNOTES_FIELD
from quber.core.figures.models import FigureValue, FigureValueRun
from quber.core.fusion.footnotes import (
FootnoteResolution,
absorb_lookup,
canonical_marker,
resolve_footnotes,
)
from quber.playground import db
from quber.playground.embedding import embed_document
from quber.settings import get_settings
DATA_DIR = get_settings().playground.data_dir
# The footnote block appended to a chunk's stored content. The answering
# model reads it; the embedding never sees it (see _embed_view) — the same
# note text appended to many records would drag their vectors toward each
# other, the boilerplate-domination failure measured on line records.
FOOTNOTE_HEADER = "\nFootnotes:\n"
# Section headings and captions are handled structurally rather than skipped: a
# retrieval unit must be able to answer a question, and a bare 3-word heading
# cannot — it can only name a thing, and as a standalone chunk it outscores the
# content it names whenever a question mentions that name. So heading text rides
# WITH the content it governs (prefixed onto every text chunk beneath it), and
# caption texts are skipped as chunks. A caption reaches a table chunk only
# through the table's header lines (see _table_head): the caption nodes fusion
# attached to its docling item, or the table's own caption field. Neither a
# picture chunk nor a SoM-missed docling table chunk carries its caption.
#: Picture classes that are identity marks with nothing inside to read: a
#: watermark, an icon, a logo. A picture of any other class is ingested when
#: it carries a description or values.
EXCLUDED_PICTURE_CLASSES = {"logo", "icon", "watermark"}
def _fetch_optional(src: Optional[str], dest: Path) -> Optional[dict]:
"""An optional artifact, read if it is there.
The figure run writes its scanned-tables file only when it read a table off
a page image, and its figure-values file whenever the value step ran,
which needs the page cells beside the parse, even if it read no values. A
document never put through the figure run has none of its artifacts. A
local path that does not exist is not an error — it says the run produced
nothing of that kind. An s3:// URI is fetched without that check, so a
missing object raises.
"""
if not src:
return None
if not src.startswith("s3://") and not Path(src).exists():
logger.info("no {} to read; the run wrote none", dest.name)
return None
return json.loads(_fetch(src, dest).read_text())
def _fetch(src: str, dest: Path) -> Path:
dest.parent.mkdir(parents=True, exist_ok=True)
if src.startswith("s3://"):
subprocess.run(["aws", "s3", "cp", src, str(dest)], check=True, capture_output=True)
else:
dest.write_bytes(Path(src).read_bytes())
return dest
def _governing(page_heading: str, heading: str) -> str:
"""The section vocabulary a chunk rides with: the page's opening heading
joined with the heading in effect at the chunk.
A presentation page opens with its subject — 'Commercial Portfolio
Geographic Diversification' — and then prints panel labels the parse also
calls headings ('U.S.', 'Europe', 'Australia'). Carrying only the latest
heading strips the page's subject from every chunk below the first panel
label, and a question phrased in the subject's words then never ranks the
chunk. Both ride along; when they agree, one.
"""
if page_heading and heading and page_heading.lower() != heading.lower():
return f"{page_heading} — {heading}"
return heading or page_heading
def _is_page_number(text: str, page: int, content_layer: str) -> bool:
"""Is this text the page printing its own number and nothing else?
The page's furniture layer is where the parse puts what it takes for page
decoration, but its guess is unreliable in the direction that matters: on a
financial deck that layer also holds the footnote definitions that qualify
the figures, so it cannot be excluded wholesale. A page number is the one
thing in it that can be named exactly rather than guessed at — a furniture
text whose whole content, read as an integer, is the number of the page it
sits on. Anything that fails to match is kept, so a deck whose printed
numbering runs off the page index loses nothing.
"""
return content_layer == "furniture" and text.isdigit() and int(text) == page
def _picture_notes(pic: dict) -> list[FootnoteDef]:
"""The notes an agent read off the page this figure is printed on."""
return [
FootnoteDef(marker=n.get("marker", ""), text=n.get("text", ""))
for n in ((pic.get("meta") or {}).get(FOOTNOTES_FIELD) or [])
if n.get("text")
]
def _picture_marks(pic: dict) -> list[dict]:
"""The footnote reference markers an agent read off this figure.
Each carries what it points at, so a cross-reference to a named part of the
document is never hunted among the lines printed near the figure.
"""
return list(((pic.get("meta") or {}).get(FOOTNOTE_MARKS_FIELD)) or [])
def _superseded_refs(run: Optional[dict]) -> set[str]:
"""The parse tables a scanned table replaced, from the figure run's record.
The fusion report names a table the table engine missed, so the ingest reads
that table out of the parse. Where the figure run has since replaced the
parse's reading with the scan's, reading it out of the parse again would
state one region twice.
"""
if not run:
return set()
return {
ref
for scan in run.get("scans", [])
for table in scan.get("tables", [])
for ref in [table.get("table_ref")]
if ref
}
def _picture_class(pic: dict) -> str:
"""The parse's top predicted class for a picture, or '' if unclassified."""
meta = pic.get("meta") or {}
preds = ((meta.get("classification") or {}).get("predictions")) or []
return preds[0].get("class_name", "") if preds else ""
def _cited_line(fid: str, name: str, value: str) -> str:
"""One value line the answering model can cite.
The `[fv-n]` tag is the citation contract: the model quotes the id and the
app resolves it to that value's own grounding box, so the overlay is the
printed value rather than whatever chunk carried the line. Every chunk
that serves figure values builds its lines here — the tag must never be
reimplemented per chunk kind, because a line without it can only be cited
at chunk granularity.
"""
return f"[{fid}] {name}: {value}" if name else f"[{fid}] {value}"
def _picture_text(pic: dict, values: Sequence[tuple[str, FigureValue]] = ()) -> str:
"""What a picture says, or '' when it says nothing worth retrieving.
A chart's plotted values live in the description the page scan wrote, which
names the axes, the units, the legend and every plotted figure, so a
picture carrying one is a retrievable record like any other.
`values` are this picture's reconciled figure values, each with its
grounding ref id. They are appended as an id-tagged line per value, so the
answering model can cite a plotted value by id, as it cites a grid cell,
and the app resolves it to an overlay.
Excluded: a watermark, an icon, a logo. They are identity marks repeated on
every page with nothing inside to read. A picture with no description and
no values is excluded too, having nothing to say.
"""
if _picture_class(pic) in EXCLUDED_PICTURE_CLASSES:
return ""
description = ((pic.get("meta") or {}).get("description") or {}).get("text") or ""
text = description.strip()
if values:
lines = []
for fid, v in values:
name = " ".join(p for p in (v.label, v.series) if p)
lines.append(_cited_line(fid, name, v.value))
text = (text + "\nValues:\n" if text else "Values:\n") + "\n".join(lines)
return text
def _unanchored_value_records(
figure_values: Sequence[tuple[str, FigureValue]],
page_first_heading: dict[int, str],
) -> list[tuple[str, str, int, Optional[dict], str, Optional[str]]]:
"""One searchable record per page of figure values no picture claims.
An anchored value rides its picture chunk's searchable text, unless the
picture is a logo, icon or watermark: that picture has no chunk, and its
values are not collected here either, so they exist only as grounding
rows. An unanchored value has no chunk to ride and its grounding row alone
is invisible to retrieval — a printed number the index never hears of.
Each page's unanchored values become one record of id-tagged label-value
lines under the page's opening heading, id `figure-values-p`. The id
tags are built by `_cited_line`, as `_picture_text` builds a picture's
values: the answering model cites the value's own
`fv-` id, which resolves to that value's tight box rather than the
record's. The record itself carries NO box, deliberately: a citation of
the record alone names the page, and the only rectangles ever drawn are
the values' own. A synthetic union box was tried and rendered as a
misleading figure-sized highlight over the tight one.
"""
by_page: dict[int, list[tuple[str, FigureValue]]] = {}
for fid, v in figure_values:
if not v.picture_ref:
by_page.setdefault(v.page, []).append((fid, v))
records: list[tuple[str, str, int, Optional[dict], str, Optional[str]]] = []
for page1 in sorted(by_page):
lines = []
for fid, v in by_page[page1]:
# The chart title leads the name here and not in the picture path:
# a picture chunk IS its chart, a page record must say which one.
name = " — ".join(part for part in (v.chart_title, v.label, v.series) if part)
lines.append(_cited_line(fid, name, v.value))
content = "\n".join(lines)
section = page_first_heading.get(page1, "")
if section and section.lower() not in content.lower():
content = f"{section} — {content}"
records.append((f"figure-values-p{page1}", "figure_values", page1 - 1, None, content, None))
return records
def _picture_bbox(
doc: dict, sizes: dict[int, tuple[float, float]], picture_ref: Optional[str]
) -> Optional[dict[str, float]]:
"""The parse picture's own normalized box, for a value with no box of its own."""
if not picture_ref:
return None
for pic in doc.get("pictures", []):
if pic.get("self_ref") == picture_ref:
prov = (pic.get("prov") or [{}])[0]
if not prov.get("bbox"):
return None
page1 = prov.get("page_no", 1)
w, h = sizes.get(page1, (1.0, 1.0))
return _norm_bbox(prov["bbox"], w, h)
return None
def _reading_order_refs(doc: dict) -> list[str]:
"""Every item ref in the document, in reading order.
Descends the whole tree. An item's own children are visited straight after
it, so a table's footnotes and a picture's labels keep their place in the
flow instead of being unreachable.
Any text, table or picture the document holds but never links into the
tree is appended at the end, so an item is dropped only by a rule that
names it, never by an accident of where it sits. Returns an empty list when
the artifact carries no body tree. The caller then walks the texts array
alone, in its own order, so no picture or group is ingested and no table
gets a reading-order position.
"""
pools = {kind: doc.get(kind, []) for kind in ("texts", "tables", "pictures", "groups")}
out: list[str] = []
seen: set[str] = set()
def visit(ref: str) -> None:
if not ref or ref in seen:
return
seen.add(ref)
kind, _, idx = ref.lstrip("#/").partition("/")
items = pools.get(kind, [])
try:
item = items[int(idx)]
except (ValueError, IndexError):
return
# A group ref rides in the flow ahead of its members. In the grouped
# stream, a group that binds a label to its value is served as one
# record and the caller skips its members. Any other group, and every
# group in the flat stream, is a transparent wrapper the caller passes
# over, and its members stand alone.
out.append(ref)
for child in item.get("children") or []:
visit(child.get("$ref", ""))
for child in doc.get("body", {}).get("children") or []:
visit(child.get("$ref", ""))
if not out:
return []
for kind in ("texts", "tables", "pictures"):
for i in range(len(pools[kind])):
ref = f"#/{kind}/{i}"
if ref not in seen:
logger.warning(
"{ref} is in the document but linked into no parent; ingesting it anyway", ref=ref
)
out.append(ref)
seen.add(ref)
return out
def _page_sizes(doc: dict) -> dict[int, tuple[float, float]]:
"""page_no -> (width, height) from the DoclingDocument pages map."""
out: dict[int, tuple[float, float]] = {}
for k, v in (doc.get("pages") or {}).items():
size = v.get("size", {})
out[int(k)] = (float(size.get("width", 0)) or 1.0, float(size.get("height", 0)) or 1.0)
return out
def _norm_bbox(prov_bbox: dict[str, Any], page_w: float, page_h: float) -> Optional[dict[str, float]]:
"""Normalize a docling prov bbox to top-left origin, 0..1.
docling boxes are PDF points with coord_origin BOTTOMLEFT, so t/b are
measured up from the page bottom (t > b). Flip Y to a top-left origin.
"""
left, right = prov_bbox.get("l"), prov_bbox.get("r")
top_pt, bottom_pt = prov_bbox.get("t"), prov_bbox.get("b")
if left is None or right is None or top_pt is None or bottom_pt is None:
return None
origin = prov_bbox.get("coord_origin", "BOTTOMLEFT")
if origin == "BOTTOMLEFT":
top, bottom = (page_h - top_pt) / page_h, (page_h - bottom_pt) / page_h
else: # TOPLEFT
top, bottom = top_pt / page_h, bottom_pt / page_h
return {
"left": max(0.0, left / page_w),
"top": max(0.0, min(top, bottom)),
"right": min(1.0, right / page_w),
"bottom": min(1.0, max(top, bottom)),
}
def _esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">")
def _pg_text(s: Optional[str]) -> Optional[str]:
"""Strip NUL characters: Postgres text fields reject 0x00. They reach us
from PDF text layers that encode glyphs like cover-page checkboxes as NUL."""
return s.replace("\x00", "") if s is not None else None
def _embed_view(content: str) -> str:
"""The text a reader reads: tags dropped, cells separated by ' | '. The
footnote block is cut before embedding. It stays in the stored content."""
content = content.split(FOOTNOTE_HEADER)[0]
text = re.sub(r" | ]*>", " | ", content)
text = re.sub(r"<[^>]+>", " ", text)
return re.sub(r"[ \t]+", " ", text).strip()
def _table_head(t: dict, item: Optional[dict], texts: list[dict]) -> list[str]:
"""The header lines of a table chunk: the caption nodes fusion attached to
the table's docling item, in order (the sentence that introduces the table,
its printed name, its units), read from the unified document so the chunk
says what the fused document says. A table with no fused item falls back to
its own fields."""
if item:
lines: list[str] = []
for ref in item.get("captions") or []:
cref = ref.get("$ref", "") if isinstance(ref, dict) else str(ref)
if not cref.startswith("#/texts/"):
continue
idx = int(cref.rsplit("/", 1)[1])
text = (texts[idx].get("text") or "").strip() if idx < len(texts) else ""
if text:
lines.append(text)
if lines:
return lines
return [p for p in (t.get("caption"), t.get("title"), t.get("subtitle"), t.get("units")) if p]
def _grid_html(table_index: int, grid: list[list[dict]], row_offsets: Optional[list[int]] = None) -> str:
"""Render a corrected_grid as an HTML table with an id on every cell.
Ids follow `t--` so they are unique per document,
short enough for the agent to cite verbatim, and resolve directly back to
the grid position they came from. `row_offsets` supplies the true row
number of each rendered row when `grid` is a subset (a line-item record
rendering header + one line keeps the line's real ids).
"""
rows = []
for k, row in enumerate(grid):
r = row_offsets[k] if row_offsets else k
tds = "".join(
f'| {_esc(cell.get("text") or "")} | '
for c, cell in enumerate(row)
)
rows.append(f"{tds}
")
return "\n" + "\n".join(rows) + "\n
"
def _cell_vicinity(
grid: list[list[dict]], r: int, c: int, region: Optional[tuple]
) -> Optional[tuple[float, float, float, float]]:
"""Approximate box for an unboxed cell, read off its boxed neighbors.
The column's x-extent (union of boxes in column c) crossed with the row's
y-extent (union of boxes in row r) marks the empty spot where the cell
sits. A fully unboxed row falls back to the band between the table's top
edge and the first boxed row — the header area. Returns None when either
axis cannot be derived; the caller falls back to the whole table region.
"""
col = [row[c]["box"] for row in grid if len(row) > c and row[c].get("box")]
rowb = [cell["box"] for cell in grid[r] if cell.get("box")]
allb = [cell["box"] for row in grid for cell in row if cell.get("box")]
if not (col or rowb):
return None
if col:
x1, x2 = min(b[0] for b in col), max(b[2] for b in col)
elif region:
x1, x2 = region[0], region[2]
else:
return None
if rowb:
y1, y2 = min(b[1] for b in rowb), max(b[3] for b in rowb)
elif region and allb and min(b[1] for b in allb) > region[1]:
y1, y2 = region[1], min(b[1] for b in allb)
else:
return None
return (x1, y1, x2, y2)
def _header_band(
grid: list[list[dict]], region: Optional[tuple]
) -> Optional[tuple[float, float, float, float]]:
"""The table's header area: from the region's top edge down to the top of
the first boxed row. Dropped printed text lives there by definition, so it
is the honest vicinity for a dropped-text flag. None without a region, when
no cell is boxed, or when the first boxed row starts at or above the
region's top edge."""
if not region:
return None
allb = [cell["box"] for row in grid for cell in row if cell.get("box")]
if not allb or min(b[1] for b in allb) <= region[1]:
return None
return (region[0], region[1], region[2], min(b[1] for b in allb))
def _docling_table_text(t: dict) -> str:
"""Reconstruct a docling table's content as pipe-separated rows."""
rows: dict[int, dict[int, str]] = {}
for cell in (t.get("data") or {}).get("table_cells", []):
text = (cell.get("text") or "").strip()
if not text:
continue
r = cell.get("start_row_offset_idx", 0)
c = cell.get("start_col_offset_idx", 0)
rows.setdefault(r, {})[c] = text
return "\n".join(" | ".join(rows[r][c] for c in sorted(rows[r])) for r in sorted(rows))
def _lookup_unresolved(resolution: FootnoteResolution, pdf_path: Path, page1: int) -> FootnoteResolution:
"""The demand-driven last tier: ask the lookup agent to find the still-
unresolved markers on the rendered pages following the table. Any failure
(no PDF, no credentials, a failed call) leaves the markers unresolved in
the exception record — the lookup never takes an ingest down."""
if not pdf_path.exists():
return resolution
try:
import asyncio
import io
from pdf2image import convert_from_path
from quber.agents.footnote_lookup import PydanticAIFootnoteLookup
agent = PydanticAIFootnoteLookup()
images = convert_from_path(str(pdf_path), dpi=150, fmt="png", first_page=page1, last_page=page1 + 2)
pngs: list[bytes] = []
for image in images:
buf = io.BytesIO()
image.save(buf, format="PNG")
pngs.append(buf.getvalue())
report = asyncio.run(agent.lookup(resolution.unresolved, pngs, page1))
except Exception as exc:
logger.warning("page {p}: footnote lookup unavailable; markers stay unresolved: {e}", p=page1, e=exc)
return resolution
if report is None:
return resolution
return absorb_lookup(resolution, report.found)
def ingest(
doc_key: str,
unified_json: str,
tables_json: str,
source_pdf: str,
fusion_json: Optional[str] = None,
scanned_tables_json: Optional[str] = None,
figures_json: Optional[str] = None,
figure_values_json: Optional[str] = None,
filename: Optional[str] = None,
chunking: str = "flat",
) -> int:
"""`chunking` selects the ingestion stream. `flat` (the default) serves
every printed line on its own: a stat panel's label lands in one chunk
and its number in another, and each text and picture chunk is tagged
with only the most recent heading the walk passed — nothing is bundled.
A table chunk carries a heading only when the fusion report matches it
to a docling table the walk reached, so a scanned table carries none. Technically: every group is a transparent wrapper,
single-heading ride-along. `grouped`: a binding group — a
key_value_area or a scan block — is served as one chunk so a label
travels with its number, and every chunk rides with the page's opening
heading joined to the local one. A per-page figure_values record carries
the page's opening heading in both streams. The upload app picks the
stream per track (`TRACKS` in `quber.playground.app`): the dpt-2 track
ingests flat, the dpt-3 track, its default, ingests grouped.
"""
grouped = chunking == "grouped"
pdf_local = _fetch(source_pdf, DATA_DIR / f"{doc_key}.pdf")
doc = json.loads(_fetch(unified_json, DATA_DIR / f"{doc_key}.unified.json").read_text())
tables = json.loads(_fetch(tables_json, DATA_DIR / f"{doc_key}.tables.json").read_text())
report = (
json.loads(_fetch(fusion_json, DATA_DIR / f"{doc_key}.fusion.json").read_text())
if fusion_json
else None
)
# A table read off a page image arrives as an ExtractedTable like any other,
# from the figure run rather than from the table engine. It is appended to the
# same list so it is chunked, grounded and footnoted the same way.
scanned = _fetch_optional(scanned_tables_json, DATA_DIR / f"{doc_key}.scanned-tables.json")
tables.extend(scanned or [])
# Where such a table replaced one the parse held, the parse's reading of that
# region is superseded and must not be ingested beside it.
superseded = _superseded_refs(_fetch_optional(figures_json, DATA_DIR / f"{doc_key}.figures.json"))
sizes = _page_sizes(doc)
# The reconciled figure values, each assigned the grounding ref id it will
# be cited by. Grouped by picture so the picture chunk can list its own.
value_run = _fetch_optional(figure_values_json, DATA_DIR / f"{doc_key}.figure-values.json")
figure_values: list[tuple[str, FigureValue]] = (
[(f"fv-{n}", v) for n, v in enumerate(FigureValueRun.model_validate(value_run).values)]
if value_run
else []
)
values_by_picture: dict[str, list[tuple[str, FigureValue]]] = {}
for fid, v in figure_values:
if v.picture_ref:
values_by_picture.setdefault(v.picture_ref, []).append((fid, v))
# --- Build chunk records: (ref_id, ref_type, page0, bbox, content, parent) ---
# parent is the whole-table record a record belongs to: a line-item
# record's table, or the SoM table a footnote text directly follows.
# None on every other record.
records: list[tuple[str, str, int, Optional[dict], str, Optional[str]]] = []
# Cell groundings: (ref_id, ref_type, page0, bbox, position, status, note, cell_text)
cell_rows: list[tuple[str, str, int, Optional[dict], dict, Optional[str], Optional[str], str]] = []
# Walk the body tree in reading order, carrying the governing section
# heading onto every text chunk beneath it. A question that names a
# section then pulls the section's content, and the heading never
# competes as a bare fragment. Caption texts are skipped: they are not
# standalone units (see the note above EXCLUDED_PICTURE_CLASSES).
# Each SoM table's docling counterpart (for the governing heading and its
# reading-order position), and the reverse map (for a footnote chunk's
# parent link back to the SoM table chunk it follows).
som_docling_ref: dict[int, str] = {}
docling_to_som: dict[str, str] = {}
if report:
for m in report.get("matches", []):
refs = m.get("docling_table_refs", [])
for si in m.get("som_indices", []):
if refs:
som_docling_ref[si] = refs[0]
docling_to_som.setdefault(refs[0], f"#/tables/{si}")
all_texts = doc.get("texts", [])
all_pictures = doc.get("pictures", [])
all_groups = doc.get("groups", [])
order = _reading_order_refs(doc)
walk_refs = order or [f"#/texts/{i}" for i in range(len(all_texts))]
heading = ""
# The heading in effect when the walk passes each docling table. Tables
# get the same ride-along as text: a table titled only by a printed date
# line is unfindable by the section vocabulary that governs it.
table_headings: dict[str, str] = {}
# The walk also captures what footnote resolution needs: every content
# line with its page in reading order, where each table sits in that
# stream (a marker's definition is searched only AFTER its table), and
# the section headings (the Notes-pointer universe).
ordered_texts: list[tuple[str, int]] = []
table_positions: dict[str, int] = {}
table_sequence: list[str] = []
headings: list[tuple[str, int]] = []
# Each page's opening heading: the page's subject, which panel labels the
# parse also calls headings would otherwise strip from the ride-along.
page_first_heading: dict[int, str] = {}
# A footnote text directly after a docling table belongs to that table: it
# gets a parent link to the SoM table chunk the fusion report matched to
# it, and none when there is no match. A section heading reached on its
# own, or a non-footnote text that survives the caption, empty-text and
# page-number skips, breaks the adjacency. A caption, a page-number line,
# a picture or a group does not.
last_table_ref: Optional[str] = None
# Where each element that owns a footnote block sits in the line stream.
# A figure's definitions are searched from its own position up to the next
# such element: a table or a figure with markers or notes.
owner_positions: list[int] = []
# Each ingested figure carrying markers or notes, as (index in `records`,
# the marks the agent read off it, its notes, its page, its position in
# the line stream), resolved after the walk.
picture_notes: list[tuple[int, list[dict], list[FootnoteDef], int, int]] = []
# In the grouped stream, members of a scan block or a key_value_area are
# served through their group's single record, so their own refs are
# skipped when the walk reaches them.
grouped_members: set[str] = set()
for ref_id in walk_refs:
if ref_id.startswith("#/groups/"):
# A group that binds a label to its value is served as ONE record,
# so the label travels with its number. Two producers write such
# groups: the parse itself, as a key_value_area over a stat
# panel's pair, and the figure run's scan blocks, which group what
# the parse left as bare page text. A list group is neither — its
# items answer questions one at a time and stay separate records.
# The flat stream treats every group as a transparent wrapper.
if not grouped:
continue
gidx = int(ref_id.rsplit("/", 1)[1])
group = all_groups[gidx] if gidx < len(all_groups) else {}
if group.get("name") != SCAN_BLOCK and group.get("label") != "key_value_area":
continue
members = []
for child in group.get("children", []):
mref = child.get("$ref") or ""
if not mref.startswith("#/texts/"):
continue
t = all_texts[int(mref.rsplit("/", 1)[1])]
text = (t.get("text") or "").strip()
if not text:
continue
# A heading grouped with its content still governs what
# follows the group: it feeds the heading stream exactly as it
# would have standing alone, and its text stays in the group's
# own chunk.
if t.get("label") == "section_header":
heading = text
mpage = (t.get("prov") or [{}])[0].get("page_no", 1)
headings.append((text, mpage))
page_first_heading.setdefault(mpage, text)
members.append((mref, t, text))
if not members:
continue
grouped_members.update(mref for mref, _t, _text in members)
prov = (members[0][1].get("prov") or [{}])[0]
page1 = prov.get("page_no", 1)
w, h = sizes.get(page1, (1.0, 1.0))
boxes = [
b
for _mref, t, _text in members
for p in [(t.get("prov") or [{}])[0]]
if p.get("bbox")
for b in [_norm_bbox(p["bbox"], w, h)]
if b
]
bbox = (
{
"left": min(b["left"] for b in boxes),
"top": min(b["top"] for b in boxes),
"right": max(b["right"] for b in boxes),
"bottom": max(b["bottom"] for b in boxes),
}
if boxes
else None
)
for _mref, _t, text in members:
ordered_texts.append((text, page1))
content = "\n".join(text for _mref, _t, text in members)
section = _governing(page_first_heading.get(page1, ""), heading)
if section and section.lower() not in content.lower():
content = f"{section} — {content}"
records.append((group.get("self_ref") or ref_id, "text", page1 - 1, bbox, content, None))
continue
if ref_id in grouped_members:
continue
if ref_id.startswith("#/tables/"):
table_headings[ref_id] = heading
table_positions[ref_id] = len(ordered_texts)
table_sequence.append(ref_id)
owner_positions.append(len(ordered_texts))
last_table_ref = ref_id
continue
if ref_id.startswith("#/pictures/"):
pic = all_pictures[int(ref_id.rsplit("/", 1)[1])]
content = _picture_text(
pic, values_by_picture.get(pic.get("self_ref") or "") or values_by_picture.get(ref_id, [])
)
if not content:
continue
prov = (pic.get("prov") or [{}])[0]
page1 = prov.get("page_no", 1)
w, h = sizes.get(page1, (1.0, 1.0))
bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
section = _governing(page_first_heading.get(page1, ""), heading) if grouped else heading
if section and section.lower() not in content.lower():
content = f"{section} — {content}"
marks = _picture_marks(pic)
defs = _picture_notes(pic)
if marks or defs:
owner_positions.append(len(ordered_texts))
# The notes these markers point at are printed below the figure
# and have not been walked yet, so the block is attached in a
# second pass once every content line is known.
picture_notes.append((len(records), marks, defs, page1, len(ordered_texts)))
records.append((pic.get("self_ref") or ref_id, "picture", page1 - 1, bbox, content, None))
continue
if not ref_id.startswith("#/texts/"):
continue
idx = int(ref_id.rsplit("/", 1)[1])
if idx >= len(all_texts):
continue
t = all_texts[idx]
text = (t.get("text") or "").strip()
label = t.get("label")
prov = (t.get("prov") or [{}])[0]
page1 = prov.get("page_no", 1)
if label == "section_header":
heading = text
if text:
headings.append((text, page1))
page_first_heading.setdefault(page1, text)
last_table_ref = None
continue
if not text or label == "caption":
continue
if _is_page_number(text, page1, t.get("content_layer") or ""):
continue
ordered_texts.append((text, page1))
parent = docling_to_som.get(last_table_ref or "") if label == "footnote" else None
if label != "footnote":
last_table_ref = None
w, h = sizes.get(page1, (1.0, 1.0))
bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
section = _governing(page_first_heading.get(page1, ""), heading) if grouped else heading
content = f"{section} — {text}" if section and section.lower() not in text.lower() else text
# page stored 0-based; the app adds one when it reads a chunk's page.
records.append((t.get("self_ref") or ref_id, "text", page1 - 1, bbox, content, parent))
# A figure states its markers the way a table does, so its notes are looked
# up the same way. The marker alone says a qualification exists and not what
# it says: a chart labelled "Undepreciated Book Equity Value" with a raised 1
# reads as a plain figure without the note saying it excludes a
# noncontrolling interest.
# Every note read on a page, and every marker printed on it, pooled across the
# figures. A page prints ONE note block at its foot and it serves each figure
# above it: on one earnings deck, three charts on a page all print a marker 1
# against a single note defining it. The agent is asked which figure each note
# belongs to and has to answer with one, so the other two figures lose it.
page_defs: dict[int, list[FootnoteDef]] = {}
page_markers: dict[int, set[str]] = {}
for _i, marks, defs, page1, _p in picture_notes:
page_defs.setdefault(page1, []).extend(defs)
page_markers.setdefault(page1, set()).update(
canonical_marker(m.get("marker", "")) for m in marks if m.get("marker")
)
orphans_seen: set[tuple[int, str]] = set()
for index, marks, defs, page1, pos in picture_notes:
markers = [m.get("marker", "") for m in marks if m.get("marker")]
if not markers and not defs:
continue
later = [p for p in owner_positions if p > pos]
end = later[0] if later else len(ordered_texts)
trailing = ordered_texts[pos:end]
section_keys = {canonical_marker(m.get("marker", "")) for m in marks if m.get("kind") == "section"}
# The agent's own pairs come first, exactly as a table's do. It read the
# notes off the page image, so a marker is answered by the line printed
# for it rather than by whichever line a scan happens to match. The scan
# over the page's lines stays behind it as a fallback.
# This figure's own notes, then the ones the page printed for its other
# figures. A page prints ONE note block and it serves every figure above
# it, so a marker this figure prints is answered by that block whichever
# figure the agent filed the note under. Both go in ahead of the
# reading-order scan: a note an agent read off the page is better
# evidence than a line whose leading character happens to match. Own
# notes are listed first, so a figure's own always wins a collision.
own_keys = {canonical_marker(d.marker) for d in defs}
pool = list(defs) + [
d for d in page_defs.get(page1, []) if canonical_marker(d.marker) not in own_keys
]
resolution = resolve_footnotes(markers, pool, page1, trailing, headings, section_keys)
# A definition taken as read from a neighbour's notes (source `table`)
# is relabelled `sibling`. Nothing reads the label: the attached block
# holds only marker and text, and the label is neither stored nor
# logged.
resolution = FootnoteResolution(
resolved=[
r
if r.source != "table" or canonical_marker(r.marker) in own_keys
else r.model_copy(update={"source": "sibling"})
for r in resolution.resolved
],
unresolved=list(resolution.unresolved),
unreferenced=list(resolution.unreferenced),
)
if resolution.unresolved:
resolution = _lookup_unresolved(resolution, DATA_DIR / f"{doc_key}.pdf", page1)
ref_id, ref_type, page0, bbox, body, parent = records[index]
notes = [f"{r.marker} — {r.text}" for r in resolution.resolved]
if notes:
records[index] = (
ref_id,
ref_type,
page0,
bbox,
body + FOOTNOTE_HEADER + "\n".join(notes),
parent,
)
logger.info("Attached {} footnote(s) to picture {} (page {})", len(notes), ref_id, page1)
# A note is orphaned only when NO marker on the page names it, and it is
# then the PAGE's exception rather than each figure's. Both halves matter:
# judged per figure, every shared note would look unreferenced to the
# figures that did not print its marker, and every genuinely orphaned note
# would then be reported once per figure on the page.
orphans: list[str] = []
for d in resolution.unreferenced:
key = canonical_marker(d.marker)
if key in page_markers.get(page1, set()) or (page1, key) in orphans_seen:
continue
orphans_seen.add((page1, key))
orphans.append(f"{d.marker} {d.text}".strip())
# Footnote exceptions, quoted verbatim, the same three a table reports.
# Without the flag, a figure whose marker finds no definition is the
# failure that cannot be seen by looking at what a run produced: a
# missing qualification and a clean figure read identically.
flags = (
[("footnote_unresolved", m) for m in resolution.unresolved]
+ [("footnote_unreferenced", quoted) for quoted in orphans]
+ [
("footnote_marker_unplaced", m.get("marker", ""))
for m in marks
if not (m.get("label") or "").strip()
]
)
for k, (status, quoted) in enumerate(flags):
logger.warning(
"picture {r} (page {p}): {status}: {q!r}", r=ref_id, p=page1, status=status, q=quoted
)
cell_rows.append(
(f"{ref_id}-fn-{k}", "tableFlag", page0, bbox, {"chunk_id": ref_id}, status, None, quoted)
)
records.extend(_unanchored_value_records(figure_values, page_first_heading))
# One grounding per reconciled figure value, the way a table serves one per
# cell. The value's own box (already normalized top-left 0..1) is its
# overlay; a value with no box of its own borrows its picture's, and only
# when the picture cannot be resolved is it served without one. Status and
# note ride along so the UI flags an unreconciled value like a flagged cell.
for fid, v in figure_values:
box = v.box or _picture_bbox(doc, sizes, v.picture_ref)
position = {
"label": v.label,
"series": v.series,
"chart": v.chart_title,
"picture_ref": v.picture_ref,
}
if not v.picture_ref:
position["chunk_id"] = f"figure-values-p{v.page}"
cell_rows.append(
(
fid,
"figureValue",
v.page - 1,
dict(box) if box else None,
position,
v.status,
v.note,
v.value,
)
)
unified_tables = {tb.get("self_ref"): tb for tb in doc.get("tables", [])}
for i, t in enumerate(tables):
page1 = t.get("page", 1)
region = t.get("content_region") or t.get("som_region")
bbox = None
if region:
x1, y1, x2, y2 = region
bbox = {"left": x1, "top": y1, "right": x2, "bottom": y2}
# The corrected grid gives one GroundedCell per markdown cell. Render
# it as HTML with per-cell ids so the agent can cite cells; fall back
# to the plain markdown for tables without a grid (charts, image
# tables).
grid = t.get("corrected_grid") or []
body = _grid_html(i, grid) if grid else t.get("markdown")
head = _table_head(t, unified_tables.get(som_docling_ref.get(i, "")), all_texts)
section = table_headings.get(som_docling_ref.get(i, ""), "")
if grouped:
section = _governing(page_first_heading.get(t.get("page", 1), ""), section)
if section and all(section.lower() not in p.lower() for p in head):
head.insert(0, section)
# Resolve this table's footnote markers to their definitions. The
# scan universe is every content line AFTER the table in reading
# order. A table the fusion report does not place in that stream —
# no report, no match, or a scanned table — falls back to every line
# from the table's page onward.
marks = t.get("footnote_marks") or []
fdefs = [
FootnoteDef(marker="", text=f) if isinstance(f, str) else FootnoteDef.model_validate(f)
for f in (t.get("footnotes") or [])
]
markers = [m.get("marker", "") for m in marks] + list(t.get("footnote_refs") or [])
# The scan universe ends where the NEXT table begins: a footnote
# printed beyond it belongs to that table, and another table's own
# coherent footnote block would otherwise satisfy this table's
# markers. The overleaf continuation case has no table in between,
# so it survives the cut; what the cut excludes falls to the lookup
# agent, which reads the page image.
own_ref = som_docling_ref.get(i, "")
pos = table_positions.get(own_ref)
if pos is not None:
seq = table_sequence.index(own_ref)
end = (
table_positions[table_sequence[seq + 1]]
if seq + 1 < len(table_sequence)
else len(ordered_texts)
)
trailing = ordered_texts[pos:end]
else:
trailing = [x for x in ordered_texts if x[1] >= page1]
section_keys = {canonical_marker(m.get("marker", "")) for m in marks if m.get("kind") == "section"}
resolution = resolve_footnotes(markers, fdefs, page1, trailing, headings, section_keys)
if resolution.unresolved:
resolution = _lookup_unresolved(resolution, DATA_DIR / f"{doc_key}.pdf", page1)
# The whole-table chunk carries every resolved footnote plus the
# unmarked general notes (those attach at table level only).
table_notes = [f"{r.marker} — {r.text}" for r in resolution.resolved] + [
d.text for d in fdefs if not d.marker and d.text.strip()
]
content = "\n".join(head + [body]) if body else "\n".join(head)
if table_notes:
content += FOOTNOTE_HEADER + "\n".join(table_notes)
records.append((f"#/tables/{i}", "table", page1 - 1, bbox, content, None))
# One searchable record per printed table line, alongside the whole-
# table record. One embedding cannot represent forty printed lines, so
# a question about a single line item gets a unit that IS that line.
# Each line rides with the table's section heading, title, units, its
# full header block, and the row-axis label governing it, so it is
# answerable alone; the cell ids are the same ones the table record
# carries, so citation resolves identically. The header block depth is
# the correction review's reading of the table image — there is no
# derived fallback. A gridded table can legitimately arrive without a
# depth: when structure correction is rejected or fails, the table
# keeps its grid but no image-read header count exists. Such a table
# is served whole (record above, cell groundings below); only its
# per-line records are skipped, loudly, rather than guessed at or
# failing the document. Header rows and row-axis label rows get no
# record of their own — a record whose whole content is header words
# is retrieval noise posing as data.
depth = t.get("header_rows")
if grid and not isinstance(depth, int):
logger.warning(
"table {i} (page {p}): gridded but carries no header depth — structure "
"correction absent; serving the whole-table record only, no per-line records",
i=i,
p=page1,
)
elif grid:
label_row: Optional[int] = None
for r in range(depth, len(grid)):
row = grid[r]
if not any((cell.get("text") or "").strip() for cell in row):
continue
# A row-axis section label ("Business Segments", "Revenues:")
# carries text only in the stub column. A row with text across
# columns is a line, whether its values are bare numbers or
# worded ranges ("6.1 to 6.4 million").
if not any((cell.get("text") or "").strip() for cell in row[1:]):
label_row = r
continue
rows = list(range(depth)) + ([label_row] if label_row is not None else []) + [r]
line_html = _grid_html(i, [grid[k] for k in rows], row_offsets=rows)
line_content = "\n".join(head + [line_html])
# A footnote rides with exactly the records that reference its
# marker: a marker in the header rows reaches every record (a
# column footnote qualifies every row); a marker on a row-axis
# label row reaches every record under that label; a marker on
# the data row reaches only this record.
rows_set = set(rows)
line_notes: list[str] = []
seen_keys: set[str] = set()
for m in marks:
if m.get("row") not in rows_set:
continue
key = canonical_marker(m.get("marker", ""))
if not key or key in seen_keys:
continue
seen_keys.add(key)
note_text = resolution.text_for(m.get("marker", ""))
if note_text:
line_notes.append(f"{m.get('marker')} — {note_text}")
if line_notes:
line_content += FOOTNOTE_HEADER + "\n".join(line_notes)
boxes = [cell["box"] for cell in row if cell.get("box")]
line_bbox = (
{
"left": min(b[0] for b in boxes),
"top": min(b[1] for b in boxes),
"right": max(b[2] for b in boxes),
"bottom": max(b[3] for b in boxes),
}
if boxes
else bbox
)
records.append(
(f"t{i}-line-{r}", "line_item", page1 - 1, line_bbox, line_content, f"#/tables/{i}")
)
# One grounding per cell. A cell whose box could not be traced is
# still served: its overlay is the vicinity read off its boxed
# neighbors (column x-extent crossed with row y-extent), and only when
# that is underivable the whole table region. Status/note ride along
# either way so the UI can flag it.
for r, row in enumerate(grid):
for c, cell in enumerate(row):
box = cell.get("box") or _cell_vicinity(grid, r, c, region)
cell_bbox = (
{"left": box[0], "top": box[1], "right": box[2], "bottom": box[3]} if box else bbox
)
cell_rows.append(
(
f"t{i}-{r}-{c}",
"tableCell",
page1 - 1,
cell_bbox,
{"row": r, "col": c, "chunk_id": f"#/tables/{i}"},
cell.get("status"),
cell.get("note"),
cell.get("text") or "",
)
)
# Printed header-area text that reached no output cell is a table-level
# review flag. It has no cell to point at, but the status means the
# text was printed in the table's header area, so the band above the
# first boxed row is its vicinity; the whole region is the fallback.
band = _header_band(grid, region)
band_bbox = {"left": band[0], "top": band[1], "right": band[2], "bottom": band[3]} if band else bbox
for k, fragment in enumerate(t.get("dropped_text") or []):
cell_rows.append(
(
f"t{i}-dropped-{k}",
"tableFlag",
page1 - 1,
band_bbox,
{"chunk_id": f"#/tables/{i}"},
"header_text_dropped",
None,
fragment,
)
)
# Footnote exceptions, quoted verbatim. A clean table adds nothing.
footnote_flags = (
[("footnote_unresolved", m) for m in resolution.unresolved]
+ [("footnote_unreferenced", f"{d.marker} {d.text}".strip()) for d in resolution.unreferenced]
+ [("footnote_marker_unplaced", m.get("marker", "")) for m in marks if m.get("row") is None]
)
for k, (status, quoted) in enumerate(footnote_flags):
logger.warning("table {i} (page {p}): {status}: {q!r}", i=i, p=page1, status=status, q=quoted)
cell_rows.append(
(
f"t{i}-fn-{k}",
"tableFlag",
page1 - 1,
bbox,
{"chunk_id": f"#/tables/{i}"},
status,
None,
quoted,
)
)
# Tables the Set-of-Mark engine missed exist only in the unified document;
# ingest them from docling's own body so their content is retrievable.
if report:
missed_refs = {
ref
for m in report.get("matches", [])
if m.get("kind") == "som_miss"
for ref in m.get("docling_table_refs", [])
}
by_ref = {t.get("self_ref"): t for t in doc.get("tables", [])}
for ref in sorted(missed_refs - superseded):
t = by_ref.get(ref)
if t is None:
logger.warning("fusion report names {ref} but the unified doc has no such table", ref=ref)
continue
content = _docling_table_text(t)
if not content:
logger.warning("SoM-missed table {ref} has no cell text; skipped", ref=ref)
continue
prov = (t.get("prov") or [{}])[0]
page1 = prov.get("page_no", 1)
section = table_headings.get(ref, "")
if grouped:
section = _governing(page_first_heading.get(page1, ""), section)
if section and section.lower() not in content.lower():
content = f"{section}\n{content}"
w, h = sizes.get(page1, (1.0, 1.0))
bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
# Prefixed ref id: '#/tables/N' is already taken by the SoM tables.
records.append((f"#/docling{ref}", "table", page1 - 1, bbox, content, None))
logger.info("Ingesting SoM-missed docling table {ref} (page {p})", ref=ref, p=page1)
# --- Embed and insert ---
# Embed what a reader reads: the stored content keeps its HTML so the
# agent can cite cell ids, but the markup is noise to the embedding — on a
# short line record the td tags and ids outweigh the words themselves.
vectors = embed_document(doc_key, [_embed_view(r[4]) for r in records]) if records else []
meta = doc.get("origin", {})
filename = filename or meta.get("filename") or f"{doc_key}.pdf"
page_count = len(sizes) or None
content_hash = hashlib.sha256(pdf_local.read_bytes()).hexdigest()
with db.connect() as conn:
conn.execute("DELETE FROM ade_playground.documents WHERE doc_key = %s", (doc_key,))
row = conn.execute(
"""INSERT INTO ade_playground.documents
(doc_key, content_hash, filename, page_count, ade_version)
VALUES (%s, %s, %s, %s, %s) RETURNING id""",
(doc_key, content_hash, filename, page_count, "quber-fusion"),
).fetchone()
assert row is not None # INSERT .. RETURNING always yields a row
doc_id = row[0]
for (ref_id, ref_type, page0, bbox, content, parent), vec in zip(records, vectors, strict=True):
conn.execute(
"""INSERT INTO ade_playground.chunks
(document_id, chunk_id, chunk_type, page, bbox, content, embedding, parent_chunk_id)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
(doc_id, ref_id, ref_type, page0, json.dumps(bbox), _pg_text(content), vec, parent),
)
# Every chunk is also a grounding under its own ref id, with the
# chunk's box. A per-page figure_values record has none.
conn.execute(
"""INSERT INTO ade_playground.groundings
(document_id, ref_id, ref_type, page, bbox, position)
VALUES (%s, %s, %s, %s, %s, NULL)
ON CONFLICT (document_id, ref_id) DO NOTHING""",
(doc_id, ref_id, ref_type, page0, json.dumps(bbox)),
)
for ref_id, ref_type, page0, bbox, position, status, note, cell_text in cell_rows:
conn.execute(
"""INSERT INTO ade_playground.groundings
(document_id, ref_id, ref_type, page, bbox, position, status, note, cell_text)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (document_id, ref_id) DO NOTHING""",
(
doc_id,
ref_id,
ref_type,
page0,
json.dumps(bbox),
json.dumps(position),
status,
_pg_text(note),
_pg_text(cell_text),
),
)
logger.success(
"Ingested {key}: {n} chunks ({t} text, {tb} table, {ln} line item), {cells} cell groundings",
key=doc_key,
n=len(records),
t=sum(1 for r in records if r[1] == "text"),
tb=sum(1 for r in records if r[1] == "table"),
ln=sum(1 for r in records if r[1] == "line_item"),
cells=len(cell_rows),
)
return len(records)
def main(argv: Optional[list[str]] = None) -> None:
import argparse
p = argparse.ArgumentParser(description="Ingest quber fusion output into the playground pgvector schema.")
p.add_argument("--doc-key", required=True, help="opaque storage key for the document")
p.add_argument("--unified-json", required=True, help="local path or s3:// URI to .unified.json")
p.add_argument("--tables-json", required=True, help="local path or s3:// URI to .tables.json")
p.add_argument("--pdf", required=True, help="local path or s3:// URI to the source PDF")
p.add_argument(
"--fusion-json",
default=None,
help="local path or s3:// URI to .fusion.json; enables ingesting "
"SoM-missed docling tables from the unified document",
)
p.add_argument(
"--scanned-tables-json",
default=None,
help="local path or s3:// URI to .scanned-tables.json; the tables the "
"figure run read off a page image, ingested alongside the table engine's",
)
p.add_argument(
"--figures-json",
default=None,
help="local path or s3:// URI to .figures.json; names the parse tables a "
"scanned table replaced, so the parse's reading of them is not ingested too",
)
p.add_argument(
"--figure-values-json",
default=None,
help="local path or s3:// URI to .figure-values.json; the reconciled "
"figure values, grounded per value and listed on their picture chunks",
)
p.add_argument("--filename", default=None, help="original PDF filename, kept for presentation")
p.add_argument(
"--chunking",
choices=["flat", "grouped"],
default="flat",
help="ingestion stream: flat serves every printed line as its own chunk — the dpt-2 "
"stream's byte-stable behavior (default); "
"grouped is the dpt-3 track's — binding groups served as one chunk, page heading "
"joined into the ride-along",
)
p.add_argument("--init-schema", action="store_true", help="(re)create the schema before ingest")
args = p.parse_args(argv)
if args.init_schema:
db.apply_schema()
logger.success("Applied schema")
ingest(
args.doc_key,
args.unified_json,
args.tables_json,
args.pdf,
fusion_json=args.fusion_json,
scanned_tables_json=args.scanned_tables_json,
figures_json=args.figures_json,
figure_values_json=args.figure_values_json,
filename=args.filename,
chunking=args.chunking,
)
if __name__ == "__main__":
main()