mARTin-B78
736aeaaeac
Magazines break the audiobook pipeline's assumptions: several unrelated articles share a page, columns run in parallel, adverts and sidebars interrupt body copy, and a feature routinely jumps twenty pages to its tail. Narrating page-by-page walks out of an article mid-sentence and into an advert, so the canonical object is the article, not the page. core/magazine.py reconstructs articles from a flat PDF: PyMuPDF extracts blocks with bboxes and font metrics, tesseract supplies a TSV fallback so scanned pages join the same column grouping, blocks are classified and ordered column-major, and articles are anchored by the parsed table of contents with "continued on page N" tails linked back. routes/magazine.py adds 13 endpoints. Ingestion is a restart-safe background job. Nothing is narrated until a human approves the boundaries; podcast generation refuses unapproved articles. Podcast mode renders an approved article as a 2-3 host discussion using the configured LLM and voices the user picks, alongside narration rather than instead of it. Verified end-to-end over HTTP on a synthetic 8-page issue: the feature came back as pages [3, 8] with its jump tail absorbed and the advert page excluded, and an image-only rendering recovered all 47 blocks via OCR. 44 new tests; full suite 90 passing. The Studio UI is not built — this is reachable only via the API. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
841 lines
32 KiB
Python
841 lines
32 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 json
|
||
import re
|
||
import statistics
|
||
import subprocess
|
||
import unicodedata
|
||
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"
|
||
|
||
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 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:
|
||
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}
|
||
|
||
|
||
# ── 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]
|
||
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
|
||
|
||
|
||
def ocr_page(png_path: Path, lang: str = "deu+eng",
|
||
min_confidence: float = 40.0) -> list[dict[str, Any]]:
|
||
"""OCR one rendered page, returning line boxes from tesseract's TSV.
|
||
|
||
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 []
|
||
|
||
lines: dict[tuple[int, 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
|
||
key = (block_num, par_num, line_num)
|
||
rec = lines.setdefault(key, {
|
||
"text": [], "x0": left, "y0": top,
|
||
"x1": left + width, "y1": top + height, "conf": [],
|
||
})
|
||
rec["text"].append(word)
|
||
rec["x0"] = min(rec["x0"], left)
|
||
rec["y0"] = min(rec["y0"], top)
|
||
rec["x1"] = max(rec["x1"], left + width)
|
||
rec["y1"] = max(rec["y1"], top + height)
|
||
rec["conf"].append(conf)
|
||
|
||
out = []
|
||
for (_, _, _), rec in sorted(lines.items()):
|
||
out.append({
|
||
"text": " ".join(rec["text"]),
|
||
"bbox": [float(rec["x0"]), float(rec["y0"]),
|
||
float(rec["x1"] - rec["x0"]), float(rec["y1"] - rec["y0"])],
|
||
"confidence": round(sum(rec["conf"]) / len(rec["conf"]) / 100.0, 3),
|
||
})
|
||
return out
|
||
|
||
|
||
def extract_pages(pdf_path: Path, out_dir: Path, render_dpi: int = 110,
|
||
ocr_lang: str = "deu+eng",
|
||
progress: Any = None) -> dict[str, Any]:
|
||
"""Run the full extraction for one PDF, writing page renders and images.
|
||
|
||
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)
|
||
|
||
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.
|
||
native_chars = sum(len(b.text) for b in blocks)
|
||
if native_chars < 120:
|
||
scale_x = rect.width / max(1.0, _png_width(png, fitz))
|
||
for li, line in enumerate(ocr_page(png, ocr_lang)):
|
||
bx, by, bw, bh = line["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=line["text"],
|
||
confidence=line["confidence"],
|
||
font_size=bh * scale_x,
|
||
source="ocr",
|
||
))
|
||
|
||
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"])
|
||
|
||
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,
|
||
}
|
||
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 = [
|
||
b 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
|