"""Ground the corrected table's cells to Camelot's measured cell boxes. The structure-correction agent reads the Camelot grid with a spreadsheet address tag printed inside every non-empty cell ('[B3] 1,637' is column B, grid row 3), so for every cell it builds by joining Camelot cells it reports the source ADDRESSES it copied from those tags — `CellMerge.source_cells`, e.g. ["B2", "B3"] — alongside the verbatim source texts. Grounding is then a pure lookup: an address names a `cell_grid` cell, and that cell's Camelot-measured box is the geometry. The model never produces a coordinate and never counts positions; it repeats printed labels, and a lookup resolves them. Every report is validated before it is trusted: the text at the claimed address must appear in the merge's result. A misread address is detected and refused rather than absorbed, so a wrong box can enter the record only if the wrong cell holds the exact same text — never as an invented or unchecked location. Refusals and unreported addresses leave the merge `partial` or `none`, and a partial merge's box is never planted. `log_ungrounded` traces them at debug level. The cell is then boxed by the alignment or closure passes, or classified by `classify_cells`. It reaches a reviewer only if its status is one registered for inspection. The same stage closes the whole table. `resolve_corrected_grid` aligns the corrected markdown to the Camelot grid in reading order — rows first, then cells within each aligned row — matching whole cells on exact text equality, so a value moved by the correction keeps its measured box and identical twins map in order. Combined cells take the box their merge resolved to, but only when the merge is fully grounded and the corrected cell at its reported position holds the merge's result. A second pass then closes the cells whole-cell matching cannot reach: Camelot sometimes glues several printed columns into one grid cell and wraps one printed label across consecutive grid rows, so the corrected cells are *fragments* of grid text. Those are matched as ordered contiguous spans of their aligned rows' text, each taking the union of the grid-cell boxes its span touches. A cell that matches nothing keeps box=None and is listed at debug level by `log_ungrounded_cells` — a token with no home in the grid (e.g. a placeholder the agent invented) never acquires a box. After grounding, `classify_cells` sets a status on every non-empty corrected cell so a missing box reads as what it is: `reconciled` (tied to a printed source with measured coordinates), `header_printed_unlocated` / `label_printed_unlocated` (printed in the source but coordinates not measured, split by row kind), `single_character` (a one-character cell left unboxed, which the span-based passes never match by design), `total_label_added` / `header_label_added` (no printed source — an authorized conventional label the correction added, split by row kind), `unverified` (no printed source and not an authorized label — the catch-all for conditions outside every cataloged classification). Added-label and unverified cells are surfaced for user inspection. The full glossary lives in this package's README.md. Three closure passes then handle the transformations reading-order matching cannot express. Header stacks: a flattened multi-row header's fragments sit in the grid's header region stacked within one column, so an unboxed header cell takes the union of the column run whose concatenation equals its text, with repeated labels pairing to candidate stacks in reading order. Unpaired rows: when the agent rebuilds two stacked printed sections side by side, each corrected row's other-section cells match as ordered spans of a grid row the alignment never consumed. Marker labels: a row label the correction extended with a footnote marker it catalogued in `footnote_refs` matches with the marker stripped. Header merges are also region-checked — a header cell's source must sit above the grid's first value row, because header text often repeats in the data region and the text check alone would let a misread address anchor a header to a data cell. Nothing downstream re-derives or reconciles geometry. """ from __future__ import annotations import re from typing import Dict, List, Literal, Optional, Sequence, Tuple from loguru import logger from quber.agents.llm_client import CellMerge, FootnoteDef, FootnoteMark from quber.core.extractors.base import AUTHORIZED_LABELS, GroundedCell, LocatedMarker, MergedCellBox Box = Tuple[float, float, float, float] _ADDRESS_RE = re.compile(r"^([A-Za-z]+)\s*(\d+)$") # A money/figure cell: currency, parenthesized negative, thousands comma, or a # decimal. Used to find where a table's data begins — its header rows sit above # the first row carrying two or more of these. _VALUE_RE = re.compile(r"\$|\(\d|\d,\d|\d+\.\d") def _norm(s: str) -> str: return re.sub(r"\s+", "", s or "") def _first_value_row(rows: Sequence[Sequence[str]]) -> Optional[int]: """Index of the first row carrying at least two money/figure cells; None when no row does (a prose table has no data boundary).""" for i, row in enumerate(rows): if sum(1 for v in row if _VALUE_RE.search(v)) >= 2: return i return None def _union(boxes: Sequence[Box]) -> Box: return ( min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes), ) def markdown_rows(markdown: str) -> List[List[str]]: """The markdown table's rows as cell lists; row 0 is the header row and the separator line is skipped, matching `CellMerge.row` addressing.""" rows: List[List[str]] = [] for line in (markdown or "").splitlines(): line = line.strip() if not line.startswith("|"): continue parts = [p.strip() for p in line.strip("|").split("|")] if parts and any("-" in p for p in parts) and all(set(p) <= {"-", ":", " "} for p in parts): continue rows.append(parts) return rows def parse_address(address: str) -> Optional[Tuple[int, int]]: """A printed spreadsheet address to 0-based (grid row, grid column). 'B3' names grid row 3 (1-based) and data column B (A=0), as printed in the inline tag `grid_to_addressed_markdown` puts inside each non-empty cell. Anything that does not parse returns None. """ m = _ADDRESS_RE.match((address or "").strip()) if not m: return None letters, number = m.group(1).upper(), int(m.group(2)) if number < 1: return None col = 0 for ch in letters: col = col * 26 + (ord(ch) - ord("A") + 1) return number - 1, col - 1 def resolve_merges( merges: Sequence[CellMerge], cell_grid: Sequence[Sequence[GroundedCell]], markdown: str = "", ) -> List[MergedCellBox]: """Resolve every merge's reported source addresses to measured boxes. Pure lookup plus validation: each address names a `cell_grid` cell, and it is trusted only if that cell's text appears in the merge's result. An address that does not parse, is out of bounds, names a blank cell, fails the text check, or has no Camelot box resolves to None. `box` is the union of the located source boxes. All boxes are `cell_grid`'s own — normalized 0..1, page top-left origin. When the corrected `markdown` is given, a merge in a HEADER row (above the corrected table's first value row) may only cite sources above the grid's own first value row. Header text often repeats in the data region (an equity statement prints 'Treasury' in both), so the text check alone would let a misread address anchor a header cell to a data cell's box; the region constraint refuses it instead. """ corrected_first = _first_value_row(markdown_rows(markdown)) if markdown else None grid_boundary = _first_value_row([[c.text for c in grow] for grow in cell_grid]) def in_data_region(address: str) -> bool: parsed = parse_address(address) return parsed is not None and grid_boundary is not None and parsed[0] >= grid_boundary out: List[MergedCellBox] = [] for m in merges: header_merge = corrected_first is not None and m.row < corrected_first pieces = m.sources if len(m.sources) == len(m.source_cells) else [""] * len(m.source_cells) source_boxes: List[Optional[Box]] = [ None if header_merge and in_data_region(address) else _lookup(address, m.result, piece, cell_grid) for address, piece in zip(m.source_cells, pieces, strict=False) ] located = [b for b in source_boxes if b is not None] if not m.source_cells or not located: grounded = "none" elif len(located) == len(source_boxes): grounded = "cell_address" else: grounded = "partial" out.append( MergedCellBox( result=m.result, row=m.row, col=m.col, sources=list(m.sources), source_cells=list(m.source_cells), source_boxes=source_boxes, box=_union(located) if located else None, grounded_by=grounded, ) ) return out def _lookup( address: str, result: str, piece: str, cell_grid: Sequence[Sequence[GroundedCell]] ) -> Optional[Box]: parsed = parse_address(address) if parsed is None: return None row, col = parsed if not (0 <= row < len(cell_grid) and 0 <= col < len(cell_grid[row])): return None cell = cell_grid[row][col] text = _norm(cell.text) if not text: return None # The claim must be text-anchored, one of two ways: the cell's text appears # in the result (the normal case), or the paired source piece appears in the # cell's text (Camelot glued extra fragments into the cell, so the cell # contains more than the merge used — the cell still IS the piece's home). # A misread address matching neither is refused, never absorbed. if text in _norm(result): return cell.box p = _norm(piece) if p and p in text: return cell.box return None def resolve_corrected_grid( markdown: str, merged: Sequence[MergedCellBox], cell_grid: Sequence[Sequence[GroundedCell]], footnote_refs: Sequence[str] = (), region_text: str = "", ) -> List[List[GroundedCell]]: """The corrected markdown's grid with a measured box on every provable cell. Combined cells take the box their merge resolved to, at the address the merge reported — but only when the merge is fully grounded and the corrected cell at that address actually holds the merge's result, so a miscounted position can never plant a box on the wrong cell. A partial merge's box is never planted. Every other cell is matched to the Camelot grid by an order-preserving alignment: corrected rows align to grid rows by shared values (correction never reorders rows), and within an aligned row pair cells match left-to-right on exact text equality (correction never reorders a row's values, so identical twins map in printed order). No global value search, no context scoring — a cell either aligns exactly or is left to the closure passes. A cell none of them boxes keeps box=None, gets its status from `classify_cells`, and is listed at debug level by `log_ungrounded_cells`. """ rows = markdown_rows(markdown) if not rows: return [] width = max(len(r) for r in rows) padded = [row + [""] * (width - len(row)) for row in rows] grid_keys = [[_norm(c.text) for c in grow] for grow in cell_grid] row_keys = [[_norm(v) for v in row] for row in padded] # A merge's box is planted only when the merge is FULLY grounded and the # corrected cell at its reported position actually holds the result. The # sources are text-validated, but row/col is the agent COUNTING positions # in its own output — a miscount would plant a correct box on the wrong # cell, which no later check could see (a planted cell reads as boxed). # A partial merge's union is just as treacherous in extent: with the # number's source refused it can cover only the '$' symbol, a box that # misses the value it claims. Both fall through to the alignment and # closure passes, which box the true cell from the grid text or leave a # gap that the cell's status reports. merged_boxes: Dict[Tuple[int, int], Box] = {} for m in merged: if m.box is None or m.grounded_by != "cell_address": continue if 0 <= m.row < len(row_keys) and 0 <= m.col < len(row_keys[m.row]): if row_keys[m.row][m.col] == _norm(m.result): merged_boxes[(m.row, m.col)] = m.box boxes: Dict[Tuple[int, int], Box] = dict(merged_boxes) pairs = _align_rows(row_keys, grid_keys) for r, g in pairs: gi = 0 grow = cell_grid[g] gkeys = grid_keys[g] for c, key in enumerate(row_keys[r]): if not key or (r, c) in boxes: continue found = _consume_run(grow, gkeys, gi, key) if found is None: continue end, matched = found if matched is not None: boxes[(r, c)] = matched gi = end # Span tiling is for VALUE rows: within one printed row, row-major char # order matches reading order. In a stacked header region it does not — # fragments interleave across columns row by row, so a stream span can # falsely bridge two columns' fragments. Header rows are closed by the # column-stack pass instead. corrected_first = _first_value_row(row_keys) _tile_windows(pairs, row_keys, grid_keys, cell_grid, boxes, skip_before=corrected_first or 0) _close_header_stacks(row_keys, grid_keys, cell_grid, boxes) _close_unpaired_rows(pairs, row_keys, grid_keys, cell_grid, boxes) _close_marker_labels(pairs, row_keys, grid_keys, cell_grid, boxes, footnote_refs) grid = [ [GroundedCell(text=text, box=boxes.get((r, c))) for c, text in enumerate(row)] for r, row in enumerate(padded) ] classify_cells(grid, cell_grid, region_text) return grid def classify_cells( grid: Sequence[Sequence[GroundedCell]], cell_grid: Sequence[Sequence[GroundedCell]], region_text: str = "", ) -> None: """Set a status on every non-empty corrected cell, so a missing box reads as what it is instead of as an undifferentiated failure. The deciding test for an unboxed cell is whether its text exists in the table's SOURCE — the Camelot grid plus the region's text layer. Present means the correction preserved printed text the matcher does not reach (split header/band rows from value rows, since the two gaps close differently). Absent splits two ways: text that is one of the conventional labels the correction is AUTHORIZED to add gets an added-label status, split by row kind exactly like the printed statuses — 'Total' on an unlabeled totals row is `total_label_added`, a generic column name on a table printed without a header row is `header_label_added`; any other unverifiable text is `unverified` — a condition outside every cataloged classification, named as such rather than folded into a known class. All three are surfaced for user inspection. An unboxed one-character cell is its own class, `single_character`. Most span-based passes never match a key that short, because a bare symbol or digit would anchor inside any unrelated number. A one-character cell can still be boxed by whole-cell alignment, or in the header rows by the column-stack closure (`_close_header_stacks`), which accepts a header fragment that contains it. A boxed one reads `reconciled` like any other. Source texts join on a sentinel so a token can never match across two cells' seam. """ source = _norm("\x00".join(c.text for grow in cell_grid for c in grow if c.text) + "\x00" + region_text) value_rows = {r for r, row in enumerate(grid) if sum(1 for c in row if _VALUE_RE.search(c.text)) >= 2} for r, row in enumerate(grid): for cell in row: if not cell.text.strip(): continue if cell.box is not None: cell.status = "reconciled" continue key = _norm(cell.text) if len(key) < 2: cell.status = "single_character" continue tokens = [t for t in (_norm(p) for p in cell.text.split()) if len(t) >= 2] or [key] if all(t in source for t in tokens): cell.status = "label_printed_unlocated" if r in value_rows else "header_printed_unlocated" elif key.casefold() in AUTHORIZED_LABELS: cell.status = "total_label_added" if r in value_rows else "header_label_added" else: cell.status = "unverified" def _tile_windows( pairs: Sequence[Tuple[int, int]], row_keys: Sequence[Sequence[str]], grid_keys: Sequence[Sequence[str]], cell_grid: Sequence[Sequence[GroundedCell]], boxes: Dict[Tuple[int, int], Box], skip_before: int = 0, ) -> None: """Box the corrected cells whole-cell matching cannot reach: fragments of glued Camelot cells, matched as ordered contiguous spans of their window. Camelot sometimes packs several printed columns into one grid cell ('(135)\\n (93)\\n (14,880)') and wraps one printed label across consecutive grid rows; the corrected cell's text is then a fragment of grid text, never equal to any whole cell. For each aligned row pair the window is the paired grid row plus the unpaired grid rows since the previous pair (where a wrapped label's other lines live). Every char of the window's text carries its grid cell's box; a still-unboxed corrected cell found as a contiguous span at or after the cursor takes the union of the boxes its span touches. The cursor only moves forward, so matches keep printed order and repeated values cannot cross. A key shorter than two normalized chars is never span-matched (a bare digit would anchor inside an unrelated number). A cell that matches nothing keeps no box — a token absent from the grid has no home and never acquires one. Corrected rows before `skip_before` (the header region) are never tiled: their fragments interleave across columns, where a stream span can falsely bridge two columns; the column-stack closure owns them. """ prev_g = -1 for r, g in pairs: window = range(prev_g + 1, g + 1) prev_g = g if r < skip_before: continue if all(not key or (r, c) in boxes for c, key in enumerate(row_keys[r])): continue stream: List[Optional[Box]] = [] text = "" for gi in window: for cell, key in zip(cell_grid[gi], grid_keys[gi], strict=False): stream.extend([cell.box] * len(key)) text += key pos = 0 for c, key in enumerate(row_keys[r]): if not key: continue if (r, c) in boxes: # Already boxed by a merge or a whole-cell match: advance the # cursor past it when findable, so later spans stay ordered. at = text.find(key, pos) if at >= 0: pos = at + len(key) continue if len(key) < 2: continue at = text.find(key, pos) if at < 0: continue span = [b for b in stream[at : at + len(key)] if b is not None] if span: boxes[(r, c)] = _union(span) pos = at + len(key) def _close_header_stacks( row_keys: Sequence[Sequence[str]], grid_keys: Sequence[Sequence[str]], cell_grid: Sequence[Sequence[GroundedCell]], boxes: Dict[Tuple[int, int], Box], ) -> None: """Box header cells the agent flattened from a stacked printed header. A multi-row header's fragments live in the grid's HEADER REGION (the rows above its first value row) stacked within ONE grid column — 'Additional' / 'Paid-In' / 'Capital' each in its own row of the same column. So an unboxed corrected header cell closes deterministically: find the grid column whose top-to-bottom fragment run concatenates to exactly the cell's text (or a single glued fragment containing it), and take the union of the run's boxes. A spanning label printed once ('Common Stock' over Shares and Dollars) serves every corrected column it spans. Repeated labels — twin period bands, identical '% of' stacks — pair with candidate stacks positionally in reading order, the same twins-in-order rule the row alignment uses. Merge reporting still takes precedence: only cells the merges and the alignment left unboxed are considered. """ corrected_first = _first_value_row(row_keys) grid_boundary = _first_value_row(grid_keys) if not corrected_first or not grid_boundary: return width = max((len(grow) for grow in cell_grid), default=0) stacks: List[Tuple[int, List[Tuple[int, GroundedCell]]]] = [] for col in range(width): frags = [ (gi, cell_grid[gi][col]) for gi in range(grid_boundary) if col < len(cell_grid[gi]) and _norm(cell_grid[gi][col].text) ] if frags: stacks.append((col, frags)) def candidates(target: str) -> List[Tuple[int, int, List[Box]]]: out: List[Tuple[int, int, List[Box]]] = [] for col, frags in stacks: for i in range(len(frags)): acc = "" run: List[Box] = [] for j in range(i, len(frags)): acc += _norm(frags[j][1].text) b = frags[j][1].box if b is not None: run.append(b) if acc == target: out.append((frags[i][0], col, run)) break if len(acc) > len(target): break for gi, cell in frags: t = _norm(cell.text) if target != t and target in t and cell.box is not None: out.append((gi, col, [cell.box])) return sorted(out, key=lambda x: (x[0], x[1])) targets: Dict[str, List[Tuple[int, int]]] = {} for r in range(corrected_first): for c, key in enumerate(row_keys[r]): if key: targets.setdefault(key, []).append((r, c)) for key, cells in targets.items(): if all((r, c) in boxes for r, c in cells): continue cands = candidates(key) if not cands: continue for i, (r, c) in enumerate(cells): if (r, c) in boxes: continue _, _, run = cands[min(i, len(cands) - 1)] if run: boxes[(r, c)] = _union(run) def _close_unpaired_rows( pairs: Sequence[Tuple[int, int]], row_keys: Sequence[Sequence[str]], grid_keys: Sequence[Sequence[str]], cell_grid: Sequence[Sequence[GroundedCell]], boxes: Dict[Tuple[int, int], Box], ) -> None: """Box corrected rows that drew cells from a grid row the alignment never consumed. When the agent rebuilds two stacked printed sections side by side, each corrected row holds one section's values in its left columns and the other's in its right — but a corrected row can align to only one grid row, so the other section's cells stay unboxed even though their grid row sits untouched. This pass offers each such corrected row the UNPAIRED grid rows, in order: it adopts the first one in which EVERY still-unboxed key (of two or more normalized chars) occurs as ordered non-overlapping spans, and each key takes the union of the grid-cell boxes its span touches. A pool row is consumed by exactly one corrected row, so twin sections cannot double-assign. All-or-nothing per row keeps the anchoring strong: a pool row must account for the row's whole remainder or none of it. """ paired = {g for _, g in pairs} pool = [g for g in range(len(cell_grid)) if g not in paired and any(grid_keys[g])] for r in range(len(row_keys)): missing = [c for c, key in enumerate(row_keys[r]) if key and (r, c) not in boxes and len(key) >= 2] if not missing: continue for pi, g in enumerate(pool): stream: List[Optional[Box]] = [] text = "" for cell, key in zip(cell_grid[g], grid_keys[g], strict=False): stream.extend([cell.box] * len(key)) text += key pos = 0 found: Dict[int, Box] = {} for c in missing: key = row_keys[r][c] at = text.find(key, pos) if at < 0: break span = [b for b in stream[at : at + len(key)] if b is not None] if not span: break found[c] = _union(span) pos = at + len(key) if len(found) == len(missing): boxes.update({(r, c): b for c, b in found.items()}) pool.pop(pi) break def _close_marker_labels( pairs: Sequence[Tuple[int, int]], row_keys: Sequence[Sequence[str]], grid_keys: Sequence[Sequence[str]], cell_grid: Sequence[Sequence[GroundedCell]], boxes: Dict[Tuple[int, int], Box], footnote_refs: Sequence[str], ) -> None: """Box row labels the correction extended with a catalogued footnote marker. A superscript marker is printed on the page but often absent from both the Camelot grid and the text layer, so the corrected label ('Transaction and integration costs(1)') has no exact home in the grid even though the label itself does. Only markers the correction itself catalogued in `footnote_refs` are considered — stripping is grounded in the correction's own report, never a guess. For each still-unboxed cell in an aligned row pair, the key with a trailing catalogued marker removed must match a grid cell (or run) of the paired row exactly; the remainder must keep at least two normalized chars, so a bare parenthesized value like '(84)' can never be consumed as marker plus empty label. """ inners = {_norm(m).strip("()") for m in footnote_refs} suffixes = sorted( {f"({i})" for i in inners if i} | {i for i in inners if i and not any(ch.isalnum() for ch in i)}, key=len, reverse=True, ) if not suffixes: return for r, g in pairs: grow = cell_grid[g] gkeys = grid_keys[g] for c, key in enumerate(row_keys[r]): if not key or (r, c) in boxes: continue stripped = next( (key[: -len(s)] for s in suffixes if key.endswith(s) and len(key) - len(s) >= 2), None ) if stripped is None: continue found = _consume_run(grow, gkeys, 0, stripped) if found is not None and found[1] is not None: boxes[(r, c)] = found[1] def locate_markers( marks: Sequence[FootnoteMark], footnote_refs: Sequence[str], markdown: str, footnotes: Sequence[FootnoteDef] = (), table_text: str = "", ) -> List[LocatedMarker]: """Resolve each quoted mark to its cell in the corrected markdown. The agent QUOTES the carrying cell (`FootnoteMark.cell_text`); it never counts positions — the same contract as `CellMerge`, and for the same reason: counted coordinates are where the model errs. The lookup matches the quote against the corrected cells on exact normalized text, then with the marker's own group stripped from both sides — a superscript the text layer dropped leaves the quote and the cell differing only by the marker. Every cell the quote matches becomes a located entry (a marker printed on twin labels carries on both). Next, a marker not yet placed that is the entire content of a cell places itself there, unless it is all digits. Markers for this pass come from the catalogue and from the `footnotes` definitions, which feed no other pass. A catalogued marker still without a position is then tried against `table_text` (the table's title and subtitle): carried there, it places at table scope — it qualifies the whole table. What remains — every catalogued marker (in `footnote_refs` or the marks themselves) left without a position — is emitted once as an unplaced entry, kept and flagged downstream, never dropped. """ rows = markdown_rows(markdown) row_keys = [[_norm(v) for v in row] for row in rows] def canon(marker: str) -> str: return _norm(marker).strip("()") # Each distinct marker's printed form and the agent's judgement of what # it points at; a marker only listed in `footnote_refs` defaults to a # footnote. catalogued: Dict[str, Tuple[str, Literal["footnote", "section"]]] = {} for m in marks: c = canon(m.marker) if c: catalogued.setdefault(c, (m.marker, m.kind)) for printed in footnote_refs: c = canon(printed) if c: catalogued.setdefault(c, (printed, "footnote")) def strip_marker(key: str, c: str) -> str: # The marker's printed forms, removed wherever they appear: the # parenthesized group for any marker, the bare glyph for a # pure-symbol marker ('*', '#') whose superscript carries no parens. out = key.replace(f"({c})", "") if not any(ch.isalnum() for ch in c): out = out.replace(c, "") return out placed: List[LocatedMarker] = [] seen: set[Tuple[str, int, int]] = set() for m in marks: c = canon(m.marker) quote = _norm(m.cell_text) if not c or not quote: continue stripped_quote = strip_marker(quote, c) for r, row in enumerate(row_keys): for col, key in enumerate(row): if not key: continue if key == quote or (stripped_quote and strip_marker(key, c) == stripped_quote): pos = (m.marker, r, col) if pos not in seen: seen.add(pos) placed.append(LocatedMarker(marker=m.marker, row=r, col=col, kind=m.kind)) # A marker printed AS a whole cell places itself: when a catalogued # marker or a captured definition's own marker is the entire content of # a table cell — a bare '#' (or '# %') standing in a variance column, a # bare '(b)' standing where the rate would print — that cell IS its # carrier: the table prints the marker in place of a value and the note # says why the value is absent, so no agent catalogue is needed to # connect them. An all-digits marker never places this way, because a # bare '(1)' cell is indistinguishable from a parenthesized negative # value; and an alphanumeric marker only matches its parenthesized form, # never the bare letter, which could be a genuine one-letter cell. A # marker that appears in no whole cell keeps its unplaced/unreferenced # flag. self_placing = dict(catalogued) for d in footnotes: c = canon(d.marker) if c: self_placing.setdefault(c, (d.marker, "footnote")) already = {canon(p.marker) for p in placed} for c, (printed, kind) in self_placing.items(): if c in already or c.isdigit(): continue forms = {f"({c})"} if any(ch.isalnum() for ch in c) else {c, f"{c}%"} for r, row in enumerate(row_keys): for col, key in enumerate(row): if key in forms: catalogued.setdefault(c, (printed, kind)) pos = (printed, r, col) if pos not in seen: seen.add(pos) placed.append(LocatedMarker(marker=printed, row=r, col=col, kind=kind)) # A marker carried by the table's own header text — a title suffix # ('Volume and Rate Analysis (a)') or a spanning band absorbed into the # subtitle ('Accounts Classified as a TDR (c)') — qualifies the whole # table: every record sits under that text, so the marker places at # table scope and reaches all of them rather than being flagged as # unplaced. Matched on the parenthesized form (or the bare glyph for a # pure-symbol marker), the same forms a cell match uses. header_key = _norm(table_text) if header_key: placed_canon = {canon(p.marker) for p in placed} for c, (printed, kind) in catalogued.items(): # Digit markers match here up to two digits — a title suffix # 'Key Financials(1)' is unambiguous, while a longer # parenthesized number is a year or a value, never a marker. if c in placed_canon or (c.isdigit() and len(c) > 2): continue found = f"({c})" in header_key if any(ch.isalnum() for ch in c) else c in header_key if found: placed.append(LocatedMarker(marker=printed, kind=kind, scope="table")) placed_canon = {canon(p.marker) for p in placed} return placed + [ LocatedMarker(marker=printed, kind=kind) for c, (printed, kind) in catalogued.items() if c not in placed_canon ] def _consume_run( grow: Sequence[GroundedCell], gkeys: Sequence[str], start: int, key: str ) -> Optional[Tuple[int, Optional[Box]]]: """Find `key` at or after `start` as one grid cell or a run of consecutive grid cells whose concatenated text equals it exactly. The run covers the correction's unreported cell joins — a rejoined symbol ('$' + '168,663') or a label Camelot split across cells — with the same exactness as a single-cell match: the concatenation must equal the target, character for character. Blank cells inside a run contribute nothing. Returns (index after the run, union box of the run's boxed cells), or None when nothing at or after `start` matches. """ for i in range(start, len(grow)): acc = "" parts: List[Box] = [] for j in range(i, len(grow)): acc += gkeys[j] if gkeys[j]: b = grow[j].box if b is not None: parts.append(b) if acc == key: return j + 1, _union(parts) if parts else None if len(acc) > len(key): break return None def _align_rows( row_keys: Sequence[Sequence[str]], grid_keys: Sequence[Sequence[str]] ) -> List[Tuple[int, int]]: """Order-preserving row alignment maximizing shared cell values. Classic weighted longest-common-subsequence over rows: a corrected row may pair with a grid row only if the corrected row's text contains at least one of the grid row's values (containment, because the correction may have joined grid cells into one corrected cell), rows never cross, and the pairing with the greatest total contained values wins. Pairing is only a routing decision — the cell step still demands exact text equality — so a generous metric here can never produce a wrong box. Deterministic, no thresholds to tune. """ def overlap(a: Sequence[str], b: Sequence[str]) -> int: joined = "".join(a) return sum(1 for k in b if k and k in joined) n, m = len(row_keys), len(grid_keys) weight = [[overlap(row_keys[i], grid_keys[j]) for j in range(m)] for i in range(n)] dp = [[0] * (m + 1) for _ in range(n + 1)] for i in range(1, n + 1): for j in range(1, m + 1): best = max(dp[i - 1][j], dp[i][j - 1]) if weight[i - 1][j - 1] > 0: best = max(best, dp[i - 1][j - 1] + weight[i - 1][j - 1]) dp[i][j] = best pairs: List[Tuple[int, int]] = [] i, j = n, m while i > 0 and j > 0: if weight[i - 1][j - 1] > 0 and dp[i][j] == dp[i - 1][j - 1] + weight[i - 1][j - 1]: pairs.append((i - 1, j - 1)) i, j = i - 1, j - 1 elif dp[i - 1][j] >= dp[i][j - 1]: i -= 1 else: j -= 1 return list(reversed(pairs)) def find_dropped_header_text( grid: Sequence[Sequence[GroundedCell]], cell_grid: Sequence[Sequence[GroundedCell]], absorbed: str, page: int, ) -> List[str]: """Return printed header-area text that appears in no corrected cell. A flattened multi-level header must carry every level, but the correction model drops a group label ('Coverage Data' printed over two sub-headers) inconsistently from run to run, even when instructed not to — so a prompt cannot be relied on to close it. The check is deterministic: a cell in the grid's header region (rows above its first value row) counts as dropped when NONE of its words appears in any corrected cell or in the table's absorbed metadata (title, subtitle, units, footnotes). Word level, because Camelot glues side-by-side stacks into single cells whose whole text exists nowhere even when every word was carried. Cells whose longest word is under four chars are skipped — stub noise, not a level. Each returned fragment becomes a table-level review flag (status 'header_text_dropped') in the flags record — the review channel; the log line here is a debugging trace only. """ boundary = _first_value_row([[c.text for c in grow] for grow in cell_grid]) if not boundary: return [] seen = _norm("\x00".join(c.text for row in grid for c in row if c.text) + "\x00" + absorbed) dropped = [] for gi in range(boundary): for cell in cell_grid[gi]: tokens = [t for t in (_norm(p) for p in cell.text.split()) if len(t) >= 3] if not tokens or max(len(t) for t in tokens) < 4: continue if not any(t in seen for t in tokens): dropped.append(cell.text.strip()) if dropped: logger.debug( "page {}: text printed above the table's first value row reached no output cell " "(flagged for review as 'header_text_dropped'): {}", page, [t[:50] for t in dropped], ) return dropped def log_ungrounded_cells(grid: Sequence[Sequence[GroundedCell]], page: int) -> None: """Debug trace of corrected cells that carry no measured box. Every gap is already accounted for by its status: expected conditions (unlocated headers and labels, one-character symbols) are pass tier, and anything needing a person reaches the flags record — that is the action channel. This line exists so a debugging session can see all of a table's gaps in one place; it asks for no action. Empty cells have no printed mark and are not gaps. """ missing = [ (r, c, cell.text, cell.status or "unclassified") for r, row in enumerate(grid) for c, cell in enumerate(row) if cell.text.strip() and cell.box is None ] if missing: logger.debug( "page {}: {} output cell(s) carry no measured location (each accounted for by its status): {}", page, len(missing), [(r, c, t[:40], s) for r, c, t, s in missing], ) def log_ungrounded(merged: Sequence[MergedCellBox], page: int) -> None: """Debug trace of merges whose reported source addresses did not all resolve. The record carries the miss (a None per unresolved address and a partial/none state), and the affected cell is then classified and — when a person should look — flagged through the flags record. This line is the step-by-step trace for debugging merge resolution; it asks for no action. """ for m in merged: if m.grounded_by == "cell_address": continue missing = [a for a, b in zip(m.source_cells, m.source_boxes, strict=False) if b is None] logger.debug( "page {}: combined cell {!r} (row {}, col {}) not fully located (grounded '{}'); " "unresolved source cells: {}", page, m.result, m.row, m.col, m.grounded_by, missing if m.source_cells else "none reported", )