mARTin-B78
fd1de80f83
Columns are found from where column-width text starts, not from the whitespace between them (loose OCR boxes shrank one gutter to 4-6 pt and a pull-quote bridged it, collapsing page 38 into one interleaved column). Headlines get their own band, pull-quotes are read at the end of the page. Measured over a 204-page issue: 89 sentence continuations kept vs 85. Each review row gets a type picker (text kinds, page number, source, ad, image) and its own delete button; text<->image conversions keep the region in its article. Fix reading order re-reads existing issues without OCR and leaves hand-ordered pages alone. The shared confirm dialog is now an alertdialog for screen readers. v1.35.0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2904 lines
131 KiB
Python
2904 lines
131 KiB
Python
"""Layout-aware magazine ingestion.
|
||
|
||
Magazines are not books: a single page routinely carries several unrelated
|
||
articles, advertisements, sidebars and pull quotes, and an article often
|
||
continues twenty pages later. Treating a page as the unit of narration
|
||
therefore produces audio that jumps mid-sentence into an advert. The unit
|
||
here is the *article*, reconstructed from classified layout blocks.
|
||
|
||
Pipeline (see architecture/magazine-ingestion.md):
|
||
|
||
PDF
|
||
-> extract_pages() native text blocks with bbox/font via PyMuPDF
|
||
-> ocr_page() tesseract TSV fallback for scanned pages
|
||
-> classify_blocks() title / deck / byline / body / caption / ...
|
||
-> assign_reading_order() column detection, then top-to-bottom per column
|
||
-> detect_contents() Stage A: locate and parse the table of contents
|
||
-> group_articles() Stage C: group blocks, link continuations
|
||
-> (human review) Stage D happens in the UI; nothing is narrated
|
||
until an article is approved.
|
||
|
||
PyMuPDF is imported lazily inside the I/O functions so that the pure geometry
|
||
and classification helpers -- which hold all the fiddly logic worth testing --
|
||
can be exercised without the native dependency or a sample PDF.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import difflib
|
||
import json
|
||
import logging
|
||
import math
|
||
import re
|
||
import statistics
|
||
import subprocess
|
||
import tempfile
|
||
import unicodedata
|
||
|
||
import requests
|
||
|
||
from core.retry import request_with_retries
|
||
from dataclasses import dataclass, field, asdict
|
||
from pathlib import Path
|
||
from typing import Any, Iterable, Sequence
|
||
from urllib.parse import urlsplit
|
||
|
||
# ── Block taxonomy ───────────────────────────────────────────────────────────
|
||
# Ordered roughly by how much they matter to narration. Only NARRATABLE types
|
||
# are ever sent to TTS; the rest are kept for display and provenance.
|
||
BLOCK_TITLE = "title"
|
||
BLOCK_HEADING = "heading"
|
||
BLOCK_DECK = "deck" # standfirst / subtitle under the headline
|
||
BLOCK_BYLINE = "byline"
|
||
BLOCK_BODY = "body"
|
||
BLOCK_CAPTION = "caption"
|
||
BLOCK_PULLQUOTE = "pullquote" # verbatim repeat of body text, blown up
|
||
BLOCK_SIDEBAR = "sidebar"
|
||
BLOCK_FURNITURE = "furniture" # running heads, folios, page numbers
|
||
BLOCK_AD = "ad"
|
||
BLOCK_FIGURE = "figure"
|
||
BLOCK_CHART = "chart"
|
||
BLOCK_TABLE = "table"
|
||
BLOCK_CODE = "code"
|
||
VISUAL_TYPES = (BLOCK_FIGURE, BLOCK_CHART, BLOCK_TABLE, BLOCK_CODE)
|
||
|
||
NARRATABLE = (BLOCK_TITLE, BLOCK_HEADING, BLOCK_DECK, BLOCK_BYLINE, BLOCK_BODY)
|
||
|
||
# Text that marks an article jumping to (or arriving from) another page.
|
||
_CONT_TO = re.compile(
|
||
r"\((?:continued|cont\.?)\s+on\s+page\s+(\d{1,4})\)"
|
||
r"|continued\s+on\s+page\s+(\d{1,4})"
|
||
r"|fortsetzung\s+auf\s+seite\s+(\d{1,4})"
|
||
r"|weiter\s+auf\s+seite\s+(\d{1,4})",
|
||
re.I,
|
||
)
|
||
_CONT_FROM = re.compile(
|
||
r"continued\s+from\s+page\s+(\d{1,4})"
|
||
r"|fortsetzung\s+von\s+seite\s+(\d{1,4})",
|
||
re.I,
|
||
)
|
||
|
||
_BYLINE = re.compile(
|
||
r"^\s*(?:by|von|text|words|story|interview)\s*[: ]\s*(.{2,80})$",
|
||
re.I,
|
||
)
|
||
|
||
# Headings that mark a contents page, and the section headers inside one.
|
||
_TOC_MARKERS = (
|
||
"contents", "table of contents", "in this issue", "inside this issue",
|
||
"features", "departments", "columns", "regulars",
|
||
"inhalt", "inhaltsverzeichnis", "in dieser ausgabe", "rubriken", "themen",
|
||
)
|
||
# "The future of batteries ....... 18" / "The future of batteries 18"
|
||
_TOC_ENTRY = re.compile(r"^(.{3,120}?)[\s.…·•\-]{2,}(\d{1,4})\s*$")
|
||
|
||
_ADVERT_MARKERS = ("advertisement", "anzeige", "sponsored", "promotion")
|
||
|
||
|
||
@dataclass
|
||
class Block:
|
||
"""One laid-out region of a page, in PDF user-space coordinates.
|
||
|
||
bbox is [x, y, width, height] with the origin at the page's top-left, which
|
||
is what the reflowed view needs to draw a highlight over the rendered page.
|
||
"""
|
||
id: str
|
||
page: int
|
||
bbox: list[float]
|
||
type: str
|
||
text: str
|
||
reading_order: int = 0
|
||
confidence: float = 1.0
|
||
font_size: float = 0.0
|
||
font_name: str = ""
|
||
bold: bool = False
|
||
column: int = 0
|
||
region_label: str = ""
|
||
continues_from_previous_page: bool = False
|
||
source: str = "native" # "native" | "ocr" | "vision_ocr" | "vision_label"
|
||
|
||
def to_json(self) -> dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass
|
||
class Article:
|
||
id: str
|
||
title: str
|
||
author: str = ""
|
||
pages: list[int] = field(default_factory=list)
|
||
blocks: list[str] = field(default_factory=list)
|
||
images: list[str] = field(default_factory=list)
|
||
status: str = "detected" # detected | approved | rejected
|
||
confidence: float = 0.0
|
||
section: str = ""
|
||
continued_to: list[int] = field(default_factory=list)
|
||
audio_status: str = "none" # none | narration | podcast | both
|
||
|
||
def to_json(self) -> dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
# ── Geometry helpers (pure) ──────────────────────────────────────────────────
|
||
|
||
def bbox_overlaps(a: Sequence[float], b: Sequence[float], tol: float = 0.0) -> bool:
|
||
"""True when two [x, y, w, h] boxes intersect, allowing a tolerance."""
|
||
ax, ay, aw, ah = a
|
||
bx, by, bw, bh = b
|
||
return not (
|
||
ax + aw + tol < bx or bx + bw + tol < ax
|
||
or ay + ah + tol < by or by + bh + tol < ay
|
||
)
|
||
|
||
|
||
def detect_columns(blocks: Sequence[Block], page_width: float,
|
||
min_gap_ratio: float = 0.015) -> list[tuple[float, float]]:
|
||
"""Infer the column grid from where blocks are *absent*.
|
||
|
||
Magazines vary the column count per page, so a fixed grid is wrong. The
|
||
reliable signal is vertical whitespace running the height of the page: we
|
||
project every block onto the x axis, then split wherever the uncovered gap
|
||
is wider than a few percent of the page. Returns (left, right) spans.
|
||
|
||
1.5%, not the rounder 3.5% this started at: measured on a real scanned
|
||
c't magazine page, the actual gutter between two genuine columns was
|
||
only ~2.1% of the page width -- narrower than the old threshold, which
|
||
made detect_columns see one merged column and, combined with a wide
|
||
caption bridging the same gap (see assign_reading_order), produced
|
||
every block on the page landing in column=0.
|
||
"""
|
||
spans = sorted(
|
||
((b.bbox[0], b.bbox[0] + b.bbox[2]) for b in blocks if b.bbox[2] > 1),
|
||
key=lambda s: s[0],
|
||
)
|
||
if not spans:
|
||
return [(0.0, page_width)]
|
||
|
||
merged: list[list[float]] = [list(spans[0])]
|
||
min_gap = page_width * min_gap_ratio
|
||
for left, right in spans[1:]:
|
||
if left - merged[-1][1] > min_gap:
|
||
merged.append([left, right])
|
||
else:
|
||
merged[-1][1] = max(merged[-1][1], right)
|
||
return [(m[0], m[1]) for m in merged]
|
||
|
||
|
||
# Blocks that are not running text: they never define a column or split the page.
|
||
_NOT_TEXT_FLOW = (BLOCK_FURNITURE, BLOCK_AD, BLOCK_FIGURE, BLOCK_CHART, BLOCK_TABLE, BLOCK_CODE)
|
||
|
||
_QUOTE_PAIRS = (("»", "«"), ("„", "“"), ("“", "”"), ("«", "»"), ('"', '"'), ("‚", "‘"))
|
||
|
||
|
||
def is_pullquote_like(block: Block) -> bool:
|
||
"""A short display quote that repeats the article: read after the page's running text.
|
||
|
||
Types are not trusted on their own: on a real scanned page the layout model typed the actual
|
||
pull-quote ("»Wenn man hundertmal die KI nutzt…«") as body text, and typed a left-column body
|
||
paragraph as a pull-quote. A pull-quote is short and wrapped in quotation marks; a long block
|
||
is running text whatever it was labelled.
|
||
"""
|
||
text = " ".join((block.text or "").split())
|
||
if not text or len(text) > 260:
|
||
return False
|
||
if block.type == BLOCK_PULLQUOTE:
|
||
return True
|
||
return any(text.startswith(open_) and text.rstrip(".!? ").endswith(close) for open_, close in _QUOTE_PAIRS)
|
||
|
||
|
||
def column_anchors(blocks: Sequence[Block], page_width: float) -> tuple[list[float], float]:
|
||
"""Where the text columns begin, and the typical column width.
|
||
|
||
Found from the LEFT EDGES of column-width blocks, which cluster tightly, rather than from
|
||
whitespace between columns. The whitespace signal failed on a real scanned c't page: the OCR
|
||
boxes were loose enough that the gutter shrank to 4-6 points (under any sane threshold), and a
|
||
pull-quote laid across the gutter bridged the columns outright -- every block came back as
|
||
column 0 and the two columns were read interleaved. Left edges were unaffected: 25-29 and
|
||
236-240. Columns must be most of a column width apart (the heavier of two close candidates
|
||
wins), so a narrow stray box starting mid-column does not invent a column of its own. The
|
||
weight floor is deliberately low: on page 193 the left column is three paragraphs under a
|
||
screenshot, 19% of the right column's text, and a 25% floor dropped it and interleaved the page.
|
||
Measured over a whole 204-page issue, 5-15% all score the same.
|
||
"""
|
||
flow = [b for b in blocks if b.bbox[2] > 1 and b.type not in _NOT_TEXT_FLOW]
|
||
if not flow:
|
||
return [0.0], page_width
|
||
# The column width comes from running text only. Measured over every block, a page-wide photo
|
||
# and the headline pushed it to 317 points against real columns of ~210 (page 88), which made
|
||
# a plain right-column paragraph look as if it spanned two columns.
|
||
body = [b for b in flow if b.type == BLOCK_BODY] or flow
|
||
widths = sorted(b.bbox[2] for b in body)
|
||
col_w = widths[len(widths) // 2]
|
||
candidates = sorted((b for b in flow if 0.55 * col_w <= b.bbox[2] <= 1.4 * col_w), key=lambda b: b.bbox[0])
|
||
tol = page_width * 0.04
|
||
clusters: list[list[Block]] = []
|
||
for b in candidates:
|
||
if clusters and b.bbox[0] - clusters[-1][-1].bbox[0] <= tol:
|
||
clusters[-1].append(b)
|
||
else:
|
||
clusters.append([b])
|
||
if not clusters:
|
||
return [min(b.bbox[0] for b in flow)], col_w
|
||
weighted = [(min(b.bbox[0] for b in c), sum(b.bbox[3] for b in c)) for c in clusters]
|
||
heaviest = max(w for _, w in weighted)
|
||
anchors: list[tuple[float, float]] = []
|
||
for x, weight in weighted:
|
||
if weight < 0.10 * heaviest:
|
||
continue
|
||
if anchors and x - anchors[-1][0] < 0.6 * col_w:
|
||
if weight > anchors[-1][1]:
|
||
anchors[-1] = (x, weight)
|
||
continue
|
||
anchors.append((x, weight))
|
||
return [x for x, _ in anchors] or [weighted[0][0]], col_w
|
||
|
||
|
||
def assign_reading_order(blocks: list[Block], page_width: float) -> list[Block]:
|
||
"""Order blocks the way a person reads a magazine page.
|
||
|
||
1. Text spanning two or more columns (headline, standfirst) is read in turn from the top,
|
||
and splits the page into bands: everything under it comes after it.
|
||
2. Within a band, each column top to bottom, columns left to right. Sorting purely by y
|
||
interleaves parallel columns and splices one column's sentence into the next.
|
||
3. Short pull-quotes after the page's running text (they repeat the article, so reading
|
||
one mid-column interrupts a sentence), then furniture (running heads, folios).
|
||
|
||
Captions never split the page into bands: a caption under a photo in the middle of a page
|
||
would otherwise tear both columns in half.
|
||
"""
|
||
anchors, col_w = column_anchors(blocks, page_width)
|
||
# Columns never overlap: each ends where the next begins.
|
||
spans = [(a, min(a + col_w, anchors[i + 1]) if i + 1 < len(anchors) else a + col_w)
|
||
for i, a in enumerate(anchors)]
|
||
|
||
def overlap(b: Block, span: tuple[float, float]) -> float:
|
||
return max(0.0, min(b.bbox[0] + b.bbox[2], span[1]) - max(b.bbox[0], span[0]))
|
||
|
||
def column_of(b: Block) -> int:
|
||
best = max(range(len(spans)), key=lambda i: (overlap(b, spans[i]), -i))
|
||
if overlap(b, spans[best]) > 0:
|
||
return best
|
||
# Outside every column (a margin note): the nearest column by left edge.
|
||
return min(range(len(anchors)), key=lambda i: abs(b.bbox[0] - anchors[i]))
|
||
|
||
def crosses_columns(b: Block) -> bool:
|
||
if len(spans) < 2:
|
||
return b.bbox[2] > page_width * 0.6
|
||
return sum(1 for s in spans if overlap(b, s) >= 0.3 * col_w) >= 2
|
||
|
||
def own_strip(b: Block) -> bool:
|
||
"""A real headline sits in a horizontal strip of its own. A wide box lying over the column
|
||
boxes (a duplicate region covering the same text, seen on pages 78 and 154) does not."""
|
||
top, bottom = b.bbox[1], b.bbox[1] + b.bbox[3]
|
||
for other in blocks:
|
||
if other is b or other.type in _NOT_TEXT_FLOW or crosses_columns(other):
|
||
continue
|
||
shared = min(bottom, other.bbox[1] + other.bbox[3]) - max(top, other.bbox[1])
|
||
if shared > 0.3 * max(1.0, min(b.bbox[3], other.bbox[3])):
|
||
return False
|
||
return True
|
||
|
||
def splits_page(b: Block) -> bool:
|
||
"""Text across two or more columns, in a strip of its own: read before what lies under it.
|
||
Captions never split the page -- one in the middle would tear both columns."""
|
||
if b.type in _NOT_TEXT_FLOW or b.type in (BLOCK_CAPTION, BLOCK_PULLQUOTE) or is_pullquote_like(b):
|
||
return False
|
||
return crosses_columns(b) and own_strip(b)
|
||
|
||
def photo_splits_page(b: Block) -> bool:
|
||
"""A photo with no running text beside it separates the text above from the text below.
|
||
|
||
Page 176: a caption beside a screenshot at the top, the two columns under it. Without this
|
||
the caption was read between the columns and split "…zumal sie auch | helfen könnte…". A
|
||
photo inset beside column text (page 164) does not split anything. Paragraph gaps are never
|
||
used for this: in neighbouring columns they often line up, and treating them as breaks
|
||
interleaves the columns again.
|
||
"""
|
||
if b.type not in _NOT_TEXT_FLOW or b.type in (BLOCK_FURNITURE, BLOCK_AD):
|
||
return False
|
||
top, bottom = b.bbox[1], b.bbox[1] + b.bbox[3]
|
||
for other in blocks:
|
||
if other.type in _NOT_TEXT_FLOW or is_pullquote_like(other):
|
||
continue
|
||
if min(bottom, other.bbox[1] + other.bbox[3]) - max(top, other.bbox[1]) > 2.0:
|
||
return False
|
||
return True
|
||
|
||
for b in blocks:
|
||
b.column = column_of(b)
|
||
|
||
breakers = {id(b) for b in blocks if splits_page(b) or photo_splits_page(b)}
|
||
band_tops = sorted(b.bbox[1] for b in blocks if id(b) in breakers)
|
||
spanning = lambda b: id(b) in breakers # noqa: E731
|
||
|
||
def key(b: Block) -> tuple:
|
||
if b.type == BLOCK_FURNITURE:
|
||
return (3, 0, 0, round(b.bbox[1], 1), b.bbox[0])
|
||
if is_pullquote_like(b):
|
||
return (2, 0, b.column, round(b.bbox[1], 1), b.bbox[0])
|
||
# The band is the number of breakers starting at or above this block: a breaker opens its
|
||
# own band, so it follows the columns above it and precedes those below.
|
||
band = sum(1 for top in band_tops if top <= b.bbox[1] + 0.5)
|
||
if spanning(b):
|
||
return (1, band, 0, round(b.bbox[1], 1), b.bbox[0])
|
||
return (1, band, 1 + b.column, round(b.bbox[1], 1), b.bbox[0])
|
||
|
||
ordered = sorted(blocks, key=key)
|
||
for i, b in enumerate(ordered):
|
||
b.reading_order = i
|
||
return ordered
|
||
|
||
|
||
# ── Classification (pure) ────────────────────────────────────────────────────
|
||
|
||
def body_font_size(blocks: Iterable[Block]) -> float:
|
||
"""The document's body size: the char-weighted median of block sizes.
|
||
|
||
Weighting by length stops a page of huge display type from redefining what
|
||
"normal" means -- body copy always dominates by character count.
|
||
"""
|
||
sizes: list[float] = []
|
||
for b in blocks:
|
||
if b.font_size > 0 and b.text.strip():
|
||
sizes.extend([b.font_size] * max(1, len(b.text) // 20))
|
||
return statistics.median(sizes) if sizes else 10.0
|
||
|
||
|
||
def _normalise(text: str) -> str:
|
||
text = unicodedata.normalize("NFKC", text or "")
|
||
return re.sub(r"\s+", " ", text).strip()
|
||
|
||
|
||
def looks_like_byline(text: str) -> tuple[bool, str]:
|
||
"""Detect 'By Jane Smith' / 'Von Jane Smith', returning the name."""
|
||
m = _BYLINE.match(_normalise(text))
|
||
if not m:
|
||
return False, ""
|
||
name = m.group(1).strip(" .,;:")
|
||
# A byline is a name, not a sentence: reject anything long or punctuated.
|
||
if len(name) > 60 or name.count(" ") > 5 or name.endswith((".", "!", "?")):
|
||
return False, ""
|
||
return True, name
|
||
|
||
|
||
def find_continuation(text: str) -> tuple[int | None, int | None]:
|
||
"""Return (continues_on_page, continued_from_page) mentioned in the text."""
|
||
to_m = _CONT_TO.search(text or "")
|
||
from_m = _CONT_FROM.search(text or "")
|
||
to_page = next((int(g) for g in to_m.groups() if g), None) if to_m else None
|
||
from_page = next((int(g) for g in from_m.groups() if g), None) if from_m else None
|
||
return to_page, from_page
|
||
|
||
|
||
def strip_continuation_markers(text: str) -> str:
|
||
"""Remove '(continued on page N)' style cues before text is narrated.
|
||
|
||
These are a print layout convention -- meaningless, and slightly absurd,
|
||
spoken aloud ("open paren continued on page eight close paren"). The
|
||
layout store keeps the original text; only the narration copy is cleaned.
|
||
"""
|
||
cleaned = _CONT_TO.sub("", text or "")
|
||
cleaned = _CONT_FROM.sub("", cleaned)
|
||
return re.sub(r"\s+", " ", cleaned).strip()
|
||
|
||
|
||
def classify_blocks(blocks: list[Block], page_height: float, page_width: float,
|
||
repeated: set[str] | None = None,
|
||
image_boxes: Sequence[Sequence[float]] = (),
|
||
base_size: float | None = None) -> list[Block]:
|
||
"""Assign a semantic type to every block on one page.
|
||
|
||
Font size is judged relative to the document body size rather than in
|
||
absolute points, so the same rules hold for a pocket magazine and a
|
||
broadsheet.
|
||
|
||
base_size must be measured across the whole document, not this page: a page
|
||
carrying a full-bleed photograph and one caption has no body copy of its
|
||
own, so a per-page baseline would make that caption its own reference and
|
||
classify it as body text. It falls back to the page when not supplied.
|
||
"""
|
||
repeated = repeated or set()
|
||
base = base_size if base_size and base_size > 0 else body_font_size(blocks)
|
||
|
||
for b in blocks:
|
||
# A vision OCR engine that explicitly labeled this region (footer,
|
||
# title, page_number, caption -- see build_vision_blocks) is a more
|
||
# direct signal than any font-size heuristic can reconstruct after
|
||
# the fact; second-guessing it here would throw that away.
|
||
if b.source == "vision_label":
|
||
continue
|
||
|
||
text = _normalise(b.text)
|
||
if not text:
|
||
b.type = BLOCK_FURNITURE
|
||
b.confidence = 0.5
|
||
continue
|
||
|
||
ratio = (b.font_size / base) if base else 1.0
|
||
near_top = b.bbox[1] < page_height * 0.07
|
||
near_bottom = b.bbox[1] + b.bbox[3] > page_height * 0.93
|
||
short = len(text) <= 90
|
||
lower = text.lower()
|
||
|
||
# Running heads, folios and page numbers repeat at the same edge on
|
||
# many pages -- that repetition is what identifies them, not position
|
||
# alone, since a headline can also sit at the top.
|
||
if (near_top or near_bottom) and (text in repeated or text.isdigit()):
|
||
b.type, b.confidence = BLOCK_FURNITURE, 0.95
|
||
continue
|
||
|
||
if any(m in lower for m in _ADVERT_MARKERS) and short:
|
||
b.type, b.confidence = BLOCK_AD, 0.8
|
||
continue
|
||
|
||
is_byline, _ = looks_like_byline(text)
|
||
if is_byline:
|
||
b.type, b.confidence = BLOCK_BYLINE, 0.9
|
||
continue
|
||
|
||
# A caption sits directly beneath or beside a figure and is set smaller
|
||
# than body copy. Both conditions are needed: small text elsewhere is
|
||
# usually a footnote, and body-sized text under an image is often the
|
||
# article continuing past it.
|
||
if ratio < 0.92 and any(
|
||
bbox_overlaps(b.bbox, box, tol=max(page_height, page_width) * 0.02)
|
||
for box in image_boxes
|
||
):
|
||
b.type, b.confidence = BLOCK_CAPTION, 0.75
|
||
continue
|
||
|
||
if ratio >= 1.6 and short:
|
||
b.type, b.confidence = BLOCK_TITLE, 0.9 if ratio >= 2.0 else 0.7
|
||
continue
|
||
|
||
if 1.18 <= ratio < 1.6 and len(text) <= 300:
|
||
# Display-sized text mid-column is a pull quote; at the head of a
|
||
# page it is the deck introducing the article.
|
||
b.type = BLOCK_DECK if b.bbox[1] < page_height * 0.45 else BLOCK_PULLQUOTE
|
||
b.confidence = 0.6
|
||
continue
|
||
|
||
b.type, b.confidence = BLOCK_BODY, 0.85
|
||
|
||
return blocks
|
||
|
||
|
||
def mark_advert_pages(blocks: list[Block]) -> list[Block]:
|
||
"""Spread an explicit advertisement label across its whole page.
|
||
|
||
Publishers print "ADVERTISEMENT" once to disclaim the page, not each block
|
||
on it. Without this the ad's own display type reads as a headline and the
|
||
page gets claimed by whichever feature precedes it.
|
||
"""
|
||
# A small ad label beside substantial prose does not establish a full-page ad.
|
||
if any(b.type == BLOCK_BODY and len(b.text) > 160 for b in blocks):
|
||
return blocks
|
||
if not any(b.type == BLOCK_AD and b.source not in ("vision_label", "llm_ad", "llm_layout") for b in blocks):
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type != BLOCK_FURNITURE:
|
||
b.type, b.confidence = BLOCK_AD, 0.8
|
||
return blocks
|
||
|
||
|
||
def mark_pullquotes(blocks: list[Block]) -> list[Block]:
|
||
"""Demote display text that merely repeats nearby body copy.
|
||
|
||
Narrating a pull quote reads the same sentence twice, a few seconds apart,
|
||
which sounds like a skipping record.
|
||
"""
|
||
body_text = _normalise(" ".join(
|
||
b.text for b in blocks if b.type == BLOCK_BODY
|
||
)).lower()
|
||
if not body_text:
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type not in (BLOCK_DECK, BLOCK_PULLQUOTE):
|
||
continue
|
||
probe = _normalise(b.text).lower().strip(' "“”«»')
|
||
if len(probe) > 25 and probe in body_text:
|
||
b.type, b.confidence = BLOCK_PULLQUOTE, 0.9
|
||
return blocks
|
||
|
||
|
||
def mark_sidebars(blocks: list[Block], page_width: float,
|
||
narrow_ratio: float = 0.7) -> list[Block]:
|
||
"""Reclassify body text sitting in an anomalously narrow column as a
|
||
sidebar, so it is excluded from an article's narration by default rather
|
||
than requiring a manual exclusion in review every time.
|
||
|
||
Geometry-only, like the rest of this module's classification: a sidebar
|
||
or pull-out box is reliably narrower than the page's regular column grid
|
||
(a boxed insert set off from the main flow), which needs no image
|
||
analysis to detect. This is a heuristic, not a certainty -- a genuinely
|
||
narrow-column page design (or a caption sitting in its own slim column)
|
||
can trigger a false positive, which is exactly why review still shows
|
||
every sidebar block and lets it be included with one click rather than
|
||
silently discarding it.
|
||
|
||
Full-width elements (a headline spanning the whole page) are excluded
|
||
from the column analysis here, the same way assign_reading_order treats
|
||
them specially: left in, one would bridge two genuinely separate columns
|
||
into a single merged span (its bbox overlaps both), hiding a real narrow
|
||
sidebar column entirely. This computes its own column partition rather
|
||
than trusting `.column` from assign_reading_order, since that partition
|
||
was built from the FULL block list and its indices would not line up
|
||
with this narrower one.
|
||
"""
|
||
candidates = [b for b in blocks if b.bbox[2] <= page_width * 0.6]
|
||
columns = detect_columns(candidates, page_width)
|
||
if len(columns) < 2:
|
||
return blocks
|
||
widths = [right - left for left, right in columns]
|
||
widest = max(widths)
|
||
narrow_spans = [columns[i] for i, w in enumerate(widths) if w < widest * narrow_ratio]
|
||
if not narrow_spans:
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type != BLOCK_BODY or b.bbox[2] > page_width * 0.6:
|
||
continue
|
||
centre = b.bbox[0] + b.bbox[2] / 2
|
||
if any(left <= centre <= right for left, right in narrow_spans):
|
||
b.type, b.confidence = BLOCK_SIDEBAR, 0.6
|
||
return blocks
|
||
|
||
|
||
def find_repeated_lines(pages_text: Sequence[Sequence[str]],
|
||
threshold: float = 0.25) -> set[str]:
|
||
"""Lines appearing near the same edge on many pages: page furniture."""
|
||
if not pages_text:
|
||
return set()
|
||
counts: dict[str, int] = {}
|
||
for lines in pages_text:
|
||
for line in {_normalise(t) for t in lines if _normalise(t)}:
|
||
counts[line] = counts.get(line, 0) + 1
|
||
minimum = max(3, int(len(pages_text) * threshold))
|
||
# Long lines that recur are more likely a repeated standfirst than a
|
||
# running head, so cap the length.
|
||
return {t for t, n in counts.items() if n >= minimum and len(t) <= 80}
|
||
|
||
|
||
def looks_like_real_title(text: str) -> bool:
|
||
"""Guards the headline-fallback article start against OCR noise.
|
||
|
||
A stray misread character or chart-axis fragment ("I", "@)X", ". |") can
|
||
classify as BLOCK_TITLE by size alone; without this it would spawn its
|
||
own article. Confirmed live on a scanned magazine: single-character OCR
|
||
artifacts at running-footer size otherwise start empty articles.
|
||
"""
|
||
letters = sum(1 for c in (text or "") if c.isalpha())
|
||
return letters >= 4
|
||
|
||
|
||
def _digit_stripped(text: str) -> str:
|
||
return re.sub(r"\d+", "", text or "")
|
||
|
||
|
||
def cluster_running_furniture(blocks: Sequence[Block], page_heights: dict[int, float],
|
||
similarity: float = 0.6, min_count: int = 3) -> None:
|
||
"""Reclassify near-edge blocks that recur across pages as furniture, even
|
||
when OCR reads the running header/footer differently each time.
|
||
|
||
find_repeated_lines() needs an EXACT match, which OCR defeats: a magazine
|
||
tag like "c't Wissen 2026" came back as "cE KI-WISSEN", "ce Wissen 2136",
|
||
"cA Wissen 2036" across different pages of the same scan -- confirmed
|
||
live, and the dominant cause of a 28-page scan producing 26 near-1-per-
|
||
page "articles" instead of grouping around real headlines. Only the page
|
||
number reliably varies page to page, so digit-stripped fuzzy similarity
|
||
catches what exact matching misses. Mutates blocks in place.
|
||
"""
|
||
candidates: list[tuple[Block, str]] = []
|
||
for b in blocks:
|
||
if b.type in (BLOCK_FURNITURE, BLOCK_AD):
|
||
continue
|
||
text = _normalise(b.text)
|
||
if not text or len(text) > 80:
|
||
continue
|
||
height = page_heights.get(b.page, 842.0)
|
||
near_top = b.bbox[1] < height * 0.07
|
||
near_bottom = b.bbox[1] + b.bbox[3] > height * 0.93
|
||
if not (near_top or near_bottom):
|
||
continue
|
||
key = _digit_stripped(text).lower().strip()
|
||
if len(key) < 3:
|
||
continue
|
||
candidates.append((b, key))
|
||
|
||
assigned = [False] * len(candidates)
|
||
for i, (_, key_i) in enumerate(candidates):
|
||
if assigned[i]:
|
||
continue
|
||
cluster = [i]
|
||
for j in range(i + 1, len(candidates)):
|
||
if assigned[j]:
|
||
continue
|
||
if difflib.SequenceMatcher(None, key_i, candidates[j][1]).ratio() >= similarity:
|
||
cluster.append(j)
|
||
if len(cluster) >= min_count:
|
||
for idx in cluster:
|
||
candidates[idx][0].type = BLOCK_FURNITURE
|
||
candidates[idx][0].confidence = 0.75
|
||
assigned[idx] = True
|
||
|
||
|
||
# ── Stage A: the table of contents ───────────────────────────────────────────
|
||
|
||
def score_contents_page(lines: Sequence[str]) -> float:
|
||
"""How strongly one page looks like a table of contents (0..1)."""
|
||
if not lines:
|
||
return 0.0
|
||
joined = " ".join(lines).lower()
|
||
marker_hits = sum(1 for m in _TOC_MARKERS if m in joined)
|
||
entry_hits = sum(1 for line in lines if _TOC_ENTRY.match(_normalise(line)))
|
||
# Either signal alone is weak: a feature can be titled "Features", and a
|
||
# dotted leader appears in ordinary tables too.
|
||
score = min(1.0, marker_hits * 0.25) * 0.4 + min(1.0, entry_hits / 6) * 0.6
|
||
return round(score, 3)
|
||
|
||
|
||
def parse_contents(lines: Sequence[str]) -> list[dict[str, Any]]:
|
||
"""Pull (title, page, section) entries out of a contents page."""
|
||
entries: list[dict[str, Any]] = []
|
||
section = ""
|
||
for raw in lines:
|
||
line = _normalise(raw)
|
||
if not line:
|
||
continue
|
||
low = line.lower().strip(": ")
|
||
if low in _TOC_MARKERS and low not in ("contents", "inhalt"):
|
||
section = line.strip(": ")
|
||
continue
|
||
m = _TOC_ENTRY.match(line)
|
||
if not m:
|
||
continue
|
||
title, page = m.group(1).strip(" .…·•-"), int(m.group(2))
|
||
if len(title) < 3 or page <= 0 or page > 2000:
|
||
continue
|
||
entries.append({"title": title, "page": page, "section": section})
|
||
return entries
|
||
|
||
|
||
# ── Stage C: grouping blocks into articles ───────────────────────────────────
|
||
|
||
def _title_key(text: str) -> str:
|
||
return re.sub(r"[^a-z0-9]+", "", _normalise(text).lower())
|
||
|
||
|
||
def group_articles(blocks: Sequence[Block],
|
||
contents: Sequence[dict[str, Any]],
|
||
page_count: int,
|
||
skip_pages: set[int] | None = None) -> list[Article]:
|
||
"""Group ordered blocks into articles.
|
||
|
||
The contents listing, when present, is the strongest available signal: it
|
||
states both the title and the starting page, so it anchors boundaries that
|
||
layout alone would guess at. Pages not claimed by any listing still get
|
||
articles from their own headlines, because a contents page never lists
|
||
everything (and some issues have no contents page at all).
|
||
"""
|
||
by_page: dict[int, list[Block]] = {}
|
||
for b in blocks:
|
||
by_page.setdefault(b.page, []).append(b)
|
||
for page_blocks in by_page.values():
|
||
page_blocks.sort(key=lambda b: b.reading_order)
|
||
|
||
# Where does each article start? A contents entry pins a start page; a
|
||
# title block on a page with no entry starts one there.
|
||
starts: dict[int, dict[str, Any]] = {}
|
||
for entry in contents:
|
||
page = entry["page"]
|
||
if 1 <= page <= page_count:
|
||
starts.setdefault(page, {
|
||
"title": entry["title"],
|
||
"section": entry.get("section", ""),
|
||
"confidence": 0.9,
|
||
})
|
||
|
||
skip_pages = skip_pages or set()
|
||
for page in sorted(by_page):
|
||
if page in starts or page in skip_pages:
|
||
continue
|
||
# A page carrying nothing but advertising and page furniture has no
|
||
# article on it -- the cover and the ad pages otherwise each become a
|
||
# spurious entry the reviewer has to reject by hand on every issue.
|
||
if not any(b.type in NARRATABLE and b.type != BLOCK_TITLE
|
||
for b in by_page[page]):
|
||
continue
|
||
titles = [b for b in by_page[page] if b.type == BLOCK_TITLE
|
||
and looks_like_real_title(b.text)]
|
||
if titles:
|
||
starts[page] = {
|
||
"title": _normalise(titles[0].text),
|
||
"section": "",
|
||
"confidence": 0.55,
|
||
}
|
||
|
||
if not starts:
|
||
return []
|
||
|
||
ordered_starts = sorted(starts)
|
||
articles: list[Article] = []
|
||
for i, start_page in enumerate(ordered_starts):
|
||
end_page = (ordered_starts[i + 1] - 1) if i + 1 < len(ordered_starts) else page_count
|
||
info = starts[start_page]
|
||
art = Article(
|
||
id=f"article-{i + 1:03d}",
|
||
title=info["title"],
|
||
section=info.get("section", ""),
|
||
confidence=info["confidence"],
|
||
)
|
||
for page in range(start_page, end_page + 1):
|
||
page_blocks = by_page.get(page, [])
|
||
if not page_blocks:
|
||
continue
|
||
# Only claim a page the article actually occupies. A full-page
|
||
# advert sitting between two features starts no article of its
|
||
# own, so it would otherwise be swallowed by the preceding one and
|
||
# highlighted as part of it in the PDF view.
|
||
#
|
||
# A sidebar/pull-out box (mark_sidebars) is excluded the same way
|
||
# advert text is: automatically kept out of narration by default,
|
||
# rather than requiring a manual exclusion in review every time.
|
||
# It still shows in review (blocksForPages lists every block on
|
||
# the article's pages) and can be ticked back in with one click.
|
||
contributed = False
|
||
for b in page_blocks:
|
||
if b.type in (BLOCK_FURNITURE, BLOCK_AD, BLOCK_SIDEBAR, BLOCK_PULLQUOTE, *VISUAL_TYPES):
|
||
continue
|
||
contributed = True
|
||
art.blocks.append(b.id)
|
||
if b.type == BLOCK_BYLINE and not art.author:
|
||
_, name = looks_like_byline(b.text)
|
||
art.author = name
|
||
to_page, _ = find_continuation(b.text)
|
||
if to_page and to_page not in art.continued_to:
|
||
art.continued_to.append(to_page)
|
||
if contributed:
|
||
art.pages.append(page)
|
||
|
||
# An article the pipeline could not fill with body copy is almost
|
||
# certainly a false start (an advert headline, a section divider).
|
||
has_body = any(
|
||
b.type == BLOCK_BODY for b in blocks if b.id in set(art.blocks)
|
||
)
|
||
if not has_body:
|
||
art.confidence = round(art.confidence * 0.4, 3)
|
||
articles.append(art)
|
||
|
||
return _link_continuations(articles, page_count)
|
||
|
||
|
||
def _link_continuations(articles: list[Article], page_count: int) -> list[Article]:
|
||
"""Attach jump pages ('continued on page 88') to their originating article.
|
||
|
||
Without this, the tail of a feature parked at the back of the issue is
|
||
orphaned and gets narrated as a separate stub article that begins
|
||
mid-sentence.
|
||
"""
|
||
by_start = {a.pages[0]: a for a in articles if a.pages}
|
||
for art in articles:
|
||
for target in art.continued_to:
|
||
if not (1 <= target <= page_count):
|
||
continue
|
||
tail = by_start.get(target)
|
||
if tail is None or tail is art or tail.status == "merged":
|
||
continue
|
||
# Only absorb a genuine continuation: a real article listed in the
|
||
# contents keeps its own identity.
|
||
if tail.confidence >= 0.9:
|
||
continue
|
||
art.pages.extend(p for p in tail.pages if p not in art.pages)
|
||
art.blocks.extend(tail.blocks)
|
||
art.images.extend(tail.images)
|
||
tail.status = "merged"
|
||
return [a for a in articles if a.status != "merged"]
|
||
|
||
|
||
def estimate_listen_seconds(blocks: Sequence[Block], article: Article,
|
||
words_per_minute: float = 155.0) -> int:
|
||
"""Rough narration length, for the 'estimated listening time' column."""
|
||
ids = set(article.blocks)
|
||
words = sum(
|
||
len(b.text.split()) for b in blocks
|
||
if b.id in ids and b.type in NARRATABLE
|
||
)
|
||
return int(round(words / words_per_minute * 60)) if words else 0
|
||
|
||
|
||
# ── PDF and OCR I/O ──────────────────────────────────────────────────────────
|
||
|
||
def _require_fitz():
|
||
"""Import PyMuPDF under its current name, falling back to the old alias.
|
||
|
||
1.24 renamed the module to ``pymupdf``; ``fitz`` still resolves but emits a
|
||
deprecation warning and is slated for removal.
|
||
"""
|
||
try:
|
||
import pymupdf
|
||
return pymupdf
|
||
except ImportError:
|
||
pass
|
||
try:
|
||
import fitz
|
||
return fitz
|
||
except ImportError as exc: # pragma: no cover - environment guard
|
||
raise RuntimeError(
|
||
"PyMuPDF is required for magazine ingestion; install pymupdf (pip install pymupdf) "
|
||
"or rebuild your Docker container."
|
||
) from exc
|
||
|
||
|
||
# Labels a vision OCR engine may return for a region, mapped onto the same
|
||
# taxonomy tesseract/native blocks use. Anything else ("text", or a label the
|
||
# engine invents) is treated as ambiguous body copy and still goes through
|
||
# classify_blocks() for finer distinctions (byline, deck, ...).
|
||
_VISION_LABEL_TYPES = {
|
||
# Headlines / Titles / Headers
|
||
"title": BLOCK_TITLE,
|
||
"headline": BLOCK_TITLE,
|
||
"article_headline": BLOCK_TITLE,
|
||
"main_title": BLOCK_TITLE,
|
||
"heading": BLOCK_HEADING,
|
||
"header": BLOCK_HEADING,
|
||
"section_heading": BLOCK_HEADING,
|
||
"subheading": BLOCK_HEADING,
|
||
"subtitle": BLOCK_HEADING,
|
||
"deck": BLOCK_HEADING,
|
||
|
||
# Paragraphs / Text / Body
|
||
"paragraph": BLOCK_BODY,
|
||
"body": BLOCK_BODY,
|
||
"content": BLOCK_BODY,
|
||
"list": BLOCK_BODY,
|
||
"list_item": BLOCK_BODY,
|
||
"author": BLOCK_BYLINE,
|
||
"byline": BLOCK_BYLINE,
|
||
|
||
# Sidenotes / Sidebars / Quotes / Footnotes
|
||
"sidenote": BLOCK_SIDEBAR,
|
||
"sidebar": BLOCK_SIDEBAR,
|
||
"aside_text": BLOCK_SIDEBAR,
|
||
"aside": BLOCK_SIDEBAR,
|
||
"footnote": BLOCK_SIDEBAR,
|
||
"footnotes": BLOCK_SIDEBAR,
|
||
"note": BLOCK_SIDEBAR,
|
||
"commentary": BLOCK_SIDEBAR,
|
||
"quote": BLOCK_PULLQUOTE,
|
||
"pullquote": BLOCK_PULLQUOTE,
|
||
|
||
# Page numbers / Furniture / Folios / Sources
|
||
"page_number": BLOCK_FURNITURE,
|
||
"pagenumber": BLOCK_FURNITURE,
|
||
"page": BLOCK_FURNITURE,
|
||
"header_text": BLOCK_FURNITURE,
|
||
"footer": BLOCK_FURNITURE,
|
||
"running_header": BLOCK_FURNITURE,
|
||
"running_footer": BLOCK_FURNITURE,
|
||
"folio": BLOCK_FURNITURE,
|
||
"head": BLOCK_FURNITURE,
|
||
"source": BLOCK_FURNITURE,
|
||
"sources": BLOCK_FURNITURE,
|
||
"citation": BLOCK_FURNITURE,
|
||
"citations": BLOCK_FURNITURE,
|
||
|
||
# Images / Figures / Photos
|
||
"image": BLOCK_FIGURE,
|
||
"figure": BLOCK_FIGURE,
|
||
"picture": BLOCK_FIGURE,
|
||
"photo": BLOCK_FIGURE,
|
||
"illustration": BLOCK_FIGURE,
|
||
"non_text": BLOCK_FIGURE,
|
||
"nontext": BLOCK_FIGURE,
|
||
"drawing": BLOCK_FIGURE,
|
||
|
||
# Graphics / Charts / Diagrams / Maps / Treemaps
|
||
"chart": BLOCK_CHART,
|
||
"graphic": BLOCK_CHART,
|
||
"infographic": BLOCK_CHART,
|
||
"statistics": BLOCK_CHART,
|
||
"diagram": BLOCK_CHART,
|
||
"map": BLOCK_CHART,
|
||
"treemap": BLOCK_CHART,
|
||
"plot": BLOCK_CHART,
|
||
"flowchart": BLOCK_CHART,
|
||
|
||
# Tables
|
||
"table": BLOCK_TABLE,
|
||
"grid": BLOCK_TABLE,
|
||
"matrix": BLOCK_TABLE,
|
||
|
||
# Code / Formulas / Math
|
||
"code": BLOCK_CODE,
|
||
"code_block": BLOCK_CODE,
|
||
"formula": BLOCK_CODE,
|
||
"equation": BLOCK_CODE,
|
||
"math": BLOCK_CODE,
|
||
|
||
# Captions
|
||
"caption": BLOCK_CAPTION,
|
||
"image_caption": BLOCK_CAPTION,
|
||
"figure_caption": BLOCK_CAPTION,
|
||
"table_caption": BLOCK_CAPTION,
|
||
|
||
# Ads
|
||
"advertisement": BLOCK_AD,
|
||
"advert": BLOCK_AD,
|
||
"ad": BLOCK_AD,
|
||
}
|
||
|
||
# One label + its four-number bbox, anchoring a region; content runs until
|
||
# the next such marker. Tolerates DeepSeek-OCR / Unlimited-OCR / PaddleOCR shapes
|
||
# like "text [241, 90, 588, 249]", "<|ref|>title<|/ref|><|det|>[[x0, y0, x1, y1]]<|/det|>", etc.
|
||
_VISION_REGION = re.compile(
|
||
r"(?m)"
|
||
r"(?:<\|ref\|>\s*(?P<ref>[^<]+?)\s*<\|/ref\|>\s*)?"
|
||
r"(?:<\|det\|>\s*)?"
|
||
r"(?P<label>[a-zA-Z0-9_\- ]+?)\s*"
|
||
r"\[\[?\s*(?P<x0>-?\d+(?:\.\d+)?)\s*,\s*(?P<y0>-?\d+(?:\.\d+)?)\s*,\s*(?P<x1>-?\d+(?:\.\d+)?)\s*,\s*(?P<y1>-?\d+(?:\.\d+)?)\s*\]\]?"
|
||
r"(?:\s*<\|/det\|>)?"
|
||
r"\s*:?"
|
||
)
|
||
|
||
_UNLIMITED_OCR_REGION = re.compile(
|
||
r"(?m)"
|
||
r"(?:<\|ref\|>\s*(?P<ref>[^<]+?)\s*<\|/ref\|>\s*)?"
|
||
r"<\|det\|>\s*"
|
||
r"(?P<label>[a-zA-Z0-9_\- ]+?)\s*"
|
||
r"(?:\[\[?\s*(?P<x0>-?\d+(?:\.\d+)?)\s*,\s*(?P<y0>-?\d+(?:\.\d+)?)\s*,\s*(?P<x1>-?\d+(?:\.\d+)?)\s*,\s*(?P<y1>-?\d+(?:\.\d+)?)\s*\]\]?)?"
|
||
r"\s*<\|/det\|>"
|
||
)
|
||
|
||
|
||
def ocr_quality_score(text: str) -> float:
|
||
"""Rough plausibility that OCR/extracted text is real prose, 0..1."""
|
||
text = (text or "").strip()
|
||
if not text:
|
||
return 0.0
|
||
words = text.split()
|
||
if not words:
|
||
return 0.0
|
||
|
||
def _stripped(w: str) -> str:
|
||
return w.strip(".,;:!?()[]{}\"'|-–—")
|
||
|
||
garbage = sum(1 for w in words if not _stripped(w))
|
||
letters = sum(1 for c in text if c.isalpha())
|
||
total = len(text.replace(" ", "")) or 1
|
||
score = (letters / total) * (1.0 - garbage / len(words))
|
||
return round(max(0.0, min(1.0, score)), 3)
|
||
|
||
|
||
def estimate_font_size_from_area(width: float, height: float, text_length: int,
|
||
char_width_ratio: float = 0.5,
|
||
line_height_ratio: float = 1.2) -> float:
|
||
"""Estimate a representative font size for a paragraph-level box."""
|
||
if width <= 0 or height <= 0 or text_length <= 0:
|
||
return 0.0
|
||
denom = text_length * char_width_ratio * line_height_ratio
|
||
return (width * height / denom) ** 0.5 if denom > 0 else 0.0
|
||
|
||
|
||
def parse_vision_ocr_regions(raw: str) -> list[dict[str, Any]]:
|
||
"""Parse a vision OCR engine's labeled-region transcription."""
|
||
matches = list(_VISION_REGION.finditer(raw or ""))
|
||
regions = []
|
||
for i, m in enumerate(matches):
|
||
gd = m.groupdict()
|
||
raw_label = (gd.get("ref") or gd.get("label") or "text").strip().strip("[]")
|
||
label = raw_label.lower().replace("-", "_").replace(" ", "_")
|
||
try:
|
||
x0, y0, x1, y1 = (float(gd[k]) for k in ("x0", "y0", "x1", "y1"))
|
||
except (ValueError, TypeError):
|
||
continue
|
||
start = m.end()
|
||
end = matches[i + 1].start() if i + 1 < len(matches) else len(raw or "")
|
||
text = (raw or "")[start:end].strip()
|
||
if (not text and _VISION_LABEL_TYPES.get(label) not in VISUAL_TYPES) or x1 <= x0 or y1 <= y0:
|
||
continue
|
||
regions.append({
|
||
"label": label,
|
||
"bbox": [x0, y0, x1 - x0, y1 - y0],
|
||
"text": text,
|
||
})
|
||
return regions
|
||
|
||
|
||
def parse_unlimited_ocr_regions(raw: str) -> list[dict[str, Any]]:
|
||
"""Parse Unlimited-OCR's native ``<|det|>`` document-layout markers."""
|
||
matches = list(_UNLIMITED_OCR_REGION.finditer(raw or ""))
|
||
regions = []
|
||
for i, match in enumerate(matches):
|
||
gd = match.groupdict()
|
||
raw_label = (gd.get("ref") or gd.get("label") or "").strip().strip("[]")
|
||
label = raw_label.lower().replace("-", "_").replace(" ", "_")
|
||
try:
|
||
x0, y0, x1, y1 = (float(gd[k]) for k in ("x0", "y0", "x1", "y1"))
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if x1 <= x0 or y1 <= y0:
|
||
continue
|
||
start = match.end()
|
||
end = matches[i + 1].start() if i + 1 < len(matches) else len(raw or "")
|
||
text = (raw or "")[start:end].strip()
|
||
if label in {"non_text", "nontext", "photo", "drawing", "picture"}:
|
||
label = "image"
|
||
elif label in {"map", "graphic", "infographic", "diagram", "treemap", "statistics"}:
|
||
label = "chart"
|
||
if not text and _VISION_LABEL_TYPES.get(label) not in VISUAL_TYPES:
|
||
continue
|
||
regions.append({
|
||
"label": label,
|
||
"bbox": [x0, y0, x1 - x0, y1 - y0],
|
||
"text": text,
|
||
"native_normalized_bbox": True,
|
||
})
|
||
return regions
|
||
|
||
|
||
def _whole_page_region(content: str, png_path: Path) -> list[dict[str, Any]] | None:
|
||
"""Fallback for an engine that answers with plain text and no coordinate
|
||
markers at all (confirmed live: KDL Frontier Parser nano -- fast, but no
|
||
"label [x0,y0,x1,y1]" structure the way ocr-baidu-unlimited-ocr gives).
|
||
|
||
Treated as one ambiguous "text" region spanning the whole page: real
|
||
text with no layout information beats no text at all, but it does mean
|
||
column/reading-order structure is lost for that page -- classify_blocks
|
||
still runs on it (source="vision_ocr", not "vision_label"), same as any
|
||
other ambiguous region.
|
||
"""
|
||
text = (content or "").strip()
|
||
# A vision model can occasionally echo a formatting instruction instead
|
||
# of reading the page. The observed failure is a long sequence of literal
|
||
# ``\\n `equation` `` tokens (not real line breaks), often prefixed with
|
||
# "Do not use special symbols". It scores highly under the generic
|
||
# letter-ratio heuristic because "equation" is a normal word, so reject
|
||
# this distinct response shape before turning it into a page-sized text
|
||
# block. Returning None lets vision_ocr_page_chain try the configured
|
||
# fallback engine, then preserves the existing OCR if both fail.
|
||
escaped_newlines = len(re.findall(r"\\n", text))
|
||
equation_tokens = len(re.findall(r"`?\s*\\?equation\b", text, re.IGNORECASE))
|
||
prompt_echo = "do not use special symbols" in text.lower()
|
||
# Another observed loop starts with genuine page text and then repeats a
|
||
# fabricated LaTex fragment (``\\therefore m = \\frac{3}{11}``) until the
|
||
# response limit. It has no instruction echo or `equation` token, so it
|
||
# needs its own narrow signature. Requiring both uncommon controls many
|
||
# times avoids rejecting ordinary prose that merely mentions one formula.
|
||
therefore_tokens = len(re.findall(r"\\therefore\b", text, re.IGNORECASE))
|
||
fraction_tokens = len(re.findall(r"\\frac\s*\{", text, re.IGNORECASE))
|
||
latex_loop = therefore_tokens >= 4 and fraction_tokens >= 4
|
||
if (escaped_newlines >= 4 and equation_tokens >= 4) or (
|
||
prompt_echo and escaped_newlines >= 2 and equation_tokens >= 2
|
||
) or latex_loop:
|
||
logging.getLogger(__name__).warning("Vision OCR returned a repeated formatting artifact; discarding response")
|
||
return None
|
||
# The two signatures above are narrow matches for specific observed
|
||
# failures; this is the general case -- any short phrase repeated many
|
||
# times back-to-back (see vision_ocr_page's docstring on the same check).
|
||
if _looks_like_repetition_loop(text):
|
||
logging.getLogger(__name__).warning("Vision OCR returned a repeated phrase loop; discarding response")
|
||
return None
|
||
if len(text) < 20 or ocr_quality_score(text) < _VISION_QUALITY_THRESHOLD:
|
||
return None
|
||
try:
|
||
from PIL import Image
|
||
with Image.open(png_path) as im:
|
||
width, height = im.size
|
||
except Exception:
|
||
return None
|
||
return [{"label": "text", "bbox": [0.0, 0.0, float(width), float(height)], "text": text}]
|
||
|
||
|
||
# A short phrase repeating many times back-to-back, unlike the two narrow
|
||
# formatting-echo signatures above, is not specific to one prompt or one
|
||
# engine -- it is the generic decoding-loop failure any local vision model
|
||
# can fall into on a busy chart or graphic. Confirmed live: a Qwen VL engine
|
||
# transcribing a population chart produced "das Produkt, das Produkt, das
|
||
# Produkt, ..." several hundred times over, filling the whole response
|
||
# budget. Genuine text essentially never repeats an exact phrase this many
|
||
# times consecutively, so this is a safe, general trigger rather than
|
||
# another one-off signature -- but a leader-dot table of contents line
|
||
# ("......................") legitimately repeats a short unit too, which is
|
||
# why the repeated unit must contain a real word, not just punctuation.
|
||
_REPETITION_RUN = re.compile(r"(.{2,80}?)\1{7,}", re.DOTALL)
|
||
_REPETITION_WORD = re.compile(r"[A-Za-zÀ-ÖØ-öø-ÿ]{2,}")
|
||
|
||
|
||
def _looks_like_repetition_loop(text: str) -> bool:
|
||
"""True when `text` degenerates into the same short phrase repeated
|
||
immediately back-to-back many times -- see vision_ocr_page."""
|
||
return any(_REPETITION_WORD.search(m.group(1))
|
||
for m in _REPETITION_RUN.finditer(text or ""))
|
||
|
||
|
||
def vision_ocr_page(png_path: Path, base_url: str, model: str,
|
||
timeout: float = 150.0) -> list[dict[str, Any]] | None:
|
||
"""OCR one rendered page with a connected vision-capable OCR engine.
|
||
|
||
Returns None on any failure (unreachable engine, timeout, unusable
|
||
output) so the caller can fall back to the next engine or tesseract --
|
||
this must never be the reason ingestion fails outright. Confirmed live
|
||
against ocr-baidu-unlimited-ocr: markedly more accurate German text than
|
||
tesseract on the same scanned page, and it labels running headers/
|
||
footers and page numbers directly, which is the dominant source of the
|
||
over-fragmentation cluster_running_furniture() otherwise has to recover
|
||
after the fact from noisy OCR variants of the same repeating line.
|
||
"""
|
||
try:
|
||
image_b64 = base64.b64encode(png_path.read_bytes()).decode("ascii")
|
||
except OSError:
|
||
return None
|
||
|
||
# Settings also accept a bare server address; OpenAI-compatible servers
|
||
# expose their API under /v1. Preserve explicit proxy/API prefixes.
|
||
base_url = base_url.strip().rstrip("/")
|
||
if not urlsplit(base_url).path:
|
||
base_url += "/v1"
|
||
|
||
unlimited_ocr = "unlimited-ocr" in model.lower()
|
||
prompt = "<|grounding|>Convert the document to markdown." if unlimited_ocr else (
|
||
"Recognise the page layout and transcribe visible text verbatim. "
|
||
"Return one region per headline, title, paragraph, sidenote, sidebar, "
|
||
"image, figure, chart, graphic, map, diagram, table, code box, "
|
||
"caption, header, footer, or page number. Use the format "
|
||
"label [x0, y0, x1, y1]transcription with pixel coordinates in the "
|
||
"supplied image, one region starting on each new line. Labels: "
|
||
"title, heading, paragraph, sidenote, sidebar, image, figure, chart, "
|
||
"graphic, map, table, code, caption, header, footer, page_number. "
|
||
"Enclose a whole chart, diagram, map, graphic, or table in one box, "
|
||
"not separate axis labels or cells. Preserve table rows and columns "
|
||
"as Markdown and code indentation and line breaks. Images, figures, and "
|
||
"charts may have empty transcription; do not invent descriptions. "
|
||
"Do not merge neighbouring articles or include commentary."
|
||
)
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": [
|
||
{"type": "text", "text": prompt},
|
||
{"type": "image_url", "image_url": {
|
||
"url": f"data:image/png;base64,{image_b64}",
|
||
}},
|
||
]}],
|
||
"max_tokens": 4000,
|
||
"temperature": 0.0,
|
||
}
|
||
# Unlimited-OCR's official vLLM recipe requires its per-request
|
||
# NGramPerReqLogitsProcessor settings. The container already exposes the
|
||
# processor, but without these fields it does nothing and the model can
|
||
# run into a repeated chart/formula loop. Keep the vendor-specific fields
|
||
# off generic OpenAI-compatible vision engines.
|
||
if unlimited_ocr:
|
||
payload.update({
|
||
"skip_special_tokens": False,
|
||
"ngram_size": 35,
|
||
"window_size": 128,
|
||
"images_config": {"image_mode": "gundam"},
|
||
})
|
||
try:
|
||
resp = request_with_retries(
|
||
"POST", f"{base_url.rstrip('/')}/chat/completions",
|
||
json=payload, timeout=timeout, attempts=2,
|
||
)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
except requests.exceptions.RequestException as exc:
|
||
status = getattr(getattr(exc, "response", None), "status_code", None)
|
||
logging.getLogger(__name__).warning("Vision OCR request failed (%s%s)",
|
||
type(exc).__name__, f", HTTP {status}" if status else "")
|
||
return None
|
||
except (ValueError, KeyError, IndexError, TypeError):
|
||
logging.getLogger(__name__).warning("Vision OCR returned an invalid completion response")
|
||
return None
|
||
if not isinstance(content, str):
|
||
return None
|
||
if _looks_like_repetition_loop(content):
|
||
# Worse than no response at all: a repeated phrase baked into a
|
||
# block's text would be narrated verbatim. Treating it as a failed
|
||
# call gives the existing fallback chain (vision_ocr_page_chain) a
|
||
# real shot instead.
|
||
logging.getLogger(__name__).warning("Vision OCR returned a repeated phrase loop; discarding response")
|
||
return None
|
||
|
||
regions = parse_unlimited_ocr_regions(content) if unlimited_ocr else parse_vision_ocr_regions(content)
|
||
if regions and unlimited_ocr:
|
||
# Native markers use the model input's 0–999-ish coordinate space;
|
||
# blocks downstream expect coordinates of the rendered PNG.
|
||
try:
|
||
from PIL import Image
|
||
with Image.open(png_path) as im:
|
||
width, height = im.size
|
||
for region in regions:
|
||
if region.pop("native_normalized_bbox", False):
|
||
x, y, w, h = region["bbox"]
|
||
region["bbox"] = [x / 1000 * width, y / 1000 * height,
|
||
w / 1000 * width, h / 1000 * height]
|
||
except Exception:
|
||
return None
|
||
return regions or _whole_page_region(content, png_path)
|
||
|
||
|
||
def vision_ocr_page_chain(png_path: Path,
|
||
engines: Sequence[tuple[str, str]],
|
||
timeout: float = 150.0) -> list[dict[str, Any]] | None:
|
||
"""Try each configured vision OCR engine in order, first success wins.
|
||
|
||
A primary engine (labeled regions, best quality) with a faster or
|
||
differently-shaped secondary as a real fallback -- not just tesseract --
|
||
when the primary is unreachable or times out. Engines are (url, model)
|
||
pairs; empty/blank entries are skipped.
|
||
"""
|
||
for url, model in engines:
|
||
if not (url and model):
|
||
continue
|
||
regions = vision_ocr_page(png_path, url, model, timeout=timeout)
|
||
if regions:
|
||
return regions
|
||
return None
|
||
|
||
|
||
# One line per visual type -- a chart needs "what trend", a table needs
|
||
# "what's listed", not the generic "what's shown" that suits a photo. Every
|
||
# variant still asks for one or two plain sentences: this is read aloud
|
||
# between the surrounding paragraphs, not a caption someone reads at their
|
||
# own pace, so it has to stay short regardless of how much detail the image
|
||
# could support.
|
||
_DESCRIBE_PROMPTS = {
|
||
BLOCK_CHART: (
|
||
"This image is a chart or graph from a magazine article. In one or "
|
||
"two plain sentences, for a listener who cannot see it, describe "
|
||
"what kind of chart it is, what it measures, and the general trend "
|
||
"or most notable values. Do not invent a number you cannot actually "
|
||
"read in the image. No markdown, no preamble -- the description "
|
||
"text only."
|
||
),
|
||
BLOCK_TABLE: (
|
||
"This image is a table from a magazine article. In one or two "
|
||
"plain sentences, for a listener who cannot see it, summarise what "
|
||
"it lists -- do not read out individual cells. No markdown, no "
|
||
"preamble -- the description text only."
|
||
),
|
||
BLOCK_CODE: (
|
||
"This image is a code listing or code box from a magazine article. "
|
||
"In one plain sentence, for a listener who cannot see it, describe "
|
||
"what it appears to show (the language and general purpose) -- do "
|
||
"not read the code line by line. No markdown, no preamble -- the "
|
||
"description text only."
|
||
),
|
||
BLOCK_FIGURE: (
|
||
"This image is a photo or illustration from a magazine article. In "
|
||
"one or two plain sentences, for a listener who cannot see it, "
|
||
"describe what it shows, including any text visible in the image "
|
||
"itself. No markdown, no preamble -- the description text only."
|
||
),
|
||
}
|
||
|
||
|
||
def describe_visual_region(image_path: Path, region_type: str, base_url: str,
|
||
model: str, timeout: float = 90.0) -> str | None:
|
||
"""Describe one already-cropped figure/chart/table/code image in plain
|
||
language, for optional inclusion in an article's spoken narration.
|
||
|
||
A deliberately different task from vision_ocr_page(): that prompt
|
||
explicitly forbids inventing a description ("Images, figures, and
|
||
charts may have empty transcription; do not invent descriptions") --
|
||
correct for transcription, wrong here, where a description is exactly
|
||
what is wanted. Reuses the same engine (already vision-capable and
|
||
already configured) with its own prompt instead, and the same
|
||
reliability guards -- a repeated-phrase decoding loop is exactly as
|
||
possible describing a busy chart as transcribing one.
|
||
|
||
Returns None on any failure (unreachable engine, timeout, empty or
|
||
degenerate response) so the caller can try a fallback engine or leave
|
||
the figure without a description -- never partway through one.
|
||
"""
|
||
try:
|
||
image_b64 = base64.b64encode(image_path.read_bytes()).decode("ascii")
|
||
except OSError:
|
||
return None
|
||
|
||
base_url = base_url.strip().rstrip("/")
|
||
if not urlsplit(base_url).path:
|
||
base_url += "/v1"
|
||
|
||
prompt = _DESCRIBE_PROMPTS.get(region_type, _DESCRIBE_PROMPTS[BLOCK_FIGURE])
|
||
unlimited_ocr = "unlimited-ocr" in model.lower()
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": [
|
||
{"type": "text", "text": prompt},
|
||
{"type": "image_url", "image_url": {
|
||
"url": f"data:image/png;base64,{image_b64}",
|
||
}},
|
||
]}],
|
||
"max_tokens": 300,
|
||
"temperature": 0.2,
|
||
}
|
||
if unlimited_ocr:
|
||
# Same anti-repetition settings as vision_ocr_page -- same model,
|
||
# same decoding-loop risk, regardless of the prompt asked of it.
|
||
payload.update({
|
||
"skip_special_tokens": False,
|
||
"ngram_size": 35,
|
||
"window_size": 128,
|
||
"images_config": {"image_mode": "gundam"},
|
||
})
|
||
try:
|
||
resp = request_with_retries(
|
||
"POST", f"{base_url.rstrip('/')}/chat/completions",
|
||
json=payload, timeout=timeout, attempts=2,
|
||
)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
except requests.exceptions.RequestException as exc:
|
||
status = getattr(getattr(exc, "response", None), "status_code", None)
|
||
logging.getLogger(__name__).warning("Figure description request failed (%s%s)",
|
||
type(exc).__name__, f", HTTP {status}" if status else "")
|
||
return None
|
||
except (ValueError, KeyError, IndexError, TypeError):
|
||
logging.getLogger(__name__).warning("Figure description returned an invalid completion response")
|
||
return None
|
||
if not isinstance(content, str):
|
||
return None
|
||
text = content.strip().strip("`").strip()
|
||
if not text or len(text) < 8:
|
||
return None
|
||
if _looks_like_repetition_loop(text):
|
||
logging.getLogger(__name__).warning("Figure description returned a repeated phrase loop; discarding response")
|
||
return None
|
||
# A long response is either the model ignoring "one or two sentences" or
|
||
# a loop this pass's own repetition check didn't happen to catch --
|
||
# either way, narration wants a short aside, not a paragraph. Cut at a
|
||
# sentence boundary rather than mid-word.
|
||
limit = 600
|
||
if len(text) > limit:
|
||
cut = text.rfind(". ", 0, limit)
|
||
text = text[:cut + 1] if cut > limit // 2 else text[:limit].rstrip() + "…"
|
||
return text
|
||
|
||
|
||
def describe_visual_region_chain(image_path: Path, region_type: str,
|
||
engines: Sequence[tuple[str, str]],
|
||
timeout: float = 90.0) -> str | None:
|
||
"""Try each configured vision engine in order, first usable description
|
||
wins -- same shape as vision_ocr_page_chain."""
|
||
for url, model in engines:
|
||
if not (url and model):
|
||
continue
|
||
text = describe_visual_region(image_path, region_type, url, model, timeout=timeout)
|
||
if text:
|
||
return text
|
||
return None
|
||
|
||
|
||
def build_vision_blocks(regions: list[dict[str, Any]], page_no: int,
|
||
scale_x: float, id_prefix: str) -> list[Block]:
|
||
"""Turn one page's vision OCR regions into Blocks.
|
||
|
||
A label the engine gives directly (footer/page_number/header/title/
|
||
caption) is trusted outright -- source="vision_label" makes
|
||
classify_blocks() skip it rather than second-guess it from font size,
|
||
which vision OCR does not reliably provide (see estimate_font_size_from_
|
||
area's docstring: it is a paragraph-area estimate, not a measurement).
|
||
An unrecognised or "text" label is left ambiguous (BLOCK_BODY,
|
||
source="vision_ocr") and still goes through the normal classification
|
||
pass for finer distinctions such as byline or deck.
|
||
"""
|
||
out = []
|
||
for i, region in enumerate(regions):
|
||
x, y, w, h = (v * scale_x for v in region["bbox"])
|
||
label = region["label"]
|
||
text = region["text"]
|
||
mapped = _VISION_LABEL_TYPES.get(label)
|
||
if mapped is not None:
|
||
block = Block(
|
||
id=f"{id_prefix}{page_no:04d}-v{i:03d}",
|
||
page=page_no, bbox=[x, y, w, h], type=mapped, text=text,
|
||
confidence=0.85, source="vision_label",
|
||
)
|
||
else:
|
||
block = Block(
|
||
id=f"{id_prefix}{page_no:04d}-v{i:03d}",
|
||
page=page_no, bbox=[x, y, w, h], type=BLOCK_BODY, text=text,
|
||
confidence=0.8,
|
||
font_size=estimate_font_size_from_area(w, h, len(text)),
|
||
source="vision_ocr",
|
||
)
|
||
block.region_label = label
|
||
out.append(block)
|
||
return out
|
||
|
||
|
||
# ── Figure bounds from the page render ───────────────────────────────────────
|
||
#
|
||
# A vision OCR engine grounds a figure far less precisely than it transcribes
|
||
# one: on c't KI Wissen 2026 the label and the read-out contents were right
|
||
# while the four numbers landed on the neighbouring text column at two thirds
|
||
# of the real size (page 23), or covered only the top third of a full-width
|
||
# illustration (page 20). The rendered page itself carries the answer -- a
|
||
# photo, a chart or a UI screenshot is a densely inked rectangle fenced off
|
||
# from the type area by white gutters -- so the model's box is used only to
|
||
# say WHICH graphic is meant and the render says where that graphic ends.
|
||
#
|
||
# Two passes over the render, not one. A strict "solid ink" pass finds and
|
||
# repositions onto a densely inked anchor -- a photo, a filled UI panel, a
|
||
# card in a tiled grid -- exactly the way a photo behaves. But a bar chart's
|
||
# thin bars, hairline gridlines and axis labels rarely reach that density
|
||
# (confirmed on page 18 of the same issue: most bar rows measured 0.25-0.5
|
||
# coverage against the 0.55 solid threshold, so the strict pass anchored
|
||
# only on the one fully-solid element -- the vertical zero-line -- and
|
||
# returned a box cropped to a third of the real chart). A second, looser
|
||
# pass grows that anchor -- or, lacking one, the seed itself -- through any
|
||
# real mark at all, which is what recovers the rest of a sparse chart.
|
||
#
|
||
# Both passes are fenced by the same `blocked` cells: real text, and (new)
|
||
# every *other* figure's own seed area, so the looser pass cannot bridge a
|
||
# real gutter between two adjacent figures the way a raw blank-run
|
||
# allowance could -- confirmed against two head-and-shoulders portraits on
|
||
# page 31 sitting only ~12pt (1.8 grid cells) apart, well inside what the
|
||
# blank-run tolerance alone would cross.
|
||
#
|
||
# Deliberately conservative: every rule below can decline, and declining
|
||
# leaves the model's box untouched. A wrong box the user can see and redraw
|
||
# is a nuisance; a box confidently snapped onto the wrong thing is worse.
|
||
|
||
_SNAP_CELL_PX = 10 # grid resolution -- ~2 mm at the 110 dpi page render
|
||
_SNAP_SOLID = 0.55 # inked fraction marking a cell as graphic, not type
|
||
_SNAP_GAP_CELLS = 3 # white gutter a tiled graphic is allowed to span
|
||
_SNAP_MIN_FILL = 0.45 # an accepted box is this solidly inked ...
|
||
_SNAP_MIN_FRAME = 0.9 # ... or framed on this share of its rows AND columns
|
||
_SNAP_MIN_SEED_AREA = 0.30 # ... and never shrinks below this much of the seed
|
||
_SNAP_MIN_OVERLAP = 0.005 # a graphic must touch the seed by at least this much --
|
||
# low on purpose: a seed can be offset far enough
|
||
# that it barely grazes the real graphic's edge
|
||
# (see page 70 of c't KI Wissen 2026, where the
|
||
# 0.02 first drafted here missed a 3-of-220-cell
|
||
# graze by a hair and needed a second snap pass
|
||
# before it could reposition at all)
|
||
_SNAP_ANY_INK = 0.15 # the loose pass's "there is a real mark here" floor --
|
||
# above the ~0.02-0.11 scan-noise floor measured in
|
||
# genuinely blank chart gaps, below the ~0.25+ every
|
||
# real bar/line/label measured at
|
||
_SNAP_MAX_BLANK_RUN = 2 # cells of true blank the loose pass may cross --
|
||
# covers a chart's own internal row gaps (measured
|
||
# at 1, occasionally 2 cells) without relying on
|
||
# this alone to keep two adjacent figures apart
|
||
_SNAP_BLOCKED_WALL = 0.7 # a row/column blocked across this much of a
|
||
# candidate box is real text the box must not
|
||
# cross, even though merging separate ink
|
||
# components into one rectangle (see _snap_seed)
|
||
# can otherwise enclose it geometrically
|
||
_SNAP_JOINT_ROUNDS = 6 # rounds of re-seeding every visual on a page from
|
||
# each other's last result (see snap_visual_bboxes)
|
||
# -- enough to settle the common case and to reach
|
||
# a deterministic stopping point for the rare pair
|
||
# that does not truly converge
|
||
|
||
|
||
def page_ink_grid(png_path: Path, cell: int = _SNAP_CELL_PX):
|
||
"""Coarse map of the page render: inked fraction of each `cell`-px square.
|
||
|
||
"Inked" means visibly darker than the paper or visibly coloured, so a
|
||
grey scan background does not count as ink. Body type covers roughly a
|
||
quarter of the cells it runs through; a photo or a filled panel covers
|
||
nearly all of them, which is the separation snap_visual_bboxes() relies
|
||
on. Returns None when the render cannot be read.
|
||
"""
|
||
try:
|
||
import numpy as np
|
||
from PIL import Image
|
||
with Image.open(png_path) as im:
|
||
rgb = np.asarray(im.convert("RGB"), dtype=np.float32)
|
||
except Exception:
|
||
return None
|
||
lightness = rgb.max(axis=2)
|
||
saturation = lightness - rgb.min(axis=2)
|
||
paper = float(np.percentile(lightness, 90)) or 255.0
|
||
ink = (lightness < paper * 0.88) | (saturation > 40)
|
||
rows, cols = (n // cell for n in ink.shape)
|
||
if rows < 4 or cols < 4:
|
||
return None
|
||
return ink[:rows * cell, :cols * cell].reshape(
|
||
rows, cell, cols, cell).mean(axis=(1, 3))
|
||
|
||
|
||
_SNAP_STABLE_IOU = 0.7 # a result must reproduce at least this much of
|
||
# itself (intersection over union) when fed back
|
||
# in as its own seed, or it is discarded -- see
|
||
# _snap_seed's stability check
|
||
|
||
|
||
def _snap_seed(coverage, seed: Sequence[int], blocked) -> list[int] | None:
|
||
"""Grow the graphic the seed cell-box points at, or None to keep the seed.
|
||
|
||
A result is only trusted once it has proven it will not move again:
|
||
since `snap_visual_bboxes` is meant to be safe to run repeatedly (an
|
||
already-corrected box seeding the same graphic again), a candidate this
|
||
function itself would revise on a second look is exactly the case that
|
||
must not be published. Two adjacent dark photos sharing a caption strip
|
||
(page 158 of c't KI Wissen 2026) is the confirmed real case: which one
|
||
"owns" the ambiguous shared area flips depending on subtle seed-size
|
||
effects, so neither box ever settles. Declining -- keeping the model's
|
||
original box -- is safer than publishing a box the very next run would
|
||
replace with a substantially different one.
|
||
"""
|
||
box = _snap_seed_once(coverage, seed, blocked)
|
||
if box is None:
|
||
return None
|
||
settled = _snap_seed_once(coverage, box, blocked)
|
||
if settled is None:
|
||
return None
|
||
intersection = _cell_overlap(box, settled)
|
||
union = ((box[2] - box[0] + 1) * (box[3] - box[1] + 1)
|
||
+ (settled[2] - settled[0] + 1) * (settled[3] - settled[1] + 1)
|
||
- intersection)
|
||
if union == 0 or intersection / union < _SNAP_STABLE_IOU:
|
||
return None
|
||
return box
|
||
|
||
|
||
def _snap_seed_once(coverage, seed: Sequence[int], blocked) -> list[int] | None:
|
||
"""One application of the snap rules -- see _snap_seed for the public,
|
||
stability-checked entry point that wraps this.
|
||
|
||
Cells already claimed by a text block -- or by another figure's own
|
||
seed area, see snap_visual_bboxes -- are excluded up front: text boxes
|
||
come back from the same engine far more accurately than figure boxes
|
||
(verified against the page render for every block of the issue above),
|
||
so they are the one trustworthy fence against a figure swallowing the
|
||
column, or the neighbouring figure, beside it.
|
||
"""
|
||
import numpy as np
|
||
from scipy import ndimage
|
||
|
||
solid = (coverage >= _SNAP_SOLID) & ~blocked
|
||
seed_area = max(1, (seed[2] - seed[0] + 1) * (seed[3] - seed[1] + 1))
|
||
seed_mask = np.zeros(solid.shape, dtype=bool)
|
||
sx0, sy0, sx1, sy1 = seed
|
||
seed_mask[sy0:sy1 + 1, sx0:sx1 + 1] = True
|
||
|
||
# A figure split by white gutters -- a grid of cards, or (as much as a
|
||
# dashed zero-line and a dense axis-label strip can be) a chart -- is
|
||
# several disconnected solid components. Bridging small gaps BEFORE
|
||
# labelling, rather than iteratively re-joining whichever component the
|
||
# seed happens to touch first, is what makes the result independent of
|
||
# the seed's own size: re-running this function on an already-corrected
|
||
# (and now larger) box must pick the same components it did the first
|
||
# time, not a different one that merely overlaps the new seed more by
|
||
# raw area. Confirmed unstable on page 18's chart before this was
|
||
# connectivity- rather than seed-driven -- the dense axis-label strip
|
||
# briefly out-competed the thin zero-line for "best overlap" once the
|
||
# seed grew to cover both, shrinking an already-correct box back down.
|
||
connected = solid
|
||
if _SNAP_GAP_CELLS > 0:
|
||
structure = ndimage.generate_binary_structure(2, 1)
|
||
for _ in range(_SNAP_GAP_CELLS):
|
||
connected = ndimage.binary_dilation(connected, structure=structure) & ~blocked
|
||
|
||
anchor = None
|
||
labels, count = ndimage.label(connected)
|
||
if count:
|
||
overlaps = ndimage.sum_labels(seed_mask, labels, range(1, count + 1))
|
||
best = max(range(count), key=lambda i: overlaps[i])
|
||
if overlaps[best] >= _SNAP_MIN_OVERLAP * seed_area:
|
||
# The bounding box of this component's ORIGINAL solid cells,
|
||
# not its dilated footprint -- dilation only decides which
|
||
# solid patches count as one figure, it must not itself widen
|
||
# the box.
|
||
ys, xs = np.where(solid & (labels == best + 1))
|
||
if xs.size:
|
||
anchor = [int(xs.min()), int(ys.min()), int(xs.max()), int(ys.max())]
|
||
# A bounding box is a rectangle, not the component's actual
|
||
# shape, so it can still enclose a blocked text wall lying
|
||
# between two merged patches even though neither patch's
|
||
# own cells ever touched it.
|
||
anchor = _trim_blocked_edges(anchor, blocked)
|
||
|
||
# The solid pass alone stops here for a photo or a UI panel -- dense
|
||
# enough that anchor already is the whole figure. A chart's sparse bars
|
||
# and axis rules need the second, looser pass below to be included at
|
||
# all; without a solid anchor to grow from, grow straight from the seed
|
||
# instead of giving up outright, since the seed usually still overlaps
|
||
# at least part of the real graphic even when none of it is solid.
|
||
grow_from = anchor if anchor is not None else seed
|
||
any_ink = (coverage >= _SNAP_ANY_INK) & ~blocked
|
||
gx0, gy0, gx1, gy1 = grow_from
|
||
if not any_ink[gy0:gy1 + 1, gx0:gx1 + 1].any():
|
||
# Nothing at all under the anchor/seed -- a blank margin, or a pale
|
||
# chart/white-on-white screenshot below even the loose floor (see
|
||
# the known limitation in Manual.md). Declining is safer than
|
||
# guessing.
|
||
if anchor is None:
|
||
return None
|
||
box = anchor
|
||
else:
|
||
box = _grow_to_bounds(coverage, blocked, grow_from)
|
||
|
||
area = (box[2] - box[0] + 1) * (box[3] - box[1] + 1)
|
||
if area < _SNAP_MIN_SEED_AREA * seed_area:
|
||
return None
|
||
|
||
if box == grow_from:
|
||
# Growth found nothing beyond the anchor itself -- fall back to the
|
||
# original density test, which only ever runs here because a real
|
||
# solid anchor was required to reach this point.
|
||
if anchor is None:
|
||
return None
|
||
inside = coverage[box[1]:box[3] + 1, box[0]:box[2] + 1]
|
||
if inside.mean() >= _SNAP_MIN_FILL:
|
||
return box
|
||
# A screenshot or a boxed insert is mostly white inside but inked
|
||
# right around its edge, so a solid frame counts as a graphic too.
|
||
edge = solid[box[1]:box[3] + 1, box[0]:box[2] + 1]
|
||
if (edge.any(axis=1).mean() >= _SNAP_MIN_FRAME
|
||
and edge.any(axis=0).mean() >= _SNAP_MIN_FRAME):
|
||
return box
|
||
return None
|
||
# Growth extended the anchor/seed through real, connected marks fenced
|
||
# by text and other figures -- trust it without re-testing density,
|
||
# which a chart's mostly-white interior fails by construction.
|
||
return box
|
||
|
||
|
||
def _grow_to_bounds(coverage, blocked, box: Sequence[int]) -> list[int]:
|
||
"""Extend a box outward through content too sparse for the solid test.
|
||
|
||
Each of the four sides advances one cell at a time through any real
|
||
mark at all (`_SNAP_ANY_INK`), which is what a bar chart's thin bars,
|
||
hairline axis and small print register as even though none of them
|
||
individually reach the solid-ink threshold above. A side stops the
|
||
instant it reaches a blocked cell -- real text, or another figure's own
|
||
seed area -- or after `_SNAP_MAX_BLANK_RUN` consecutive cells with no
|
||
mark at all, whichever comes first, so growth cannot bridge a real
|
||
gutter between two adjacent figures or wander off into open margin.
|
||
"""
|
||
rows, cols = coverage.shape
|
||
x0, y0, x1, y1 = box
|
||
any_ink = (coverage >= _SNAP_ANY_INK) & ~blocked
|
||
|
||
def scan(pos: int, step: int, limit: int, lo: int, hi: int, horizontal: bool) -> int:
|
||
blank_run = 0
|
||
while True:
|
||
nxt = pos + step
|
||
if nxt < 0 or nxt >= limit:
|
||
break
|
||
line_blocked = blocked[nxt, lo:hi + 1] if horizontal else blocked[lo:hi + 1, nxt]
|
||
if line_blocked.any():
|
||
break
|
||
line_ink = any_ink[nxt, lo:hi + 1] if horizontal else any_ink[lo:hi + 1, nxt]
|
||
if line_ink.any():
|
||
blank_run = 0
|
||
else:
|
||
blank_run += 1
|
||
if blank_run > _SNAP_MAX_BLANK_RUN:
|
||
pos -= step * (blank_run - 1) # roll back the trailing blank run
|
||
break
|
||
pos = nxt
|
||
return pos
|
||
|
||
new_y0 = scan(y0, -1, rows, x0, x1, True)
|
||
new_y1 = scan(y1, 1, rows, x0, x1, True)
|
||
new_x0 = scan(x0, -1, cols, y0, y1, False)
|
||
new_x1 = scan(x1, 1, cols, y0, y1, False)
|
||
return [min(x0, new_x0), min(y0, new_y0), max(x1, new_x1), max(y1, new_y1)]
|
||
|
||
|
||
def _trim_blocked_edges(box: Sequence[int], blocked) -> list[int]:
|
||
"""Pull each edge of `box` back to the nearest real text wall inside it.
|
||
|
||
Merging separate ink components into one bounding rectangle (see
|
||
_snap_seed) can enclose a text block that happens to sit between two
|
||
pieces of the same graphic, even though that block's own cells were
|
||
excluded from the ink search itself -- the merge only produces a
|
||
rectangle, not the components' actual shape, so the rectangle can still
|
||
geometrically span over blocked cells lying between them. Confirmed on
|
||
page 22 of c't KI Wissen 2026: a chart's candidate box reached two text
|
||
lines above the chart itself, because a small mark on the chart's own
|
||
top edge and the graphic's body were merged across a gap that also
|
||
happened to contain the tail of an unrelated paragraph.
|
||
|
||
A row or column blocked across most of the box's width/height is
|
||
unambiguously real text, not a stray corner overlap, and is trusted
|
||
over the rectangle's own extent -- searched outward from the box's own
|
||
centre so a wall near one edge trims that edge without disturbing the
|
||
other three.
|
||
"""
|
||
x0, y0, x1, y1 = box
|
||
width, height = x1 - x0 + 1, y1 - y0 + 1
|
||
sub = blocked[y0:y1 + 1, x0:x1 + 1]
|
||
row_frac = sub.mean(axis=1)
|
||
col_frac = sub.mean(axis=0)
|
||
|
||
new_y0, new_y1 = y0, y1
|
||
for r in range(height // 2, -1, -1):
|
||
if row_frac[r] >= _SNAP_BLOCKED_WALL:
|
||
new_y0 = y0 + r + 1
|
||
break
|
||
for r in range(height // 2, height):
|
||
if row_frac[r] >= _SNAP_BLOCKED_WALL:
|
||
new_y1 = y0 + r - 1
|
||
break
|
||
new_x0, new_x1 = x0, x1
|
||
for c in range(width // 2, -1, -1):
|
||
if col_frac[c] >= _SNAP_BLOCKED_WALL:
|
||
new_x0 = x0 + c + 1
|
||
break
|
||
for c in range(width // 2, width):
|
||
if col_frac[c] >= _SNAP_BLOCKED_WALL:
|
||
new_x1 = x0 + c - 1
|
||
break
|
||
if new_y0 > new_y1 or new_x0 > new_x1:
|
||
return list(box) # trimming would invert the box -- leave it be
|
||
return [new_x0, new_y0, new_x1, new_y1]
|
||
|
||
|
||
def _cell_overlap(a: Sequence[int], b: Sequence[int]) -> int:
|
||
"""Area, in grid cells, shared by two inclusive [x0, y0, x1, y1] boxes."""
|
||
return (max(0, min(a[2], b[2]) - max(a[0], b[0]) + 1)
|
||
* max(0, min(a[3], b[3]) - max(a[1], b[1]) + 1))
|
||
|
||
|
||
def snap_visual_bboxes(png_path: Path, blocks: Sequence[Block],
|
||
page_meta: dict[str, Any]) -> int:
|
||
"""Correct every visual block's bbox against the page render, in place.
|
||
|
||
Returns how many boxes moved. Silent no-op when the render is missing or
|
||
the imaging stack is unavailable -- the layout stays exactly as the OCR
|
||
engine reported it, which is what happened before this pass existed.
|
||
"""
|
||
visuals = [b for b in blocks if b.type in VISUAL_TYPES]
|
||
if not visuals:
|
||
return 0
|
||
try:
|
||
import numpy as np
|
||
import scipy.ndimage # noqa: F401 -- required by _snap_seed
|
||
except ImportError:
|
||
return 0
|
||
coverage = page_ink_grid(png_path)
|
||
if coverage is None:
|
||
return 0
|
||
rows, cols = coverage.shape
|
||
page_w = float(page_meta.get("width") or 0) or 1.0
|
||
page_h = float(page_meta.get("height") or 0) or 1.0
|
||
cells_x, cells_y = cols / page_w, rows / page_h
|
||
|
||
def to_cells(bbox: Sequence[float]) -> list[int]:
|
||
x, y, w, h = bbox
|
||
return [min(cols - 1, max(0, int(x * cells_x))),
|
||
min(rows - 1, max(0, int(y * cells_y))),
|
||
min(cols - 1, max(0, int((x + w) * cells_x))),
|
||
min(rows - 1, max(0, int((y + h) * cells_y)))]
|
||
|
||
text_blocked = np.zeros(coverage.shape, dtype=bool)
|
||
for b in blocks:
|
||
if b.type in VISUAL_TYPES or not b.text.strip():
|
||
continue
|
||
x0, y0, x1, y1 = to_cells(b.bbox)
|
||
text_blocked[y0:y1 + 1, x0:x1 + 1] = True
|
||
|
||
visual_seeds = [to_cells(b.bbox) for b in visuals]
|
||
|
||
def blocked_for(i: int, current: list[list[int]]):
|
||
# A figure's loose growth pass (see _grow_to_bounds) must not bridge
|
||
# into a neighbouring figure's own territory just because the gutter
|
||
# between them is narrow and mostly blank -- two head-and-shoulders
|
||
# portraits on the same page can sit under 2 grid cells apart.
|
||
# Blocking every *other* figure's own current area, not just text,
|
||
# fences that gutter regardless of how little ink is in it.
|
||
blocked = text_blocked.copy()
|
||
for j, other in enumerate(current):
|
||
if j == i:
|
||
continue
|
||
ox0, oy0, ox1, oy1 = other
|
||
blocked[oy0:oy1 + 1, ox0:ox1 + 1] = True
|
||
return blocked
|
||
|
||
def colliding_pairs(boxes: list[list[int]]) -> set[int]:
|
||
# Two figures pulled onto the same graphic means the page was read
|
||
# as one picture where the engine saw two (or the reverse); neither
|
||
# reading can be trusted over the other, so both keep their
|
||
# original box.
|
||
hit: set[int] = set()
|
||
for i, a in enumerate(boxes):
|
||
for j in range(i + 1, len(boxes)):
|
||
c = boxes[j]
|
||
smaller = min((a[2] - a[0] + 1) * (a[3] - a[1] + 1),
|
||
(c[2] - c[0] + 1) * (c[3] - c[1] + 1))
|
||
if _cell_overlap(a, c) > 0.5 * smaller:
|
||
hit.add(i)
|
||
hit.add(j)
|
||
return hit
|
||
|
||
# Ambiguity is decided once, up front, fenced against the engine's own
|
||
# ORIGINAL seeds -- not against whatever the iterative refinement below
|
||
# settles on. Iterating first and only then checking for collisions
|
||
# would let two seeds legitimately ambiguous about the same graphic
|
||
# mutually fence each other into a tidy, confident-looking split of it
|
||
# instead of ever being recognised as the same disputed claim.
|
||
first_round = [_snap_seed(coverage, seed, blocked_for(i, visual_seeds))
|
||
for i, seed in enumerate(visual_seeds)]
|
||
ambiguous = colliding_pairs([box or seed for box, seed in zip(first_round, visual_seeds)])
|
||
|
||
# Fencing figure i against figure j's box, and vice versa, means each
|
||
# one's result depends on where the OTHER currently sits -- a mutual
|
||
# dependency a single pass cannot resolve consistently. Repeating the
|
||
# whole page's worth of visuals, each round seeded by the previous
|
||
# round's output, converges that to a joint fixed point in the common
|
||
# case, and -- for the rare pair that never truly settles (two dark
|
||
# photos sharing a caption strip, confirmed on page 158 of c't KI
|
||
# Wissen 2026) -- reliably lands on the SAME deterministic answer every
|
||
# time this function is called, because a later call starts from
|
||
# exactly the state this loop stops at rather than re-diverging. An
|
||
# already-ambiguous visual is frozen at its original seed throughout,
|
||
# both as a participant (it will be discarded anyway) and as a fence
|
||
# for its neighbours (its disputed reading must not move anyone else).
|
||
current = list(visual_seeds)
|
||
for _ in range(_SNAP_JOINT_ROUNDS):
|
||
nxt = [visual_seeds[i] if i in ambiguous else
|
||
(_snap_seed(coverage, seed, blocked_for(i, current)) or seed)
|
||
for i, seed in enumerate(current)]
|
||
if nxt == current:
|
||
break
|
||
current = nxt
|
||
|
||
reverted = ambiguous | colliding_pairs(current)
|
||
final = [None if i in reverted or current[i] == visual_seeds[i] else current[i]
|
||
for i in range(len(current))]
|
||
|
||
moved = 0
|
||
for block, box in zip(visuals, final):
|
||
if box is None:
|
||
continue
|
||
bbox = [box[0] / cells_x, box[1] / cells_y,
|
||
(box[2] + 1 - box[0]) / cells_x, (box[3] + 1 - box[1]) / cells_y]
|
||
if any(abs(a - b) > 1.0 for a, b in zip(bbox, block.bbox)):
|
||
moved += 1
|
||
block.bbox = bbox
|
||
return moved
|
||
|
||
|
||
def ocr_page(png_path: Path, lang: str = "deu+eng",
|
||
min_confidence: float = 40.0) -> list[dict[str, Any]]:
|
||
"""OCR one rendered page, returning paragraph boxes from tesseract's TSV.
|
||
|
||
Grouped by (block, paragraph), not by line: PyMuPDF's native path already
|
||
aggregates a paragraph's spans into one Block with one representative font
|
||
size, and OCR must match that shape or the two paths behave inconsistently
|
||
downstream. Confirmed live on a scanned magazine -- one Block per LINE let
|
||
a single anomalous line (a chart axis number, a headline's wrapped second
|
||
line, scan noise) independently trigger BLOCK_TITLE classification, and a
|
||
28-page scan produced ~11 "title" blocks per page instead of grouping
|
||
around real headlines. font_size uses the MEDIAN of the paragraph's own
|
||
line heights rather than the paragraph's total bbox height (which would
|
||
scale with line count, not type size) -- a genuine multi-line headline
|
||
still measures large because each of its few lines is uniformly large;
|
||
one noisy line inside an otherwise normal paragraph no longer dominates.
|
||
TSV rather than plain text because the coordinates are the whole point:
|
||
they are what lets an OCR'd page take part in the same column and article
|
||
grouping as a native-text page.
|
||
"""
|
||
try:
|
||
proc = subprocess.run(
|
||
["tesseract", str(png_path), "stdout", "-l", lang, "tsv"],
|
||
capture_output=True, text=True, timeout=180, check=True,
|
||
)
|
||
except (subprocess.SubprocessError, FileNotFoundError, OSError):
|
||
return []
|
||
|
||
paragraphs: dict[tuple[int, int], dict[str, Any]] = {}
|
||
for row in proc.stdout.splitlines()[1:]:
|
||
parts = row.split("\t")
|
||
if len(parts) < 12:
|
||
continue
|
||
try:
|
||
block_num, par_num, line_num = int(parts[2]), int(parts[3]), int(parts[4])
|
||
left, top, width, height = (int(parts[6]), int(parts[7]),
|
||
int(parts[8]), int(parts[9]))
|
||
conf = float(parts[10])
|
||
except ValueError:
|
||
continue
|
||
word = parts[11].strip()
|
||
if not word or conf < min_confidence:
|
||
continue
|
||
para_key = (block_num, par_num)
|
||
para = paragraphs.setdefault(para_key, {"lines": {}})
|
||
line = para["lines"].setdefault(line_num, {
|
||
"text": [], "x0": left, "y0": top,
|
||
"x1": left + width, "y1": top + height, "conf": [],
|
||
})
|
||
line["text"].append(word)
|
||
line["x0"] = min(line["x0"], left)
|
||
line["y0"] = min(line["y0"], top)
|
||
line["x1"] = max(line["x1"], left + width)
|
||
line["y1"] = max(line["y1"], top + height)
|
||
line["conf"].append(conf)
|
||
|
||
out = []
|
||
for _, para in sorted(paragraphs.items()):
|
||
lines = [l for _, l in sorted(para["lines"].items()) if l["text"]]
|
||
if not lines:
|
||
continue
|
||
x0 = min(l["x0"] for l in lines)
|
||
y0 = min(l["y0"] for l in lines)
|
||
x1 = max(l["x1"] for l in lines)
|
||
y1 = max(l["y1"] for l in lines)
|
||
heights = [float(l["y1"] - l["y0"]) for l in lines]
|
||
confs = [c for l in lines for c in l["conf"]]
|
||
out.append({
|
||
"text": " ".join(" ".join(l["text"]) for l in lines),
|
||
"bbox": [float(x0), float(y0), float(x1 - x0), float(y1 - y0)],
|
||
"line_height": statistics.median(heights),
|
||
"confidence": round(sum(confs) / len(confs) / 100.0, 3),
|
||
})
|
||
return out
|
||
|
||
|
||
_VISION_QUALITY_THRESHOLD = 0.5 # below this, existing native/OCR text is escalated
|
||
|
||
|
||
def extract_pages(pdf_path: Path, out_dir: Path, render_dpi: int = 110,
|
||
ocr_lang: str = "deu+eng",
|
||
vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
vision_ocr_timeout: float = 150.0,
|
||
force_vision_ocr: bool = False,
|
||
progress: Any = None) -> dict[str, Any]:
|
||
"""Run the full extraction for one PDF, writing page renders and images.
|
||
|
||
When vision_ocr_url/model are given, a page whose text needs OCR (or
|
||
whose already-extracted text scores as mostly garbage, native or
|
||
tesseract alike -- see ocr_quality_score) is escalated to that engine
|
||
first, with a second vision engine (vision_ocr_fallback_*) tried next if
|
||
given, and tesseract only as the last resort if every vision call fails
|
||
or every engine is unreachable. Confirmed live against
|
||
ocr-baidu-unlimited-ocr: markedly more accurate transcription of a
|
||
scanned German magazine than tesseract, and it labels running headers/
|
||
footers/page numbers directly (see build_vision_blocks), which is the
|
||
dominant source of the over-fragmentation cluster_running_furniture()
|
||
otherwise has to recover after the fact from noisy OCR variants of the
|
||
same repeating line. KDL Frontier Parser nano, tested as the fallback
|
||
engine, answers faster but returns plain text with no layout coordinates
|
||
at all -- handled by _whole_page_region as one ambiguous full-page block.
|
||
|
||
Returns the layout document: every block on every page, classified and in
|
||
reading order, plus the raw contents-page candidates.
|
||
"""
|
||
fitz = _require_fitz()
|
||
pages_dir = out_dir / "pages"
|
||
images_dir = out_dir / "images"
|
||
pages_dir.mkdir(parents=True, exist_ok=True)
|
||
images_dir.mkdir(parents=True, exist_ok=True)
|
||
vision_engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
vision_enabled = bool(vision_ocr_url and vision_ocr_model)
|
||
vision_page_count = 0
|
||
|
||
doc = fitz.open(str(pdf_path))
|
||
try:
|
||
raw_pages: list[dict[str, Any]] = []
|
||
|
||
for pno in range(doc.page_count):
|
||
page = doc.load_page(pno)
|
||
rect = page.rect
|
||
page_no = pno + 1
|
||
|
||
png = pages_dir / f"{page_no:04d}.png"
|
||
if not png.exists():
|
||
pix = page.get_pixmap(dpi=render_dpi)
|
||
pix.save(str(png))
|
||
|
||
image_boxes: list[list[float]] = []
|
||
page_images: list[dict[str, Any]] = []
|
||
for img_index, info in enumerate(page.get_images(full=True)):
|
||
rects: list[list[float]] = []
|
||
try:
|
||
for r in page.get_image_rects(info[0]):
|
||
rects.append([r.x0, r.y0, r.width, r.height])
|
||
except Exception:
|
||
continue
|
||
image_boxes.extend(rects)
|
||
name = f"p{page_no:04d}-{img_index:02d}"
|
||
if _extract_image(fitz, doc, info[0], images_dir, name) and rects:
|
||
# A few embedded images repeat (a logo stamped on every
|
||
# page); each placement still gets its own bbox so a
|
||
# caption or figure position can be matched precisely.
|
||
for ri, box in enumerate(rects):
|
||
page_images.append({
|
||
"id": name if ri == 0 else f"{name}-{ri}",
|
||
"page": page_no,
|
||
"bbox": box,
|
||
"filename": f"{name}.png",
|
||
})
|
||
|
||
blocks: list[Block] = []
|
||
data = page.get_text("dict")
|
||
for bi, blk in enumerate(data.get("blocks", [])):
|
||
if blk.get("type") != 0: # 0 = text, 1 = image
|
||
continue
|
||
texts, sizes, fonts, bolds = [], [], [], []
|
||
for line in blk.get("lines", []):
|
||
for span in line.get("spans", []):
|
||
if not span.get("text", "").strip():
|
||
continue
|
||
texts.append(span["text"])
|
||
sizes.append(float(span.get("size", 0)))
|
||
fonts.append(span.get("font", ""))
|
||
bolds.append(bool(span.get("flags", 0) & 2 ** 4))
|
||
text = " ".join(texts).strip()
|
||
if not text:
|
||
continue
|
||
x0, y0, x1, y1 = blk["bbox"]
|
||
blocks.append(Block(
|
||
id=f"b{page_no:04d}-{bi:03d}",
|
||
page=page_no,
|
||
bbox=[x0, y0, x1 - x0, y1 - y0],
|
||
type=BLOCK_BODY,
|
||
text=text,
|
||
font_size=statistics.median(sizes) if sizes else 0.0,
|
||
font_name=max(set(fonts), key=fonts.count) if fonts else "",
|
||
bold=sum(bolds) > len(bolds) / 2 if bolds else False,
|
||
))
|
||
|
||
# A page whose native text is negligible is a scan (or is drawn as
|
||
# outlines); OCR is the only way to reach its words. A page WITH
|
||
# native/embedded text is not necessarily good text -- a
|
||
# pre-OCR'd scan (e.g. from scanner-bundled OCR software) can
|
||
# carry a poor text layer that this same escalation should reach.
|
||
native_chars = sum(len(b.text) for b in blocks)
|
||
needs_ocr = native_chars < 120
|
||
native_quality = (
|
||
ocr_quality_score(" ".join(b.text for b in blocks))
|
||
if not needs_ocr else 0.0
|
||
)
|
||
needs_escalation = force_vision_ocr or needs_ocr or native_quality < _VISION_QUALITY_THRESHOLD
|
||
|
||
vision_blocks: list[Block] | None = None
|
||
if vision_enabled and needs_escalation:
|
||
regions = vision_ocr_page_chain(png, vision_engines,
|
||
timeout=vision_ocr_timeout)
|
||
if regions:
|
||
scale_x = rect.width / max(1.0, _png_width(png, fitz))
|
||
vision_blocks = build_vision_blocks(
|
||
regions, page_no, scale_x, id_prefix="b")
|
||
|
||
if vision_blocks is not None:
|
||
# Replaces whatever was already there -- including a poor
|
||
# native/pre-OCR'd text layer, rather than layering a second,
|
||
# conflicting reading on top of it.
|
||
vision_page_count += 1
|
||
blocks = vision_blocks
|
||
page_meta = {"width": rect.width, "height": rect.height}
|
||
# Before cropping: the crop is taken from the block's bbox,
|
||
# so a box the engine placed loosely would be saved as a
|
||
# mis-framed figure file as well as a mis-drawn review box.
|
||
snap_visual_bboxes(png, blocks, page_meta)
|
||
detected_images = crop_visual_regions(out_dir, page_no,
|
||
page_meta, blocks)
|
||
if detected_images:
|
||
page_images = detected_images
|
||
elif needs_ocr:
|
||
scale_x = rect.width / max(1.0, _png_width(png, fitz))
|
||
for li, para in enumerate(ocr_page(png, ocr_lang)):
|
||
bx, by, bw, bh = para["bbox"]
|
||
blocks.append(Block(
|
||
id=f"b{page_no:04d}-o{li:03d}",
|
||
page=page_no,
|
||
bbox=[bx * scale_x, by * scale_x, bw * scale_x, bh * scale_x],
|
||
type=BLOCK_BODY,
|
||
text=para["text"],
|
||
confidence=para["confidence"],
|
||
# The paragraph's own line height, not its full bbox
|
||
# height -- that would scale with line count instead
|
||
# of type size (see ocr_page's docstring).
|
||
font_size=para["line_height"] * scale_x,
|
||
source="ocr",
|
||
))
|
||
# else: native text present and scored well enough -- unchanged.
|
||
|
||
raw_pages.append({
|
||
"page": page_no,
|
||
"width": rect.width,
|
||
"height": rect.height,
|
||
"blocks": blocks,
|
||
"image_boxes": image_boxes,
|
||
"images": page_images,
|
||
})
|
||
if progress:
|
||
# Raw block text, pre-classification -- good enough for a
|
||
# live "here is what was just read" preview during ingestion
|
||
# (mirrors ocr-baidu-unlimited-ocr's own demo: source page on
|
||
# one side, recognised text on the other, filled in as each
|
||
# page finishes) without waiting for the document-wide passes
|
||
# below.
|
||
try:
|
||
progress(page_no, doc.page_count, blocks, page_images)
|
||
except TypeError:
|
||
progress(page_no, doc.page_count, blocks)
|
||
|
||
# Furniture can only be identified across the whole document, so
|
||
# classification waits until every page has been read.
|
||
repeated = find_repeated_lines(
|
||
[[b.text for b in p["blocks"]] for p in raw_pages]
|
||
)
|
||
all_blocks: list[Block] = []
|
||
contents_candidates: list[dict[str, Any]] = []
|
||
|
||
# One baseline for the whole issue (see classify_blocks).
|
||
doc_base = body_font_size(
|
||
[b for p in raw_pages for b in p["blocks"]]
|
||
)
|
||
|
||
for p in raw_pages:
|
||
classify_blocks(p["blocks"], p["height"], p["width"],
|
||
repeated=repeated, image_boxes=p["image_boxes"],
|
||
base_size=doc_base)
|
||
mark_advert_pages(p["blocks"])
|
||
mark_pullquotes(p["blocks"])
|
||
assign_reading_order(p["blocks"], p["width"])
|
||
mark_sidebars(p["blocks"], p["width"])
|
||
all_blocks.extend(p["blocks"])
|
||
|
||
# Cross-page, so it runs once over every block rather than per page
|
||
# (see cluster_running_furniture's docstring for why exact-match
|
||
# find_repeated_lines above is not enough once OCR is involved).
|
||
page_heights = {p["page"]: p["height"] for p in raw_pages}
|
||
cluster_running_furniture(all_blocks, page_heights)
|
||
|
||
for p in raw_pages:
|
||
lines = [b.text for b in sorted(p["blocks"], key=lambda b: b.reading_order)]
|
||
score = score_contents_page(lines)
|
||
if score >= 0.45:
|
||
contents_candidates.append({
|
||
"page": p["page"],
|
||
"score": score,
|
||
"entries": parse_contents(lines),
|
||
})
|
||
|
||
all_images = [img for p in raw_pages for img in p["images"]]
|
||
for img in all_images:
|
||
img["reading_order"] = image_reading_position(img, all_blocks)
|
||
|
||
return {
|
||
"page_count": doc.page_count,
|
||
"pages": [{"page": p["page"], "width": p["width"], "height": p["height"]}
|
||
for p in raw_pages],
|
||
"blocks": [b.to_json() for b in all_blocks],
|
||
"images": all_images,
|
||
"contents_candidates": contents_candidates,
|
||
"vision_ocr_pages": vision_page_count,
|
||
}
|
||
finally:
|
||
doc.close()
|
||
|
||
|
||
def _png_width(png: Path, fitz) -> float:
|
||
try:
|
||
return float(fitz.Pixmap(str(png)).width)
|
||
except Exception:
|
||
return 1.0
|
||
|
||
|
||
def _extract_image(fitz, doc, xref: int, images_dir: Path, name: str) -> bool:
|
||
"""Save one embedded image, skipping the tiny rules and spacers.
|
||
|
||
Returns whether a file now exists at that name, so the caller can decide
|
||
whether to record it as a real figure -- without this, a page's rules and
|
||
spacers were silently skipped on disk but still implicitly "existed" as
|
||
far as anything downstream knew, since nothing checked.
|
||
"""
|
||
try:
|
||
pix = fitz.Pixmap(doc, xref)
|
||
if pix.width < 64 or pix.height < 64:
|
||
return False
|
||
if pix.n - pix.alpha >= 4: # CMYK needs converting first
|
||
pix = fitz.Pixmap(fitz.csRGB, pix)
|
||
target = images_dir / f"{name}.png"
|
||
if not target.exists():
|
||
pix.save(str(target))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ── Images: linking a figure to the article it illustrates ──────────────────
|
||
|
||
def assign_images_to_articles(images: Sequence[dict[str, Any]],
|
||
blocks: Sequence[Block],
|
||
articles: Sequence[Article]) -> None:
|
||
"""Attach each extracted figure to the article whose column it falls in.
|
||
|
||
A figure belongs to whichever article occupies its page; when two
|
||
articles share a page (a feature ending, a new one starting lower down),
|
||
the figure goes to whichever article has the nearer block by vertical
|
||
position, so a photo is not shown under the wrong headline. Mutates
|
||
articles in place.
|
||
"""
|
||
by_page: dict[int, list[Article]] = {}
|
||
for art in articles:
|
||
for page in art.pages:
|
||
by_page.setdefault(page, []).append(art)
|
||
|
||
blocks_by_page: dict[int, list[Block]] = {}
|
||
for b in blocks:
|
||
blocks_by_page.setdefault(b.page, []).append(b)
|
||
|
||
for img in images:
|
||
if img.get("type") == BLOCK_AD:
|
||
continue
|
||
candidates = by_page.get(img["page"], [])
|
||
if not candidates:
|
||
continue
|
||
if len(candidates) == 1:
|
||
candidates[0].images.append(img["id"])
|
||
continue
|
||
# Several articles share this page: pick by nearest block vertically,
|
||
# since that is what a reader would associate the figure with.
|
||
img_y = img["bbox"][1]
|
||
best, best_dist = candidates[0], float("inf")
|
||
for art in candidates:
|
||
own_blocks = [b for b in blocks_by_page.get(img["page"], [])
|
||
if b.id in set(art.blocks)]
|
||
if not own_blocks:
|
||
continue
|
||
dist = min(abs(b.bbox[1] - img_y) for b in own_blocks)
|
||
if dist < best_dist:
|
||
best, best_dist = art, dist
|
||
best.images.append(img["id"])
|
||
|
||
|
||
def image_reading_position(image: dict[str, Any], blocks: Sequence[Block]) -> int:
|
||
"""Approximate reading_order for an image, for ordering it among sentences.
|
||
|
||
Uses the nearest block on the same page whose top sits at or above the
|
||
image (so a photo is shown once the surrounding text has been reached,
|
||
not before), falling back to the closest block on the page if the image
|
||
sits above all of them.
|
||
"""
|
||
same_page = [b for b in blocks if b.page == image["page"]]
|
||
if not same_page:
|
||
return 0
|
||
above = [b for b in same_page if b.bbox[1] <= image["bbox"][1]]
|
||
pool = above or same_page
|
||
nearest = min(pool, key=lambda b: abs(b.bbox[1] - image["bbox"][1]))
|
||
return nearest.reading_order
|
||
|
||
|
||
# ── Persistence ──────────────────────────────────────────────────────────────
|
||
|
||
def build_articles(layout: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Turn a stored layout document into detected articles with estimates."""
|
||
blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
candidates = layout.get("contents_candidates") or []
|
||
best = max(candidates, key=lambda c: c["score"], default=None)
|
||
entries = best["entries"] if best else []
|
||
|
||
# The contents page lists articles; it is not one.
|
||
skip = {c["page"] for c in candidates if c.get("score", 0) >= 0.45}
|
||
articles = group_articles(blocks, entries, layout.get("page_count", 0),
|
||
skip_pages=skip)
|
||
assign_images_to_articles(layout.get("images") or [], blocks, articles)
|
||
out = []
|
||
for art in articles:
|
||
rec = art.to_json()
|
||
rec["estimated_seconds"] = estimate_listen_seconds(blocks, art)
|
||
out.append(rec)
|
||
return out
|
||
|
||
|
||
def reocr_page(doc_dir: Path, layout: dict[str, Any], page_no: int,
|
||
vision_ocr_url: str, vision_ocr_model: str,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> dict[str, Any] | None:
|
||
"""Force one page through the vision OCR chain again with the engine the
|
||
user picked, replacing only that page's blocks in an already-ingested
|
||
layout.
|
||
|
||
Unlike extract_pages()'s automatic escalation, this always calls the
|
||
engine regardless of the page's existing text quality -- a user pressing
|
||
"OCR again" has already judged the result unusable and is not asking for
|
||
another quality gate. Deliberately does NOT call build_articles(): an
|
||
article the user already reviewed and approved must not be silently
|
||
re-grouped and lose that approval just because one of its pages was
|
||
re-read. The caller decides whether to re-run detection separately.
|
||
|
||
Returns the updated layout, or None if the page has no rendered PNG or
|
||
the engine returned nothing usable (network failure, empty response) --
|
||
the caller keeps the existing blocks in that case, never discards good
|
||
data for a failed retry.
|
||
"""
|
||
fitz = _require_fitz()
|
||
png = doc_dir / "pages" / f"{page_no:04d}.png"
|
||
if not png.is_file():
|
||
return None
|
||
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
return None
|
||
|
||
engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
regions = vision_ocr_page_chain(png, engines, timeout=timeout)
|
||
if not regions:
|
||
return None
|
||
|
||
scale_x = page_meta["width"] / max(1.0, _png_width(png, fitz))
|
||
new_blocks = build_vision_blocks(regions, page_no, scale_x, id_prefix="r")
|
||
snap_visual_bboxes(png, new_blocks, page_meta)
|
||
|
||
all_blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
kept = [b for b in all_blocks if b.page != page_no]
|
||
|
||
# The baseline must come from the rest of the document, not this page
|
||
# alone -- see classify_blocks' docstring on why a single page (which may
|
||
# have no body copy of its own) cannot be its own reference.
|
||
doc_base = body_font_size(kept) or body_font_size(all_blocks)
|
||
classify_blocks(new_blocks, page_meta["height"], page_meta["width"], base_size=doc_base)
|
||
mark_advert_pages(new_blocks)
|
||
mark_pullquotes(new_blocks)
|
||
assign_reading_order(new_blocks, page_meta["width"])
|
||
mark_sidebars(new_blocks, page_meta["width"])
|
||
|
||
merged = kept + new_blocks
|
||
page_heights = {p["page"]: p["height"] for p in layout.get("pages", [])}
|
||
cluster_running_furniture(merged, page_heights)
|
||
|
||
layout = dict(layout)
|
||
layout["blocks"] = [b.to_json() for b in merged]
|
||
detected_images = crop_visual_regions(doc_dir, page_no, page_meta, new_blocks)
|
||
if detected_images:
|
||
layout["images"] = [i for i in layout.get("images", []) if i["page"] != page_no or "-manual-" in i["id"]] + detected_images
|
||
layout["vision_ocr_pages"] = int(layout.get("vision_ocr_pages") or 0) + 1
|
||
return layout
|
||
|
||
|
||
def apply_semantic_regions(blocks: list[dict], regions: list[dict], page_meta: dict | None = None) -> list[dict]:
|
||
"""Group existing OCR fragments without allowing invented IDs or text."""
|
||
known = {b["id"]: b for b in blocks}
|
||
used = set()
|
||
out = []
|
||
allowed = set(NARRATABLE) | set(VISUAL_TYPES) | {BLOCK_AD, BLOCK_SIDEBAR, BLOCK_CAPTION, BLOCK_FURNITURE, BLOCK_PULLQUOTE}
|
||
for region in regions:
|
||
if not isinstance(region, dict):
|
||
raise ValueError("Invalid semantic region")
|
||
ids = region.get("block_ids")
|
||
label = region.get("type")
|
||
kind = _VISION_LABEL_TYPES.get(label, label)
|
||
visual_box = None
|
||
if "bbox" in region and kind in (*VISUAL_TYPES, BLOCK_AD):
|
||
bbox = region["bbox"]
|
||
if not page_meta or not isinstance(bbox, list) or len(bbox) != 4 or not all(
|
||
isinstance(v, (int, float)) and math.isfinite(v) and 0 <= v <= 1000 for v in bbox):
|
||
raise ValueError("Invalid visual bounding box")
|
||
x0, y0, x1, y1 = bbox
|
||
if x1 <= x0 or y1 <= y0:
|
||
raise ValueError("Empty visual region")
|
||
visual_box = [x0 / 1000 * page_meta["width"], y0 / 1000 * page_meta["height"],
|
||
(x1 - x0) / 1000 * page_meta["width"], (y1 - y0) / 1000 * page_meta["height"]]
|
||
if not ids and kind in VISUAL_TYPES and visual_box:
|
||
out.append(Block(id=f"p{page_meta['page']:04d}-visual-{len(out)}", page=page_meta['page'],
|
||
bbox=visual_box, type=kind, text="", reading_order=len(out),
|
||
source="llm_layout", region_label=label).to_json())
|
||
continue
|
||
if not isinstance(ids, list) or not ids or not all(isinstance(i, str) for i in ids):
|
||
raise ValueError("Invalid region IDs")
|
||
if kind not in allowed or len(set(ids)) != len(ids) or any(i not in known or i in used for i in ids):
|
||
raise ValueError("Unknown or repeated semantic region")
|
||
members = [known[i] for i in ids]
|
||
if len({b["page"] for b in members}) != 1:
|
||
raise ValueError("A region cannot span pages")
|
||
if any(b["type"] == BLOCK_AD for b in members):
|
||
if not all(b["type"] == BLOCK_AD for b in members):
|
||
raise ValueError("Cannot merge advertising and editorial text")
|
||
kind = BLOCK_AD
|
||
used.update(ids)
|
||
x = min(b["bbox"][0] for b in members)
|
||
y = min(b["bbox"][1] for b in members)
|
||
right = max(b["bbox"][0] + b["bbox"][2] for b in members)
|
||
bottom = max(b["bbox"][1] + b["bbox"][3] for b in members)
|
||
separator = "\n" if kind in (BLOCK_CODE, BLOCK_TABLE) else " "
|
||
text = separator.join(b["text"] for b in members)
|
||
if kind in NARRATABLE:
|
||
text = re.sub(r"(?<=\w)-\s+(?=[a-zäöüß])", "", text)
|
||
out.append(dict(members[0], bbox=visual_box or [x, y, right - x, bottom - y], text=text,
|
||
type=kind, region_label=label, source="llm_layout",
|
||
reading_order=len(out),
|
||
continues_from_previous_page=region.get("continues_from_previous_page") is True))
|
||
# Incomplete model output must not silently remove document content.
|
||
return out + [b for b in blocks if b["id"] not in used]
|
||
|
||
|
||
def remap_reviewed_page(articles: list[dict], old_layout: dict, layout: dict, page: int) -> None:
|
||
"""Reconnect reviewed articles when OCR replaces the page's region IDs."""
|
||
old = {b["id"]: b for b in old_layout.get("blocks", []) if b["page"] == page}
|
||
owners = {a["id"]: [old[i] for i in a.get("blocks", []) if i in old] for a in articles}
|
||
affected = [a for a in articles if page in a.get("pages", []) or owners[a["id"]]]
|
||
for a in affected:
|
||
a["blocks"] = [i for i in a.get("blocks", []) if i not in old]
|
||
known_images = {i["id"] for i in layout.get("images", [])}
|
||
a["images"] = [i for i in a.get("images", []) if i in known_images]
|
||
if a.get("status") == "approved":
|
||
a["status"] = "detected"
|
||
for block in layout.get("blocks", []):
|
||
if block["page"] != page or block["type"] not in NARRATABLE or not affected:
|
||
continue
|
||
x, y, w, h = block["bbox"]
|
||
def overlap(a):
|
||
return sum(max(0, min(x + w, b["bbox"][0] + b["bbox"][2]) - max(x, b["bbox"][0])) *
|
||
max(0, min(y + h, b["bbox"][1] + b["bbox"][3]) - max(y, b["bbox"][1]))
|
||
for b in owners[a["id"]])
|
||
best = max(affected, key=overlap)
|
||
if overlap(best) >= w * h * 0.35:
|
||
best["blocks"].append(block["id"])
|
||
# Newly recognised visual regions follow the closest article on the page.
|
||
objects = [Article(**{k: v for k, v in a.items() if k in Article.__dataclass_fields__}) for a in affected]
|
||
new_images = [i for i in layout.get("images", []) if i["page"] == page and i.get("source_block")]
|
||
for a in objects:
|
||
a.images = [i for i in a.images if i not in {im["id"] for im in new_images}]
|
||
assign_images_to_articles(new_images, [Block(**b) for b in layout.get("blocks", [])], objects)
|
||
for a, obj in zip(affected, objects):
|
||
a["images"] = obj.images
|
||
|
||
|
||
def _crop_page_png(doc_dir: Path, page_no: int, page_meta: dict[str, Any],
|
||
bbox: Sequence[float], dest: Path) -> None:
|
||
"""Crop one region (PDF-point bbox) out of a page's rendered PNG."""
|
||
from PIL import Image
|
||
png = doc_dir / "pages" / f"{page_no:04d}.png"
|
||
with Image.open(png) as im:
|
||
sx, sy = im.width / page_meta["width"], im.height / page_meta["height"]
|
||
box_px = (
|
||
max(0, int(bbox[0] * sx)), max(0, int(bbox[1] * sy)),
|
||
min(im.width, int((bbox[0] + bbox[2]) * sx)),
|
||
min(im.height, int((bbox[1] + bbox[3]) * sy)),
|
||
)
|
||
im.crop(box_px).save(dest)
|
||
|
||
|
||
def refresh_visual_regions(doc_dir: Path, layout: dict, page: int | None = None) -> None:
|
||
"""Re-snap figure bounds, regenerate the crops taken from them, and avoid
|
||
treating a whole scanned page as one figure."""
|
||
for meta in layout.get("pages", []):
|
||
page_no = meta["page"]
|
||
if page is not None and page_no != page:
|
||
continue
|
||
blocks = [Block(**b) for b in layout.get("blocks", []) if b["page"] == page_no]
|
||
# Re-snapping here, not only at OCR time, is what lets an issue
|
||
# ingested before this pass existed pick up corrected figure boxes
|
||
# without paying for a second OCR run of the whole document. It is
|
||
# idempotent: an already-correct box seeds the same graphic again.
|
||
if snap_visual_bboxes(doc_dir / "pages" / f"{page_no:04d}.png", blocks, meta):
|
||
snapped = {b.id: b.bbox for b in blocks if b.type in VISUAL_TYPES}
|
||
for saved in layout["blocks"]:
|
||
if saved["id"] in snapped:
|
||
saved["bbox"] = snapped[saved["id"]]
|
||
crops = crop_visual_regions(doc_dir, page_no, meta, blocks)
|
||
kept = []
|
||
for img in layout.get("images", []):
|
||
if img["page"] != page_no or "-manual-" in img["id"]:
|
||
kept.append(img)
|
||
continue
|
||
if img.get("source_block"):
|
||
continue
|
||
x, y, w, h = img["bbox"]
|
||
if w * h >= meta["width"] * meta["height"] * .85 and sum(len(b.text) for b in blocks) > 120:
|
||
continue
|
||
if any(abs(x - c["bbox"][0]) < 5 and abs(y - c["bbox"][1]) < 5 and
|
||
abs(w - c["bbox"][2]) < 10 and abs(h - c["bbox"][3]) < 10 for c in crops):
|
||
continue
|
||
# Embedded graphics inside an identified advert are advertising too.
|
||
if any(b.type == BLOCK_AD and x >= b.bbox[0] and y >= b.bbox[1] and
|
||
x + w <= b.bbox[0] + b.bbox[2] and y + h <= b.bbox[1] + b.bbox[3] for b in blocks):
|
||
img = dict(img, type=BLOCK_AD)
|
||
kept.append(img)
|
||
for crop in crops:
|
||
crop["reading_order"] = image_reading_position(crop, blocks)
|
||
layout["images"] = kept + crops
|
||
|
||
|
||
def crop_visual_regions(doc_dir: Path, page_no: int, page_meta: dict,
|
||
blocks: Sequence[Block]) -> list[dict[str, Any]]:
|
||
"""Keep charts, tables and code boxes intact as page crops for playback."""
|
||
images = []
|
||
for b in blocks:
|
||
if b.type not in VISUAL_TYPES:
|
||
continue
|
||
name = f"p{page_no:04d}-region-{b.id}"
|
||
(doc_dir / "images").mkdir(parents=True, exist_ok=True)
|
||
_crop_page_png(doc_dir, page_no, page_meta, b.bbox, doc_dir / "images" / f"{name}.png")
|
||
images.append({"id": name, "page": page_no, "bbox": b.bbox,
|
||
"filename": f"{name}.png", "type": b.type,
|
||
"source_block": b.id, "text": b.text,
|
||
"reading_order": b.reading_order})
|
||
return images
|
||
|
||
|
||
def add_manual_region(doc_dir: Path, layout: dict[str, Any], page_no: int,
|
||
bbox: Sequence[float], kind: str, text: str = "",
|
||
vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> dict[str, Any]:
|
||
"""Add a region the user drew by hand -- for text or a figure OCR/layout
|
||
detection missed entirely, not for correcting one already found (use
|
||
review's tick/untick or reocr_page for that).
|
||
|
||
kind "image" crops the region into its own figure file and adds it to
|
||
layout["images"]. kind "text" adds a new Block: if `text` is supplied it
|
||
is used verbatim (the user typed it themselves), otherwise the crop is
|
||
sent through the vision OCR chain if one is configured -- best-effort,
|
||
the block is still added with a placeholder if that returns nothing,
|
||
since the user drawing this region at all means they judged the text
|
||
important enough to include even without a clean read.
|
||
|
||
Deliberately does not re-run classify_blocks on the new region: the user
|
||
is drawing it precisely because automatic detection already failed on
|
||
it, so re-applying the same heuristics would not be trustworthy. Text is
|
||
trusted as stated (BLOCK_BODY, narratable); explicit review labels map to
|
||
their matching semantic block types and only then receive a fresh
|
||
reading-order position on its page.
|
||
"""
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
raise ValueError("Unknown page")
|
||
if len(bbox) != 4 or bbox[2] <= 0 or bbox[3] <= 0:
|
||
raise ValueError("bbox must be [x, y, width, height] with a positive size")
|
||
layout = dict(layout)
|
||
|
||
if kind == "image":
|
||
images_dir = doc_dir / "images"
|
||
images_dir.mkdir(parents=True, exist_ok=True)
|
||
existing = list(images_dir.glob(f"p{page_no:04d}-manual-*.png"))
|
||
name = f"p{page_no:04d}-manual-{len(existing):03d}"
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, images_dir / f"{name}.png")
|
||
images = list(layout.get("images", []))
|
||
# Placed after every auto-detected image on the page by reading
|
||
# order -- a reasonable default for a figure the pipeline missed;
|
||
# the user can still see and move past it like any other figure.
|
||
page_images = [i for i in images if i["page"] == page_no]
|
||
next_ro = max([i.get("reading_order", 0) for i in page_images], default=0) + 1
|
||
images.append({"id": name, "page": page_no, "bbox": list(bbox),
|
||
"filename": f"{name}.png", "reading_order": next_ro})
|
||
layout["images"] = images
|
||
return layout
|
||
|
||
manual_type, region_label = {
|
||
"text": (BLOCK_BODY, None),
|
||
"heading": (BLOCK_HEADING, None),
|
||
"page_number": (BLOCK_FURNITURE, "page_number"),
|
||
"source": (BLOCK_FURNITURE, "source"),
|
||
"ad": (BLOCK_AD, None),
|
||
}.get(kind, (None, None))
|
||
if manual_type is None:
|
||
raise ValueError("Unsupported manual region kind")
|
||
|
||
final_text = text.strip()
|
||
if not final_text and vision_ocr_url and vision_ocr_model:
|
||
crop_path = doc_dir / "pages" / f".manual-crop-{page_no}.png"
|
||
try:
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, crop_path)
|
||
engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
regions = vision_ocr_page_chain(crop_path, engines, timeout=timeout)
|
||
if regions:
|
||
final_text = " ".join(r["text"] for r in regions).strip()
|
||
finally:
|
||
crop_path.unlink(missing_ok=True)
|
||
if not final_text:
|
||
final_text = "(no text recognised -- click to edit)"
|
||
|
||
blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
existing_ids = {b.id for b in blocks}
|
||
i = 0
|
||
while f"m{page_no:04d}-{i:03d}" in existing_ids:
|
||
i += 1
|
||
new_block = Block(id=f"m{page_no:04d}-{i:03d}", page=page_no, bbox=list(bbox),
|
||
type=manual_type, text=final_text, confidence=0.5, source="manual",
|
||
region_label=region_label)
|
||
|
||
page_blocks = [b for b in blocks if b.page == page_no] + [new_block]
|
||
assign_reading_order(page_blocks, page_meta["width"])
|
||
others = [b for b in blocks if b.page != page_no]
|
||
layout["blocks"] = [b.to_json() for b in others + page_blocks]
|
||
return layout
|
||
|
||
|
||
def extract_region_text(doc_dir: Path, layout: dict[str, Any], page_no: int,
|
||
bbox: Sequence[float], vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> str:
|
||
"""OCR a user-selected rectangle without changing the saved layout.
|
||
|
||
Used for metadata fields such as article title/author: unlike a manual
|
||
region, the selection is a one-off extraction and must not become prose
|
||
that is later narrated.
|
||
"""
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta or len(bbox) != 4 or bbox[2] <= 0 or bbox[3] <= 0:
|
||
raise ValueError("Invalid page selection")
|
||
with tempfile.NamedTemporaryFile(prefix=f".extract-{page_no}-", suffix=".png",
|
||
dir=doc_dir / "pages", delete=False) as tmp:
|
||
crop_path = Path(tmp.name)
|
||
try:
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, crop_path)
|
||
regions = None
|
||
if vision_ocr_url and vision_ocr_model:
|
||
regions = vision_ocr_page_chain(crop_path, [
|
||
(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model),
|
||
], timeout=timeout)
|
||
if regions:
|
||
return " ".join(r.get("text", "") for r in regions if r.get("text")).strip()
|
||
return " ".join(p.get("text", "") for p in ocr_page(crop_path) if p.get("text")).strip()
|
||
finally:
|
||
crop_path.unlink(missing_ok=True)
|
||
|
||
|
||
def update_region(doc_dir: Path, layout: dict[str, Any], page_no: int, block_id: str,
|
||
bbox: Sequence[float] | None, reocr: bool = False,
|
||
vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> tuple[dict[str, Any], str | None]:
|
||
"""Move/resize a region the layout engine already found, after the user
|
||
dragged its box to correct a misplaced or mis-sized detection.
|
||
|
||
Not for a region detection missed entirely (add_manual_region covers
|
||
that) -- block_id must already exist in this page's blocks or images.
|
||
A text block can also be re-OCRed at its corrected geometry (`reocr`);
|
||
an image region is always re-cropped from the page render when its bbox
|
||
changes, since its saved file is exactly that crop. Returns the updated
|
||
layout and, when reocr produced text, that text (None otherwise, so the
|
||
caller can tell "no re-OCR requested" apart from "re-OCR found nothing").
|
||
"""
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
raise ValueError("Unknown page")
|
||
if bbox is not None and (len(bbox) != 4 or bbox[2] <= 0 or bbox[3] <= 0):
|
||
raise ValueError("bbox must be [x, y, width, height] with a positive size")
|
||
layout = dict(layout)
|
||
|
||
blocks = list(layout.get("blocks", []))
|
||
block_idx = next((i for i, b in enumerate(blocks) if b["id"] == block_id), None)
|
||
if block_idx is not None:
|
||
block = dict(blocks[block_idx])
|
||
if bbox is not None:
|
||
block["bbox"] = list(bbox)
|
||
new_text = None
|
||
if reocr:
|
||
new_text = extract_region_text(
|
||
doc_dir, layout, page_no, block["bbox"],
|
||
vision_ocr_url, vision_ocr_model,
|
||
vision_ocr_fallback_url, vision_ocr_fallback_model, timeout)
|
||
if new_text:
|
||
block["text"] = new_text
|
||
blocks[block_idx] = block
|
||
layout["blocks"] = blocks
|
||
return layout, new_text
|
||
|
||
images = list(layout.get("images", []))
|
||
image_idx = next((i for i, im in enumerate(images) if im["id"] == block_id), None)
|
||
if image_idx is None:
|
||
raise ValueError("Region not found on this page")
|
||
image = dict(images[image_idx])
|
||
if bbox is not None:
|
||
image["bbox"] = list(bbox)
|
||
filename = image.get("filename") or f"{block_id}.png"
|
||
(doc_dir / "images").mkdir(parents=True, exist_ok=True)
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, doc_dir / "images" / filename)
|
||
image["filename"] = filename
|
||
images[image_idx] = image
|
||
layout["images"] = images
|
||
return layout, None
|
||
|
||
|
||
# What a region can be changed into from the review screen: text kinds map to a block type (and,
|
||
# for furniture, which kind of furniture); "image" turns the region into a figure crop.
|
||
REGION_TYPES: dict[str, tuple[str, str]] = {
|
||
"body": (BLOCK_BODY, ""),
|
||
"title": (BLOCK_TITLE, ""),
|
||
"heading": (BLOCK_HEADING, ""),
|
||
"deck": (BLOCK_DECK, ""),
|
||
"byline": (BLOCK_BYLINE, ""),
|
||
"caption": (BLOCK_CAPTION, ""),
|
||
"pullquote": (BLOCK_PULLQUOTE, ""),
|
||
"sidebar": (BLOCK_SIDEBAR, ""),
|
||
"page_number": (BLOCK_FURNITURE, "page_number"),
|
||
"source": (BLOCK_FURNITURE, "source"),
|
||
"ad": (BLOCK_AD, ""),
|
||
"image": ("", ""),
|
||
}
|
||
|
||
|
||
def change_region_type(doc_dir: Path, layout: dict[str, Any], page_no: int, region_id: str,
|
||
new_type: str,
|
||
vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> tuple[dict[str, Any], str]:
|
||
"""Change what a detected region is. Returns the new layout and the region's id afterwards.
|
||
|
||
Between text kinds only the type changes; the text and the box stay. Text to image crops the
|
||
box into a figure and drops the text block; image to text reads the box with OCR (a placeholder
|
||
when nothing is recognised, as for a hand-drawn region) and drops the figure. The id changes
|
||
when a region crosses between text and image, so the caller can move article membership over.
|
||
The page's reading order is recomputed, since a type decides where a region is read.
|
||
"""
|
||
if new_type not in REGION_TYPES:
|
||
raise ValueError(f"Unknown region type: {new_type}")
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
raise ValueError("Unknown page")
|
||
block = next((b for b in layout.get("blocks", []) if b["id"] == region_id and b["page"] == page_no), None)
|
||
image = next((im for im in layout.get("images", []) if im["id"] == region_id and im["page"] == page_no), None)
|
||
if block is None and image is None:
|
||
raise ValueError("Region not found on this page")
|
||
|
||
if new_type == "image":
|
||
if image is not None:
|
||
return layout, region_id
|
||
before = {im["id"] for im in layout.get("images", [])}
|
||
layout = add_manual_region(doc_dir, layout, page_no, block["bbox"], "image")
|
||
new_id = next(im["id"] for im in layout["images"] if im["id"] not in before)
|
||
return _reorder_page(delete_region(layout, region_id), page_no), new_id
|
||
|
||
block_type, region_label = REGION_TYPES[new_type]
|
||
if block is not None:
|
||
layout = dict(layout)
|
||
layout["blocks"] = [dict(b, type=block_type, region_label=region_label, confidence=1.0)
|
||
if b["id"] == region_id else b for b in layout.get("blocks", [])]
|
||
return _reorder_page(layout, page_no), region_id
|
||
|
||
# Image to text: read the box, then give the new block the chosen type.
|
||
before = {b["id"] for b in layout.get("blocks", [])}
|
||
layout = add_manual_region(doc_dir, delete_region(layout, region_id), page_no, image["bbox"], "text", "",
|
||
vision_ocr_url, vision_ocr_model,
|
||
vision_ocr_fallback_url, vision_ocr_fallback_model, timeout)
|
||
new_id = next(b["id"] for b in layout["blocks"] if b["id"] not in before)
|
||
layout["blocks"] = [dict(b, type=block_type, region_label=region_label) if b["id"] == new_id else b
|
||
for b in layout["blocks"]]
|
||
return _reorder_page(layout, page_no), new_id
|
||
|
||
|
||
def recompute_reading_order(layout: dict[str, Any]) -> tuple[dict[str, Any], list[int]]:
|
||
"""Re-derive the reading order of every page in an issue, without any OCR.
|
||
|
||
For an issue ingested before the column detection improved. Only the order changes: text,
|
||
boxes, types, articles and approvals stay as they are, and a page whose order the reader set by
|
||
hand is left alone. Returns the layout and the pages whose order actually changed.
|
||
"""
|
||
changed: list[int] = []
|
||
for page_meta in layout.get("pages", []):
|
||
page_no = page_meta["page"]
|
||
before = [b["id"] for b in sorted((b for b in layout.get("blocks", []) if b["page"] == page_no),
|
||
key=lambda b: b.get("reading_order", 0))]
|
||
layout = _reorder_page(layout, page_no)
|
||
after = [b["id"] for b in sorted((b for b in layout.get("blocks", []) if b["page"] == page_no),
|
||
key=lambda b: b.get("reading_order", 0))]
|
||
if before != after:
|
||
changed.append(page_no)
|
||
return layout, changed
|
||
|
||
|
||
def _reorder_page(layout: dict[str, Any], page_no: int) -> dict[str, Any]:
|
||
"""Recompute one page's reading order -- unless the reader fixed that order by hand."""
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta or page_meta.get("order_locked"):
|
||
return layout
|
||
blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
page_blocks = [b for b in blocks if b.page == page_no]
|
||
assign_reading_order(page_blocks, page_meta["width"])
|
||
layout = dict(layout)
|
||
layout["blocks"] = [b.to_json() for b in [b for b in blocks if b.page != page_no] + page_blocks]
|
||
return layout
|
||
|
||
|
||
def delete_region(layout: dict[str, Any], block_id: str) -> dict[str, Any]:
|
||
"""Remove a region the layout engine got wrong entirely (not just
|
||
misplaced). An article referencing this id in its own blocks/images
|
||
list is left as-is -- article_text() and the review screen already
|
||
treat an id no longer present in the layout as simply absent, the same
|
||
way a page re-OCR can already drop ids an article still lists."""
|
||
layout = dict(layout)
|
||
layout["blocks"] = [b for b in layout.get("blocks", []) if b["id"] != block_id]
|
||
layout["images"] = [im for im in layout.get("images", []) if im["id"] != block_id]
|
||
return layout
|
||
|
||
|
||
def save_layout(doc_dir: Path, layout: dict[str, Any]) -> None:
|
||
(doc_dir / "layout.json").write_text(
|
||
json.dumps(layout, ensure_ascii=False), "utf-8")
|
||
|
||
|
||
def load_layout(doc_dir: Path) -> dict[str, Any]:
|
||
path = doc_dir / "layout.json"
|
||
if not path.is_file():
|
||
return {}
|
||
try:
|
||
return json.loads(path.read_text("utf-8"))
|
||
except (ValueError, OSError):
|
||
return {}
|
||
|
||
|
||
def save_articles(doc_dir: Path, articles: list[dict[str, Any]]) -> None:
|
||
(doc_dir / "articles.json").write_text(
|
||
json.dumps({"articles": articles}, ensure_ascii=False), "utf-8")
|
||
|
||
|
||
def load_articles(doc_dir: Path) -> list[dict[str, Any]]:
|
||
path = doc_dir / "articles.json"
|
||
if not path.is_file():
|
||
return []
|
||
try:
|
||
return json.loads(path.read_text("utf-8")).get("articles", [])
|
||
except (ValueError, OSError):
|
||
return []
|
||
|
||
|
||
BLOCK_FIGURE_DESCRIPTION = "figure_description"
|
||
|
||
|
||
def article_text(layout: dict[str, Any], article: dict[str, Any],
|
||
include_captions: bool = False,
|
||
include_descriptions: bool = False) -> list[dict[str, Any]]:
|
||
"""The narratable text of one article, in reading order.
|
||
|
||
Returned per block so the player can highlight the source region and jump
|
||
to the originating PDF page.
|
||
|
||
A figure/chart/table/code region has no text block of its own to
|
||
narrate -- see describe_visual_region() -- so an already-described
|
||
image belonging to this article is synthesised into a
|
||
BLOCK_FIGURE_DESCRIPTION entry at that image's own reading position when
|
||
`include_descriptions` is set. Optional and off by default: generating
|
||
a description costs a vision-model call the user may not have wanted
|
||
for every figure, and a description read aloud mid-article changes the
|
||
listening experience enough that it should be an explicit choice, not a
|
||
silent side effect of one being available.
|
||
"""
|
||
wanted = set(article.get("blocks") or [])
|
||
types = set(NARRATABLE) | ({BLOCK_CAPTION} if include_captions else set())
|
||
blocks = [
|
||
dict(b, text=strip_continuation_markers(b["text"]))
|
||
for b in layout.get("blocks", [])
|
||
if b["id"] in wanted and b["type"] in types
|
||
]
|
||
if include_descriptions:
|
||
wanted_images = set(article.get("images") or [])
|
||
for img in layout.get("images", []):
|
||
description = (img.get("description") or "").strip()
|
||
if img["id"] not in wanted_images or not description:
|
||
continue
|
||
blocks.append({
|
||
"id": f"desc-{img['id']}", "page": img["page"], "bbox": img["bbox"],
|
||
"type": BLOCK_FIGURE_DESCRIPTION, "text": description,
|
||
"reading_order": img.get("reading_order", 0),
|
||
})
|
||
blocks.sort(key=lambda b: (b["page"], b["reading_order"]))
|
||
return blocks
|