"""Layout-aware magazine ingestion. Magazines are not books: a single page routinely carries several unrelated articles, advertisements, sidebars and pull quotes, and an article often continues twenty pages later. Treating a page as the unit of narration therefore produces audio that jumps mid-sentence into an advert. The unit here is the *article*, reconstructed from classified layout blocks. Pipeline (see architecture/magazine-ingestion.md): PDF -> extract_pages() native text blocks with bbox/font via PyMuPDF -> ocr_page() tesseract TSV fallback for scanned pages -> classify_blocks() title / deck / byline / body / caption / ... -> assign_reading_order() column detection, then top-to-bottom per column -> detect_contents() Stage A: locate and parse the table of contents -> group_articles() Stage C: group blocks, link continuations -> (human review) Stage D happens in the UI; nothing is narrated until an article is approved. PyMuPDF is imported lazily inside the I/O functions so that the pure geometry and classification helpers -- which hold all the fiddly logic worth testing -- can be exercised without the native dependency or a sample PDF. """ from __future__ import annotations import base64 import difflib import json import logging import math import re import statistics import subprocess import tempfile import unicodedata import requests from core.retry import request_with_retries from core.pronunciations import ( _SUSPENDED_COMPOUND_WORDS, detect_article_language, language_for_block, normalize_ocr_text, prepare_text_for_tts, recurring_tokens, remove_linebreak_hyphens, suspect_ocr_tokens, ) 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_QUOTE = "quote" # quoted speech that belongs in narration 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, BLOCK_QUOTE) # 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" # The author line's text before an author line that ran into the first sentence was split # (clean_up_regions). Part of the model so every path that rebuilds blocks keeps it -- and does # not crash on it (drawing a region did, "unexpected keyword argument 'original_text'"). original_text: str = "" # The text as recognised, kept when the reader first edits it by hand (set_region_text), so the # review can show what changed and put it back. Empty = never edited. ocr_text: str = "" # box_manual: the reviewer placed or resized this box, so snap_text_boxes never moves it. # box_snapped: snap_text_boxes fitted it to its text. Both are model fields for the same # reason as original_text: every path that rebuilds blocks with Block(**b) must accept them. box_manual: bool = False box_snapped: bool = False def to_json(self) -> dict[str, Any]: data = asdict(self) for optional in ("original_text", "ocr_text", "box_manual", "box_snapped"): if not data[optional]: del data[optional] return data @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) continued_from: list[int] = field(default_factory=list) # 'continued from page N' seen on this article's pages audio_status: str = "none" # none | narration | podcast | both def to_json(self) -> dict[str, Any]: return asdict(self) # ── Geometry helpers (pure) ────────────────────────────────────────────────── def bbox_overlaps(a: Sequence[float], b: Sequence[float], tol: float = 0.0) -> bool: """True when two [x, y, w, h] boxes intersect, allowing a tolerance.""" ax, ay, aw, ah = a bx, by, bw, bh = b return not ( ax + aw + tol < bx or bx + bw + tol < ax or ay + ah + tol < by or by + bh + tol < ay ) def detect_columns(blocks: Sequence[Block], page_width: float, min_gap_ratio: float = 0.015) -> list[tuple[float, float]]: """Infer the column grid from where blocks are *absent*. Magazines vary the column count per page, so a fixed grid is wrong. The reliable signal is vertical whitespace running the height of the page: we project every block onto the x axis, then split wherever the uncovered gap is wider than a few percent of the page. Returns (left, right) spans. 1.5%, not the rounder 3.5% this started at: measured on a real scanned c't magazine page, the actual gutter between two genuine columns was only ~2.1% of the page width -- narrower than the old threshold, which made detect_columns see one merged column and, combined with a wide caption bridging the same gap (see assign_reading_order), produced every block on the page landing in column=0. """ spans = sorted( ((b.bbox[0], b.bbox[0] + b.bbox[2]) for b in blocks if b.bbox[2] > 1), key=lambda s: s[0], ) if not spans: return [(0.0, page_width)] merged: list[list[float]] = [list(spans[0])] min_gap = page_width * min_gap_ratio for left, right in spans[1:]: if left - merged[-1][1] > min_gap: merged.append([left, right]) else: merged[-1][1] = max(merged[-1][1], right) return [(m[0], m[1]) for m in merged] # Blocks that are not running text: they never define a column or split the page. _NOT_TEXT_FLOW = (BLOCK_FURNITURE, BLOCK_AD, BLOCK_FIGURE, BLOCK_CHART, BLOCK_TABLE, BLOCK_CODE) _QUOTE_PAIRS = (("»", "«"), ("„", "“"), ("“", "”"), ("«", "»"), ('"', '"'), ("‚", "‘")) def is_pullquote_like(block: Block) -> bool: """A short display quote that repeats the article: read after the page's running text. Types are not trusted on their own: on a real scanned page the layout model typed the actual pull-quote ("»Wenn man hundertmal die KI nutzt…«") as body text, and typed a left-column body paragraph as a pull-quote. A pull-quote is short and wrapped in quotation marks; a long block is running text whatever it was labelled. """ text = " ".join((block.text or "").split()) if not text or len(text) > 260: return False if block.type == BLOCK_PULLQUOTE: return True return any(text.startswith(open_) and text.rstrip(".!? ").endswith(close) for open_, close in _QUOTE_PAIRS) def column_anchors(blocks: Sequence[Block], page_width: float) -> tuple[list[float], float]: """Where the text columns begin, and the typical column width. Found from the LEFT EDGES of column-width blocks, which cluster tightly, rather than from whitespace between columns. The whitespace signal failed on a real scanned c't page: the OCR boxes were loose enough that the gutter shrank to 4-6 points (under any sane threshold), and a pull-quote laid across the gutter bridged the columns outright -- every block came back as column 0 and the two columns were read interleaved. Left edges were unaffected: 25-29 and 236-240. Columns must be most of a column width apart (the heavier of two close candidates wins), so a narrow stray box starting mid-column does not invent a column of its own. The weight floor is deliberately low: on page 193 the left column is three paragraphs under a screenshot, 19% of the right column's text, and a 25% floor dropped it and interleaved the page. Measured over a whole 204-page issue, 5-15% all score the same. """ flow = [b for b in blocks if b.bbox[2] > 1 and b.type not in _NOT_TEXT_FLOW] if not flow: return [0.0], page_width # The column width comes from running text only. Measured over every block, a page-wide photo # and the headline pushed it to 317 points against real columns of ~210 (page 88), which made # a plain right-column paragraph look as if it spanned two columns. body = [b for b in flow if b.type == BLOCK_BODY] or flow widths = sorted(b.bbox[2] for b in body) col_w = widths[len(widths) // 2] candidates = sorted((b for b in flow if 0.55 * col_w <= b.bbox[2] <= 1.4 * col_w), key=lambda b: b.bbox[0]) tol = page_width * 0.04 clusters: list[list[Block]] = [] for b in candidates: if clusters and b.bbox[0] - clusters[-1][-1].bbox[0] <= tol: clusters[-1].append(b) else: clusters.append([b]) if not clusters: return [min(b.bbox[0] for b in flow)], col_w weighted = [(min(b.bbox[0] for b in c), sum(b.bbox[3] for b in c)) for c in clusters] heaviest = max(w for _, w in weighted) anchors: list[tuple[float, float]] = [] for x, weight in weighted: if weight < 0.10 * heaviest: continue if anchors and x - anchors[-1][0] < 0.6 * col_w: if weight > anchors[-1][1]: anchors[-1] = (x, weight) continue anchors.append((x, weight)) return [x for x, _ in anchors] or [weighted[0][0]], col_w def assign_reading_order(blocks: list[Block], page_width: float) -> list[Block]: """Order blocks the way a person reads a magazine page. 1. Text spanning two or more columns (headline, standfirst) is read in turn from the top, and splits the page into bands: everything under it comes after it. 2. Within a band, each column top to bottom, columns left to right. Sorting purely by y interleaves parallel columns and splices one column's sentence into the next. 3. Short pull-quotes after the page's running text (they repeat the article, so reading one mid-column interrupts a sentence), then furniture (running heads, folios). Captions never split the page into bands: a caption under a photo in the middle of a page would otherwise tear both columns in half. """ anchors, col_w = column_anchors(blocks, page_width) # Columns never overlap: each ends where the next begins. spans = [(a, min(a + col_w, anchors[i + 1]) if i + 1 < len(anchors) else a + col_w) for i, a in enumerate(anchors)] def overlap(b: Block, span: tuple[float, float]) -> float: return max(0.0, min(b.bbox[0] + b.bbox[2], span[1]) - max(b.bbox[0], span[0])) def column_of(b: Block) -> int: best = max(range(len(spans)), key=lambda i: (overlap(b, spans[i]), -i)) if overlap(b, spans[best]) > 0: return best # Outside every column (a margin note): the nearest column by left edge. return min(range(len(anchors)), key=lambda i: abs(b.bbox[0] - anchors[i])) def crosses_columns(b: Block) -> bool: if len(spans) < 2: return b.bbox[2] > page_width * 0.6 return sum(1 for s in spans if overlap(b, s) >= 0.3 * col_w) >= 2 def own_strip(b: Block) -> bool: """A real headline sits in a horizontal strip of its own. A wide box lying over the column boxes (a duplicate region covering the same text, seen on pages 78 and 154) does not.""" top, bottom = b.bbox[1], b.bbox[1] + b.bbox[3] for other in blocks: if other is b or other.type in _NOT_TEXT_FLOW or crosses_columns(other): continue shared = min(bottom, other.bbox[1] + other.bbox[3]) - max(top, other.bbox[1]) if shared > 0.3 * max(1.0, min(b.bbox[3], other.bbox[3])): return False return True def splits_page(b: Block) -> bool: """Text across two or more columns, in a strip of its own: read before what lies under it. Captions never split the page -- one in the middle would tear both columns.""" if b.type in _NOT_TEXT_FLOW or b.type in (BLOCK_CAPTION, BLOCK_PULLQUOTE) or is_pullquote_like(b): return False return crosses_columns(b) and own_strip(b) def photo_splits_page(b: Block) -> bool: """A photo with no running text beside it separates the text above from the text below. Page 176: a caption beside a screenshot at the top, the two columns under it. Without this the caption was read between the columns and split "…zumal sie auch | helfen könnte…". A photo inset beside column text (page 164) does not split anything. Paragraph gaps are never used for this: in neighbouring columns they often line up, and treating them as breaks interleaves the columns again. """ if b.type not in _NOT_TEXT_FLOW or b.type in (BLOCK_FURNITURE, BLOCK_AD): return False top, bottom = b.bbox[1], b.bbox[1] + b.bbox[3] for other in blocks: if other.type in _NOT_TEXT_FLOW or is_pullquote_like(other): continue if min(bottom, other.bbox[1] + other.bbox[3]) - max(top, other.bbox[1]) > 2.0: return False return True for b in blocks: b.column = column_of(b) breakers = {id(b) for b in blocks if splits_page(b) or photo_splits_page(b)} band_tops = sorted(b.bbox[1] for b in blocks if id(b) in breakers) spanning = lambda b: id(b) in breakers # noqa: E731 def key(b: Block) -> tuple: if b.type == BLOCK_FURNITURE: return (3, 0, 0, round(b.bbox[1], 1), b.bbox[0]) if is_pullquote_like(b): return (2, 0, b.column, round(b.bbox[1], 1), b.bbox[0]) # The band is the number of breakers starting at or above this block: a breaker opens its # own band, so it follows the columns above it and precedes those below. band = sum(1 for top in band_tops if top <= b.bbox[1] + 0.5) if spanning(b): return (1, band, 0, round(b.bbox[1], 1), b.bbox[0]) return (1, band, 1 + b.column, round(b.bbox[1], 1), b.bbox[0]) ordered = sorted(blocks, key=key) for i, b in enumerate(ordered): b.reading_order = i return ordered # ── Classification (pure) ──────────────────────────────────────────────────── def body_font_size(blocks: Iterable[Block]) -> float: """The document's body size: the char-weighted median of block sizes. Weighting by length stops a page of huge display type from redefining what "normal" means -- body copy always dominates by character count. """ sizes: list[float] = [] for b in blocks: if b.font_size > 0 and b.text.strip(): sizes.extend([b.font_size] * max(1, len(b.text) // 20)) return statistics.median(sizes) if sizes else 10.0 def _normalise(text: str) -> str: text = unicodedata.normalize("NFKC", text or "") return re.sub(r"\s+", " ", text).strip() def looks_like_byline(text: str) -> tuple[bool, str]: """Detect 'By Jane Smith' / 'Von Jane Smith', returning the name.""" m = _BYLINE.match(_normalise(text)) if not m: return False, "" name = m.group(1).strip(" .,;:") # A byline is a name, not a sentence: reject anything long or punctuated. if len(name) > 60 or name.count(" ") > 5 or name.endswith((".", "!", "?")): return False, "" return True, name def find_continuation(text: str) -> tuple[int | None, int | None]: """Return (continues_on_page, continued_from_page) mentioned in the text.""" to_m = _CONT_TO.search(text or "") from_m = _CONT_FROM.search(text or "") to_page = next((int(g) for g in to_m.groups() if g), None) if to_m else None from_page = next((int(g) for g in from_m.groups() if g), None) if from_m else None return to_page, from_page def strip_continuation_markers(text: str) -> str: """Remove '(continued on page N)' style cues before text is narrated. These are a print layout convention -- meaningless, and slightly absurd, spoken aloud ("open paren continued on page eight close paren"). The layout store keeps the original text; only the narration copy is cleaned. """ cleaned = _CONT_TO.sub("", text or "") cleaned = _CONT_FROM.sub("", cleaned) return re.sub(r"\s+", " ", cleaned).strip() def classify_blocks(blocks: list[Block], page_height: float, page_width: float, repeated: set[str] | None = None, image_boxes: Sequence[Sequence[float]] = (), base_size: float | None = None) -> list[Block]: """Assign a semantic type to every block on one page. Font size is judged relative to the document body size rather than in absolute points, so the same rules hold for a pocket magazine and a broadsheet. base_size must be measured across the whole document, not this page: a page carrying a full-bleed photograph and one caption has no body copy of its own, so a per-page baseline would make that caption its own reference and classify it as body text. It falls back to the page when not supplied. """ repeated = repeated or set() base = base_size if base_size and base_size > 0 else body_font_size(blocks) for b in blocks: # A vision OCR engine that explicitly labeled this region (footer, # title, page_number, caption -- see build_vision_blocks) is a more # direct signal than any font-size heuristic can reconstruct after # the fact; second-guessing it here would throw that away. if b.source == "vision_label": continue text = _normalise(b.text) if not text: b.type = BLOCK_FURNITURE b.confidence = 0.5 continue ratio = (b.font_size / base) if base else 1.0 near_top = b.bbox[1] < page_height * 0.07 near_bottom = b.bbox[1] + b.bbox[3] > page_height * 0.93 short = len(text) <= 90 lower = text.lower() # Running heads, folios and page numbers repeat at the same edge on # many pages -- that repetition is what identifies them, not position # alone, since a headline can also sit at the top. if (near_top or near_bottom) and (text in repeated or text.isdigit()): b.type, b.confidence = BLOCK_FURNITURE, 0.95 continue if any(m in lower for m in _ADVERT_MARKERS) and short: b.type, b.confidence = BLOCK_AD, 0.8 continue is_byline, _ = looks_like_byline(text) if is_byline: b.type, b.confidence = BLOCK_BYLINE, 0.9 continue # A caption sits directly beneath or beside a figure and is set smaller # than body copy. Both conditions are needed: small text elsewhere is # usually a footnote, and body-sized text under an image is often the # article continuing past it. if ratio < 0.92 and any( bbox_overlaps(b.bbox, box, tol=max(page_height, page_width) * 0.02) for box in image_boxes ): b.type, b.confidence = BLOCK_CAPTION, 0.75 continue if ratio >= 1.6 and short: b.type, b.confidence = BLOCK_TITLE, 0.9 if ratio >= 2.0 else 0.7 continue if 1.18 <= ratio < 1.6 and len(text) <= 300: # Display-sized text mid-column is a pull quote; at the head of a # page it is the deck introducing the article. b.type = BLOCK_DECK if b.bbox[1] < page_height * 0.45 else BLOCK_PULLQUOTE b.confidence = 0.6 continue b.type, b.confidence = BLOCK_BODY, 0.85 return blocks def mark_advert_pages(blocks: list[Block]) -> list[Block]: """Spread an explicit advertisement label across its whole page. Publishers print "ADVERTISEMENT" once to disclaim the page, not each block on it. Without this the ad's own display type reads as a headline and the page gets claimed by whichever feature precedes it. """ # A small ad label beside substantial prose does not establish a full-page ad. if any(b.type == BLOCK_BODY and len(b.text) > 160 for b in blocks): return blocks if not any(b.type == BLOCK_AD and b.source not in ("vision_label", "llm_ad", "llm_layout") for b in blocks): return blocks for b in blocks: if b.type != BLOCK_FURNITURE: b.type, b.confidence = BLOCK_AD, 0.8 return blocks def mark_pullquotes(blocks: list[Block]) -> list[Block]: """Demote display text that merely repeats nearby body copy. Narrating a pull quote reads the same sentence twice, a few seconds apart, which sounds like a skipping record. """ body_text = _normalise(" ".join( b.text for b in blocks if b.type == BLOCK_BODY )).lower() if not body_text: return blocks for b in blocks: if b.type not in (BLOCK_DECK, BLOCK_PULLQUOTE): continue probe = _normalise(b.text).lower().strip(' "“”«»') if len(probe) > 25 and probe in body_text: b.type, b.confidence = BLOCK_PULLQUOTE, 0.9 return blocks def mark_sidebars(blocks: list[Block], page_width: float, narrow_ratio: float = 0.7) -> list[Block]: """Reclassify body text sitting in an anomalously narrow column as a sidebar, so it is excluded from an article's narration by default rather than requiring a manual exclusion in review every time. Geometry-only, like the rest of this module's classification: a sidebar or pull-out box is reliably narrower than the page's regular column grid (a boxed insert set off from the main flow), which needs no image analysis to detect. This is a heuristic, not a certainty -- a genuinely narrow-column page design (or a caption sitting in its own slim column) can trigger a false positive, which is exactly why review still shows every sidebar block and lets it be included with one click rather than silently discarding it. Full-width elements (a headline spanning the whole page) are excluded from the column analysis here, the same way assign_reading_order treats them specially: left in, one would bridge two genuinely separate columns into a single merged span (its bbox overlaps both), hiding a real narrow sidebar column entirely. This computes its own column partition rather than trusting `.column` from assign_reading_order, since that partition was built from the FULL block list and its indices would not line up with this narrower one. """ candidates = [b for b in blocks if b.bbox[2] <= page_width * 0.6] columns = detect_columns(candidates, page_width) if len(columns) < 2: return blocks widths = [right - left for left, right in columns] widest = max(widths) narrow_spans = [columns[i] for i, w in enumerate(widths) if w < widest * narrow_ratio] if not narrow_spans: return blocks for b in blocks: if b.type != BLOCK_BODY or b.bbox[2] > page_width * 0.6: continue centre = b.bbox[0] + b.bbox[2] / 2 if any(left <= centre <= right for left, right in narrow_spans): b.type, b.confidence = BLOCK_SIDEBAR, 0.6 return blocks def find_repeated_lines(pages_text: Sequence[Sequence[str]], threshold: float = 0.25) -> set[str]: """Lines appearing near the same edge on many pages: page furniture.""" if not pages_text: return set() counts: dict[str, int] = {} for lines in pages_text: for line in {_normalise(t) for t in lines if _normalise(t)}: counts[line] = counts.get(line, 0) + 1 minimum = max(3, int(len(pages_text) * threshold)) # Long lines that recur are more likely a repeated standfirst than a # running head, so cap the length. return {t for t, n in counts.items() if n >= minimum and len(t) <= 80} def looks_like_real_title(text: str) -> bool: """Guards the headline-fallback article start against OCR noise. A stray misread character or chart-axis fragment ("I", "@)X", ". |") can classify as BLOCK_TITLE by size alone; without this it would spawn its own article. Confirmed live on a scanned magazine: single-character OCR artifacts at running-footer size otherwise start empty articles. """ letters = sum(1 for c in (text or "") if c.isalpha()) return letters >= 4 def _digit_stripped(text: str) -> str: return re.sub(r"\d+", "", text or "") def cluster_running_furniture(blocks: Sequence[Block], page_heights: dict[int, float], similarity: float = 0.6, min_count: int = 3) -> None: """Reclassify near-edge blocks that recur across pages as furniture, even when OCR reads the running header/footer differently each time. find_repeated_lines() needs an EXACT match, which OCR defeats: a magazine tag like "c't Wissen 2026" came back as "cE KI-WISSEN", "ce Wissen 2136", "cA Wissen 2036" across different pages of the same scan -- confirmed live, and the dominant cause of a 28-page scan producing 26 near-1-per- page "articles" instead of grouping around real headlines. Only the page number reliably varies page to page, so digit-stripped fuzzy similarity catches what exact matching misses. Mutates blocks in place. """ candidates: list[tuple[Block, str]] = [] for b in blocks: if b.type in (BLOCK_FURNITURE, BLOCK_AD): continue text = _normalise(b.text) if not text or len(text) > 80: continue height = page_heights.get(b.page, 842.0) near_top = b.bbox[1] < height * 0.07 near_bottom = b.bbox[1] + b.bbox[3] > height * 0.93 if not (near_top or near_bottom): continue key = _digit_stripped(text).lower().strip() if len(key) < 3: continue candidates.append((b, key)) assigned = [False] * len(candidates) for i, (_, key_i) in enumerate(candidates): if assigned[i]: continue cluster = [i] for j in range(i + 1, len(candidates)): if assigned[j]: continue if difflib.SequenceMatcher(None, key_i, candidates[j][1]).ratio() >= similarity: cluster.append(j) if len(cluster) >= min_count: for idx in cluster: candidates[idx][0].type = BLOCK_FURNITURE candidates[idx][0].confidence = 0.75 assigned[idx] = True # ── Stage A: the table of contents ─────────────────────────────────────────── def score_contents_page(lines: Sequence[str]) -> float: """How strongly one page looks like a table of contents (0..1).""" if not lines: return 0.0 joined = " ".join(lines).lower() marker_hits = sum(1 for m in _TOC_MARKERS if m in joined) entry_hits = sum(1 for line in lines if _TOC_ENTRY.match(_normalise(line))) # Either signal alone is weak: a feature can be titled "Features", and a # dotted leader appears in ordinary tables too. score = min(1.0, marker_hits * 0.25) * 0.4 + min(1.0, entry_hits / 6) * 0.6 return round(score, 3) def parse_contents(lines: Sequence[str]) -> list[dict[str, Any]]: """Pull (title, page, section) entries out of a contents page.""" entries: list[dict[str, Any]] = [] section = "" for raw in lines: line = _normalise(raw) if not line: continue low = line.lower().strip(": ") if low in _TOC_MARKERS and low not in ("contents", "inhalt"): section = line.strip(": ") continue m = _TOC_ENTRY.match(line) if not m: continue title, page = m.group(1).strip(" .…·•-"), int(m.group(2)) if len(title) < 3 or page <= 0 or page > 2000: continue entries.append({"title": title, "page": page, "section": section}) return entries # ── Stage C: grouping blocks into articles ─────────────────────────────────── def _title_key(text: str) -> str: return re.sub(r"[^a-z0-9]+", "", _normalise(text).lower()) def group_articles(blocks: Sequence[Block], contents: Sequence[dict[str, Any]], page_count: int, skip_pages: set[int] | None = None) -> list[Article]: """Group ordered blocks into articles. The contents listing, when present, is the strongest available signal: it states both the title and the starting page, so it anchors boundaries that layout alone would guess at. Pages not claimed by any listing still get articles from their own headlines, because a contents page never lists everything (and some issues have no contents page at all). """ by_page: dict[int, list[Block]] = {} for b in blocks: by_page.setdefault(b.page, []).append(b) for page_blocks in by_page.values(): page_blocks.sort(key=lambda b: b.reading_order) # Where does each article start? A contents entry pins a start page; a # title block on a page with no entry starts one there. starts: dict[int, dict[str, Any]] = {} for entry in contents: page = entry["page"] if 1 <= page <= page_count: starts.setdefault(page, { "title": entry["title"], "section": entry.get("section", ""), "confidence": 0.9, }) skip_pages = skip_pages or set() for page in sorted(by_page): if page in starts or page in skip_pages: continue # A page carrying nothing but advertising and page furniture has no # article on it -- the cover and the ad pages otherwise each become a # spurious entry the reviewer has to reject by hand on every issue. if not any(b.type in NARRATABLE and b.type != BLOCK_TITLE for b in by_page[page]): continue page_blocks = by_page[page] titles = [b for b in page_blocks if b.type == BLOCK_TITLE and looks_like_real_title(b.text)] # A large subheading, chart title or pull-out on a continuation page # is frequently classified as a title. It is not a new article when # body copy already precedes it in reading order. Genuine page-level # article leads either come first, or have a nearby byline/deck. if titles: title = titles[0] before = [b for b in page_blocks if b.reading_order < title.reading_order and b.type == BLOCK_BODY] following = [b for b in page_blocks if title.reading_order < b.reading_order <= title.reading_order + 3] has_lead_context = any(b.type in (BLOCK_BYLINE, BLOCK_DECK) for b in following) if before and not has_lead_context: titles = [] 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) # "Continued from page N" is usually set as small print the classifier files as furniture or a # sidebar, so look at every short line on the page, not only at what is narrated. A long block that # merely mentions the phrase is prose, not a jump marker. for b in page_blocks: if len(b.text or "") > _MAX_JUMP_MARKER_CHARS: continue _, from_page = find_continuation(b.text) if from_page and from_page != page and from_page not in art.continued_from: art.continued_from.append(from_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) _MAX_JUMP_MARKER_CHARS = 160 # a "continued from" line is a few words; longer text is prose that quotes the phrase def _absorb(head: Article, tail: Article) -> None: head.pages.extend(p for p in tail.pages if p not in head.pages) head.blocks.extend(tail.blocks) head.images.extend(tail.images) tail.status = "merged" def _link_continuations(articles: list[Article], page_count: int) -> list[Article]: """Attach jump pages to their originating article, in both directions. Forward: the head says "continued on page 88" -- the article that starts on page 88 is its tail. Backward: the tail says "continued from page 27" -- the article that occupies page 27 is its head. Without this, the tail of a feature parked at the back of the issue is orphaned and narrated as a separate stub that begins mid-sentence. A forward link wins when both exist (it is processed first, and a merged article is never absorbed twice). An article a contents entry anchors (confidence >= 0.9) keeps its own identity either way. """ 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 if tail.confidence >= 0.9: continue _absorb(art, tail) for tail in articles: if tail.status == "merged" or tail.confidence >= 0.9 or not tail.pages: continue for source in tail.continued_from: if not (1 <= source <= page_count) or source >= tail.pages[0]: continue # a tail can only follow a page before it heads = [a for a in articles if a is not tail and a.status != "merged" and source in a.pages] if len(heads) != 1: continue # no article there, or ambiguous (a page shared by several): leave it to the reviewer _absorb(heads[0], tail) break return [a for a in articles if a.status != "merged"] def estimate_listen_seconds(blocks: Sequence[Block], article: Article, words_per_minute: float = 155.0) -> int: """Rough narration length, for the 'estimated listening time' column.""" ids = set(article.blocks) words = sum( len(b.text.split()) for b in blocks if b.id in ids and b.type in NARRATABLE ) return int(round(words / words_per_minute * 60)) if words else 0 # ── PDF and OCR I/O ────────────────────────────────────────────────────────── def _require_fitz(): """Import PyMuPDF under its current name, falling back to the old alias. 1.24 renamed the module to ``pymupdf``; ``fitz`` still resolves but emits a deprecation warning and is slated for removal. """ try: import pymupdf return pymupdf except ImportError: pass try: import fitz return fitz except ImportError as exc: # pragma: no cover - environment guard raise RuntimeError( "PyMuPDF is required for magazine ingestion; install pymupdf (pip install pymupdf) " "or rebuild your Docker container." ) from exc # Labels a vision OCR engine may return for a region, mapped onto the same # taxonomy tesseract/native blocks use. Anything else ("text", or a label the # engine invents) is treated as ambiguous body copy and still goes through # classify_blocks() for finer distinctions (byline, deck, ...). _VISION_LABEL_TYPES = { # Headlines / Titles / Headers "title": BLOCK_TITLE, "headline": BLOCK_TITLE, "article_headline": BLOCK_TITLE, "main_title": BLOCK_TITLE, "heading": BLOCK_HEADING, "header": BLOCK_HEADING, "section_heading": BLOCK_HEADING, "subheading": BLOCK_HEADING, "subtitle": BLOCK_HEADING, "deck": BLOCK_HEADING, # Paragraphs / Text / Body "paragraph": BLOCK_BODY, "body": BLOCK_BODY, "content": BLOCK_BODY, "list": BLOCK_BODY, "list_item": BLOCK_BODY, "author": BLOCK_BYLINE, "byline": BLOCK_BYLINE, # Sidenotes / Sidebars / Quotes / Footnotes "sidenote": BLOCK_SIDEBAR, "sidebar": BLOCK_SIDEBAR, "aside_text": BLOCK_SIDEBAR, "aside": BLOCK_SIDEBAR, "footnote": BLOCK_SIDEBAR, "footnotes": BLOCK_SIDEBAR, "note": BLOCK_SIDEBAR, "commentary": BLOCK_SIDEBAR, "quote": BLOCK_QUOTE, "blockquote": BLOCK_QUOTE, "pullquote": BLOCK_PULLQUOTE, # Page numbers / Furniture / Folios / Sources "page_number": BLOCK_FURNITURE, "pagenumber": BLOCK_FURNITURE, "page": BLOCK_FURNITURE, "header_text": BLOCK_FURNITURE, "footer": BLOCK_FURNITURE, "running_header": BLOCK_FURNITURE, "running_footer": BLOCK_FURNITURE, "folio": BLOCK_FURNITURE, "head": BLOCK_FURNITURE, "source": BLOCK_FURNITURE, "sources": BLOCK_FURNITURE, "citation": BLOCK_FURNITURE, "citations": BLOCK_FURNITURE, # Images / Figures / Photos "image": BLOCK_FIGURE, "figure": BLOCK_FIGURE, "picture": BLOCK_FIGURE, "photo": BLOCK_FIGURE, "illustration": BLOCK_FIGURE, "non_text": BLOCK_FIGURE, "nontext": BLOCK_FIGURE, "drawing": BLOCK_FIGURE, # Graphics / Charts / Diagrams / Maps / Treemaps "chart": BLOCK_CHART, "graphic": BLOCK_CHART, "infographic": BLOCK_CHART, "statistics": BLOCK_CHART, "diagram": BLOCK_CHART, "map": BLOCK_CHART, "treemap": BLOCK_CHART, "plot": BLOCK_CHART, "flowchart": BLOCK_CHART, # Tables "table": BLOCK_TABLE, "grid": BLOCK_TABLE, "matrix": BLOCK_TABLE, # Code / Formulas / Math "code": BLOCK_CODE, "code_block": BLOCK_CODE, "formula": BLOCK_CODE, "equation": BLOCK_CODE, "math": BLOCK_CODE, # Captions "caption": BLOCK_CAPTION, "image_caption": BLOCK_CAPTION, "figure_caption": BLOCK_CAPTION, "table_caption": BLOCK_CAPTION, # Ads "advertisement": BLOCK_AD, "advert": BLOCK_AD, "ad": BLOCK_AD, } # One label + its four-number bbox, anchoring a region; content runs until # the next such marker. Tolerates DeepSeek-OCR / Unlimited-OCR / PaddleOCR shapes # like "text [241, 90, 588, 249]", "<|ref|>title<|/ref|><|det|>[[x0, y0, x1, y1]]<|/det|>", etc. _VISION_REGION = re.compile( r"(?m)" r"(?:<\|ref\|>\s*(?P[^<]+?)\s*<\|/ref\|>\s*)?" r"(?:<\|det\|>\s*)?" r"(?P