mARTin-B78
813e06c914
Magazine P1:
- Images shown automatically during playback (assign_images_to_articles
links each figure to the article whose column it falls in; the client
shows the last image reached in reading order as playback advances).
- "Play all articles" queues every approved article and advances
automatically at the end of each rather than stopping.
- Word-level highlighting, explicitly documented as an ESTIMATE (no
forced-alignment/timestamp source exists): each word's share of the
sentence's real measured audio duration is proportional to its
character length. A checkbox turns it off; sentence highlighting
(measured, not estimated) is unaffected.
- Driving mode: full-screen overlay, two oversized controls, reuses
whatever is already playing/queued.
- Sidebars/pull-out boxes excluded from narration automatically
(mark_sidebars: an anomalously narrow column vs. the page's regular
grid). Fixed a real defect found while building this: a full-width
headline was left in the column-width analysis and could bridge two
genuinely separate columns into one, hiding a real sidebar column
entirely -- caught by a failing test before the fix.
Second OCR engine + fallback ladder:
- KDL Frontier Parser nano wired in via vision_ocr_page_chain (ordered
engine list, first usable result wins, tesseract last resort).
Confirmed live: ~3x faster than Baidu (~12s vs ~40s/page) but returns
plain text with NO coordinate markers, unlike Baidu's labeled regions.
_whole_page_region treats the whole response as one ambiguous
full-page block (gated by ocr_quality_score) rather than discarding
usable text just because it lacks layout structure.
OCR engines page + a real UI bug:
- The OCR & Document Parsing page had no .s-engines-page[data-page="ocr"]
markup at all -- renderLocalContainers already supported an OCR grid
generically, but $('dc-grid-ocr') was always null, so the card grid
silently never rendered. Added an Active OCR Engine panel mirroring
Active Language Model exactly, plus the Local grid -- both configured
engines now appear as ready-made cards, no "Add Custom" dialog needed.
- Fixed the audiobook pipeline's "Previous/Next" nav bar bleeding into
every .section-head in the document (confirmed live on the Engines
page: a "Cast Audiobook" button and "STEP 1/8" hint that made no sense
there). Changed from a blocklist (naming each unrelated section as
noticed) to an allowlist of the pipeline's three actual destinations.
Verified live end-to-end: play-all queue building and manual advance,
word-span markup, driving-mode toggle, OCR page showing both engines as
local cards, zero stray nav bars on the Engines page, no console errors.
20 new backend tests (156 total in the magazine module).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1375 lines
57 KiB
Python
1375 lines
57 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 re
|
||
import statistics
|
||
import subprocess
|
||
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
|
||
|
||
# ── 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_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"
|
||
|
||
NARRATABLE = (BLOCK_TITLE, 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
|
||
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.035) -> 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.
|
||
"""
|
||
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]
|
||
|
||
|
||
def assign_reading_order(blocks: list[Block], page_width: float) -> list[Block]:
|
||
"""Order blocks the way a person reads the page: column by column.
|
||
|
||
Sorting purely by y interleaves parallel columns, which is precisely the
|
||
failure that splices one article's sentence into another's.
|
||
"""
|
||
columns = detect_columns(blocks, page_width)
|
||
|
||
def column_of(b: Block) -> int:
|
||
centre = b.bbox[0] + b.bbox[2] / 2
|
||
for i, (left, right) in enumerate(columns):
|
||
if left <= centre <= right:
|
||
return i
|
||
# Spanning blocks (a headline across all columns) belong to the first
|
||
# column they touch, so they precede the body they introduce.
|
||
for i, (left, right) in enumerate(columns):
|
||
if bbox_overlaps(b.bbox, [left, b.bbox[1], right - left, b.bbox[3]]):
|
||
return i
|
||
return len(columns)
|
||
|
||
for b in blocks:
|
||
b.column = column_of(b)
|
||
|
||
# A full-width headline must not be dragged to the bottom of column 0, so
|
||
# blocks spanning most of the page sort by y before anything below them.
|
||
def key(b: Block) -> tuple[float, int, float]:
|
||
spans_page = b.bbox[2] > page_width * 0.6
|
||
return (0 if spans_page else 1, b.column, round(b.bbox[1], 1))
|
||
|
||
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.
|
||
"""
|
||
if not any(b.type == BLOCK_AD 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):
|
||
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"
|
||
) 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 = {
|
||
"footer": BLOCK_FURNITURE,
|
||
"header": BLOCK_FURNITURE,
|
||
"page_number": BLOCK_FURNITURE,
|
||
"title": BLOCK_TITLE,
|
||
"heading": BLOCK_TITLE,
|
||
"caption": BLOCK_CAPTION,
|
||
}
|
||
|
||
# One label + its four-number bbox, anchoring a region; content runs until
|
||
# the next such marker. Confirmed live against Baidu Unlimited-OCR's actual
|
||
# output shape: "text [241, 90, 588, 249]Battery technology is entering...".
|
||
_VISION_REGION = re.compile(r"(?m)^(\w+)\s*\[\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*\]")
|
||
|
||
|
||
def ocr_quality_score(text: str) -> float:
|
||
"""Rough plausibility that OCR/extracted text is real prose, 0..1.
|
||
|
||
Coarse by design: it catches text that is mostly symbol noise (confirmed
|
||
live on tesseract line-height artifacts -- ". |", "@)X" scored 0.0/0.33)
|
||
well enough to gate an expensive vision-OCR escalation, but it cannot and
|
||
is not meant to catch a plausible-looking character substitution inside
|
||
an otherwise normal sentence (confirmed live: "K | soll Arbeit abnehmen"
|
||
-- a single misread "KI" -- still scored 0.81, since 7 of its 8 words are
|
||
genuine German). Catching that class of error needs a dictionary or
|
||
language-model check, which is out of scope for a cheap heuristic.
|
||
"""
|
||
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 with no
|
||
per-line data (a vision OCR engine gives one box per paragraph, not per
|
||
line, unlike tesseract -- see ocr_page()).
|
||
|
||
Derived from the standard area relationship for a wrapped paragraph:
|
||
box_height ~= (text_length / (box_width / (char_width_ratio*size))) *
|
||
(line_height_ratio*size), which solves to size ~= sqrt(area / (text_length
|
||
* char_width_ratio * line_height_ratio)). Using total bbox height directly
|
||
(as if it were one line) would scale with paragraph length instead of
|
||
type size -- the same bug already fixed for tesseract's line grouping.
|
||
"""
|
||
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.
|
||
|
||
Tolerates a preamble the model sometimes prepends before the first region
|
||
(confirmed live: instruction-echoing junk, and the literal decoy phrase
|
||
"[No text detected]" ahead of real output) by only ever anchoring on the
|
||
"label [x0, y0, x1, y1]" marker -- anything before the first match, or
|
||
between calls with zero matches, is discarded rather than misread as
|
||
content.
|
||
"""
|
||
matches = list(_VISION_REGION.finditer(raw or ""))
|
||
regions = []
|
||
for i, m in enumerate(matches):
|
||
label = m.group(1).lower()
|
||
x0, y0, x1, y1 = (int(m.group(g)) for g in (2, 3, 4, 5))
|
||
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 or x1 <= x0 or y1 <= y0:
|
||
continue
|
||
regions.append({
|
||
"label": label,
|
||
"bbox": [float(x0), float(y0), float(x1 - x0), float(y1 - y0)],
|
||
"text": text,
|
||
})
|
||
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()
|
||
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}]
|
||
|
||
|
||
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
|
||
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": [
|
||
{"type": "text", "text": (
|
||
"Transcribe all visible text on this page verbatim, preserving "
|
||
"paragraph breaks. Output only the transcription, no commentary."
|
||
)},
|
||
{"type": "image_url", "image_url": {
|
||
"url": f"data:image/png;base64,{image_b64}",
|
||
}},
|
||
]}],
|
||
"max_tokens": 4000,
|
||
"temperature": 0.0,
|
||
}
|
||
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, ValueError, KeyError, IndexError):
|
||
return None
|
||
|
||
regions = parse_vision_ocr_regions(content)
|
||
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
|
||
|
||
|
||
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",
|
||
)
|
||
out.append(block)
|
||
return out
|
||
|
||
|
||
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,
|
||
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 = 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
|
||
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:
|
||
progress(page_no, doc.page_count)
|
||
|
||
# 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:
|
||
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 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 []
|
||
|
||
|
||
def article_text(layout: dict[str, Any], article: dict[str, Any],
|
||
include_captions: 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.
|
||
"""
|
||
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
|
||
]
|
||
blocks.sort(key=lambda b: (b["page"], b["reading_order"]))
|
||
return blocks
|