1808 lines
79 KiB
Python
1808 lines
79 KiB
Python
"""Layout-aware magazine ingestion.
|
||
|
||
Magazines are not books: a single page routinely carries several unrelated
|
||
articles, advertisements, sidebars and pull quotes, and an article often
|
||
continues twenty pages later. Treating a page as the unit of narration
|
||
therefore produces audio that jumps mid-sentence into an advert. The unit
|
||
here is the *article*, reconstructed from classified layout blocks.
|
||
|
||
Pipeline (see architecture/magazine-ingestion.md):
|
||
|
||
PDF
|
||
-> extract_pages() native text blocks with bbox/font via PyMuPDF
|
||
-> ocr_page() tesseract TSV fallback for scanned pages
|
||
-> classify_blocks() title / deck / byline / body / caption / ...
|
||
-> assign_reading_order() column detection, then top-to-bottom per column
|
||
-> detect_contents() Stage A: locate and parse the table of contents
|
||
-> group_articles() Stage C: group blocks, link continuations
|
||
-> (human review) Stage D happens in the UI; nothing is narrated
|
||
until an article is approved.
|
||
|
||
PyMuPDF is imported lazily inside the I/O functions so that the pure geometry
|
||
and classification helpers -- which hold all the fiddly logic worth testing --
|
||
can be exercised without the native dependency or a sample PDF.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import difflib
|
||
import json
|
||
import logging
|
||
import math
|
||
import re
|
||
import statistics
|
||
import subprocess
|
||
import unicodedata
|
||
|
||
import requests
|
||
|
||
from core.retry import request_with_retries
|
||
from dataclasses import dataclass, field, asdict
|
||
from pathlib import Path
|
||
from typing import Any, Iterable, Sequence
|
||
from urllib.parse import urlsplit
|
||
|
||
# ── Block taxonomy ───────────────────────────────────────────────────────────
|
||
# Ordered roughly by how much they matter to narration. Only NARRATABLE types
|
||
# are ever sent to TTS; the rest are kept for display and provenance.
|
||
BLOCK_TITLE = "title"
|
||
BLOCK_HEADING = "heading"
|
||
BLOCK_DECK = "deck" # standfirst / subtitle under the headline
|
||
BLOCK_BYLINE = "byline"
|
||
BLOCK_BODY = "body"
|
||
BLOCK_CAPTION = "caption"
|
||
BLOCK_PULLQUOTE = "pullquote" # verbatim repeat of body text, blown up
|
||
BLOCK_SIDEBAR = "sidebar"
|
||
BLOCK_FURNITURE = "furniture" # running heads, folios, page numbers
|
||
BLOCK_AD = "ad"
|
||
BLOCK_FIGURE = "figure"
|
||
BLOCK_CHART = "chart"
|
||
BLOCK_TABLE = "table"
|
||
BLOCK_CODE = "code"
|
||
VISUAL_TYPES = (BLOCK_FIGURE, BLOCK_CHART, BLOCK_TABLE, BLOCK_CODE)
|
||
|
||
NARRATABLE = (BLOCK_TITLE, BLOCK_HEADING, BLOCK_DECK, BLOCK_BYLINE, BLOCK_BODY)
|
||
|
||
# Text that marks an article jumping to (or arriving from) another page.
|
||
_CONT_TO = re.compile(
|
||
r"\((?:continued|cont\.?)\s+on\s+page\s+(\d{1,4})\)"
|
||
r"|continued\s+on\s+page\s+(\d{1,4})"
|
||
r"|fortsetzung\s+auf\s+seite\s+(\d{1,4})"
|
||
r"|weiter\s+auf\s+seite\s+(\d{1,4})",
|
||
re.I,
|
||
)
|
||
_CONT_FROM = re.compile(
|
||
r"continued\s+from\s+page\s+(\d{1,4})"
|
||
r"|fortsetzung\s+von\s+seite\s+(\d{1,4})",
|
||
re.I,
|
||
)
|
||
|
||
_BYLINE = re.compile(
|
||
r"^\s*(?:by|von|text|words|story|interview)\s*[: ]\s*(.{2,80})$",
|
||
re.I,
|
||
)
|
||
|
||
# Headings that mark a contents page, and the section headers inside one.
|
||
_TOC_MARKERS = (
|
||
"contents", "table of contents", "in this issue", "inside this issue",
|
||
"features", "departments", "columns", "regulars",
|
||
"inhalt", "inhaltsverzeichnis", "in dieser ausgabe", "rubriken", "themen",
|
||
)
|
||
# "The future of batteries ....... 18" / "The future of batteries 18"
|
||
_TOC_ENTRY = re.compile(r"^(.{3,120}?)[\s.…·•\-]{2,}(\d{1,4})\s*$")
|
||
|
||
_ADVERT_MARKERS = ("advertisement", "anzeige", "sponsored", "promotion")
|
||
|
||
|
||
@dataclass
|
||
class Block:
|
||
"""One laid-out region of a page, in PDF user-space coordinates.
|
||
|
||
bbox is [x, y, width, height] with the origin at the page's top-left, which
|
||
is what the reflowed view needs to draw a highlight over the rendered page.
|
||
"""
|
||
id: str
|
||
page: int
|
||
bbox: list[float]
|
||
type: str
|
||
text: str
|
||
reading_order: int = 0
|
||
confidence: float = 1.0
|
||
font_size: float = 0.0
|
||
font_name: str = ""
|
||
bold: bool = False
|
||
column: int = 0
|
||
region_label: str = ""
|
||
continues_from_previous_page: bool = False
|
||
source: str = "native" # "native" | "ocr" | "vision_ocr" | "vision_label"
|
||
|
||
def to_json(self) -> dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass
|
||
class Article:
|
||
id: str
|
||
title: str
|
||
author: str = ""
|
||
pages: list[int] = field(default_factory=list)
|
||
blocks: list[str] = field(default_factory=list)
|
||
images: list[str] = field(default_factory=list)
|
||
status: str = "detected" # detected | approved | rejected
|
||
confidence: float = 0.0
|
||
section: str = ""
|
||
continued_to: list[int] = field(default_factory=list)
|
||
audio_status: str = "none" # none | narration | podcast | both
|
||
|
||
def to_json(self) -> dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
# ── Geometry helpers (pure) ──────────────────────────────────────────────────
|
||
|
||
def bbox_overlaps(a: Sequence[float], b: Sequence[float], tol: float = 0.0) -> bool:
|
||
"""True when two [x, y, w, h] boxes intersect, allowing a tolerance."""
|
||
ax, ay, aw, ah = a
|
||
bx, by, bw, bh = b
|
||
return not (
|
||
ax + aw + tol < bx or bx + bw + tol < ax
|
||
or ay + ah + tol < by or by + bh + tol < ay
|
||
)
|
||
|
||
|
||
def detect_columns(blocks: Sequence[Block], page_width: float,
|
||
min_gap_ratio: float = 0.015) -> list[tuple[float, float]]:
|
||
"""Infer the column grid from where blocks are *absent*.
|
||
|
||
Magazines vary the column count per page, so a fixed grid is wrong. The
|
||
reliable signal is vertical whitespace running the height of the page: we
|
||
project every block onto the x axis, then split wherever the uncovered gap
|
||
is wider than a few percent of the page. Returns (left, right) spans.
|
||
|
||
1.5%, not the rounder 3.5% this started at: measured on a real scanned
|
||
c't magazine page, the actual gutter between two genuine columns was
|
||
only ~2.1% of the page width -- narrower than the old threshold, which
|
||
made detect_columns see one merged column and, combined with a wide
|
||
caption bridging the same gap (see assign_reading_order), produced
|
||
every block on the page landing in column=0.
|
||
"""
|
||
spans = sorted(
|
||
((b.bbox[0], b.bbox[0] + b.bbox[2]) for b in blocks if b.bbox[2] > 1),
|
||
key=lambda s: s[0],
|
||
)
|
||
if not spans:
|
||
return [(0.0, page_width)]
|
||
|
||
merged: list[list[float]] = [list(spans[0])]
|
||
min_gap = page_width * min_gap_ratio
|
||
for left, right in spans[1:]:
|
||
if left - merged[-1][1] > min_gap:
|
||
merged.append([left, right])
|
||
else:
|
||
merged[-1][1] = max(merged[-1][1], right)
|
||
return [(m[0], m[1]) for m in merged]
|
||
|
||
|
||
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.
|
||
|
||
Column BOUNDARIES are detected from blocks close to the page's typical
|
||
width only. Confirmed live on a real scanned magazine: a caption running
|
||
under a photo, at 49.9% of the page width -- under the "spans the whole
|
||
page" threshold below, so not given special handling -- sat squarely in
|
||
the whitespace gap between two real text columns and bridged them into
|
||
one merged span, which meant EVERY block on the page got column=0 and
|
||
reading order fell back to pure top-to-bottom, splicing the two
|
||
columns' sentences together (the exact failure this function exists to
|
||
prevent). Every block, wide ones included, is still assigned to
|
||
whichever column its centre or overlap lands in once boundaries are
|
||
known -- only the boundary DETECTION excludes them.
|
||
"""
|
||
widths = sorted(b.bbox[2] for b in blocks if b.bbox[2] > 1)
|
||
median_w = widths[len(widths) // 2] if widths else 0
|
||
narrow = [b for b in blocks if median_w <= 0 or b.bbox[2] <= median_w * 1.4]
|
||
columns = detect_columns(narrow if len(narrow) >= 2 else 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.
|
||
"""
|
||
# A small ad label beside substantial prose does not establish a full-page ad.
|
||
if any(b.type == BLOCK_BODY and len(b.text) > 160 for b in blocks):
|
||
return blocks
|
||
if not any(b.type == BLOCK_AD and b.source not in ("vision_label", "llm_ad", "llm_layout") for b in blocks):
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type != BLOCK_FURNITURE:
|
||
b.type, b.confidence = BLOCK_AD, 0.8
|
||
return blocks
|
||
|
||
|
||
def mark_pullquotes(blocks: list[Block]) -> list[Block]:
|
||
"""Demote display text that merely repeats nearby body copy.
|
||
|
||
Narrating a pull quote reads the same sentence twice, a few seconds apart,
|
||
which sounds like a skipping record.
|
||
"""
|
||
body_text = _normalise(" ".join(
|
||
b.text for b in blocks if b.type == BLOCK_BODY
|
||
)).lower()
|
||
if not body_text:
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type not in (BLOCK_DECK, BLOCK_PULLQUOTE):
|
||
continue
|
||
probe = _normalise(b.text).lower().strip(' "“”«»')
|
||
if len(probe) > 25 and probe in body_text:
|
||
b.type, b.confidence = BLOCK_PULLQUOTE, 0.9
|
||
return blocks
|
||
|
||
|
||
def mark_sidebars(blocks: list[Block], page_width: float,
|
||
narrow_ratio: float = 0.7) -> list[Block]:
|
||
"""Reclassify body text sitting in an anomalously narrow column as a
|
||
sidebar, so it is excluded from an article's narration by default rather
|
||
than requiring a manual exclusion in review every time.
|
||
|
||
Geometry-only, like the rest of this module's classification: a sidebar
|
||
or pull-out box is reliably narrower than the page's regular column grid
|
||
(a boxed insert set off from the main flow), which needs no image
|
||
analysis to detect. This is a heuristic, not a certainty -- a genuinely
|
||
narrow-column page design (or a caption sitting in its own slim column)
|
||
can trigger a false positive, which is exactly why review still shows
|
||
every sidebar block and lets it be included with one click rather than
|
||
silently discarding it.
|
||
|
||
Full-width elements (a headline spanning the whole page) are excluded
|
||
from the column analysis here, the same way assign_reading_order treats
|
||
them specially: left in, one would bridge two genuinely separate columns
|
||
into a single merged span (its bbox overlaps both), hiding a real narrow
|
||
sidebar column entirely. This computes its own column partition rather
|
||
than trusting `.column` from assign_reading_order, since that partition
|
||
was built from the FULL block list and its indices would not line up
|
||
with this narrower one.
|
||
"""
|
||
candidates = [b for b in blocks if b.bbox[2] <= page_width * 0.6]
|
||
columns = detect_columns(candidates, page_width)
|
||
if len(columns) < 2:
|
||
return blocks
|
||
widths = [right - left for left, right in columns]
|
||
widest = max(widths)
|
||
narrow_spans = [columns[i] for i, w in enumerate(widths) if w < widest * narrow_ratio]
|
||
if not narrow_spans:
|
||
return blocks
|
||
for b in blocks:
|
||
if b.type != BLOCK_BODY or b.bbox[2] > page_width * 0.6:
|
||
continue
|
||
centre = b.bbox[0] + b.bbox[2] / 2
|
||
if any(left <= centre <= right for left, right in narrow_spans):
|
||
b.type, b.confidence = BLOCK_SIDEBAR, 0.6
|
||
return blocks
|
||
|
||
|
||
def find_repeated_lines(pages_text: Sequence[Sequence[str]],
|
||
threshold: float = 0.25) -> set[str]:
|
||
"""Lines appearing near the same edge on many pages: page furniture."""
|
||
if not pages_text:
|
||
return set()
|
||
counts: dict[str, int] = {}
|
||
for lines in pages_text:
|
||
for line in {_normalise(t) for t in lines if _normalise(t)}:
|
||
counts[line] = counts.get(line, 0) + 1
|
||
minimum = max(3, int(len(pages_text) * threshold))
|
||
# Long lines that recur are more likely a repeated standfirst than a
|
||
# running head, so cap the length.
|
||
return {t for t, n in counts.items() if n >= minimum and len(t) <= 80}
|
||
|
||
|
||
def looks_like_real_title(text: str) -> bool:
|
||
"""Guards the headline-fallback article start against OCR noise.
|
||
|
||
A stray misread character or chart-axis fragment ("I", "@)X", ". |") can
|
||
classify as BLOCK_TITLE by size alone; without this it would spawn its
|
||
own article. Confirmed live on a scanned magazine: single-character OCR
|
||
artifacts at running-footer size otherwise start empty articles.
|
||
"""
|
||
letters = sum(1 for c in (text or "") if c.isalpha())
|
||
return letters >= 4
|
||
|
||
|
||
def _digit_stripped(text: str) -> str:
|
||
return re.sub(r"\d+", "", text or "")
|
||
|
||
|
||
def cluster_running_furniture(blocks: Sequence[Block], page_heights: dict[int, float],
|
||
similarity: float = 0.6, min_count: int = 3) -> None:
|
||
"""Reclassify near-edge blocks that recur across pages as furniture, even
|
||
when OCR reads the running header/footer differently each time.
|
||
|
||
find_repeated_lines() needs an EXACT match, which OCR defeats: a magazine
|
||
tag like "c't Wissen 2026" came back as "cE KI-WISSEN", "ce Wissen 2136",
|
||
"cA Wissen 2036" across different pages of the same scan -- confirmed
|
||
live, and the dominant cause of a 28-page scan producing 26 near-1-per-
|
||
page "articles" instead of grouping around real headlines. Only the page
|
||
number reliably varies page to page, so digit-stripped fuzzy similarity
|
||
catches what exact matching misses. Mutates blocks in place.
|
||
"""
|
||
candidates: list[tuple[Block, str]] = []
|
||
for b in blocks:
|
||
if b.type in (BLOCK_FURNITURE, BLOCK_AD):
|
||
continue
|
||
text = _normalise(b.text)
|
||
if not text or len(text) > 80:
|
||
continue
|
||
height = page_heights.get(b.page, 842.0)
|
||
near_top = b.bbox[1] < height * 0.07
|
||
near_bottom = b.bbox[1] + b.bbox[3] > height * 0.93
|
||
if not (near_top or near_bottom):
|
||
continue
|
||
key = _digit_stripped(text).lower().strip()
|
||
if len(key) < 3:
|
||
continue
|
||
candidates.append((b, key))
|
||
|
||
assigned = [False] * len(candidates)
|
||
for i, (_, key_i) in enumerate(candidates):
|
||
if assigned[i]:
|
||
continue
|
||
cluster = [i]
|
||
for j in range(i + 1, len(candidates)):
|
||
if assigned[j]:
|
||
continue
|
||
if difflib.SequenceMatcher(None, key_i, candidates[j][1]).ratio() >= similarity:
|
||
cluster.append(j)
|
||
if len(cluster) >= min_count:
|
||
for idx in cluster:
|
||
candidates[idx][0].type = BLOCK_FURNITURE
|
||
candidates[idx][0].confidence = 0.75
|
||
assigned[idx] = True
|
||
|
||
|
||
# ── Stage A: the table of contents ───────────────────────────────────────────
|
||
|
||
def score_contents_page(lines: Sequence[str]) -> float:
|
||
"""How strongly one page looks like a table of contents (0..1)."""
|
||
if not lines:
|
||
return 0.0
|
||
joined = " ".join(lines).lower()
|
||
marker_hits = sum(1 for m in _TOC_MARKERS if m in joined)
|
||
entry_hits = sum(1 for line in lines if _TOC_ENTRY.match(_normalise(line)))
|
||
# Either signal alone is weak: a feature can be titled "Features", and a
|
||
# dotted leader appears in ordinary tables too.
|
||
score = min(1.0, marker_hits * 0.25) * 0.4 + min(1.0, entry_hits / 6) * 0.6
|
||
return round(score, 3)
|
||
|
||
|
||
def parse_contents(lines: Sequence[str]) -> list[dict[str, Any]]:
|
||
"""Pull (title, page, section) entries out of a contents page."""
|
||
entries: list[dict[str, Any]] = []
|
||
section = ""
|
||
for raw in lines:
|
||
line = _normalise(raw)
|
||
if not line:
|
||
continue
|
||
low = line.lower().strip(": ")
|
||
if low in _TOC_MARKERS and low not in ("contents", "inhalt"):
|
||
section = line.strip(": ")
|
||
continue
|
||
m = _TOC_ENTRY.match(line)
|
||
if not m:
|
||
continue
|
||
title, page = m.group(1).strip(" .…·•-"), int(m.group(2))
|
||
if len(title) < 3 or page <= 0 or page > 2000:
|
||
continue
|
||
entries.append({"title": title, "page": page, "section": section})
|
||
return entries
|
||
|
||
|
||
# ── Stage C: grouping blocks into articles ───────────────────────────────────
|
||
|
||
def _title_key(text: str) -> str:
|
||
return re.sub(r"[^a-z0-9]+", "", _normalise(text).lower())
|
||
|
||
|
||
def group_articles(blocks: Sequence[Block],
|
||
contents: Sequence[dict[str, Any]],
|
||
page_count: int,
|
||
skip_pages: set[int] | None = None) -> list[Article]:
|
||
"""Group ordered blocks into articles.
|
||
|
||
The contents listing, when present, is the strongest available signal: it
|
||
states both the title and the starting page, so it anchors boundaries that
|
||
layout alone would guess at. Pages not claimed by any listing still get
|
||
articles from their own headlines, because a contents page never lists
|
||
everything (and some issues have no contents page at all).
|
||
"""
|
||
by_page: dict[int, list[Block]] = {}
|
||
for b in blocks:
|
||
by_page.setdefault(b.page, []).append(b)
|
||
for page_blocks in by_page.values():
|
||
page_blocks.sort(key=lambda b: b.reading_order)
|
||
|
||
# Where does each article start? A contents entry pins a start page; a
|
||
# title block on a page with no entry starts one there.
|
||
starts: dict[int, dict[str, Any]] = {}
|
||
for entry in contents:
|
||
page = entry["page"]
|
||
if 1 <= page <= page_count:
|
||
starts.setdefault(page, {
|
||
"title": entry["title"],
|
||
"section": entry.get("section", ""),
|
||
"confidence": 0.9,
|
||
})
|
||
|
||
skip_pages = skip_pages or set()
|
||
for page in sorted(by_page):
|
||
if page in starts or page in skip_pages:
|
||
continue
|
||
# A page carrying nothing but advertising and page furniture has no
|
||
# article on it -- the cover and the ad pages otherwise each become a
|
||
# spurious entry the reviewer has to reject by hand on every issue.
|
||
if not any(b.type in NARRATABLE and b.type != BLOCK_TITLE
|
||
for b in by_page[page]):
|
||
continue
|
||
titles = [b for b in by_page[page] if b.type == BLOCK_TITLE
|
||
and looks_like_real_title(b.text)]
|
||
if titles:
|
||
starts[page] = {
|
||
"title": _normalise(titles[0].text),
|
||
"section": "",
|
||
"confidence": 0.55,
|
||
}
|
||
|
||
if not starts:
|
||
return []
|
||
|
||
ordered_starts = sorted(starts)
|
||
articles: list[Article] = []
|
||
for i, start_page in enumerate(ordered_starts):
|
||
end_page = (ordered_starts[i + 1] - 1) if i + 1 < len(ordered_starts) else page_count
|
||
info = starts[start_page]
|
||
art = Article(
|
||
id=f"article-{i + 1:03d}",
|
||
title=info["title"],
|
||
section=info.get("section", ""),
|
||
confidence=info["confidence"],
|
||
)
|
||
for page in range(start_page, end_page + 1):
|
||
page_blocks = by_page.get(page, [])
|
||
if not page_blocks:
|
||
continue
|
||
# Only claim a page the article actually occupies. A full-page
|
||
# advert sitting between two features starts no article of its
|
||
# own, so it would otherwise be swallowed by the preceding one and
|
||
# highlighted as part of it in the PDF view.
|
||
#
|
||
# A sidebar/pull-out box (mark_sidebars) is excluded the same way
|
||
# advert text is: automatically kept out of narration by default,
|
||
# rather than requiring a manual exclusion in review every time.
|
||
# It still shows in review (blocksForPages lists every block on
|
||
# the article's pages) and can be ticked back in with one click.
|
||
contributed = False
|
||
for b in page_blocks:
|
||
if b.type in (BLOCK_FURNITURE, BLOCK_AD, BLOCK_SIDEBAR, BLOCK_PULLQUOTE, *VISUAL_TYPES):
|
||
continue
|
||
contributed = True
|
||
art.blocks.append(b.id)
|
||
if b.type == BLOCK_BYLINE and not art.author:
|
||
_, name = looks_like_byline(b.text)
|
||
art.author = name
|
||
to_page, _ = find_continuation(b.text)
|
||
if to_page and to_page not in art.continued_to:
|
||
art.continued_to.append(to_page)
|
||
if contributed:
|
||
art.pages.append(page)
|
||
|
||
# An article the pipeline could not fill with body copy is almost
|
||
# certainly a false start (an advert headline, a section divider).
|
||
has_body = any(
|
||
b.type == BLOCK_BODY for b in blocks if b.id in set(art.blocks)
|
||
)
|
||
if not has_body:
|
||
art.confidence = round(art.confidence * 0.4, 3)
|
||
articles.append(art)
|
||
|
||
return _link_continuations(articles, page_count)
|
||
|
||
|
||
def _link_continuations(articles: list[Article], page_count: int) -> list[Article]:
|
||
"""Attach jump pages ('continued on page 88') to their originating article.
|
||
|
||
Without this, the tail of a feature parked at the back of the issue is
|
||
orphaned and gets narrated as a separate stub article that begins
|
||
mid-sentence.
|
||
"""
|
||
by_start = {a.pages[0]: a for a in articles if a.pages}
|
||
for art in articles:
|
||
for target in art.continued_to:
|
||
if not (1 <= target <= page_count):
|
||
continue
|
||
tail = by_start.get(target)
|
||
if tail is None or tail is art or tail.status == "merged":
|
||
continue
|
||
# Only absorb a genuine continuation: a real article listed in the
|
||
# contents keeps its own identity.
|
||
if tail.confidence >= 0.9:
|
||
continue
|
||
art.pages.extend(p for p in tail.pages if p not in art.pages)
|
||
art.blocks.extend(tail.blocks)
|
||
art.images.extend(tail.images)
|
||
tail.status = "merged"
|
||
return [a for a in articles if a.status != "merged"]
|
||
|
||
|
||
def estimate_listen_seconds(blocks: Sequence[Block], article: Article,
|
||
words_per_minute: float = 155.0) -> int:
|
||
"""Rough narration length, for the 'estimated listening time' column."""
|
||
ids = set(article.blocks)
|
||
words = sum(
|
||
len(b.text.split()) for b in blocks
|
||
if b.id in ids and b.type in NARRATABLE
|
||
)
|
||
return int(round(words / words_per_minute * 60)) if words else 0
|
||
|
||
|
||
# ── PDF and OCR I/O ──────────────────────────────────────────────────────────
|
||
|
||
def _require_fitz():
|
||
"""Import PyMuPDF under its current name, falling back to the old alias.
|
||
|
||
1.24 renamed the module to ``pymupdf``; ``fitz`` still resolves but emits a
|
||
deprecation warning and is slated for removal.
|
||
"""
|
||
try:
|
||
import pymupdf
|
||
return pymupdf
|
||
except ImportError:
|
||
pass
|
||
try:
|
||
import fitz
|
||
return fitz
|
||
except ImportError as exc: # pragma: no cover - environment guard
|
||
raise RuntimeError(
|
||
"PyMuPDF is required for magazine ingestion; install pymupdf"
|
||
) 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_HEADING,
|
||
"caption": BLOCK_CAPTION,
|
||
"headline": BLOCK_TITLE,
|
||
"article_headline": BLOCK_TITLE,
|
||
"paragraph": BLOCK_BODY,
|
||
"section_heading": BLOCK_HEADING,
|
||
"author": BLOCK_BYLINE,
|
||
"byline": BLOCK_BYLINE,
|
||
"sidenote": BLOCK_SIDEBAR,
|
||
"quote": BLOCK_PULLQUOTE,
|
||
"pullquote": BLOCK_PULLQUOTE,
|
||
"image": BLOCK_FIGURE,
|
||
"figure": BLOCK_FIGURE,
|
||
"picture": BLOCK_FIGURE,
|
||
"chart": BLOCK_CHART,
|
||
"graphic": BLOCK_CHART,
|
||
"diagram": BLOCK_CHART,
|
||
"table": BLOCK_TABLE,
|
||
"code": BLOCK_CODE,
|
||
"code_block": BLOCK_CODE,
|
||
"advertisement": BLOCK_AD,
|
||
"ad": BLOCK_AD,
|
||
"aside_text": BLOCK_SIDEBAR,
|
||
}
|
||
|
||
# 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 and _VISION_LABEL_TYPES.get(label) not in VISUAL_TYPES) 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
|
||
|
||
# Settings also accept a bare server address; OpenAI-compatible servers
|
||
# expose their API under /v1. Preserve explicit proxy/API prefixes.
|
||
base_url = base_url.strip().rstrip("/")
|
||
if not urlsplit(base_url).path:
|
||
base_url += "/v1"
|
||
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": [
|
||
{"type": "text", "text": (
|
||
"Recognise the page layout and transcribe visible text verbatim. "
|
||
"Return one region per headline, paragraph, image, chart, table, "
|
||
"code box, caption, header, footer or page number. Use the format "
|
||
"label [x0, y0, x1, y1]transcription with pixel coordinates in the "
|
||
"supplied image, one region starting on each new line. Labels: "
|
||
"title, paragraph, image, chart, table, code, caption, header, "
|
||
"footer, page_number, aside_text. Enclose a whole chart or table "
|
||
"in one box, not separate axis labels or cells. Preserve table "
|
||
"rows and columns as Markdown and code indentation and line breaks. "
|
||
"Images may have empty transcription; do not invent descriptions. "
|
||
"Do not merge neighbouring articles or include 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 as exc:
|
||
status = getattr(getattr(exc, "response", None), "status_code", None)
|
||
logging.getLogger(__name__).warning("Vision OCR request failed (%s%s)",
|
||
type(exc).__name__, f", HTTP {status}" if status else "")
|
||
return None
|
||
except (ValueError, KeyError, IndexError, TypeError):
|
||
logging.getLogger(__name__).warning("Vision OCR returned an invalid completion response")
|
||
return None
|
||
if not isinstance(content, str):
|
||
return None
|
||
|
||
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",
|
||
)
|
||
block.region_label = label
|
||
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,
|
||
force_vision_ocr: bool = False,
|
||
progress: Any = None) -> dict[str, Any]:
|
||
"""Run the full extraction for one PDF, writing page renders and images.
|
||
|
||
When vision_ocr_url/model are given, a page whose text needs OCR (or
|
||
whose already-extracted text scores as mostly garbage, native or
|
||
tesseract alike -- see ocr_quality_score) is escalated to that engine
|
||
first, with a second vision engine (vision_ocr_fallback_*) tried next if
|
||
given, and tesseract only as the last resort if every vision call fails
|
||
or every engine is unreachable. Confirmed live against
|
||
ocr-baidu-unlimited-ocr: markedly more accurate transcription of a
|
||
scanned German magazine than tesseract, and it labels running headers/
|
||
footers/page numbers directly (see build_vision_blocks), which is the
|
||
dominant source of the over-fragmentation cluster_running_furniture()
|
||
otherwise has to recover after the fact from noisy OCR variants of the
|
||
same repeating line. KDL Frontier Parser nano, tested as the fallback
|
||
engine, answers faster but returns plain text with no layout coordinates
|
||
at all -- handled by _whole_page_region as one ambiguous full-page block.
|
||
|
||
Returns the layout document: every block on every page, classified and in
|
||
reading order, plus the raw contents-page candidates.
|
||
"""
|
||
fitz = _require_fitz()
|
||
pages_dir = out_dir / "pages"
|
||
images_dir = out_dir / "images"
|
||
pages_dir.mkdir(parents=True, exist_ok=True)
|
||
images_dir.mkdir(parents=True, exist_ok=True)
|
||
vision_engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
vision_enabled = bool(vision_ocr_url and vision_ocr_model)
|
||
vision_page_count = 0
|
||
|
||
doc = fitz.open(str(pdf_path))
|
||
try:
|
||
raw_pages: list[dict[str, Any]] = []
|
||
|
||
for pno in range(doc.page_count):
|
||
page = doc.load_page(pno)
|
||
rect = page.rect
|
||
page_no = pno + 1
|
||
|
||
png = pages_dir / f"{page_no:04d}.png"
|
||
if not png.exists():
|
||
pix = page.get_pixmap(dpi=render_dpi)
|
||
pix.save(str(png))
|
||
|
||
image_boxes: list[list[float]] = []
|
||
page_images: list[dict[str, Any]] = []
|
||
for img_index, info in enumerate(page.get_images(full=True)):
|
||
rects: list[list[float]] = []
|
||
try:
|
||
for r in page.get_image_rects(info[0]):
|
||
rects.append([r.x0, r.y0, r.width, r.height])
|
||
except Exception:
|
||
continue
|
||
image_boxes.extend(rects)
|
||
name = f"p{page_no:04d}-{img_index:02d}"
|
||
if _extract_image(fitz, doc, info[0], images_dir, name) and rects:
|
||
# A few embedded images repeat (a logo stamped on every
|
||
# page); each placement still gets its own bbox so a
|
||
# caption or figure position can be matched precisely.
|
||
for ri, box in enumerate(rects):
|
||
page_images.append({
|
||
"id": name if ri == 0 else f"{name}-{ri}",
|
||
"page": page_no,
|
||
"bbox": box,
|
||
"filename": f"{name}.png",
|
||
})
|
||
|
||
blocks: list[Block] = []
|
||
data = page.get_text("dict")
|
||
for bi, blk in enumerate(data.get("blocks", [])):
|
||
if blk.get("type") != 0: # 0 = text, 1 = image
|
||
continue
|
||
texts, sizes, fonts, bolds = [], [], [], []
|
||
for line in blk.get("lines", []):
|
||
for span in line.get("spans", []):
|
||
if not span.get("text", "").strip():
|
||
continue
|
||
texts.append(span["text"])
|
||
sizes.append(float(span.get("size", 0)))
|
||
fonts.append(span.get("font", ""))
|
||
bolds.append(bool(span.get("flags", 0) & 2 ** 4))
|
||
text = " ".join(texts).strip()
|
||
if not text:
|
||
continue
|
||
x0, y0, x1, y1 = blk["bbox"]
|
||
blocks.append(Block(
|
||
id=f"b{page_no:04d}-{bi:03d}",
|
||
page=page_no,
|
||
bbox=[x0, y0, x1 - x0, y1 - y0],
|
||
type=BLOCK_BODY,
|
||
text=text,
|
||
font_size=statistics.median(sizes) if sizes else 0.0,
|
||
font_name=max(set(fonts), key=fonts.count) if fonts else "",
|
||
bold=sum(bolds) > len(bolds) / 2 if bolds else False,
|
||
))
|
||
|
||
# A page whose native text is negligible is a scan (or is drawn as
|
||
# outlines); OCR is the only way to reach its words. A page WITH
|
||
# native/embedded text is not necessarily good text -- a
|
||
# pre-OCR'd scan (e.g. from scanner-bundled OCR software) can
|
||
# carry a poor text layer that this same escalation should reach.
|
||
native_chars = sum(len(b.text) for b in blocks)
|
||
needs_ocr = native_chars < 120
|
||
native_quality = (
|
||
ocr_quality_score(" ".join(b.text for b in blocks))
|
||
if not needs_ocr else 0.0
|
||
)
|
||
needs_escalation = force_vision_ocr or needs_ocr or native_quality < _VISION_QUALITY_THRESHOLD
|
||
|
||
vision_blocks: list[Block] | None = None
|
||
if vision_enabled and needs_escalation:
|
||
regions = vision_ocr_page_chain(png, vision_engines,
|
||
timeout=vision_ocr_timeout)
|
||
if regions:
|
||
scale_x = rect.width / max(1.0, _png_width(png, fitz))
|
||
vision_blocks = build_vision_blocks(
|
||
regions, page_no, scale_x, id_prefix="b")
|
||
|
||
if vision_blocks is not None:
|
||
# Replaces whatever was already there -- including a poor
|
||
# native/pre-OCR'd text layer, rather than layering a second,
|
||
# conflicting reading on top of it.
|
||
vision_page_count += 1
|
||
blocks = vision_blocks
|
||
detected_images = crop_visual_regions(out_dir, page_no,
|
||
{"width": rect.width, "height": rect.height}, blocks)
|
||
if detected_images:
|
||
page_images = detected_images
|
||
elif needs_ocr:
|
||
scale_x = rect.width / max(1.0, _png_width(png, fitz))
|
||
for li, para in enumerate(ocr_page(png, ocr_lang)):
|
||
bx, by, bw, bh = para["bbox"]
|
||
blocks.append(Block(
|
||
id=f"b{page_no:04d}-o{li:03d}",
|
||
page=page_no,
|
||
bbox=[bx * scale_x, by * scale_x, bw * scale_x, bh * scale_x],
|
||
type=BLOCK_BODY,
|
||
text=para["text"],
|
||
confidence=para["confidence"],
|
||
# The paragraph's own line height, not its full bbox
|
||
# height -- that would scale with line count instead
|
||
# of type size (see ocr_page's docstring).
|
||
font_size=para["line_height"] * scale_x,
|
||
source="ocr",
|
||
))
|
||
# else: native text present and scored well enough -- unchanged.
|
||
|
||
raw_pages.append({
|
||
"page": page_no,
|
||
"width": rect.width,
|
||
"height": rect.height,
|
||
"blocks": blocks,
|
||
"image_boxes": image_boxes,
|
||
"images": page_images,
|
||
})
|
||
if progress:
|
||
# Raw block text, pre-classification -- good enough for a
|
||
# live "here is what was just read" preview during ingestion
|
||
# (mirrors ocr-baidu-unlimited-ocr's own demo: source page on
|
||
# one side, recognised text on the other, filled in as each
|
||
# page finishes) without waiting for the document-wide passes
|
||
# below.
|
||
progress(page_no, doc.page_count, blocks)
|
||
|
||
# Furniture can only be identified across the whole document, so
|
||
# classification waits until every page has been read.
|
||
repeated = find_repeated_lines(
|
||
[[b.text for b in p["blocks"]] for p in raw_pages]
|
||
)
|
||
all_blocks: list[Block] = []
|
||
contents_candidates: list[dict[str, Any]] = []
|
||
|
||
# One baseline for the whole issue (see classify_blocks).
|
||
doc_base = body_font_size(
|
||
[b for p in raw_pages for b in p["blocks"]]
|
||
)
|
||
|
||
for p in raw_pages:
|
||
classify_blocks(p["blocks"], p["height"], p["width"],
|
||
repeated=repeated, image_boxes=p["image_boxes"],
|
||
base_size=doc_base)
|
||
mark_advert_pages(p["blocks"])
|
||
mark_pullquotes(p["blocks"])
|
||
assign_reading_order(p["blocks"], p["width"])
|
||
mark_sidebars(p["blocks"], p["width"])
|
||
all_blocks.extend(p["blocks"])
|
||
|
||
# Cross-page, so it runs once over every block rather than per page
|
||
# (see cluster_running_furniture's docstring for why exact-match
|
||
# find_repeated_lines above is not enough once OCR is involved).
|
||
page_heights = {p["page"]: p["height"] for p in raw_pages}
|
||
cluster_running_furniture(all_blocks, page_heights)
|
||
|
||
for p in raw_pages:
|
||
lines = [b.text for b in sorted(p["blocks"], key=lambda b: b.reading_order)]
|
||
score = score_contents_page(lines)
|
||
if score >= 0.45:
|
||
contents_candidates.append({
|
||
"page": p["page"],
|
||
"score": score,
|
||
"entries": parse_contents(lines),
|
||
})
|
||
|
||
all_images = [img for p in raw_pages for img in p["images"]]
|
||
for img in all_images:
|
||
img["reading_order"] = image_reading_position(img, all_blocks)
|
||
|
||
return {
|
||
"page_count": doc.page_count,
|
||
"pages": [{"page": p["page"], "width": p["width"], "height": p["height"]}
|
||
for p in raw_pages],
|
||
"blocks": [b.to_json() for b in all_blocks],
|
||
"images": all_images,
|
||
"contents_candidates": contents_candidates,
|
||
"vision_ocr_pages": vision_page_count,
|
||
}
|
||
finally:
|
||
doc.close()
|
||
|
||
|
||
def _png_width(png: Path, fitz) -> float:
|
||
try:
|
||
return float(fitz.Pixmap(str(png)).width)
|
||
except Exception:
|
||
return 1.0
|
||
|
||
|
||
def _extract_image(fitz, doc, xref: int, images_dir: Path, name: str) -> bool:
|
||
"""Save one embedded image, skipping the tiny rules and spacers.
|
||
|
||
Returns whether a file now exists at that name, so the caller can decide
|
||
whether to record it as a real figure -- without this, a page's rules and
|
||
spacers were silently skipped on disk but still implicitly "existed" as
|
||
far as anything downstream knew, since nothing checked.
|
||
"""
|
||
try:
|
||
pix = fitz.Pixmap(doc, xref)
|
||
if pix.width < 64 or pix.height < 64:
|
||
return False
|
||
if pix.n - pix.alpha >= 4: # CMYK needs converting first
|
||
pix = fitz.Pixmap(fitz.csRGB, pix)
|
||
target = images_dir / f"{name}.png"
|
||
if not target.exists():
|
||
pix.save(str(target))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ── Images: linking a figure to the article it illustrates ──────────────────
|
||
|
||
def assign_images_to_articles(images: Sequence[dict[str, Any]],
|
||
blocks: Sequence[Block],
|
||
articles: Sequence[Article]) -> None:
|
||
"""Attach each extracted figure to the article whose column it falls in.
|
||
|
||
A figure belongs to whichever article occupies its page; when two
|
||
articles share a page (a feature ending, a new one starting lower down),
|
||
the figure goes to whichever article has the nearer block by vertical
|
||
position, so a photo is not shown under the wrong headline. Mutates
|
||
articles in place.
|
||
"""
|
||
by_page: dict[int, list[Article]] = {}
|
||
for art in articles:
|
||
for page in art.pages:
|
||
by_page.setdefault(page, []).append(art)
|
||
|
||
blocks_by_page: dict[int, list[Block]] = {}
|
||
for b in blocks:
|
||
blocks_by_page.setdefault(b.page, []).append(b)
|
||
|
||
for img in images:
|
||
if img.get("type") == BLOCK_AD:
|
||
continue
|
||
candidates = by_page.get(img["page"], [])
|
||
if not candidates:
|
||
continue
|
||
if len(candidates) == 1:
|
||
candidates[0].images.append(img["id"])
|
||
continue
|
||
# Several articles share this page: pick by nearest block vertically,
|
||
# since that is what a reader would associate the figure with.
|
||
img_y = img["bbox"][1]
|
||
best, best_dist = candidates[0], float("inf")
|
||
for art in candidates:
|
||
own_blocks = [b for b in blocks_by_page.get(img["page"], [])
|
||
if b.id in set(art.blocks)]
|
||
if not own_blocks:
|
||
continue
|
||
dist = min(abs(b.bbox[1] - img_y) for b in own_blocks)
|
||
if dist < best_dist:
|
||
best, best_dist = art, dist
|
||
best.images.append(img["id"])
|
||
|
||
|
||
def image_reading_position(image: dict[str, Any], blocks: Sequence[Block]) -> int:
|
||
"""Approximate reading_order for an image, for ordering it among sentences.
|
||
|
||
Uses the nearest block on the same page whose top sits at or above the
|
||
image (so a photo is shown once the surrounding text has been reached,
|
||
not before), falling back to the closest block on the page if the image
|
||
sits above all of them.
|
||
"""
|
||
same_page = [b for b in blocks if b.page == image["page"]]
|
||
if not same_page:
|
||
return 0
|
||
above = [b for b in same_page if b.bbox[1] <= image["bbox"][1]]
|
||
pool = above or same_page
|
||
nearest = min(pool, key=lambda b: abs(b.bbox[1] - image["bbox"][1]))
|
||
return nearest.reading_order
|
||
|
||
|
||
# ── Persistence ──────────────────────────────────────────────────────────────
|
||
|
||
def build_articles(layout: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Turn a stored layout document into detected articles with estimates."""
|
||
blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
candidates = layout.get("contents_candidates") or []
|
||
best = max(candidates, key=lambda c: c["score"], default=None)
|
||
entries = best["entries"] if best else []
|
||
|
||
# The contents page lists articles; it is not one.
|
||
skip = {c["page"] for c in candidates if c.get("score", 0) >= 0.45}
|
||
articles = group_articles(blocks, entries, layout.get("page_count", 0),
|
||
skip_pages=skip)
|
||
assign_images_to_articles(layout.get("images") or [], blocks, articles)
|
||
out = []
|
||
for art in articles:
|
||
rec = art.to_json()
|
||
rec["estimated_seconds"] = estimate_listen_seconds(blocks, art)
|
||
out.append(rec)
|
||
return out
|
||
|
||
|
||
def reocr_page(doc_dir: Path, layout: dict[str, Any], page_no: int,
|
||
vision_ocr_url: str, vision_ocr_model: str,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> dict[str, Any] | None:
|
||
"""Force one page through the vision OCR chain again with the engine the
|
||
user picked, replacing only that page's blocks in an already-ingested
|
||
layout.
|
||
|
||
Unlike extract_pages()'s automatic escalation, this always calls the
|
||
engine regardless of the page's existing text quality -- a user pressing
|
||
"OCR again" has already judged the result unusable and is not asking for
|
||
another quality gate. Deliberately does NOT call build_articles(): an
|
||
article the user already reviewed and approved must not be silently
|
||
re-grouped and lose that approval just because one of its pages was
|
||
re-read. The caller decides whether to re-run detection separately.
|
||
|
||
Returns the updated layout, or None if the page has no rendered PNG or
|
||
the engine returned nothing usable (network failure, empty response) --
|
||
the caller keeps the existing blocks in that case, never discards good
|
||
data for a failed retry.
|
||
"""
|
||
fitz = _require_fitz()
|
||
png = doc_dir / "pages" / f"{page_no:04d}.png"
|
||
if not png.is_file():
|
||
return None
|
||
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
return None
|
||
|
||
engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
regions = vision_ocr_page_chain(png, engines, timeout=timeout)
|
||
if not regions:
|
||
return None
|
||
|
||
scale_x = page_meta["width"] / max(1.0, _png_width(png, fitz))
|
||
new_blocks = build_vision_blocks(regions, page_no, scale_x, id_prefix="r")
|
||
|
||
all_blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
kept = [b for b in all_blocks if b.page != page_no]
|
||
|
||
# The baseline must come from the rest of the document, not this page
|
||
# alone -- see classify_blocks' docstring on why a single page (which may
|
||
# have no body copy of its own) cannot be its own reference.
|
||
doc_base = body_font_size(kept) or body_font_size(all_blocks)
|
||
classify_blocks(new_blocks, page_meta["height"], page_meta["width"], base_size=doc_base)
|
||
mark_advert_pages(new_blocks)
|
||
mark_pullquotes(new_blocks)
|
||
assign_reading_order(new_blocks, page_meta["width"])
|
||
mark_sidebars(new_blocks, page_meta["width"])
|
||
|
||
merged = kept + new_blocks
|
||
page_heights = {p["page"]: p["height"] for p in layout.get("pages", [])}
|
||
cluster_running_furniture(merged, page_heights)
|
||
|
||
layout = dict(layout)
|
||
layout["blocks"] = [b.to_json() for b in merged]
|
||
detected_images = crop_visual_regions(doc_dir, page_no, page_meta, new_blocks)
|
||
if detected_images:
|
||
layout["images"] = [i for i in layout.get("images", []) if i["page"] != page_no or "-manual-" in i["id"]] + detected_images
|
||
layout["vision_ocr_pages"] = int(layout.get("vision_ocr_pages") or 0) + 1
|
||
return layout
|
||
|
||
|
||
def apply_semantic_regions(blocks: list[dict], regions: list[dict], page_meta: dict | None = None) -> list[dict]:
|
||
"""Group existing OCR fragments without allowing invented IDs or text."""
|
||
known = {b["id"]: b for b in blocks}
|
||
used = set()
|
||
out = []
|
||
allowed = set(NARRATABLE) | set(VISUAL_TYPES) | {BLOCK_AD, BLOCK_SIDEBAR, BLOCK_CAPTION, BLOCK_FURNITURE, BLOCK_PULLQUOTE}
|
||
for region in regions:
|
||
if not isinstance(region, dict):
|
||
raise ValueError("Invalid semantic region")
|
||
ids = region.get("block_ids")
|
||
label = region.get("type")
|
||
kind = _VISION_LABEL_TYPES.get(label, label)
|
||
visual_box = None
|
||
if "bbox" in region and kind in (*VISUAL_TYPES, BLOCK_AD):
|
||
bbox = region["bbox"]
|
||
if not page_meta or not isinstance(bbox, list) or len(bbox) != 4 or not all(
|
||
isinstance(v, (int, float)) and math.isfinite(v) and 0 <= v <= 1000 for v in bbox):
|
||
raise ValueError("Invalid visual bounding box")
|
||
x0, y0, x1, y1 = bbox
|
||
if x1 <= x0 or y1 <= y0:
|
||
raise ValueError("Empty visual region")
|
||
visual_box = [x0 / 1000 * page_meta["width"], y0 / 1000 * page_meta["height"],
|
||
(x1 - x0) / 1000 * page_meta["width"], (y1 - y0) / 1000 * page_meta["height"]]
|
||
if not ids and kind in VISUAL_TYPES and visual_box:
|
||
out.append(Block(id=f"p{page_meta['page']:04d}-visual-{len(out)}", page=page_meta['page'],
|
||
bbox=visual_box, type=kind, text="", reading_order=len(out),
|
||
source="llm_layout", region_label=label).to_json())
|
||
continue
|
||
if not isinstance(ids, list) or not ids or not all(isinstance(i, str) for i in ids):
|
||
raise ValueError("Invalid region IDs")
|
||
if kind not in allowed or len(set(ids)) != len(ids) or any(i not in known or i in used for i in ids):
|
||
raise ValueError("Unknown or repeated semantic region")
|
||
members = [known[i] for i in ids]
|
||
if len({b["page"] for b in members}) != 1:
|
||
raise ValueError("A region cannot span pages")
|
||
if any(b["type"] == BLOCK_AD for b in members):
|
||
if not all(b["type"] == BLOCK_AD for b in members):
|
||
raise ValueError("Cannot merge advertising and editorial text")
|
||
kind = BLOCK_AD
|
||
used.update(ids)
|
||
x = min(b["bbox"][0] for b in members)
|
||
y = min(b["bbox"][1] for b in members)
|
||
right = max(b["bbox"][0] + b["bbox"][2] for b in members)
|
||
bottom = max(b["bbox"][1] + b["bbox"][3] for b in members)
|
||
separator = "\n" if kind in (BLOCK_CODE, BLOCK_TABLE) else " "
|
||
text = separator.join(b["text"] for b in members)
|
||
if kind in NARRATABLE:
|
||
text = re.sub(r"(?<=\w)-\s+(?=[a-zäöüß])", "", text)
|
||
out.append(dict(members[0], bbox=visual_box or [x, y, right - x, bottom - y], text=text,
|
||
type=kind, region_label=label, source="llm_layout",
|
||
reading_order=len(out),
|
||
continues_from_previous_page=region.get("continues_from_previous_page") is True))
|
||
# Incomplete model output must not silently remove document content.
|
||
return out + [b for b in blocks if b["id"] not in used]
|
||
|
||
|
||
def remap_reviewed_page(articles: list[dict], old_layout: dict, layout: dict, page: int) -> None:
|
||
"""Reconnect reviewed articles when OCR replaces the page's region IDs."""
|
||
old = {b["id"]: b for b in old_layout.get("blocks", []) if b["page"] == page}
|
||
owners = {a["id"]: [old[i] for i in a.get("blocks", []) if i in old] for a in articles}
|
||
affected = [a for a in articles if page in a.get("pages", []) or owners[a["id"]]]
|
||
for a in affected:
|
||
a["blocks"] = [i for i in a.get("blocks", []) if i not in old]
|
||
known_images = {i["id"] for i in layout.get("images", [])}
|
||
a["images"] = [i for i in a.get("images", []) if i in known_images]
|
||
if a.get("status") == "approved":
|
||
a["status"] = "detected"
|
||
for block in layout.get("blocks", []):
|
||
if block["page"] != page or block["type"] not in NARRATABLE or not affected:
|
||
continue
|
||
x, y, w, h = block["bbox"]
|
||
def overlap(a):
|
||
return sum(max(0, min(x + w, b["bbox"][0] + b["bbox"][2]) - max(x, b["bbox"][0])) *
|
||
max(0, min(y + h, b["bbox"][1] + b["bbox"][3]) - max(y, b["bbox"][1]))
|
||
for b in owners[a["id"]])
|
||
best = max(affected, key=overlap)
|
||
if overlap(best) >= w * h * 0.35:
|
||
best["blocks"].append(block["id"])
|
||
# Newly recognised visual regions follow the closest article on the page.
|
||
objects = [Article(**{k: v for k, v in a.items() if k in Article.__dataclass_fields__}) for a in affected]
|
||
new_images = [i for i in layout.get("images", []) if i["page"] == page and i.get("source_block")]
|
||
for a in objects:
|
||
a.images = [i for i in a.images if i not in {im["id"] for im in new_images}]
|
||
assign_images_to_articles(new_images, [Block(**b) for b in layout.get("blocks", [])], objects)
|
||
for a, obj in zip(affected, objects):
|
||
a["images"] = obj.images
|
||
|
||
|
||
def _crop_page_png(doc_dir: Path, page_no: int, page_meta: dict[str, Any],
|
||
bbox: Sequence[float], dest: Path) -> None:
|
||
"""Crop one region (PDF-point bbox) out of a page's rendered PNG."""
|
||
from PIL import Image
|
||
png = doc_dir / "pages" / f"{page_no:04d}.png"
|
||
with Image.open(png) as im:
|
||
sx, sy = im.width / page_meta["width"], im.height / page_meta["height"]
|
||
box_px = (
|
||
max(0, int(bbox[0] * sx)), max(0, int(bbox[1] * sy)),
|
||
min(im.width, int((bbox[0] + bbox[2]) * sx)),
|
||
min(im.height, int((bbox[1] + bbox[3]) * sy)),
|
||
)
|
||
im.crop(box_px).save(dest)
|
||
|
||
|
||
def refine_picture_bounds(png: Path, page_meta: dict, bbox: Sequence[float]) -> list[float]:
|
||
"""Snap an approximate picture box to a large, solid coloured panel.
|
||
|
||
Text and small decorations do not qualify. Grayscale or fragmented
|
||
artwork keeps the model's box instead of guessing a new boundary.
|
||
"""
|
||
import numpy as np
|
||
from PIL import Image
|
||
from scipy import ndimage
|
||
with Image.open(png) as image:
|
||
rgb = np.asarray(image.convert("RGB"), dtype=np.int16)
|
||
height, width = rgb.shape[:2]
|
||
mask = (rgb.max(axis=2) - rgb.min(axis=2) > 28) & (rgb.min(axis=2) < 225)
|
||
mask = ndimage.binary_closing(mask, iterations=3)
|
||
labels, _ = ndimage.label(mask)
|
||
candidates = []
|
||
x, y, w, h = bbox
|
||
for i, slices in enumerate(ndimage.find_objects(labels), 1):
|
||
if slices is None:
|
||
continue
|
||
sy, sx = slices
|
||
area = (sy.stop - sy.start) * (sx.stop - sx.start)
|
||
if area < width * height * .025 or area > width * height * .85:
|
||
continue
|
||
if np.count_nonzero(labels[slices] == i) / area < .7:
|
||
continue
|
||
box = [sx.start / width * page_meta["width"], sy.start / height * page_meta["height"],
|
||
(sx.stop - sx.start) / width * page_meta["width"], (sy.stop - sy.start) / height * page_meta["height"]]
|
||
overlap = max(0, min(x+w, box[0]+box[2])-max(x, box[0])) * max(0, min(y+h, box[1]+box[3])-max(y, box[1]))
|
||
score = overlap / max(1, min(w*h, box[2]*box[3]))
|
||
if score > .6:
|
||
candidates.append((score, box))
|
||
return max(candidates, key=lambda c: c[0])[1] if candidates else list(bbox)
|
||
|
||
|
||
def refresh_visual_regions(doc_dir: Path, layout: dict, page: int | None = None) -> None:
|
||
"""Regenerate semantic crops and avoid treating a scanned page as a figure."""
|
||
for meta in layout.get("pages", []):
|
||
page_no = meta["page"]
|
||
if page is not None and page_no != page:
|
||
continue
|
||
blocks = [Block(**b) for b in layout.get("blocks", []) if b["page"] == page_no]
|
||
for block in blocks:
|
||
if block.type == BLOCK_FIGURE and block.source == "llm_layout":
|
||
block.bbox = refine_picture_bounds(doc_dir / "pages" / f"{page_no:04d}.png", meta, block.bbox)
|
||
for saved in layout["blocks"]:
|
||
if saved["id"] == block.id:
|
||
saved["bbox"] = block.bbox
|
||
crops = crop_visual_regions(doc_dir, page_no, meta, blocks)
|
||
kept = []
|
||
for img in layout.get("images", []):
|
||
if img["page"] != page_no or "-manual-" in img["id"]:
|
||
kept.append(img)
|
||
continue
|
||
if img.get("source_block"):
|
||
continue
|
||
x, y, w, h = img["bbox"]
|
||
if w * h >= meta["width"] * meta["height"] * .85 and sum(len(b.text) for b in blocks) > 120:
|
||
continue
|
||
if any(abs(x - c["bbox"][0]) < 5 and abs(y - c["bbox"][1]) < 5 and
|
||
abs(w - c["bbox"][2]) < 10 and abs(h - c["bbox"][3]) < 10 for c in crops):
|
||
continue
|
||
# Embedded graphics inside an identified advert are advertising too.
|
||
if any(b.type == BLOCK_AD and x >= b.bbox[0] and y >= b.bbox[1] and
|
||
x + w <= b.bbox[0] + b.bbox[2] and y + h <= b.bbox[1] + b.bbox[3] for b in blocks):
|
||
img = dict(img, type=BLOCK_AD)
|
||
kept.append(img)
|
||
for crop in crops:
|
||
crop["reading_order"] = image_reading_position(crop, blocks)
|
||
layout["images"] = kept + crops
|
||
|
||
|
||
def crop_visual_regions(doc_dir: Path, page_no: int, page_meta: dict,
|
||
blocks: Sequence[Block]) -> list[dict[str, Any]]:
|
||
"""Keep charts, tables and code boxes intact as page crops for playback."""
|
||
images = []
|
||
for b in blocks:
|
||
if b.type not in VISUAL_TYPES:
|
||
continue
|
||
name = f"p{page_no:04d}-region-{b.id}"
|
||
(doc_dir / "images").mkdir(parents=True, exist_ok=True)
|
||
_crop_page_png(doc_dir, page_no, page_meta, b.bbox, doc_dir / "images" / f"{name}.png")
|
||
images.append({"id": name, "page": page_no, "bbox": b.bbox,
|
||
"filename": f"{name}.png", "type": b.type,
|
||
"source_block": b.id, "text": b.text,
|
||
"reading_order": b.reading_order})
|
||
return images
|
||
|
||
|
||
def add_manual_region(doc_dir: Path, layout: dict[str, Any], page_no: int,
|
||
bbox: Sequence[float], kind: str, text: str = "",
|
||
vision_ocr_url: str | None = None,
|
||
vision_ocr_model: str | None = None,
|
||
vision_ocr_fallback_url: str | None = None,
|
||
vision_ocr_fallback_model: str | None = None,
|
||
timeout: float = 150.0) -> dict[str, Any]:
|
||
"""Add a region the user drew by hand -- for text or a figure OCR/layout
|
||
detection missed entirely, not for correcting one already found (use
|
||
review's tick/untick or reocr_page for that).
|
||
|
||
kind "image" crops the region into its own figure file and adds it to
|
||
layout["images"]. kind "text" adds a new Block: if `text` is supplied it
|
||
is used verbatim (the user typed it themselves), otherwise the crop is
|
||
sent through the vision OCR chain if one is configured -- best-effort,
|
||
the block is still added with a placeholder if that returns nothing,
|
||
since the user drawing this region at all means they judged the text
|
||
important enough to include even without a clean read.
|
||
|
||
Deliberately does not re-run classify_blocks on the new region: the user
|
||
is drawing it precisely because automatic detection already failed on
|
||
it, so re-applying the same heuristics would not be trustworthy. It is
|
||
trusted as stated (BLOCK_BODY, narratable) and only given a fresh
|
||
reading-order position on its page.
|
||
"""
|
||
page_meta = next((p for p in layout.get("pages", []) if p["page"] == page_no), None)
|
||
if not page_meta:
|
||
raise ValueError("Unknown page")
|
||
if len(bbox) != 4 or bbox[2] <= 0 or bbox[3] <= 0:
|
||
raise ValueError("bbox must be [x, y, width, height] with a positive size")
|
||
layout = dict(layout)
|
||
|
||
if kind == "image":
|
||
images_dir = doc_dir / "images"
|
||
images_dir.mkdir(parents=True, exist_ok=True)
|
||
existing = list(images_dir.glob(f"p{page_no:04d}-manual-*.png"))
|
||
name = f"p{page_no:04d}-manual-{len(existing):03d}"
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, images_dir / f"{name}.png")
|
||
images = list(layout.get("images", []))
|
||
# Placed after every auto-detected image on the page by reading
|
||
# order -- a reasonable default for a figure the pipeline missed;
|
||
# the user can still see and move past it like any other figure.
|
||
page_images = [i for i in images if i["page"] == page_no]
|
||
next_ro = max([i.get("reading_order", 0) for i in page_images], default=0) + 1
|
||
images.append({"id": name, "page": page_no, "bbox": list(bbox),
|
||
"filename": f"{name}.png", "reading_order": next_ro})
|
||
layout["images"] = images
|
||
return layout
|
||
|
||
if kind != "text":
|
||
raise ValueError("kind must be 'text' or 'image'")
|
||
|
||
final_text = text.strip()
|
||
if not final_text and vision_ocr_url and vision_ocr_model:
|
||
crop_path = doc_dir / "pages" / f".manual-crop-{page_no}.png"
|
||
try:
|
||
_crop_page_png(doc_dir, page_no, page_meta, bbox, crop_path)
|
||
engines = [(vision_ocr_url, vision_ocr_model),
|
||
(vision_ocr_fallback_url, vision_ocr_fallback_model)]
|
||
regions = vision_ocr_page_chain(crop_path, engines, timeout=timeout)
|
||
if regions:
|
||
final_text = " ".join(r["text"] for r in regions).strip()
|
||
finally:
|
||
crop_path.unlink(missing_ok=True)
|
||
if not final_text:
|
||
final_text = "(no text recognised -- click to edit)"
|
||
|
||
blocks = [Block(**b) for b in layout.get("blocks", [])]
|
||
existing_ids = {b.id for b in blocks}
|
||
i = 0
|
||
while f"m{page_no:04d}-{i:03d}" in existing_ids:
|
||
i += 1
|
||
new_block = Block(id=f"m{page_no:04d}-{i:03d}", page=page_no, bbox=list(bbox),
|
||
type=BLOCK_BODY, text=final_text, confidence=0.5, source="manual")
|
||
|
||
page_blocks = [b for b in blocks if b.page == page_no] + [new_block]
|
||
assign_reading_order(page_blocks, page_meta["width"])
|
||
others = [b for b in blocks if b.page != page_no]
|
||
layout["blocks"] = [b.to_json() for b in others + page_blocks]
|
||
return layout
|
||
|
||
|
||
def 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
|