"""Resolve figure footnotes from two sources and carry them into the document. The correction agent reads each scanned page's image and reports both ends of the association: the markers printed on the figures' labels and the note lines printed on the page. The correction sweep in `quber.core.figures.correct` records them on the pictures. This module ties marker to note and writes the result where a reader finds it. Two sources answer each marker. The agent's image read is one; the parse's own footnote-labelled text on the page — retained by the track's graft — is the other. Both are keyed by the same canonical marker, so `(1)`, `1` and a superscript one meet. Agreement corroborates the note. Disagreement, an unresolved marker, and a note no marker points at all become review flags on the run, never silent drops. A marker the agent judged a cross-reference to a named section resolves to a heading pointer, the same way a table's does. Notes are pooled across a page's figures because one printed note block serves several charts; the marker on each figure decides which notes are its. The document carries the result the way it does for tables: each resolved note is inserted as a footnote node right after the picture, so a marker inside a chart and its note at the page foot are linked in the document itself. The parse's own footnote text stays where the page prints it. Placement then goes to the value, not just the chart. The agent reports which printed label carries each marker, and that label is matched against the figure's value table; the note attaches to the matching cells' records in the digest. A marker whose label matches no value is flagged unplaced, and the note still sits after the picture. No flag is raised when the figure has no digest record or no value table, or when none of its labels keeps a word other than a stop word once the marker is stripped. Such a note is left unplaced silently. """ from __future__ import annotations from typing import Any, Dict, List, Optional, Sequence, Tuple from docling_core.types.doc.document import DoclingDocument, PictureItem from docling_core.types.doc.labels import DocItemLabel from loguru import logger from quber.core.figures.dpt3.models import DigestedFigure, PageDigest, PlacedNote, ResolvedNote from quber.core.figures.graft import FOOTNOTE_MARKS_FIELD, FOOTNOTES_FIELD from quber.core.figures.models import PageScan from quber.core.figures.values import label_words from quber.core.fusion.footnotes import canonical_marker, resolve_footnotes, split_leading_marker def resolve_figure_footnotes( document: DoclingDocument, scans: Sequence[PageScan], digests: Dict[int, PageDigest], ) -> List[str]: """Resolve every scanned page's figure markers and write the results. Mutates `document` — each picture's resolved notes are inserted after it — and the digests, whose figures receive their resolutions and whose value cells receive the notes placed on them. Returns the review flags. """ flags: List[str] = [] headings = _headings(document) by_ref = {picture.self_ref: picture for picture in document.pictures} for scan in scans: if not scan.figures: continue page = scan.page digest_figures = {figure.node_id: figure for figure in _page_figures(digests, page) if figure.node_id} # Both ends are read before anything is inserted: inserting footnote # nodes adds footnote-labelled items to the page being read. page_notes = _document_notes(document, page) agent_notes, general_by_ref = _agent_notes(scan, by_ref) referenced: set[str] = set() # One resolution per marker per page. The same marker routinely sits on # several labels and several figures — the note behind it is still one # printed line, and an unresolvable marker is flagged once, not once # per place it is printed. resolutions: Dict[str, Optional[ResolvedNote]] = {} for record in scan.figures: picture = by_ref.get(record.picture_ref or "") if picture is None: continue resolved: List[ResolvedNote] = list(general_by_ref.get(picture.self_ref, [])) figure = digest_figures.get(record.chunk_id or "") mine: Dict[str, ResolvedNote] = {} labels: Dict[str, List[str]] = {} for mark in getattr(picture.meta, FOOTNOTE_MARKS_FIELD, None) or []: marker = str(mark.get("marker") or "") key = canonical_marker(marker) if not key: continue referenced.add(key) if key not in resolutions: note, flag = _resolve( marker, key, str(mark.get("kind") or "footnote"), agent_notes, page_notes, headings, page, ) if flag: flags.append(flag) resolutions[key] = note template = resolutions[key] if template is None: continue # The figure's own copy: the same marker on another figure # places against that figure's values independently. if key not in mine: mine[key] = template.model_copy(deep=True) resolved.append(mine[key]) labels.setdefault(key, []).append(str(mark.get("label") or "")) for key, note in mine.items(): flags.extend(_place_on_values(note, labels.get(key, []), figure, page)) if resolved: _insert_notes(document, picture, resolved) if figure is not None: figure.footnotes = resolved for key, (marker, _text) in agent_notes.items(): if key and key not in referenced: flags.append( f"page {page}: the page prints note {marker!r} and no marker on its " "figures points at it" ) return flags def _resolve( marker: str, key: str, kind: str, agent_notes: Dict[str, Tuple[str, str]], page_notes: Dict[str, str], headings: List[Tuple[str, int]], page: int, ) -> Tuple[Optional[ResolvedNote], Optional[str]]: """One marker's resolution and, when something needs review, its flag.""" if kind == "section": pointer = resolve_footnotes([marker], [], page, headings=headings, section_keys={key}) if pointer.resolved: return ResolvedNote(marker=marker, text=pointer.resolved[0].text, sources=["headings"]), None return None, ( f"page {page}: figure marker {marker!r} names a section of the document and no " "heading opens with it" ) agent = agent_notes.get(key) printed = page_notes.get(key) if agent is None and printed is None: return None, ( f"page {page}: figure marker {marker!r} is defined by no note — neither the " "agent's read of the page nor the page's own footnote text has it" ) if agent is None: return ResolvedNote(marker=marker, text=printed or "", sources=["document"]), None if printed is None: return ResolvedNote(marker=marker, text=agent[1], sources=["agent"]), None if _agree(agent[1], printed): return ( ResolvedNote(marker=marker, text=agent[1], sources=["agent", "document"], corroborated=True), None, ) return ( ResolvedNote(marker=marker, text=agent[1], sources=["agent", "document"]), f"page {page}: note {marker!r} reads differently in the two sources — the agent read " f"{agent[1]!r}, the page's own text says {printed!r}", ) def _agree(one: str, other: str) -> bool: """Do two reads of one note say the same thing? Whitespace and case are rendering; one read extending the other is the same note cut at a different point, not a different note. """ a = " ".join(one.split()).casefold().rstrip(".") b = " ".join(other.split()).casefold().rstrip(".") return bool(a) and bool(b) and (a == b or a in b or b in a) def _place_on_values( note: ResolvedNote, labels: Sequence[str], figure: Optional[DigestedFigure], page: int, ) -> List[str]: """Attach the note to the value cells whose text carries one of its labels. A label the marker sits on is the join: every word of it (minus the marker itself) must appear in the cell's text. The marker follows its label whether the model drew one figure or two around it. A note none of its labels place is flagged, and it stays on the picture; a marker printed on a figure's title has no value to sit on, which is what the flag records. Nothing is flagged when the figure has no value table, or when no label keeps a word other than a stop word once the marker is removed. """ if figure is None or not figure.values: return [] tried = False for label in labels: words = label_words(label.replace(note.marker, "")) if not words: continue tried = True for table in figure.values: for row in table.rows: for cell in row: if words <= label_words(cell.text): cell.footnotes.append(PlacedNote(marker=note.marker, text=note.text)) note.placed = True if note.placed or not tried: return [] printed = ", ".join(repr(label) for label in labels if label) return [ f"page {page}: note {note.marker!r} sits on {printed} and the figure's value table " "has no value under that label; the note is on the picture only" ] def _insert_notes(document: DoclingDocument, picture: PictureItem, notes: Sequence[ResolvedNote]) -> None: """Insert each resolved note as a footnote node right after the picture. The same carriage a table's footnotes get: a body node after the element renders as a paragraph below it and keeps the reading order. The node carries no provenance box, so nothing sweeping printed regions matches it. A note already sitting after the picture is not inserted again, so a rerun over an enriched parse — which is what reusing stored scans produces — leaves the document as it was rather than doubling every note. """ anchor, existing = _trailing_notes(document, picture) for note in notes: text = f"{note.marker} {note.text}".strip() if note.marker else note.text if not text.strip() or text in existing: continue anchor = document.insert_text(label=DocItemLabel.FOOTNOTE, text=text, sibling=anchor, after=True) logger.info( "Figure footnotes: picture {} carries {} resolved note(s) ({})", picture.self_ref, len(notes), ", ".join(note.marker or "unmarked" for note in notes), ) def _trailing_notes(document: DoclingDocument, picture: PictureItem) -> Tuple[Any, set[str]]: """The run of footnote nodes already following the picture: the last one, to anchor new insertions behind it, and their texts, to skip re-inserting. Only a prov-less footnote belongs to the run — one this module inserted on an earlier pass. A printed footnote the parse positioned on the page never matches, so it is never mistaken for an insertion. """ anchor: Any = picture existing: set[str] = set() parent = picture.parent.resolve(document) if picture.parent else None if parent is None: return anchor, existing refs = [child.cref for child in parent.children] if picture.self_ref not in refs: return anchor, existing for ref in refs[refs.index(picture.self_ref) + 1 :]: item = document.texts[int(ref.rsplit("/", 1)[1])] if ref.startswith("#/texts/") else None if item is None or item.label != DocItemLabel.FOOTNOTE or item.prov: break existing.add(item.text) anchor = item return anchor, existing def _agent_notes( scan: PageScan, by_ref: Dict[str, PictureItem] ) -> Tuple[Dict[str, Tuple[str, str]], Dict[str, List[ResolvedNote]]]: """The notes the agent read off this page, pooled, plus the unmarked ones. Marked notes pool across the page's figures — one printed note block serves several charts — keyed by canonical marker, each keeping its printed marker and text. An unmarked general note has no marker to pool under and stays with the figure the agent read it for. """ pooled: Dict[str, Tuple[str, str]] = {} general: Dict[str, List[ResolvedNote]] = {} for record in scan.figures: picture = by_ref.get(record.picture_ref or "") if picture is None: continue for entry in getattr(picture.meta, FOOTNOTES_FIELD, None) or []: marker = str(entry.get("marker") or "") text = str(entry.get("text") or "").strip() if not text: continue key = canonical_marker(marker) if key: pooled.setdefault(key, (marker, text)) else: general.setdefault(picture.self_ref, []).append( ResolvedNote(marker="", text=text, sources=["agent"]) ) return pooled, general def _document_notes(document: DoclingDocument, page: int) -> Dict[str, str]: """The page's own footnote text, keyed by the marker each line opens with. These are the footnote-labelled items the parse produced and the track's graft retained. A line opening with no marker defines nothing to key on. """ notes: Dict[str, str] = {} for item in document.texts: if item.label != DocItemLabel.FOOTNOTE: continue if not item.prov or item.prov[0].page_no != page: continue parsed = split_leading_marker(item.text or "") if parsed is not None: notes.setdefault(parsed[0], parsed[1]) return notes def _headings(document: DoclingDocument) -> List[Tuple[str, int]]: """The document's section headings with their pages, for section pointers.""" found: List[Tuple[str, int]] = [] for item in document.texts: if item.label != DocItemLabel.SECTION_HEADER or not (item.text or "").strip(): continue found.append((item.text, item.prov[0].page_no if item.prov else 0)) return found def _page_figures(digests: Dict[int, PageDigest], page: int) -> List[DigestedFigure]: digest = digests.get(page) return list(digest.figures) if digest is not None else []