mARTin-B78
477cb43d4a
A page needing OCR (or whose already-extracted text -- native or
tesseract alike -- scores as mostly garbage via ocr_quality_score) is
now escalated to a connected vision OCR engine before falling back to
tesseract. Falls back automatically on any failure (unreachable engine,
timeout, unparseable output); never fatal to ingestion.
Verified live against ocr-baidu-unlimited-ocr on a real 28-page scanned
c't magazine excerpt (Epson ES-580W, no scanner OCR): every garbled
article title tesseract produced ("K | soll Arbeit abnehmen", ". |",
"@)X", "Owns mee") was replaced by the real headline ("1. Irrtum: KI
ist ein allgemeiner Jobkiller", etc.), and furniture detection jumped
from 44 to 151 blocks because the engine labels running headers/
footers/page numbers directly instead of needing fuzzy post-hoc
matching. 26 of 28 pages used the vision path; the remaining 2 fell
back to tesseract automatically.
Also fixes two real defects found investigating the same real-world
data, both regression-tested against the actual garbled strings
observed:
- Exact-match repeated-line detection (running header/footer) silently
failed once tesseract OCR'd the same line differently each time --
cluster_running_furniture fuzzy-matches near-edge blocks instead.
- ocr_page() built one Block per OCR text-LINE rather than per
paragraph, so a single anomalous line (a chart number, scan noise, a
wrapped headline continuation) could independently trigger title
classification -- now grouped by paragraph like the native PyMuPDF
path, using the paragraph's median line height rather than its total
bbox height for font size.
A vision-engine label (footer/title/page_number/caption) is trusted
outright and skips classify_blocks' font-size heuristics entirely,
since the engine gives no reliable font metric; an ambiguous "text"
region gets a synthesized estimate from its paragraph area and length
(estimate_font_size_from_area) and still goes through normal
classification.
Off unless vision_ocr_url/vision_ocr_model are configured
(vision_ocr_enabled setting, default on if reachable); per-request
override via the ingest endpoint body. Default engine URL uses
host.docker.internal, confirmed live -- the OCR container runs on its
own Docker network, not dgx_net, so a container-name default would not
resolve from the app container.
28 new tests (136 total in the module). ocr_quality_score is
deliberately coarse and documented as such: it catches text that is
mostly symbol noise, not a plausible single-word misread inside an
otherwise normal sentence (confirmed live: a misread "KI" -> "K |"
still scored 0.81) -- that class of error needs a dictionary or
language-model check, out of scope here.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1188 lines
48 KiB
Python
1188 lines
48 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 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.
|
||
contributed = False
|
||
for b in page_blocks:
|
||
if b.type in (BLOCK_FURNITURE, BLOCK_AD):
|
||
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 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 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 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_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 tesseract only as the fallback if the call fails or the
|
||
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.
|
||
|
||
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_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]] = []
|
||
for img_index, info in enumerate(page.get_images(full=True)):
|
||
try:
|
||
for r in page.get_image_rects(info[0]):
|
||
image_boxes.append([r.x0, r.y0, r.width, r.height])
|
||
except Exception:
|
||
continue
|
||
_extract_image(fitz, doc, info[0], images_dir,
|
||
f"p{page_no:04d}-{img_index:02d}")
|
||
|
||
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(png, vision_ocr_url, vision_ocr_model,
|
||
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,
|
||
})
|
||
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"])
|
||
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),
|
||
})
|
||
|
||
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],
|
||
"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) -> None:
|
||
"""Save one embedded image, skipping the tiny rules and spacers."""
|
||
try:
|
||
pix = fitz.Pixmap(doc, xref)
|
||
if pix.width < 64 or pix.height < 64:
|
||
return
|
||
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))
|
||
except Exception:
|
||
return
|
||
|
||
|
||
# ── 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)
|
||
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
|