Coverage for src / quber / playground / ingest_fusion.py: 12%
550 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-09-23 22:14 -0400
1"""Ingest quber *fusion* output into the same pgvector playground schema.
3Where `ingest.py` consumes Landing.ai ADE JSON, this consumes the two
4artifacts the quber fusion workflow writes:
6 <base>.unified.json — the unified DoclingDocument (spine: reading-order
7 text + headings, each with page + PDF-point bbox)
8 <base>.tables.json — the corrected ExtractedTable list (LLM-cleaned
9 markdown, title, units, page, normalized regions)
11Mapping to the RAG schema:
12 - Text chunks <- unified.json `texts` (bbox normalized to top-left 0..1)
13 - Table chunks <- tables.json (content = title/units/markdown; region is
14 already normalized top-left 0..1)
16Grounding is cell-level, matching the ADE ingester: each table's
17`corrected_grid` supplies one GroundedCell per markdown cell — text, a box
18normalized 0..1 top-left (the same frame ADE uses), a provenance status, and
19the status inspector's note where one was taken. The table chunk's content is
20an HTML table with an id on every cell (like ADE's `<td id="0-8">`), so the
21agent can cite individual cells and the app resolves them to overlays. Every
22cell is served regardless of status; a cell whose box could not be traced
23falls back to the table's `content_region` as its overlay and keeps its
24status/note so the UI can flag it.
26Footnotes attach to the chunks that reference them: each table's markers are
27resolved to their definition text (in-crop pairs, then a reading-order scan,
28then the demand-driven lookup agent), and the resolved text is appended as a
29display-only block — on the whole-table chunk (all notes, plus unmarked
30general notes) and on each line record whose rendered rows carry the marker.
31The block is never embedded. A footnote's own text chunk gets a parent link
32to the table it follows. Unresolved markers, unreferenced definitions, and
33unplaced markers land as tableFlag groundings, quoted verbatim.
35The fusion report (`<base>.fusion.json`), when given, names any docling
36table the Set-of-Mark engine missed (`som_miss`). Those tables exist only in
37the unified document, so they are ingested from there as plain table chunks
38— docling's own cell text, table-level grounding — rather than dropped.
40The figure run writes two more artifacts, both optional here. Its scanned
41tables (`<base>.scanned-tables.json`) are `ExtractedTable`s like any other and
42join the table engine's own, so a table read off a page image is chunked,
43grounded and footnoted the same way. Its run record (`<base>.figures.json`)
44names the parse tables those replaced, and the parse's reading of a replaced
45table is then not ingested beside the scan's.
47The figure-value run (`<base>.figure-values.json`, optional) carries the
48reconciled per-value reading of each figure with per-value provenance. Each
49value becomes a grounding of its own — ref id `fv-<n>`, its reconciliation
50status and note, and the value's box on the page (already normalized top-left
510..1; a value with no box of its own borrows its picture's). The values are
52also listed in id-tagged plain text on the picture chunk they belong to, the
53same citation style the grid HTML gives cells, so the answering model can cite
54a plotted value as precisely as a table cell.
55"""
57from __future__ import annotations
59import hashlib
60import json
61import re
62import subprocess
63import sys
64from pathlib import Path
65from typing import Any, Optional, Sequence
67from loguru import logger
69# Run at INFO: loguru's unconfigured default is DEBUG, which floods the
70# upload job's log view with per-batch embedding lines.
71logger.remove()
72logger.add(sys.stderr, level="INFO")
74from quber.agents.llm_client import FootnoteDef
75from quber.core.figures.dpt3.blocks import SCAN_BLOCK
76from quber.core.figures.graft import FOOTNOTE_MARKS_FIELD, FOOTNOTES_FIELD
77from quber.core.figures.models import FigureValue, FigureValueRun
78from quber.core.fusion.footnotes import (
79 FootnoteResolution,
80 absorb_lookup,
81 canonical_marker,
82 resolve_footnotes,
83)
84from quber.playground import db
85from quber.playground.embedding import embed_document
86from quber.settings import get_settings
88DATA_DIR = get_settings().playground.data_dir
90# The footnote block appended to a chunk's stored content. The answering
91# model reads it; the embedding never sees it (see _embed_view) — the same
92# note text appended to many records would drag their vectors toward each
93# other, the boilerplate-domination failure measured on line records.
94FOOTNOTE_HEADER = "\nFootnotes:\n"
96# Section headings and captions are handled structurally rather than skipped: a
97# retrieval unit must be able to answer a question, and a bare 3-word heading
98# cannot — it can only name a thing, and as a standalone chunk it outscores the
99# content it names whenever a question mentions that name. So heading text rides
100# WITH the content it governs (prefixed onto every text chunk beneath it), and
101# caption text rides with the object it captions (a table chunk already leads
102# with its title). Nothing is dropped; it is relocated to where it can support
103# an answer.
104#: Picture classes that are identity marks with nothing inside to read: a
105#: watermark, an icon, a logo. Every other class is content and is ingested.
106EXCLUDED_PICTURE_CLASSES = {"logo", "icon", "watermark"}
109def _fetch_optional(src: Optional[str], dest: Path) -> Optional[dict]:
110 """An artifact a run writes only when it produced one, read if it is there.
112 A document with no table read off a page image produces no scanned-tables
113 file, and a document never put through the figure run produces neither of its
114 artifacts. Naming one that was not written is not an error — it says the run
115 produced nothing of that kind.
116 """
117 if not src:
118 return None
119 if not src.startswith("s3://") and not Path(src).exists():
120 logger.info("no {} to read; the run wrote none", dest.name)
121 return None
122 return json.loads(_fetch(src, dest).read_text())
125def _fetch(src: str, dest: Path) -> Path:
126 dest.parent.mkdir(parents=True, exist_ok=True)
127 if src.startswith("s3://"):
128 subprocess.run(["aws", "s3", "cp", src, str(dest)], check=True, capture_output=True)
129 else:
130 dest.write_bytes(Path(src).read_bytes())
131 return dest
134def _governing(page_heading: str, heading: str) -> str:
135 """The section vocabulary a chunk rides with: the page's opening heading
136 joined with the heading in effect at the chunk.
138 A presentation page opens with its subject — 'Commercial Portfolio
139 Geographic Diversification' — and then prints panel labels the parse also
140 calls headings ('U.S.', 'Europe', 'Australia'). Carrying only the latest
141 heading strips the page's subject from every chunk below the first panel
142 label, and a question phrased in the subject's words then never ranks the
143 chunk. Both ride along; when they agree, one.
144 """
145 if page_heading and heading and page_heading.lower() != heading.lower():
146 return f"{page_heading} — {heading}"
147 return heading or page_heading
150def _is_page_number(text: str, page: int, content_layer: str) -> bool:
151 """Is this text the page printing its own number and nothing else?
153 The page's furniture layer is where the parse puts what it takes for page
154 decoration, but its guess is unreliable in the direction that matters: on a
155 financial deck that layer also holds the footnote definitions that qualify
156 the figures, so it cannot be excluded wholesale. A page number is the one
157 thing in it that can be named exactly rather than guessed at — a furniture
158 text whose whole content, read as an integer, is the number of the page it
159 sits on. Anything that fails to match is kept, so a deck whose printed
160 numbering runs off the page index loses nothing.
161 """
162 return content_layer == "furniture" and text.isdigit() and int(text) == page
165def _picture_notes(pic: dict) -> list[FootnoteDef]:
166 """The notes an agent read off the page this figure is printed on."""
167 return [
168 FootnoteDef(marker=n.get("marker", ""), text=n.get("text", ""))
169 for n in ((pic.get("meta") or {}).get(FOOTNOTES_FIELD) or [])
170 if n.get("text")
171 ]
174def _picture_marks(pic: dict) -> list[dict]:
175 """The footnote reference markers an agent read off this figure.
177 Each carries what it points at, so a cross-reference to a named part of the
178 document is never hunted among the lines printed near the figure.
179 """
180 return list(((pic.get("meta") or {}).get(FOOTNOTE_MARKS_FIELD)) or [])
183def _superseded_refs(run: Optional[dict]) -> set[str]:
184 """The parse tables a scanned table replaced, from the figure run's record.
186 The fusion report names a table the table engine missed, so the ingest reads
187 that table out of the parse. Where the figure run has since replaced the
188 parse's reading with the scan's, reading it out of the parse again would
189 state one region twice.
190 """
191 if not run:
192 return set()
193 return {
194 ref
195 for scan in run.get("scans", [])
196 for table in scan.get("tables", [])
197 for ref in [table.get("table_ref")]
198 if ref
199 }
202def _picture_class(pic: dict) -> str:
203 """The parse's top predicted class for a picture, or '' if unclassified."""
204 meta = pic.get("meta") or {}
205 preds = ((meta.get("classification") or {}).get("predictions")) or []
206 return preds[0].get("class_name", "") if preds else ""
209def _cited_line(fid: str, name: str, value: str) -> str:
210 """One value line the answering model can cite.
212 The `[fv-n]` tag is the citation contract: the model quotes the id and the
213 app resolves it to that value's own grounding box, so the overlay is the
214 printed value rather than whatever chunk carried the line. Every chunk
215 that serves figure values builds its lines here — the tag must never be
216 reimplemented per chunk kind, because a line without it can only be cited
217 at chunk granularity.
218 """
219 return f"[{fid}] {name}: {value}" if name else f"[{fid}] {value}"
222def _picture_text(pic: dict, values: Sequence[tuple[str, FigureValue]] = ()) -> str:
223 """What a picture says, or '' when it says nothing worth retrieving.
225 A chart's plotted values live in the description the page scan wrote, which
226 names the axes, the units, the legend and every plotted figure. That is the
227 only place those numbers appear in the document — nothing else in the
228 pipeline reads a chart — so a picture carrying one is a retrievable record
229 like any other.
231 `values` are this picture's reconciled figure values, each with its
232 grounding ref id. They are appended as an id-tagged line per value, the
233 same citation style the grid HTML gives cells, so the answering model can
234 cite a plotted value by id and the app resolves it to an overlay.
236 Excluded: a watermark, an icon, a logo. They are identity marks repeated on
237 every page with nothing inside to read. A picture with no description and
238 no values is excluded too, having nothing to say.
239 """
240 if _picture_class(pic) in EXCLUDED_PICTURE_CLASSES:
241 return ""
242 description = ((pic.get("meta") or {}).get("description") or {}).get("text") or ""
243 text = description.strip()
244 if values:
245 lines = []
246 for fid, v in values:
247 name = " ".join(p for p in (v.label, v.series) if p)
248 lines.append(_cited_line(fid, name, v.value))
249 text = (text + "\nValues:\n" if text else "Values:\n") + "\n".join(lines)
250 return text
253def _unanchored_value_records(
254 figure_values: Sequence[tuple[str, FigureValue]],
255 page_first_heading: dict[int, str],
256) -> list[tuple[str, str, int, Optional[dict], str, Optional[str]]]:
257 """One searchable record per page of figure values no picture claims.
259 Anchored values ride their picture chunk's searchable text; an unanchored
260 value has no chunk to ride and its grounding row alone is invisible to
261 retrieval — a printed number the index never hears of. Each page's
262 unanchored values become one record of id-tagged label-value lines under
263 the page's opening heading, id `figure-values-p<page>`. The id tags are
264 the same citation style the grid HTML gives cells and `_picture_text`
265 gives a picture's values: the answering model cites the value's own
266 `fv-<n>` id, which resolves to that value's tight box rather than the
267 record's. The record itself carries NO box, deliberately: a citation of
268 the record alone names the page, and the only rectangles ever drawn are
269 the values' own. A synthetic union box was tried and rendered as a
270 misleading figure-sized highlight over the tight one.
271 """
272 by_page: dict[int, list[tuple[str, FigureValue]]] = {}
273 for fid, v in figure_values:
274 if not v.picture_ref:
275 by_page.setdefault(v.page, []).append((fid, v))
276 records: list[tuple[str, str, int, Optional[dict], str, Optional[str]]] = []
277 for page1 in sorted(by_page):
278 lines = []
279 for fid, v in by_page[page1]:
280 # The chart title leads the name here and not in the picture path:
281 # a picture chunk IS its chart, a page record must say which one.
282 name = " — ".join(part for part in (v.chart_title, v.label, v.series) if part)
283 lines.append(_cited_line(fid, name, v.value))
284 content = "\n".join(lines)
285 section = page_first_heading.get(page1, "")
286 if section and section.lower() not in content.lower():
287 content = f"{section} — {content}"
288 records.append((f"figure-values-p{page1}", "figure_values", page1 - 1, None, content, None))
289 return records
292def _picture_bbox(
293 doc: dict, sizes: dict[int, tuple[float, float]], picture_ref: Optional[str]
294) -> Optional[dict[str, float]]:
295 """The parse picture's own normalized box, for a value with no box of its own."""
296 if not picture_ref:
297 return None
298 for pic in doc.get("pictures", []):
299 if pic.get("self_ref") == picture_ref:
300 prov = (pic.get("prov") or [{}])[0]
301 if not prov.get("bbox"):
302 return None
303 page1 = prov.get("page_no", 1)
304 w, h = sizes.get(page1, (1.0, 1.0))
305 return _norm_bbox(prov["bbox"], w, h)
306 return None
309def _reading_order_refs(doc: dict) -> list[str]:
310 """Every item ref in the document, in reading order.
312 Descends the whole tree. An item's own children are visited straight after
313 it, so a table's footnotes and a picture's labels keep their place in the
314 flow instead of being unreachable — the walk used to read only the body's
315 direct children, which left every text hanging off a table or a picture
316 invisible to ingestion, judged by no rule at all.
318 Anything the document holds but never links into the tree is appended at the
319 end, so an item is dropped only by a rule that names it, never by an
320 accident of where it sits. Returns an empty list when the artifact carries no
321 body tree; the caller falls back to the texts array's own order.
322 """
323 pools = {kind: doc.get(kind, []) for kind in ("texts", "tables", "pictures", "groups")}
324 out: list[str] = []
325 seen: set[str] = set()
327 def visit(ref: str) -> None:
328 if not ref or ref in seen:
329 return
330 seen.add(ref)
331 kind, _, idx = ref.lstrip("#/").partition("/")
332 items = pools.get(kind, [])
333 try:
334 item = items[int(idx)]
335 except (ValueError, IndexError):
336 return
337 # A group ref rides in the flow ahead of its members. A group that
338 # binds a label to its value is served as one record and the caller
339 # then skips the members; any other group is a transparent wrapper
340 # the caller passes over, and its members stand alone as before.
341 out.append(ref)
342 for child in item.get("children") or []:
343 visit(child.get("$ref", ""))
345 for child in doc.get("body", {}).get("children") or []:
346 visit(child.get("$ref", ""))
347 if not out:
348 return []
349 for kind in ("texts", "tables", "pictures"):
350 for i in range(len(pools[kind])):
351 ref = f"#/{kind}/{i}"
352 if ref not in seen:
353 logger.warning(
354 "{ref} is in the document but linked into no parent; ingesting it anyway", ref=ref
355 )
356 out.append(ref)
357 seen.add(ref)
358 return out
361def _page_sizes(doc: dict) -> dict[int, tuple[float, float]]:
362 """page_no -> (width, height) from the DoclingDocument pages map."""
363 out: dict[int, tuple[float, float]] = {}
364 for k, v in (doc.get("pages") or {}).items():
365 size = v.get("size", {})
366 out[int(k)] = (float(size.get("width", 0)) or 1.0, float(size.get("height", 0)) or 1.0)
367 return out
370def _norm_bbox(prov_bbox: dict[str, Any], page_w: float, page_h: float) -> Optional[dict[str, float]]:
371 """Normalize a docling prov bbox to top-left origin, 0..1.
373 docling boxes are PDF points with coord_origin BOTTOMLEFT, so t/b are
374 measured up from the page bottom (t > b). Flip Y to a top-left origin.
375 """
376 left, right = prov_bbox.get("l"), prov_bbox.get("r")
377 top_pt, bottom_pt = prov_bbox.get("t"), prov_bbox.get("b")
378 if left is None or right is None or top_pt is None or bottom_pt is None:
379 return None
380 origin = prov_bbox.get("coord_origin", "BOTTOMLEFT")
381 if origin == "BOTTOMLEFT":
382 top, bottom = (page_h - top_pt) / page_h, (page_h - bottom_pt) / page_h
383 else: # TOPLEFT
384 top, bottom = top_pt / page_h, bottom_pt / page_h
385 return {
386 "left": max(0.0, left / page_w),
387 "top": max(0.0, min(top, bottom)),
388 "right": min(1.0, right / page_w),
389 "bottom": min(1.0, max(top, bottom)),
390 }
393def _esc(s: str) -> str:
394 return s.replace("&", "&").replace("<", "<").replace(">", ">")
397def _pg_text(s: Optional[str]) -> Optional[str]:
398 """Strip NUL characters: Postgres text fields reject 0x00. They reach us
399 from PDF text layers that encode glyphs like cover-page checkboxes as NUL."""
400 return s.replace("\x00", "") if s is not None else None
403def _embed_view(content: str) -> str:
404 """The text a reader reads: tags dropped, cells separated by ' | '. The
405 footnote block is display-only ride-along and is cut before embedding."""
406 content = content.split(FOOTNOTE_HEADER)[0]
407 text = re.sub(r"</td><td[^>]*>", " | ", content)
408 text = re.sub(r"<[^>]+>", " ", text)
409 return re.sub(r"[ \t]+", " ", text).strip()
412def _table_head(t: dict, item: Optional[dict], texts: list[dict]) -> list[str]:
413 """The header lines of a table chunk: the caption nodes fusion attached to
414 the table's docling item, in order (the sentence that introduces the table,
415 its printed name, its units), read from the unified document so the chunk
416 says what the fused document says. A table with no fused item falls back to
417 its own fields."""
418 if item:
419 lines: list[str] = []
420 for ref in item.get("captions") or []:
421 cref = ref.get("$ref", "") if isinstance(ref, dict) else str(ref)
422 if not cref.startswith("#/texts/"):
423 continue
424 idx = int(cref.rsplit("/", 1)[1])
425 text = (texts[idx].get("text") or "").strip() if idx < len(texts) else ""
426 if text:
427 lines.append(text)
428 if lines:
429 return lines
430 return [p for p in (t.get("caption"), t.get("title"), t.get("subtitle"), t.get("units")) if p]
433def _grid_html(table_index: int, grid: list[list[dict]], row_offsets: Optional[list[int]] = None) -> str:
434 """Render a corrected_grid as an HTML table with an id on every cell.
436 Ids follow `t<table_index>-<row>-<col>` so they are unique per document,
437 short enough for the agent to cite verbatim, and resolve directly back to
438 the grid position they came from. `row_offsets` supplies the true row
439 number of each rendered row when `grid` is a subset (a line-item record
440 rendering header + one line keeps the line's real ids).
441 """
442 rows = []
443 for k, row in enumerate(grid):
444 r = row_offsets[k] if row_offsets else k
445 tds = "".join(
446 f'<td id="t{table_index}-{r}-{c}">{_esc(cell.get("text") or "")}</td>'
447 for c, cell in enumerate(row)
448 )
449 rows.append(f"<tr>{tds}</tr>")
450 return "<table>\n" + "\n".join(rows) + "\n</table>"
453def _cell_vicinity(
454 grid: list[list[dict]], r: int, c: int, region: Optional[tuple]
455) -> Optional[tuple[float, float, float, float]]:
456 """Approximate box for an unboxed cell, read off its boxed neighbors.
458 The column's x-extent (union of boxes in column c) crossed with the row's
459 y-extent (union of boxes in row r) marks the empty spot where the cell
460 sits. A fully unboxed row falls back to the band between the table's top
461 edge and the first boxed row — the header area. Returns None when neither
462 axis is derivable; the caller falls back to the whole table region.
463 """
464 col = [row[c]["box"] for row in grid if len(row) > c and row[c].get("box")]
465 rowb = [cell["box"] for cell in grid[r] if cell.get("box")]
466 allb = [cell["box"] for row in grid for cell in row if cell.get("box")]
467 if not (col or rowb):
468 return None
469 if col:
470 x1, x2 = min(b[0] for b in col), max(b[2] for b in col)
471 elif region:
472 x1, x2 = region[0], region[2]
473 else:
474 return None
475 if rowb:
476 y1, y2 = min(b[1] for b in rowb), max(b[3] for b in rowb)
477 elif region and allb and min(b[1] for b in allb) > region[1]:
478 y1, y2 = region[1], min(b[1] for b in allb)
479 else:
480 return None
481 return (x1, y1, x2, y2)
484def _header_band(
485 grid: list[list[dict]], region: Optional[tuple]
486) -> Optional[tuple[float, float, float, float]]:
487 """The table's header area: from the region's top edge down to the top of
488 the first boxed row. Dropped printed text lives there by definition, so it
489 is the honest vicinity for a dropped-text flag. None without a region or
490 when no cell is boxed."""
491 if not region:
492 return None
493 allb = [cell["box"] for row in grid for cell in row if cell.get("box")]
494 if not allb or min(b[1] for b in allb) <= region[1]:
495 return None
496 return (region[0], region[1], region[2], min(b[1] for b in allb))
499def _docling_table_text(t: dict) -> str:
500 """Reconstruct a docling table's content as pipe-separated rows."""
501 rows: dict[int, dict[int, str]] = {}
502 for cell in (t.get("data") or {}).get("table_cells", []):
503 text = (cell.get("text") or "").strip()
504 if not text:
505 continue
506 r = cell.get("start_row_offset_idx", 0)
507 c = cell.get("start_col_offset_idx", 0)
508 rows.setdefault(r, {})[c] = text
509 return "\n".join(" | ".join(rows[r][c] for c in sorted(rows[r])) for r in sorted(rows))
512def _lookup_unresolved(resolution: FootnoteResolution, pdf_path: Path, page1: int) -> FootnoteResolution:
513 """The demand-driven last tier: ask the lookup agent to find the still-
514 unresolved markers on the rendered pages following the table. Any failure
515 (no PDF, no credentials, a failed call) leaves the markers unresolved in
516 the exception record — the lookup never takes an ingest down."""
517 if not pdf_path.exists():
518 return resolution
519 try:
520 import asyncio
521 import io
523 from pdf2image import convert_from_path
525 from quber.agents.footnote_lookup import PydanticAIFootnoteLookup
527 agent = PydanticAIFootnoteLookup()
528 images = convert_from_path(str(pdf_path), dpi=150, fmt="png", first_page=page1, last_page=page1 + 2)
529 pngs: list[bytes] = []
530 for image in images:
531 buf = io.BytesIO()
532 image.save(buf, format="PNG")
533 pngs.append(buf.getvalue())
534 report = asyncio.run(agent.lookup(resolution.unresolved, pngs, page1))
535 except Exception as exc:
536 logger.warning("page {p}: footnote lookup unavailable; markers stay unresolved: {e}", p=page1, e=exc)
537 return resolution
538 if report is None:
539 return resolution
540 return absorb_lookup(resolution, report.found)
543def ingest(
544 doc_key: str,
545 unified_json: str,
546 tables_json: str,
547 source_pdf: str,
548 fusion_json: Optional[str] = None,
549 scanned_tables_json: Optional[str] = None,
550 figures_json: Optional[str] = None,
551 figure_values_json: Optional[str] = None,
552 filename: Optional[str] = None,
553 chunking: str = "flat",
554) -> int:
555 """`chunking` selects the ingestion stream. `flat` (the default) serves
556 every printed line on its own: a stat panel's label lands in one chunk
557 and its number in another, and each chunk is tagged with only the most
558 recent heading the walk passed — nothing is bundled. Technically: every
559 group is a transparent wrapper, single-heading ride-along. This is the
560 dpt-2 stream's behavior, byte-stable from before the dpt-3 track.
561 `grouped` is the dpt-3 track's stream: a binding group — a
562 key_value_area or a scan block — is served as one chunk so a label
563 travels with its number, and every chunk rides with the page's opening
564 heading joined to the local one. The grouped behavior migrates to (or
565 replaces) the flat stream only once the track is vetted; keying it on
566 the caller rather than the artifacts keeps the production upload flow
567 frozen until then.
568 """
569 grouped = chunking == "grouped"
570 pdf_local = _fetch(source_pdf, DATA_DIR / f"{doc_key}.pdf")
571 doc = json.loads(_fetch(unified_json, DATA_DIR / f"{doc_key}.unified.json").read_text())
572 tables = json.loads(_fetch(tables_json, DATA_DIR / f"{doc_key}.tables.json").read_text())
573 report = (
574 json.loads(_fetch(fusion_json, DATA_DIR / f"{doc_key}.fusion.json").read_text())
575 if fusion_json
576 else None
577 )
578 # A table read off a page image arrives as an ExtractedTable like any other,
579 # from the figure run rather than from the table engine. It is appended to the
580 # same list so it is chunked, grounded and footnoted the same way.
581 scanned = _fetch_optional(scanned_tables_json, DATA_DIR / f"{doc_key}.scanned-tables.json")
582 tables.extend(scanned or [])
583 # Where such a table replaced one the parse held, the parse's reading of that
584 # region is superseded and must not be ingested beside it.
585 superseded = _superseded_refs(_fetch_optional(figures_json, DATA_DIR / f"{doc_key}.figures.json"))
586 sizes = _page_sizes(doc)
588 # The reconciled figure values, each assigned the grounding ref id it will
589 # be cited by. Grouped by picture so the picture chunk can list its own.
590 value_run = _fetch_optional(figure_values_json, DATA_DIR / f"{doc_key}.figure-values.json")
591 figure_values: list[tuple[str, FigureValue]] = (
592 [(f"fv-{n}", v) for n, v in enumerate(FigureValueRun.model_validate(value_run).values)]
593 if value_run
594 else []
595 )
596 values_by_picture: dict[str, list[tuple[str, FigureValue]]] = {}
597 for fid, v in figure_values:
598 if v.picture_ref:
599 values_by_picture.setdefault(v.picture_ref, []).append((fid, v))
601 # --- Build chunk records: (ref_id, ref_type, page0, bbox, content, parent) ---
602 # parent is the whole-table record a line-item record belongs to; None on
603 # every other kind of record.
604 records: list[tuple[str, str, int, Optional[dict], str, Optional[str]]] = []
605 # Cell groundings: (ref_id, ref_type, page0, bbox, position, status, note, cell_text)
606 cell_rows: list[tuple[str, str, int, Optional[dict], dict, Optional[str], Optional[str], str]] = []
608 # Walk the body tree in reading order, carrying the governing section
609 # heading onto every text chunk beneath it. A question that names a
610 # section then pulls the section's content, and the heading never
611 # competes as a bare fragment. Captions attach to the object they
612 # caption instead (a table chunk already leads with its title), so they
613 # are not standalone units either.
614 # Each SoM table's docling counterpart (for the governing heading and its
615 # reading-order position), and the reverse map (for a footnote chunk's
616 # parent link back to the SoM table chunk it follows).
617 som_docling_ref: dict[int, str] = {}
618 docling_to_som: dict[str, str] = {}
619 if report:
620 for m in report.get("matches", []):
621 refs = m.get("docling_table_refs", [])
622 for si in m.get("som_indices", []):
623 if refs:
624 som_docling_ref[si] = refs[0]
625 docling_to_som.setdefault(refs[0], f"#/tables/{si}")
627 all_texts = doc.get("texts", [])
628 all_pictures = doc.get("pictures", [])
629 all_groups = doc.get("groups", [])
630 order = _reading_order_refs(doc)
631 walk_refs = order or [f"#/texts/{i}" for i in range(len(all_texts))]
632 heading = ""
633 # The heading in effect when the walk passes each docling table. Tables
634 # get the same ride-along as text: a table titled only by a printed date
635 # line is unfindable by the section vocabulary that governs it.
636 table_headings: dict[str, str] = {}
637 # The walk also captures what footnote resolution needs: every content
638 # line with its page in reading order, where each table sits in that
639 # stream (a marker's definition is searched only AFTER its table), and
640 # the section headings (the Notes-pointer universe).
641 ordered_texts: list[tuple[str, int]] = []
642 table_positions: dict[str, int] = {}
643 table_sequence: list[str] = []
644 headings: list[tuple[str, int]] = []
645 # Each page's opening heading: the page's subject, which panel labels the
646 # parse also calls headings would otherwise strip from the ride-along.
647 page_first_heading: dict[int, str] = {}
648 # A footnote node directly after a table belongs to that table: it gets a
649 # parent link to the table's chunk. Any other content text in between
650 # breaks the adjacency.
651 last_table_ref: Optional[str] = None
652 # Where each element that owns a footnote block sits in the line stream.
653 # A figure's definitions are searched from its own position up to the next
654 # such element, the same universe a table gets.
655 owner_positions: list[int] = []
656 # Each described figure carrying markers, as (index in `records`, the marks
657 # the agent read off it, its page), resolved after the walk.
658 picture_notes: list[tuple[int, list[dict], list[FootnoteDef], int, int]] = []
659 # Members of a scan block are served through their group's single record,
660 # so their own refs are skipped when the walk reaches them.
661 grouped_members: set[str] = set()
662 for ref_id in walk_refs:
663 if ref_id.startswith("#/groups/"):
664 # A group that binds a label to its value is served as ONE record,
665 # so the label travels with its number. Two producers write such
666 # groups: the parse itself, as a key_value_area over a stat
667 # panel's pair, and the figure run's scan blocks, which group what
668 # the parse left as bare page text. A list group is neither — its
669 # items answer questions one at a time and stay separate records.
670 # The flat stream treats every group as a transparent wrapper.
671 if not grouped:
672 continue
673 gidx = int(ref_id.rsplit("/", 1)[1])
674 group = all_groups[gidx] if gidx < len(all_groups) else {}
675 if group.get("name") != SCAN_BLOCK and group.get("label") != "key_value_area":
676 continue
677 members = []
678 for child in group.get("children", []):
679 mref = child.get("$ref") or ""
680 if not mref.startswith("#/texts/"):
681 continue
682 t = all_texts[int(mref.rsplit("/", 1)[1])]
683 text = (t.get("text") or "").strip()
684 if not text:
685 continue
686 # A heading grouped with its content still governs what
687 # follows the group: it feeds the heading stream exactly as it
688 # would have standing alone, and its text stays in the group's
689 # own chunk.
690 if t.get("label") == "section_header":
691 heading = text
692 mpage = (t.get("prov") or [{}])[0].get("page_no", 1)
693 headings.append((text, mpage))
694 page_first_heading.setdefault(mpage, text)
695 members.append((mref, t, text))
696 if not members:
697 continue
698 grouped_members.update(mref for mref, _t, _text in members)
699 prov = (members[0][1].get("prov") or [{}])[0]
700 page1 = prov.get("page_no", 1)
701 w, h = sizes.get(page1, (1.0, 1.0))
702 boxes = [
703 b
704 for _mref, t, _text in members
705 for p in [(t.get("prov") or [{}])[0]]
706 if p.get("bbox")
707 for b in [_norm_bbox(p["bbox"], w, h)]
708 if b
709 ]
710 bbox = (
711 {
712 "left": min(b["left"] for b in boxes),
713 "top": min(b["top"] for b in boxes),
714 "right": max(b["right"] for b in boxes),
715 "bottom": max(b["bottom"] for b in boxes),
716 }
717 if boxes
718 else None
719 )
720 for _mref, _t, text in members:
721 ordered_texts.append((text, page1))
722 content = "\n".join(text for _mref, _t, text in members)
723 section = _governing(page_first_heading.get(page1, ""), heading)
724 if section and section.lower() not in content.lower():
725 content = f"{section} — {content}"
726 records.append((group.get("self_ref") or ref_id, "text", page1 - 1, bbox, content, None))
727 continue
728 if ref_id in grouped_members:
729 continue
730 if ref_id.startswith("#/tables/"):
731 table_headings[ref_id] = heading
732 table_positions[ref_id] = len(ordered_texts)
733 table_sequence.append(ref_id)
734 owner_positions.append(len(ordered_texts))
735 last_table_ref = ref_id
736 continue
737 if ref_id.startswith("#/pictures/"):
738 pic = all_pictures[int(ref_id.rsplit("/", 1)[1])]
739 content = _picture_text(
740 pic, values_by_picture.get(pic.get("self_ref") or "") or values_by_picture.get(ref_id, [])
741 )
742 if not content:
743 continue
744 prov = (pic.get("prov") or [{}])[0]
745 page1 = prov.get("page_no", 1)
746 w, h = sizes.get(page1, (1.0, 1.0))
747 bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
748 section = _governing(page_first_heading.get(page1, ""), heading) if grouped else heading
749 if section and section.lower() not in content.lower():
750 content = f"{section} — {content}"
751 marks = _picture_marks(pic)
752 defs = _picture_notes(pic)
753 if marks or defs:
754 owner_positions.append(len(ordered_texts))
755 # The notes these markers point at are printed below the figure
756 # and have not been walked yet, so the block is attached in a
757 # second pass once every content line is known.
758 picture_notes.append((len(records), marks, defs, page1, len(ordered_texts)))
759 records.append((pic.get("self_ref") or ref_id, "picture", page1 - 1, bbox, content, None))
760 continue
761 if not ref_id.startswith("#/texts/"):
762 continue
763 idx = int(ref_id.rsplit("/", 1)[1])
764 if idx >= len(all_texts):
765 continue
766 t = all_texts[idx]
767 text = (t.get("text") or "").strip()
768 label = t.get("label")
769 prov = (t.get("prov") or [{}])[0]
770 page1 = prov.get("page_no", 1)
771 if label == "section_header":
772 heading = text
773 if text:
774 headings.append((text, page1))
775 page_first_heading.setdefault(page1, text)
776 last_table_ref = None
777 continue
778 if not text or label == "caption":
779 continue
780 if _is_page_number(text, page1, t.get("content_layer") or ""):
781 continue
782 ordered_texts.append((text, page1))
783 parent = docling_to_som.get(last_table_ref or "") if label == "footnote" else None
784 if label != "footnote":
785 last_table_ref = None
786 w, h = sizes.get(page1, (1.0, 1.0))
787 bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
788 section = _governing(page_first_heading.get(page1, ""), heading) if grouped else heading
789 content = f"{section} — {text}" if section and section.lower() not in text.lower() else text
790 # page stored 0-based to match the ADE convention used elsewhere.
791 records.append((t.get("self_ref") or ref_id, "text", page1 - 1, bbox, content, parent))
793 # A figure states its markers the way a table does, so its notes are looked
794 # up the same way. The marker alone says a qualification exists and not what
795 # it says: a chart labelled "Undepreciated Book Equity Value" with a raised 1
796 # reads as a plain figure without the note saying it excludes a
797 # noncontrolling interest.
798 # Every note read on a page, and every marker printed on it, pooled across the
799 # figures. A page prints ONE note block at its foot and it serves each figure
800 # above it: on one earnings deck, three charts on a page all print a marker 1
801 # against a single note defining it. The agent is asked which figure each note
802 # belongs to and has to answer with one, so the other two figures lose it.
803 page_defs: dict[int, list[FootnoteDef]] = {}
804 page_markers: dict[int, set[str]] = {}
805 for _i, marks, defs, page1, _p in picture_notes:
806 page_defs.setdefault(page1, []).extend(defs)
807 page_markers.setdefault(page1, set()).update(
808 canonical_marker(m.get("marker", "")) for m in marks if m.get("marker")
809 )
811 orphans_seen: set[tuple[int, str]] = set()
812 for index, marks, defs, page1, pos in picture_notes:
813 markers = [m.get("marker", "") for m in marks if m.get("marker")]
814 if not markers and not defs:
815 continue
816 later = [p for p in owner_positions if p > pos]
817 end = later[0] if later else len(ordered_texts)
818 trailing = ordered_texts[pos:end]
819 section_keys = {canonical_marker(m.get("marker", "")) for m in marks if m.get("kind") == "section"}
820 # The agent's own pairs come first, exactly as a table's do. It read the
821 # notes off the page image, so a marker is answered by the line printed
822 # for it rather than by whichever line a scan happens to match. The scan
823 # over the page's lines stays behind it as a fallback.
824 # This figure's own notes, then the ones the page printed for its other
825 # figures. A page prints ONE note block and it serves every figure above
826 # it, so a marker this figure prints is answered by that block whichever
827 # figure the agent filed the note under. Both go in ahead of the
828 # reading-order scan: a note an agent read off the page is better
829 # evidence than a line whose leading character happens to match. Own
830 # notes are listed first, so a figure's own always wins a collision.
831 own_keys = {canonical_marker(d.marker) for d in defs}
832 pool = list(defs) + [
833 d for d in page_defs.get(page1, []) if canonical_marker(d.marker) not in own_keys
834 ]
835 resolution = resolve_footnotes(markers, pool, page1, trailing, headings, section_keys)
836 # A definition read for a neighbour is marked as borrowed rather than
837 # folded in silently, so it stays distinguishable from one read here.
838 resolution = FootnoteResolution(
839 resolved=[
840 r
841 if r.source != "table" or canonical_marker(r.marker) in own_keys
842 else r.model_copy(update={"source": "sibling"})
843 for r in resolution.resolved
844 ],
845 unresolved=list(resolution.unresolved),
846 unreferenced=list(resolution.unreferenced),
847 )
848 if resolution.unresolved:
849 resolution = _lookup_unresolved(resolution, DATA_DIR / f"{doc_key}.pdf", page1)
850 ref_id, ref_type, page0, bbox, body, parent = records[index]
851 notes = [f"{r.marker} — {r.text}" for r in resolution.resolved]
852 if notes:
853 records[index] = (
854 ref_id,
855 ref_type,
856 page0,
857 bbox,
858 body + FOOTNOTE_HEADER + "\n".join(notes),
859 parent,
860 )
861 logger.info("Attached {} footnote(s) to picture {} (page {})", len(notes), ref_id, page1)
863 # A note is orphaned only when NO marker on the page names it, and it is
864 # then the PAGE's exception rather than each figure's. Both halves matter:
865 # judged per figure, every shared note would look unreferenced to the
866 # figures that did not print its marker, and every genuinely orphaned note
867 # would then be reported once per figure on the page.
868 orphans: list[str] = []
869 for d in resolution.unreferenced:
870 key = canonical_marker(d.marker)
871 if key in page_markers.get(page1, set()) or (page1, key) in orphans_seen:
872 continue
873 orphans_seen.add((page1, key))
874 orphans.append(f"{d.marker} {d.text}".strip())
876 # Footnote exceptions, quoted verbatim, the same three a table reports. A
877 # figure whose marker finds no definition was previously silent, which is
878 # the failure that cannot be seen by looking at what a run produced: a
879 # missing qualification and a clean figure read identically.
880 flags = (
881 [("footnote_unresolved", m) for m in resolution.unresolved]
882 + [("footnote_unreferenced", quoted) for quoted in orphans]
883 + [
884 ("footnote_marker_unplaced", m.get("marker", ""))
885 for m in marks
886 if not (m.get("label") or "").strip()
887 ]
888 )
889 for k, (status, quoted) in enumerate(flags):
890 logger.warning(
891 "picture {r} (page {p}): {status}: {q!r}", r=ref_id, p=page1, status=status, q=quoted
892 )
893 cell_rows.append(
894 (f"{ref_id}-fn-{k}", "tableFlag", page0, bbox, {"chunk_id": ref_id}, status, None, quoted)
895 )
897 records.extend(_unanchored_value_records(figure_values, page_first_heading))
899 # One grounding per reconciled figure value, the way a table serves one per
900 # cell. The value's own box (already normalized top-left 0..1) is its
901 # overlay; a value with no box of its own borrows its picture's, and only
902 # when the picture cannot be resolved is it served without one. Status and
903 # note ride along so the UI flags an unreconciled value like a flagged cell.
904 for fid, v in figure_values:
905 box = v.box or _picture_bbox(doc, sizes, v.picture_ref)
906 position = {
907 "label": v.label,
908 "series": v.series,
909 "chart": v.chart_title,
910 "picture_ref": v.picture_ref,
911 }
912 if not v.picture_ref:
913 position["chunk_id"] = f"figure-values-p{v.page}"
914 cell_rows.append(
915 (
916 fid,
917 "figureValue",
918 v.page - 1,
919 dict(box) if box else None,
920 position,
921 v.status,
922 v.note,
923 v.value,
924 )
925 )
927 unified_tables = {tb.get("self_ref"): tb for tb in doc.get("tables", [])}
928 for i, t in enumerate(tables):
929 page1 = t.get("page", 1)
930 region = t.get("content_region") or t.get("som_region")
931 bbox = None
932 if region:
933 x1, y1, x2, y2 = region
934 bbox = {"left": x1, "top": y1, "right": x2, "bottom": y2}
936 # The corrected grid gives one GroundedCell per markdown cell. Render
937 # it as HTML with per-cell ids so the agent can cite cells; fall back
938 # to the plain markdown for tables without a grid (charts, image
939 # tables).
940 grid = t.get("corrected_grid") or []
941 body = _grid_html(i, grid) if grid else t.get("markdown")
942 head = _table_head(t, unified_tables.get(som_docling_ref.get(i, "")), all_texts)
943 section = table_headings.get(som_docling_ref.get(i, ""), "")
944 if grouped:
945 section = _governing(page_first_heading.get(t.get("page", 1), ""), section)
946 if section and all(section.lower() not in p.lower() for p in head):
947 head.insert(0, section)
949 # Resolve this table's footnote markers to their definitions. The
950 # scan universe is every content line AFTER the table in reading
951 # order; without a fusion report to place the table in that stream,
952 # the fallback is every line from the table's page onward.
953 marks = t.get("footnote_marks") or []
954 fdefs = [
955 FootnoteDef(marker="", text=f) if isinstance(f, str) else FootnoteDef.model_validate(f)
956 for f in (t.get("footnotes") or [])
957 ]
958 markers = [m.get("marker", "") for m in marks] + list(t.get("footnote_refs") or [])
959 # The scan universe ends where the NEXT table begins: a footnote
960 # printed beyond it belongs to that table, and another table's own
961 # coherent footnote block would otherwise satisfy this table's
962 # markers. The overleaf continuation case has no table in between,
963 # so it survives the cut; what the cut excludes falls to the lookup
964 # agent, which reads the page image.
965 own_ref = som_docling_ref.get(i, "")
966 pos = table_positions.get(own_ref)
967 if pos is not None:
968 seq = table_sequence.index(own_ref)
969 end = (
970 table_positions[table_sequence[seq + 1]]
971 if seq + 1 < len(table_sequence)
972 else len(ordered_texts)
973 )
974 trailing = ordered_texts[pos:end]
975 else:
976 trailing = [x for x in ordered_texts if x[1] >= page1]
977 section_keys = {canonical_marker(m.get("marker", "")) for m in marks if m.get("kind") == "section"}
978 resolution = resolve_footnotes(markers, fdefs, page1, trailing, headings, section_keys)
979 if resolution.unresolved:
980 resolution = _lookup_unresolved(resolution, DATA_DIR / f"{doc_key}.pdf", page1)
982 # The whole-table chunk carries every resolved footnote plus the
983 # unmarked general notes (those attach at table level only).
984 table_notes = [f"{r.marker} — {r.text}" for r in resolution.resolved] + [
985 d.text for d in fdefs if not d.marker and d.text.strip()
986 ]
987 content = "\n".join(head + [body]) if body else "\n".join(head)
988 if table_notes:
989 content += FOOTNOTE_HEADER + "\n".join(table_notes)
990 records.append((f"#/tables/{i}", "table", page1 - 1, bbox, content, None))
992 # One searchable record per printed table line, alongside the whole-
993 # table record. One embedding cannot represent forty printed lines, so
994 # a question about a single line item gets a unit that IS that line.
995 # Each line rides with the table's section heading, title, units, its
996 # full header block, and the row-axis label governing it, so it is
997 # answerable alone; the cell ids are the same ones the table record
998 # carries, so citation resolves identically. The header block depth is
999 # the correction review's reading of the table image — there is no
1000 # derived fallback. A gridded table can legitimately arrive without a
1001 # depth: when structure correction is rejected or fails, the table
1002 # keeps its grid but no image-read header count exists. Such a table
1003 # is served whole (record above, cell groundings below); only its
1004 # per-line records are skipped, loudly, rather than guessed at or
1005 # failing the document. Header rows and row-axis label rows get no
1006 # record of their own — a record whose whole content is header words
1007 # is retrieval noise posing as data.
1008 depth = t.get("header_rows")
1009 if grid and not isinstance(depth, int):
1010 logger.warning(
1011 "table {i} (page {p}): gridded but carries no header depth — structure "
1012 "correction absent; serving the whole-table record only, no per-line records",
1013 i=i,
1014 p=page1,
1015 )
1016 elif grid:
1017 label_row: Optional[int] = None
1018 for r in range(depth, len(grid)):
1019 row = grid[r]
1020 if not any((cell.get("text") or "").strip() for cell in row):
1021 continue
1022 # A row-axis section label ("Business Segments", "Revenues:")
1023 # carries text only in the stub column. A row with text across
1024 # columns is a line, whether its values are bare numbers or
1025 # worded ranges ("6.1 to 6.4 million").
1026 if not any((cell.get("text") or "").strip() for cell in row[1:]):
1027 label_row = r
1028 continue
1029 rows = list(range(depth)) + ([label_row] if label_row is not None else []) + [r]
1030 line_html = _grid_html(i, [grid[k] for k in rows], row_offsets=rows)
1031 line_content = "\n".join(head + [line_html])
1032 # A footnote rides with exactly the records that reference its
1033 # marker: a marker in the header rows reaches every record (a
1034 # column footnote qualifies every row); a marker on the stub
1035 # label or the data row reaches only this record.
1036 rows_set = set(rows)
1037 line_notes: list[str] = []
1038 seen_keys: set[str] = set()
1039 for m in marks:
1040 if m.get("row") not in rows_set:
1041 continue
1042 key = canonical_marker(m.get("marker", ""))
1043 if not key or key in seen_keys:
1044 continue
1045 seen_keys.add(key)
1046 note_text = resolution.text_for(m.get("marker", ""))
1047 if note_text:
1048 line_notes.append(f"{m.get('marker')} — {note_text}")
1049 if line_notes:
1050 line_content += FOOTNOTE_HEADER + "\n".join(line_notes)
1051 boxes = [cell["box"] for cell in row if cell.get("box")]
1052 line_bbox = (
1053 {
1054 "left": min(b[0] for b in boxes),
1055 "top": min(b[1] for b in boxes),
1056 "right": max(b[2] for b in boxes),
1057 "bottom": max(b[3] for b in boxes),
1058 }
1059 if boxes
1060 else bbox
1061 )
1062 records.append(
1063 (f"t{i}-line-{r}", "line_item", page1 - 1, line_bbox, line_content, f"#/tables/{i}")
1064 )
1066 # One grounding per cell. A cell whose box could not be traced is
1067 # still served: its overlay is the vicinity read off its boxed
1068 # neighbors (column x-extent crossed with row y-extent), and only when
1069 # that is underivable the whole table region. Status/note ride along
1070 # either way so the UI can flag it.
1071 for r, row in enumerate(grid):
1072 for c, cell in enumerate(row):
1073 box = cell.get("box") or _cell_vicinity(grid, r, c, region)
1074 cell_bbox = (
1075 {"left": box[0], "top": box[1], "right": box[2], "bottom": box[3]} if box else bbox
1076 )
1077 cell_rows.append(
1078 (
1079 f"t{i}-{r}-{c}",
1080 "tableCell",
1081 page1 - 1,
1082 cell_bbox,
1083 {"row": r, "col": c, "chunk_id": f"#/tables/{i}"},
1084 cell.get("status"),
1085 cell.get("note"),
1086 cell.get("text") or "",
1087 )
1088 )
1090 # Printed header-area text that reached no output cell is a table-level
1091 # review flag. It has no cell to point at, but the status means the
1092 # text was printed in the table's header area, so the band above the
1093 # first boxed row is its vicinity; the whole region is the fallback.
1094 band = _header_band(grid, region)
1095 band_bbox = {"left": band[0], "top": band[1], "right": band[2], "bottom": band[3]} if band else bbox
1096 for k, fragment in enumerate(t.get("dropped_text") or []):
1097 cell_rows.append(
1098 (
1099 f"t{i}-dropped-{k}",
1100 "tableFlag",
1101 page1 - 1,
1102 band_bbox,
1103 {"chunk_id": f"#/tables/{i}"},
1104 "header_text_dropped",
1105 None,
1106 fragment,
1107 )
1108 )
1110 # Footnote exceptions, quoted verbatim. A clean table adds nothing.
1111 footnote_flags = (
1112 [("footnote_unresolved", m) for m in resolution.unresolved]
1113 + [("footnote_unreferenced", f"{d.marker} {d.text}".strip()) for d in resolution.unreferenced]
1114 + [("footnote_marker_unplaced", m.get("marker", "")) for m in marks if m.get("row") is None]
1115 )
1116 for k, (status, quoted) in enumerate(footnote_flags):
1117 logger.warning("table {i} (page {p}): {status}: {q!r}", i=i, p=page1, status=status, q=quoted)
1118 cell_rows.append(
1119 (
1120 f"t{i}-fn-{k}",
1121 "tableFlag",
1122 page1 - 1,
1123 bbox,
1124 {"chunk_id": f"#/tables/{i}"},
1125 status,
1126 None,
1127 quoted,
1128 )
1129 )
1131 # Tables the Set-of-Mark engine missed exist only in the unified document;
1132 # ingest them from docling's own body so their content is retrievable.
1133 if report:
1134 missed_refs = {
1135 ref
1136 for m in report.get("matches", [])
1137 if m.get("kind") == "som_miss"
1138 for ref in m.get("docling_table_refs", [])
1139 }
1140 by_ref = {t.get("self_ref"): t for t in doc.get("tables", [])}
1141 for ref in sorted(missed_refs - superseded):
1142 t = by_ref.get(ref)
1143 if t is None:
1144 logger.warning("fusion report names {ref} but the unified doc has no such table", ref=ref)
1145 continue
1146 content = _docling_table_text(t)
1147 if not content:
1148 logger.warning("SoM-missed table {ref} has no cell text; skipped", ref=ref)
1149 continue
1150 prov = (t.get("prov") or [{}])[0]
1151 page1 = prov.get("page_no", 1)
1152 section = table_headings.get(ref, "")
1153 if grouped:
1154 section = _governing(page_first_heading.get(page1, ""), section)
1155 if section and section.lower() not in content.lower():
1156 content = f"{section}\n{content}"
1157 w, h = sizes.get(page1, (1.0, 1.0))
1158 bbox = _norm_bbox(prov.get("bbox", {}), w, h) if prov.get("bbox") else None
1159 # Prefixed ref id: '#/tables/N' is already taken by the SoM tables.
1160 records.append((f"#/docling{ref}", "table", page1 - 1, bbox, content, None))
1161 logger.info("Ingesting SoM-missed docling table {ref} (page {p})", ref=ref, p=page1)
1163 # --- Embed and insert ---
1164 # Embed what a reader reads: the stored content keeps its HTML so the
1165 # agent can cite cell ids, but the markup is noise to the embedding — on a
1166 # short line record the td tags and ids outweigh the words themselves.
1167 vectors = embed_document(doc_key, [_embed_view(r[4]) for r in records]) if records else []
1168 meta = doc.get("origin", {})
1169 filename = filename or meta.get("filename") or f"{doc_key}.pdf"
1170 page_count = len(sizes) or None
1171 content_hash = hashlib.sha256(pdf_local.read_bytes()).hexdigest()
1173 with db.connect() as conn:
1174 conn.execute("DELETE FROM ade_playground.documents WHERE doc_key = %s", (doc_key,))
1175 row = conn.execute(
1176 """INSERT INTO ade_playground.documents
1177 (doc_key, content_hash, filename, page_count, ade_version)
1178 VALUES (%s, %s, %s, %s, %s) RETURNING id""",
1179 (doc_key, content_hash, filename, page_count, "quber-fusion"),
1180 ).fetchone()
1181 assert row is not None # INSERT .. RETURNING always yields a row
1182 doc_id = row[0]
1184 for (ref_id, ref_type, page0, bbox, content, parent), vec in zip(records, vectors, strict=True):
1185 conn.execute(
1186 """INSERT INTO ade_playground.chunks
1187 (document_id, chunk_id, chunk_type, page, bbox, content, embedding, parent_chunk_id)
1188 VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
1189 (doc_id, ref_id, ref_type, page0, json.dumps(bbox), _pg_text(content), vec, parent),
1190 )
1191 # Every chunk is its own grounding (table box or text box).
1192 conn.execute(
1193 """INSERT INTO ade_playground.groundings
1194 (document_id, ref_id, ref_type, page, bbox, position)
1195 VALUES (%s, %s, %s, %s, %s, NULL)
1196 ON CONFLICT (document_id, ref_id) DO NOTHING""",
1197 (doc_id, ref_id, ref_type, page0, json.dumps(bbox)),
1198 )
1200 for ref_id, ref_type, page0, bbox, position, status, note, cell_text in cell_rows:
1201 conn.execute(
1202 """INSERT INTO ade_playground.groundings
1203 (document_id, ref_id, ref_type, page, bbox, position, status, note, cell_text)
1204 VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
1205 ON CONFLICT (document_id, ref_id) DO NOTHING""",
1206 (
1207 doc_id,
1208 ref_id,
1209 ref_type,
1210 page0,
1211 json.dumps(bbox),
1212 json.dumps(position),
1213 status,
1214 _pg_text(note),
1215 _pg_text(cell_text),
1216 ),
1217 )
1219 logger.success(
1220 "Ingested {key}: {n} chunks ({t} text, {tb} table, {ln} line item), {cells} cell groundings",
1221 key=doc_key,
1222 n=len(records),
1223 t=sum(1 for r in records if r[1] == "text"),
1224 tb=sum(1 for r in records if r[1] == "table"),
1225 ln=sum(1 for r in records if r[1] == "line_item"),
1226 cells=len(cell_rows),
1227 )
1228 return len(records)
1231def main(argv: Optional[list[str]] = None) -> None:
1232 import argparse
1234 p = argparse.ArgumentParser(description="Ingest quber fusion output into the playground pgvector schema.")
1235 p.add_argument("--doc-key", required=True, help="opaque storage key for the document")
1236 p.add_argument("--unified-json", required=True, help="local path or s3:// URI to <base>.unified.json")
1237 p.add_argument("--tables-json", required=True, help="local path or s3:// URI to <base>.tables.json")
1238 p.add_argument("--pdf", required=True, help="local path or s3:// URI to the source PDF")
1239 p.add_argument(
1240 "--fusion-json",
1241 default=None,
1242 help="local path or s3:// URI to <base>.fusion.json; enables ingesting "
1243 "SoM-missed docling tables from the unified document",
1244 )
1245 p.add_argument(
1246 "--scanned-tables-json",
1247 default=None,
1248 help="local path or s3:// URI to <base>.scanned-tables.json; the tables the "
1249 "figure run read off a page image, ingested alongside the table engine's",
1250 )
1251 p.add_argument(
1252 "--figures-json",
1253 default=None,
1254 help="local path or s3:// URI to <base>.figures.json; names the parse tables a "
1255 "scanned table replaced, so the parse's reading of them is not ingested too",
1256 )
1257 p.add_argument(
1258 "--figure-values-json",
1259 default=None,
1260 help="local path or s3:// URI to <base>.figure-values.json; the reconciled "
1261 "figure values, grounded per value and listed on their picture chunks",
1262 )
1263 p.add_argument("--filename", default=None, help="original PDF filename, kept for presentation")
1264 p.add_argument(
1265 "--chunking",
1266 choices=["flat", "grouped"],
1267 default="flat",
1268 help="ingestion stream: flat serves every printed line as its own chunk — the dpt-2 "
1269 "stream's byte-stable behavior (default); "
1270 "grouped is the dpt-3 track's — binding groups served as one chunk, page heading "
1271 "joined into the ride-along",
1272 )
1273 p.add_argument("--init-schema", action="store_true", help="(re)create the schema before ingest")
1274 args = p.parse_args(argv)
1276 if args.init_schema:
1277 db.apply_schema()
1278 logger.success("Applied schema")
1279 ingest(
1280 args.doc_key,
1281 args.unified_json,
1282 args.tables_json,
1283 args.pdf,
1284 fusion_json=args.fusion_json,
1285 scanned_tables_json=args.scanned_tables_json,
1286 figures_json=args.figures_json,
1287 figure_values_json=args.figure_values_json,
1288 filename=args.filename,
1289 chunking=args.chunking,
1290 )
1293if __name__ == "__main__":
1294 main()