`` content is kept as markdown-ish text (``#`` headings, paragraphs, ``-`` list items, ``|`` table rows). * **Text / Markdown**: decoded (UTF-8, else cp1252/latin-1) and whitespace-normalized. Layout handling shared by the text-layer and OCR paths: side-by-side narrow blocks whose lines line up (a borderless table that OCR / the PDF writer emits column by column) are zipped back into rows; cells on one line separated by a wide gap are joined with `` | ``; lines repeated at the top or bottom of most pages (running headers, "Page 3 of 9") are removed from multi-page documents. Cleanup: ligatures / soft hyphens / zero-width characters, de-hyphenation of words split across lines, re-flowing of wrapped lines into paragraphs (table rows and list items are kept as lines), OCR noise lines without any letter or digit, whitespace normalization. """ from __future__ import annotations import argparse import inspect import io import json import os import re import shutil import statistics import subprocess import sys import time import unicodedata from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Mapping, Optional, Sequence, Union import numpy as np Source = Union[str, os.PathLike, bytes, bytearray] IMAGE_EXT = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp", ".bmp", ".gif"} HTML_EXT = {".html", ".htm", ".xhtml"} TEXT_EXT = {".txt", ".md", ".markdown", ".text", ".csv", ".tsv", ".rst", ".log"} KINDS = ("pdf", "image", "html", "text") OCR_ENGINES = ("tesseract", "rapidocr") @dataclass class IngestResult: text: str # the whole document, pages separated by a blank line pages: list[str] # text per page (one entry for html / text) method_per_page: list[str] # "text", "text+tables(2)", "ocr:tesseract", "ocr:rapidocr", "html", "plain" seconds: float # wall time of the whole conversion warnings: list[str] = field(default_factory=list) kind: str = "" # "pdf" | "image" | "html" | "text" ocr_confidence: list[Optional[float]] = field(default_factory=list) # mean word confidence 0-100 per page (None: not OCR'd / unknown) def to_dict(self, with_text: bool = True) -> dict: d = asdict(self) if not with_text: d.pop("text"), d.pop("pages") d["chars"] = len(self.text) d["n_pages"] = len(self.pages) return d # ================================================================ public API def to_text(src: "Source | Sequence[Source]", kind: Optional[str] = None, *, ocr: str = "tesseract", lang: str = "eng", dpi: int = 300, min_chars: int = 30, tables: bool = True, preprocess: bool = True, strip_headers: bool = True, max_pages: Optional[int] = None, ocr_timeout: float = 120.0) -> IngestResult: """Convert a document to text. ``src``: a path, raw bytes, or a list of them (the pages of one document, e.g. page scans). ``kind``: "pdf" | "image" | "html" | "text" (default: from the extension, else sniffed from the content). ``ocr``: "tesseract" or "rapidocr". ``dpi``: render resolution for PDF pages that need OCR. ``min_chars``: a PDF page with fewer non-space characters in its text layer is OCR'd. ``tables``: markdown rows for ruled PDF tables (pdfplumber). ``preprocess``: image cleanup before OCR (see module doc). ``strip_headers``: drop running headers/footers of multi-page documents. ``max_pages``: stop after this many pages (with a warning). """ if ocr not in OCR_ENGINES: raise ValueError(f"ocr must be one of {OCR_ENGINES}, got {ocr!r}") if kind is not None and kind not in KINDS: raise ValueError(f"kind must be one of {KINDS}, got {kind!r}") t0 = time.perf_counter() opts = dict(ocr=ocr, lang=lang, dpi=dpi, min_chars=min_chars, tables=tables, preprocess=preprocess, max_pages=max_pages, ocr_timeout=ocr_timeout) srcs = list(src) if isinstance(src, (list, tuple)) else [src] if not srcs: raise ValueError("nothing to ingest") pages: list[str] = [] methods: list[str] = [] confs: list[Optional[float]] = [] warnings: list[str] = [] kinds: list[str] = [] for s in srcs: data, name = _read(s) k = kind or detect_kind(data, name) kinds.append(k) if k == "pdf": p, m, c, w = _ingest_pdf(data, **opts) elif k == "image": p, m, c, w = _ingest_image(data, **opts) elif k == "html": p, m, c, w = [html_to_text(_decode(data))], ["html"], [None], [] else: p, m, c, w = [normalize_text(_decode(data))], ["plain"], [None], [] if len(srcs) > 1 and w: w = [f"{name or 'input'}: {x}" for x in w] pages += p methods += m confs += c warnings += w if max_pages and len(pages) >= max_pages: if len(pages) > max_pages or s is not srcs[-1]: warnings.append(f"stopped after max_pages={max_pages} pages") pages, methods, confs = pages[:max_pages], methods[:max_pages], confs[:max_pages] break if strip_headers and len(pages) >= 2: pages, removed = strip_repeated_lines(pages) if removed: warnings.append(f"removed {removed} repeated header/footer line(s)") text = "\n\n".join(p for p in pages if p.strip()) if not text.strip(): warnings.append("no text extracted") ocr_conf = [c for c in confs if c is not None] if ocr_conf and min(ocr_conf) < 60: warnings.append(f"low OCR confidence on some pages (min {min(ocr_conf):.0f}/100): check the text") k = kinds[0] if len(set(kinds)) == 1 else "mixed" return IngestResult(text, pages, methods, time.perf_counter() - t0, warnings, k, confs) def predict_file(model, path: "Source | Sequence[Source]", questions: Mapping[str, Any], *, ingest_kwargs: Optional[dict] = None, **predict_kwargs) -> dict: """Ingest ``path`` with :func:`to_text` and answer ``questions`` with ``model.predict``. When the text is longer than the model's state budget (``max_len`` minus the question budget) and the caller did not choose ``long_state``, it is read in overlapping windows (``long_state="chunk"``) instead of being cut in the middle. The result is the usual predict output plus ``"ingest"``: {pages, method_per_page, seconds, warnings, chars, kind, ocr_confidence, long_state}. """ ing = to_text(path, **(ingest_kwargs or {})) text = ing.text if ing.text.strip() else "(no text could be extracted from the document)" kw = dict(predict_kwargs) params = _params(model.predict) if "long_state" in params and "long_state" not in kw and _too_long(model, text): kw["long_state"] = "chunk" res = model.predict(text, questions, **kw) info = ing.to_dict(with_text=False) info["long_state"] = kw.get("long_state", getattr(model, "long_state", None)) res["ingest"] = info return res def detect_kind(data: bytes, name: Optional[str] = None) -> str: """"pdf" | "image" | "html" | "text" from the file extension, else from the leading bytes.""" ext = Path(name).suffix.lower() if name else "" if ext == ".pdf": return "pdf" if ext in IMAGE_EXT: return "image" if ext in HTML_EXT: return "html" if ext in TEXT_EXT: return "text" head = bytes(data[:1024]) if head.lstrip().startswith(b"%PDF") or b"%PDF-" in head[:1024]: return "pdf" if (head.startswith(b"\x89PNG") or head.startswith(b"\xff\xd8\xff") or head[:4] in (b"II*\x00", b"MM\x00*") or (head.startswith(b"RIFF") and head[8:12] == b"WEBP") or head.startswith(b"GIF8") or head.startswith(b"BM")): return "image" low = head.lower().lstrip() if low.startswith((b" tuple[bytes, Optional[str]]: if isinstance(s, (bytes, bytearray)): return bytes(s), None p = Path(s) return p.read_bytes(), p.name def _decode(data: bytes) -> str: if data.startswith(b"\xef\xbb\xbf"): data = data[3:] for enc in ("utf-8", "cp1252"): try: return data.decode(enc) except UnicodeDecodeError: pass return data.decode("latin-1") def _params(fn) -> set[str]: try: return set(inspect.signature(fn).parameters) except (TypeError, ValueError): return set() def _too_long(model, text: str) -> bool: ser = getattr(model, "ser", None) tok, max_len = getattr(ser, "tok", None), getattr(model, "max_len", None) if tok is None or not max_len: return False budget = max_len - getattr(ser, "head_max_len", 0) - 2 # cheap bound first: ~>=2 chars per token for English if len(text) < 2 * budget: return False return len(tok(text, add_special_tokens=False)["input_ids"]) > budget # ================================================================ text cleanup _LIGATURES = {"ﬀ": "ff", "ﬁ": "fi", "ﬂ": "fl", "ﬃ": "ffi", "ﬄ": "ffl", "ﬅ": "st", "ﬆ": "st"} _INVISIBLE = dict.fromkeys(map(ord, "‌‍⁠"), None) _BULLET = re.compile(r"^\s*(?:[-*•▪●‣⁃◦]|\(?\d{1,3}[.)]|\(?[a-z][.)])\s+") _ROW = re.compile(r"(^\s*\|)|( \| )") def _is_row(line: str) -> bool: return bool(_ROW.search(line)) def normalize_text(s: str) -> str: """Whitespace/unicode normalization only (no re-flow): for text and markdown inputs.""" s = s.replace("\r\n", "\n").replace("\r", "\n").translate(_INVISIBLE) for k, v in _LIGATURES.items(): s = s.replace(k, v) s = s.replace(" ", " ").replace("\t", " ") lines = [re.sub(r"[ \f\v]+$", "", ln) for ln in s.split("\n")] s = "\n".join(lines) s = re.sub(r"\n{3,}", "\n\n", s) return s.strip() def clean_text(s: str, reflow: bool = True) -> str: """Cleanup for extracted / OCR'd text: unicode fixes, de-hyphenation, paragraph re-flow, removal of lines without any letter or digit (OCR specks), whitespace normalization.""" s = s.replace("\r\n", "\n").replace("\r", "\n").translate(_INVISIBLE) for k, v in _LIGATURES.items(): s = s.replace(k, v) s = s.replace(" ", " ").replace("\t", " ").replace("\f", "\n") lines = [] for ln in s.split("\n"): ln = re.sub(r" {2,}", " ", ln).strip() if ln and not re.search(r"[^\W_]", ln): # no letter/digit at all: rule, speck, "|", "~ ." continue lines.append(ln) s = "\n".join(lines) # de-hyphenate a word split across a line break: "infor-\nmation" -> "information" s = re.sub(r"([A-Za-z]{2,})[-‐‑]\n[ ]*([a-z]{2,})", r"\1\2", s) paras = re.split(r"\n\s*\n", s) out = [] for p in paras: pl = [ln for ln in p.split("\n") if ln.strip()] if not pl: continue if not reflow: out.append("\n".join(pl)) continue merged: list[str] = [] for ln in pl: keep_break = (not merged or _is_row(ln) or _is_row(merged[-1]) or _BULLET.match(ln) or ln.startswith("#") or merged[-1].startswith("#")) if keep_break: merged.append(ln) else: merged[-1] = merged[-1] + " " + ln out.append("\n".join(merged)) return "\n\n".join(out).strip() def _hf_key(line: str) -> str: """Header/footer comparison key: lowercase letters and '#' for each digit run.""" return re.sub(r"\d+", "#", re.sub(r"[^a-z0-9]+", "", line.lower())) def _similar(a: str, b: str) -> bool: if a == b: return True if min(len(a), len(b)) < 6: return False import difflib return difflib.SequenceMatcher(None, a, b, autojunk=False).ratio() >= 0.8 def strip_repeated_lines(pages: list[str], edge: int = 2, min_frac: float = 0.5) -> tuple[list[str], int]: """Remove running headers / footers: a line among the first or last ``edge`` lines of a page that (digits masked, fuzzy-matched so OCR noise does not hide it) occurs at a page edge on >= ``min_frac`` of the pages (and on >= 2 pages).""" if len(pages) < 2: return pages, 0 split = [p.split("\n") for p in pages] edges = [] for lines in split: idx = [i for i, ln in enumerate(lines) if ln.strip()] e = sorted(set(idx[:edge] + idx[-edge:])) # headers/footers are short lines; long ones are (re-flowed) body text edges.append([(i, _hf_key(lines[i])) for i in e if 0 < len(_hf_key(lines[i])) and len(lines[i]) <= 100]) need = max(2, int(np.ceil(min_frac * len(pages)))) drop: list[set[int]] = [set() for _ in pages] for p, ed in enumerate(edges): for i, k in ed: hits = sum(any(_similar(k, k2) for _, k2 in edges[q]) for q in range(len(pages)) if q != p) + 1 if hits >= need: drop[p].add(i) if len(drop[p]) >= sum(1 for ln in split[p] if ln.strip()): drop[p] = set() # never empty a page (duplicate or near-empty pages) removed = sum(len(d) for d in drop) if not removed: return pages, 0 out = [] for lines, d in zip(split, drop): keep = [ln for i, ln in enumerate(lines) if i not in d] out.append(re.sub(r"\n{3,}", "\n\n", "\n".join(keep)).strip()) return out, removed # ================================================================ layout: lines -> text @dataclass class _Line: x0: float y0: float x1: float y1: float text: str par: int = 0 # paragraph id inside its block; a change starts a new paragraph blk: int = 0 # source block id (set when blocks are zipped) @property def h(self) -> float: return max(1.0, self.y1 - self.y0) @property def yc(self) -> float: return (self.y0 + self.y1) / 2 def _numeric(text: str) -> bool: t = re.sub(r"\s", "", text) return bool(t) and sum(ch.isdigit() or ch in "$%.,()-–—+/:€£" for ch in t) >= 0.6 * len(t) _NUMBER = re.compile(r"^[(\-\u2013\u2212+$\u20ac\u00a3]*\d[\d.,]*%?\)?[%,;:]?$") def _cell_breaks(texts: list[str], gaps: list[float], h: float, prose: bool = False) -> list[bool]: """Which horizontal gaps inside one text line separate table cells (joined with " | "). Inside a prose block only very wide gaps (> 4 word heights) count, so stretched justified lines stay prose. Elsewhere a gap must be wide (> 1.5) and then either very wide, next to a number token, or much wider than the line's narrowest gap.""" if not gaps: return [] gmin = min(gaps) out = [] for j, g in enumerate(gaps): if prose: out.append(g > 4.0 * h) continue a, b = _NUMBER.match(texts[j].strip()), _NUMBER.match(texts[j + 1].strip()) if a and b: # two numbers side by side: adjacent numeric cells (padding ~1 word height) out.append(g > 1.0 * h) continue out.append(g > 1.5 * h and (g > 4.0 * h or bool(a or b) or (len(gaps) > 1 and g > 3.0 * max(gmin, 0.05 * h)))) return out def _nwords(text: str) -> int: return sum(1 for w in text.split() if w != "|") def _prose_lines(word_counts: list[int]) -> bool: """A block of running text: median >= 5 words per line over >= 3 lines.""" return len(word_counts) >= 3 and statistics.median(word_counts) >= 5 def _y_overlap(a0, a1, b0, b1) -> float: return max(0.0, min(a1, b1) - max(a0, b0)) def _bbox(lines: list[_Line]) -> tuple[float, float, float, float]: return (min(l.x0 for l in lines), min(l.y0 for l in lines), max(l.x1 for l in lines), max(l.y1 for l in lines)) def _aligned(a: list[_Line], b: list[_Line]) -> float: """Fraction of lines of the smaller block with a y-matching line in the other block.""" small, big = (a, b) if len(a) <= len(b) else (b, a) hit = sum(any(abs(s.yc - t.yc) < 0.5 * min(s.h, t.h) for t in big) for s in small) return hit / max(1, len(small)) def _zip_blocks(blocks: list[list[_Line]], page_w: float) -> list[tuple[list[_Line], bool]]: """Merge runs of side-by-side blocks whose lines line up (table columns emitted one after another) into one table block. Returns [(lines, is_table)] in reading order.""" out: list[tuple[list[_Line], bool]] = [] group: list[list[_Line]] = [] def narrow(b): return max(l.x1 - l.x0 for l in b) < 0.4 * page_w def prose(b): # columns of running text (>= 5 words per line) are never zipped into rows return statistics.median(_nwords(l.text) for l in b) >= 5 def flush(): if not group: return if len(group) == 1: out.append((group[0], False)) else: lines = [] for i, b in enumerate(group): for l in b: l.blk = i lines.append(l) out.append((lines, True)) group.clear() for b in blocks: if group: prev = group[-1] gx0, gy0, gx1, gy1 = _bbox([l for g in group for l in g]) bx0, by0, bx1, by1 = _bbox(b) px0, py0, px1, py1 = _bbox(prev) hmin = min(statistics.median(l.h for l in b), statistics.median(l.h for l in prev)) right = bx0 >= px1 - 0.5 * hmin ov = _y_overlap(gy0, gy1, by0, by1) / max(1.0, min(gy1 - gy0, by1 - by0)) both_single = len(b) == 1 and len(prev) == 1 numeric = sum(_numeric(l.text) for l in b) >= 0.6 * len(b) if right and ov >= 0.5 and _aligned(prev, b) >= 0.6 and not (prose(b) or prose(prev)) and ( both_single or numeric or (narrow(b) and narrow(prev))): group.append(b) continue flush() group.append(b) flush() return out def _join_row(cells: list[_Line], table: bool, prose: bool = False) -> str: cells = sorted(cells, key=lambda l: l.x0) h = min(l.h for l in cells) gaps = [b.x0 - a.x1 for a, b in zip(cells, cells[1:])] # separate line boxes on one row: cells of a zipped table (different source blocks), or a # wide gap by the rule of _cell_breaks (boxes are measured in box heights, ~1.4 word heights) breaks = _cell_breaks([c.text for c in cells], gaps, 0.7 * h, prose=prose and not table) txt = cells[0].text for a, b, br in zip(cells, cells[1:], breaks): txt += (" | " if (table and a.blk != b.blk) or br else " ") + b.text return txt def _rows(lines: list[_Line]) -> list[str]: """Cluster lines by vertical position into rows (for zipped table blocks).""" hmed = statistics.median(l.h for l in lines) rows: list[list[_Line]] = [] for l in sorted(lines, key=lambda l: l.yc): if rows and abs(l.yc - statistics.mean(x.yc for x in rows[-1])) < 0.5 * hmed: rows[-1].append(l) else: rows.append([l]) return [_join_row(r, True) for r in rows] def _seq_text(lines: list[_Line]) -> str: """Lines in reading order; a line that continues the previous one on the same baseline (a table cell to its right) is appended to it; paragraph changes give a blank line.""" out: list[str] = [] cur: list[_Line] = [] prose = _prose_lines([_nwords(l.text) for l in lines]) for l in lines: if cur: last = cur[-1] same_row = (l.x0 >= last.x1 - 0.2 * l.h and _y_overlap(last.y0, last.y1, l.y0, l.y1) >= 0.5 * min(last.h, l.h)) if same_row: cur.append(l) continue out.append(_join_row(cur, False, prose)) if l.par != last.par: out.append("") cur = [l] if cur: out.append(_join_row(cur, False, prose)) return "\n".join(out) def layout_text(blocks: list[list[_Line]], page_w: float) -> str: parts: list[str] = [] for lines, is_table in _zip_blocks([b for b in blocks if b], page_w): p = "\n".join(_rows(lines)) if is_table else _seq_text(lines) if not p.strip(): continue if parts and _is_row(parts[-1].rsplit("\n", 1)[-1]) and _is_row(p.split("\n", 1)[0]): parts[-1] += "\n" + p # consecutive table rows stay one block else: parts.append(p) return "\n\n".join(parts) def markdown_table(rows: list[list[Optional[str]]]) -> str: rows = [[re.sub(r"\s+", " ", c or "").strip() for c in r] for r in rows] rows = [r for r in rows if any(r)] if not rows: return "" n = max(len(r) for r in rows) rows = [r + [""] * (n - len(r)) for r in rows] # drop columns that are empty everywhere (merged-cell artefacts) keep = [j for j in range(n) if any(r[j] for r in rows)] rows = [[r[j] for j in keep] for r in rows] # no "| --- |" separator row: it carries no information for the model return "\n".join("| " + " | ".join(r) + " |" for r in rows) # ================================================================ PDF def _ingest_pdf(data: bytes, *, ocr, lang, dpi, min_chars, tables, preprocess, max_pages, ocr_timeout): import fitz # PyMuPDF warnings: list[str] = [] try: doc = fitz.open(stream=data, filetype="pdf") except Exception as e: raise ValueError(f"cannot open PDF: {type(e).__name__}: {e}") from None if doc.needs_pass and not doc.authenticate(""): raise ValueError("PDF is password-protected") plumber = None if tables: try: import pdfplumber plumber = pdfplumber.open(io.BytesIO(data)) except Exception as e: # pragma: no cover - broken/unsupported file for pdfplumber warnings.append(f"table detection unavailable: {type(e).__name__}") pages, methods, confs = [], [], [] n = doc.page_count if not max_pages else min(doc.page_count, max_pages) try: for i in range(n): page = doc[i] ppage = plumber.pages[i] if plumber is not None and i < len(plumber.pages) else None text, n_tab = _pdf_page_text(page, ppage, warnings) reason = _needs_ocr(page, text, min_chars) if reason: img = _render(page, dpi) otext, conf = _ocr_image(img, ocr=ocr, lang=lang, dpi=dpi, preprocess=preprocess, timeout=ocr_timeout) if len(otext.strip()) >= len(text.strip()): pages.append(otext) methods.append(f"ocr:{ocr}") confs.append(conf) continue pages.append(text) methods.append(f"text+tables({n_tab})" if n_tab else "text") confs.append(None) finally: if plumber is not None: plumber.close() if max_pages and doc.page_count > max_pages: warnings.append(f"PDF has {doc.page_count} pages; read the first {max_pages}") return pages, methods, confs, warnings def _needs_ocr(page, text: str, min_chars: int) -> Optional[str]: body = re.sub(r"\s", "", text) if len(body) < min_chars: return "little text" bad = body.count("�") + len(re.findall(r"$cid:\d+$", text)) * 6 if bad > 0.2 * len(body): return "unmapped glyphs" if len(body) < 300: try: area = abs(page.rect) img_area = sum(abs(r) for info in page.get_images(full=True) for r in page.get_image_rects(info[0])) if area and img_area > 0.5 * area: return "image page with a thin text layer" except Exception: pass return None def _render(page, dpi: int) -> np.ndarray: import fitz pix = page.get_pixmap(dpi=dpi, colorspace=fitz.csGRAY, alpha=False) a = np.frombuffer(pix.samples, dtype=np.uint8).reshape(pix.height, pix.stride)[:, : pix.width] return a.copy() def _pdf_page_text(page, ppage, warnings: list[str]) -> tuple[str, int]: tabs: list[tuple[tuple[float, float, float, float], str]] = [] if ppage is not None and page.rotation == 0: try: for t in ppage.find_tables(): rows = t.extract() if len(rows) >= 2 and max(len(r) for r in rows) >= 2: md = markdown_table(rows) if md: tabs.append((t.bbox, md)) except Exception as e: # pragma: no cover warnings.append(f"table detection failed on page {page.number + 1}: {type(e).__name__}") d = page.get_text("dict", flags=0) # no images; content-stream order blocks: list[list[_Line]] = [] placed: set[int] = set() items: list[Any] = [] # list of block line-lists or ("table", md) for b in d.get("blocks", []): if b.get("type", 0) != 0: continue cur: list[_Line] = [] blk_prose = _prose_lines([len("".join(sp.get("text", "") for sp in li.get("spans", [])).split()) for li in b.get("lines", [])]) for li in b.get("lines", []): spans = [s for s in li.get("spans", []) if s.get("text", "").strip()] if not spans: continue x0, y0, x1, y1 = li["bbox"] cx, cy = (x0 + x1) / 2, (y0 + y1) / 2 ti = next((k for k, (bb, _) in enumerate(tabs) if bb[0] - 1 <= cx <= bb[2] + 1 and bb[1] - 1 <= cy <= bb[3] + 1), None) if ti is not None: if cur: items.append(cur) cur = [] if ti not in placed: placed.add(ti) items.append(("table", tabs[ti][1])) continue txt = spans[0]["text"] sgaps = [s2["bbox"][0] - s1["bbox"][2] for s1, s2 in zip(spans, spans[1:])] size = max(1.0, statistics.median(sp.get("size", 10) for sp in spans)) breaks = _cell_breaks([sp["text"] for sp in spans], sgaps, 0.7 * size, prose=blk_prose) for sp, g, br in zip(spans[1:], sgaps, breaks): sep = " | " if br else ("" if g < 0.15 * size else " ") txt += sep + sp["text"] cur.append(_Line(x0, y0, x1, y1, re.sub(r"\s+", " ", txt).strip())) if cur: items.append(cur) for ti, (_, md) in enumerate(tabs): if ti not in placed: items.append(("table", md)) # layout: consecutive line blocks go through layout_text; tables are inserted verbatim parts: list[str] = [] run: list[list[_Line]] = [] for it in items: if isinstance(it, tuple): if run: parts.append(layout_text(run, page.rect.width)) run = [] parts.append(it[1]) else: run.append(it) if run: parts.append(layout_text(run, page.rect.width)) return clean_text("\n\n".join(parts)), len(tabs) # ================================================================ images + OCR def _ingest_image(data: bytes, *, ocr, lang, dpi, min_chars, tables, preprocess, max_pages, ocr_timeout): from PIL import Image, ImageOps Image.MAX_IMAGE_PIXELS = 200_000_000 warnings: list[str] = [] try: im = Image.open(io.BytesIO(data)) except Exception as e: raise ValueError(f"cannot open image: {type(e).__name__}: {e}") from None n_frames = getattr(im, "n_frames", 1) pages, methods, confs = [], [], [] for f in range(n_frames if not max_pages else min(n_frames, max_pages)): if n_frames > 1: im.seek(f) fr = ImageOps.exif_transpose(im) if f == 0 else im src_dpi = None try: src_dpi = float(fr.info.get("dpi", (0, 0))[0]) or None except Exception: pass gray = np.asarray(fr.convert("L")) text, conf = _ocr_image(gray, ocr=ocr, lang=lang, dpi=src_dpi, preprocess=preprocess, timeout=ocr_timeout) pages.append(text) methods.append(f"ocr:{ocr}") confs.append(conf) return pages, methods, confs, warnings def _ocr_image(gray: np.ndarray, *, ocr: str, lang: str, dpi: Optional[float], preprocess: bool, timeout: float) -> tuple[str, Optional[float]]: if preprocess: gray, scale = prepare_image(gray, upscale=ocr == "tesseract") # RapidOCR rescales internally dpi = (dpi or 300) * scale if dpi else None if ocr == "rapidocr": blocks, conf = _rapidocr_blocks(gray) else: blocks, conf = _tesseract_blocks(gray, lang=lang, dpi=dpi, timeout=timeout) return clean_text(layout_text(blocks, gray.shape[1])), conf # ---------------------------------------------------------------- preprocessing def prepare_image(gray: np.ndarray, upscale: bool = True) -> tuple[np.ndarray, float]: """Photo/scan cleanup before OCR: perspective correction, denoising, illumination flattening, deskew and (``upscale``) enlarging small text. Returns (image, scale factor applied).""" import cv2 g = gray g = _dewarp_page(g) g = _denoise(g) g = _flatten_illumination(g) g = _deskew(g) xh = _text_height(g) if upscale else None scale = 1.0 if xh and xh < 20: scale = min(3.0, 26.0 / xh, 6000.0 / max(g.shape)) if scale > 1.15: g = cv2.resize(g, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC) else: scale = 1.0 return g, scale def _small(g: np.ndarray, target: int = 1000) -> tuple[np.ndarray, float]: import cv2 f = target / max(g.shape) if f >= 1: return g, 1.0 return cv2.resize(g, None, fx=f, fy=f, interpolation=cv2.INTER_AREA), f def _quad_angles_ok(q: np.ndarray, tol: float = 25.0) -> bool: """Every interior angle of the quadrilateral within 90 +- tol degrees (a page seen in perspective; a region cut off by a shadow edge or a fold is rejected).""" for i in range(4): a, b, c = q[i - 1], q[i], q[(i + 1) % 4] v1, v2 = a - b, c - b cos = float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2) + 1e-9)) if abs(np.degrees(np.arccos(np.clip(cos, -1, 1))) - 90) > tol: return False return True def _order_quad(q: np.ndarray) -> np.ndarray: s, d = q.sum(1), np.diff(q, axis=1).ravel() return np.array([q[np.argmin(s)], q[np.argmin(d)], q[np.argmax(s)], q[np.argmax(d)]], np.float32) def _page_quad(s: np.ndarray) -> Optional[np.ndarray]: """Largest plausible page quadrilateral in a (downscaled) photo, from several binarizations: page-vs-background edges (robust to shadows falling across the page) and global thresholds.""" import cv2 H, W = s.shape blur = cv2.GaussianBlur(s, (5, 5), 0) masks = [] edges = cv2.Canny(blur, 30, 90) masks.append(cv2.dilate(edges, np.ones((5, 5), np.uint8))) otsu, _ = cv2.threshold(blur, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) for t in (otsu, 0.75 * otsu, 0.55 * otsu): _, th = cv2.threshold(blur, t, 255, cv2.THRESH_BINARY) masks.append(cv2.morphologyEx(th, cv2.MORPH_CLOSE, np.ones((9, 9), np.uint8))) border = 0.006 * max(H, W) for m in masks: # priority order: edges first, then thresholds from strict to loose cnts, _ = cv2.findContours(m, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) for c in sorted(cnts, key=cv2.contourArea, reverse=True)[:2]: hull = cv2.convexHull(c) area = cv2.contourArea(hull) if area < 0.25 * H * W or area > 0.97 * H * W: continue peri = cv2.arcLength(hull, True) for eps in (0.02, 0.03, 0.05): ap = cv2.approxPolyDP(hull, eps * peri, True) if len(ap) == 4: q = _order_quad(ap.reshape(4, 2).astype(np.float32)) inside = bool(((q[:, 0] > border) & (q[:, 0] < W - 1 - border) & (q[:, 1] > border) & (q[:, 1] < H - 1 - border)).all()) if inside and _quad_angles_ok(q) and cv2.contourArea(q) >= 0.9 * area: return q # a corner on the frame border = leaked into the background break return None def _dewarp_page(g: np.ndarray) -> np.ndarray: """Perspective-correct a photographed page: the largest plausible page quadrilateral (>= 25% of the frame, near-right angles) that is clearly brighter than its surroundings.""" import cv2 s, f = _small(g) quad = _page_quad(s) if quad is None: return g mask = np.zeros_like(s) cv2.fillConvexPoly(mask, quad.astype(np.int32), 255) inside, outside = s[mask > 0], s[mask == 0] if outside.size < 0.03 * s.size or float(np.median(inside)) - float(np.median(outside)) < 40: return g tl, tr, br, bl = quad / f w = int(max(np.linalg.norm(tr - tl), np.linalg.norm(br - bl))) h = int(max(np.linalg.norm(bl - tl), np.linalg.norm(br - tr))) if w < 100 or h < 100: return g M = cv2.getPerspectiveTransform(np.array([tl, tr, br, bl], np.float32), np.array([[0, 0], [w - 1, 0], [w - 1, h - 1], [0, h - 1]], np.float32)) return cv2.warpPerspective(g, M, (w, h), flags=cv2.INTER_CUBIC, borderValue=255) def _denoise(g: np.ndarray) -> np.ndarray: """3x3 median filter when the paper shows pixel noise / specks (noisy scans, grainy photos).""" import cv2 med = cv2.medianBlur(g, 3) thr, _ = cv2.threshold(med, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) bg = med > thr + 10 if bg.mean() < 0.2: return g noise = float(np.std(g[bg].astype(np.float32) - med[bg])) return med if noise > 3.0 else g def _flatten_illumination(g: np.ndarray) -> np.ndarray: """Divide by a smooth background estimate when the paper brightness varies across the page.""" import cv2 s, f = _small(g, 600) bg = cv2.medianBlur(cv2.morphologyEx(s, cv2.MORPH_CLOSE, np.ones((15, 15), np.uint8)), 31) lo, hi = np.percentile(bg, [5, 95]) if hi - lo < 30: return g bg_full = cv2.resize(bg, (g.shape[1], g.shape[0]), interpolation=cv2.INTER_LINEAR).astype(np.float32) out = np.clip(g.astype(np.float32) / np.maximum(bg_full, 1) * 245.0, 0, 255) return out.astype(np.uint8) def _deskew(g: np.ndarray, max_angle: float = 5.0) -> np.ndarray: """Projection-profile deskew: the angle that maximizes the variance of row ink sums.""" import cv2 s, _ = _small(g, 1200) _, th = cv2.threshold(s, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) if th.mean() < 0.5: # (almost) blank return g h, w = th.shape center = (w / 2, h / 2) def score(a): M = cv2.getRotationMatrix2D(center, a, 1.0) r = cv2.warpAffine(th, M, (w, h), flags=cv2.INTER_NEAREST, borderValue=0) return float(np.var(r.sum(1, dtype=np.float64))) angles = np.arange(-max_angle, max_angle + 1e-6, 0.5) best = max(angles, key=score) fine = np.arange(best - 0.4, best + 0.41, 0.1) best = float(max(fine, key=score)) if abs(best) < 0.25: return g M = cv2.getRotationMatrix2D((g.shape[1] / 2, g.shape[0] / 2), best, 1.0) return cv2.warpAffine(g, M, (g.shape[1], g.shape[0]), flags=cv2.INTER_CUBIC, borderValue=255) def _text_height(g: np.ndarray) -> Optional[float]: """Median height of character-sized connected components (~ x-height), in pixels.""" import cv2 _, th = cv2.threshold(g, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) n, _, stats, _ = cv2.connectedComponentsWithStats(th, connectivity=8) if n < 20: return None hs = stats[1:, cv2.CC_STAT_HEIGHT] ws = stats[1:, cv2.CC_STAT_WIDTH] ok = (hs >= 3) & (hs <= 150) & (ws <= 3 * hs) & (ws >= 1) if ok.sum() < 20: return None return float(np.median(hs[ok])) # ---------------------------------------------------------------- engines def _tesseract_blocks(gray: np.ndarray, lang: str = "eng", dpi: Optional[float] = None, psm: int = 3, timeout: float = 120.0) -> tuple[list[list[_Line]], Optional[float]]: import cv2 exe = shutil.which("tesseract") if not exe: raise RuntimeError("tesseract is not installed (apt install tesseract-ocr), or use ocr='rapidocr'") ok, png = cv2.imencode(".png", gray) env = dict(os.environ) env.setdefault("OMP_THREAD_LIMIT", "1") cmd = [exe, "stdin", "stdout", "-l", lang, "--psm", str(psm)] if dpi: cmd += ["--dpi", str(int(round(dpi)))] else: cmd += ["--dpi", "300"] cmd.append("tsv") r = subprocess.run(cmd, input=png.tobytes(), capture_output=True, timeout=timeout, env=env) if r.returncode != 0: raise RuntimeError(f"tesseract failed: {r.stderr.decode(errors='replace')[:300]}") rows = r.stdout.decode("utf-8", errors="replace").splitlines()[1:] lines: dict[tuple[int, int, int], list[tuple[int, int, int, int, float, str]]] = {} order: list[tuple[int, int, int]] = [] confs, weights = [], [] for row in rows: f = row.split("\t") if len(f) < 12 or f[0] != "5": continue txt = f[11].strip() conf = float(f[10]) if not txt or conf < 0: continue key = (int(f[2]), int(f[3]), int(f[4])) if key not in lines: lines[key] = [] order.append(key) x, y, w, h = int(f[6]), int(f[7]), int(f[8]), int(f[9]) lines[key].append((x, y, x + w, y + h, conf, txt)) confs.append(conf) weights.append(len(txt)) blocks: dict[int, list[_Line]] = {} border: list[int] = [] wc: dict[int, list[int]] = {} for key in order: wc.setdefault(key[0], []).append(len(lines[key])) for key in order: ws = _fix_pipes(sorted(lines[key], key=lambda w: w[0])) if not ws: continue hs = sorted(w[3] - w[1] for w in ws) hmed = hs[len(hs) // 2] txt = ws[0][5] breaks = _cell_breaks([w[5] for w in ws], [b[0] - a[2] for a, b in zip(ws, ws[1:])], hmed, prose=_prose_lines(wc[key[0]])) for b, br in zip(ws[1:], breaks): txt += (" | " if br else " ") + b[5] ln = _Line(min(w[0] for w in ws), min(w[1] for w in ws), max(w[2] for w in ws), max(w[3] for w in ws), txt, par=key[1]) if key[0] not in blocks: blocks[key[0]] = [] border.append(key[0]) blocks[key[0]].append(ln) conf = float(np.average(confs, weights=weights)) if confs else None return [blocks[b] for b in border], conf def _fix_pipes(ws: list[tuple]) -> list[tuple]: """Tesseract reads a capital I as "|": a lone "|" followed at a normal word gap by a lowercase word becomes "I"; other lone bars (table rules) are dropped (cells are split by gaps anyway).""" out = [] for j, w in enumerate(ws): if w[5] in ("|", "||", "|,"): nxt = ws[j + 1] if j + 1 < len(ws) else None h = w[3] - w[1] if nxt is not None and nxt[5][:1].islower() and nxt[0] - w[2] < 1.2 * h: out.append(w[:5] + ("I" + w[5][1:].replace("|", ""),)) continue out.append(w) return out _RAPID = None def _rapidocr_engine(): global _RAPID if _RAPID is None: try: from rapidocr_onnxruntime import RapidOCR except ImportError as e: # pragma: no cover raise RuntimeError("ocr='rapidocr' needs `pip install rapidocr_onnxruntime`") from e n = int(os.environ.get("JEVLIKE_OCR_THREADS", "0") or 0) # no dilation of the detector's text map: it merges adjacent lines of blurred text (photos) kw: dict[str, Any] = {"det_use_dilation": False} if n > 0: kw.update(intra_op_num_threads=n, inter_op_num_threads=1) rec = os.environ.get("JEVLIKE_RAPIDOCR_REC") if rec: kw["rec_model_path"] = rec _RAPID = RapidOCR(**kw) return _RAPID def _rapidocr_blocks(gray: np.ndarray) -> tuple[list[list[_Line]], Optional[float]]: import cv2 eng = _rapidocr_engine() # no 180-degree line classifier: pages are upright after deskew, and the classifier flips # (and then garbles or drops) some long text lines res, _ = eng(cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR), use_cls=False) lines = [] confs, weights = [], [] for box, txt, score in res or []: txt = str(txt).strip() if not txt: continue pts = np.asarray(box, dtype=float) lines.append(_Line(pts[:, 0].min(), pts[:, 1].min(), pts[:, 0].max(), pts[:, 1].max(), txt)) confs.append(float(score) * 100) weights.append(len(txt)) conf = float(np.average(confs, weights=weights)) if confs else None if not lines: return [], conf return _xy_cut(lines), conf def _gaps(iv: list[tuple[float, float]]) -> list[tuple[float, float]]: """Whitespace gaps between the union of 1-D intervals: [(width, midpoint)].""" iv = sorted(iv) out, end = [], iv[0][1] for a, b in iv[1:]: if a > end: out.append((a - end, (end + a) / 2)) end = max(end, b) return out def _interior(ls: list[_Line], a: float, b: float) -> bool: """A band [a, b] crossed by some boxes is a real gutter only if every crossing box has column text above and below it on both sides (a merged line inside two columns), not a full-width line at the top or bottom of the region (a title or a paragraph above a table: those are split off by a horizontal cut instead).""" cross = [l for l in ls if l.x0 < a and l.x1 > b] left = [l for l in ls if l.x1 <= a] right = [l for l in ls if l.x0 >= b] if not left or not right: return False for c in cross: for side in (left, right): if not (any(l.yc < c.y0 for l in side) and any(l.yc > c.y1 for l in side)): return False return True def _xy_cut(lines: list[_Line], depth: int = 0) -> list[list[_Line]]: """Reading order for OCR line boxes without layout info (RapidOCR). Recursive XY-cut: split into columns at the widest vertical whitespace gap (>= 1 line height); otherwise split horizontally at the widest gap that lets one side split into columns (a full-width title above a two-column body, a paragraph above a table). Leaves are blocks read top-to-bottom, with a new paragraph at vertical gaps > 0.6 line height.""" h = statistics.median(l.h for l in lines) def xgap(ls): """Widest vertical whitespace band; a band crossed by a few boxes (text merged across a narrow gutter by the detector) still counts when the rest of the region splits there.""" if len(ls) < 2: return (0.0, None) g = _gaps([(l.x0, l.x1) for l in ls]) best = max(g) if g else (0.0, None) tol = int(0.03 * len(ls)) if tol >= 1: x0, x1 = int(min(l.x0 for l in ls)), int(max(l.x1 for l in ls)) + 1 cov = np.zeros(x1 - x0 + 1, np.int32) for l in ls: cov[int(l.x0) - x0:int(l.x1) - x0 + 1] += 1 low = cov <= tol run, start = 0, 0 for i, v in enumerate(np.append(low, False)): if v: if run == 0: start = i run += 1 elif run: if start > 0 and i < len(low) and run > best[0] and _interior(ls, x0 + start, x0 + i): best = (float(run), x0 + start + run / 2) run = 0 return best def split(ls): """Column split of a region: a gutter >= 1 line height with running text (>= 4 words per box) on both sides. Table columns are not split: their rows are read across.""" gx, px = xgap(ls) if gx < 1.0 * h or px is None: return None left = [l for l in ls if (l.x0 + l.x1) / 2 < px] right = [l for l in ls if (l.x0 + l.x1) / 2 >= px] if not left or not right: return None if min(statistics.median(_nwords(l.text) for l in side) for side in (left, right)) < 4: return None return left, right if len(lines) > 1 and depth < 60: cols = split(lines) if cols: return _xy_cut(cols[0], depth + 1) + _xy_cut(cols[1], depth + 1) for gy, py in sorted(_gaps([(l.y0, l.y1) for l in lines]), reverse=True): if gy < 0.1 * h: break top = [l for l in lines if l.yc < py] bot = [l for l in lines if l.yc >= py] if top and bot and (split(top) or split(bot)): return _xy_cut(top, depth + 1) + _xy_cut(bot, depth + 1) # OCR cell boxes on the same printed row can differ by a few pixels in y. # Cluster their baselines before sorting horizontally, or a number can be # emitted before its row label and break the table association. row_groups: list[list[_Line]] = [] for line in sorted(lines, key=lambda l: l.yc): if row_groups and abs(line.yc - statistics.mean(x.yc for x in row_groups[-1])) < 0.5 * h: row_groups[-1].append(line) else: row_groups.append([line]) leaf = [line for row in row_groups for line in sorted(row, key=lambda l: l.x0)] par, prev = 0, None for l in leaf: if prev is not None and l.y0 > prev.yc and ( l.y0 - prev.y1 > 0.5 * min(l.h, prev.h) or max(l.h, prev.h) > 1.3 * min(l.h, prev.h)): par += 1 l.par = par if prev is None or l.y1 > prev.y1: prev = l return [leaf] # ================================================================ HTML _DROP_TAGS = ["script", "style", "noscript", "iframe", "svg", "canvas", "form", "button", "nav", "aside", "template", "select", "input", "object", "embed", "dialog", "menu"] _BOILER = re.compile(r"(^|[-_\s])(nav|navbar|menu|footer|sidebar|side-bar|cookie|consent|subscribe|newsletter|" r"share|sharing|social|breadcrumb|advert|ads?|promo|related|recommend|comments?|popup|modal|" r"banner|skip-link|signup|paywall)([-_\s]|$)", re.I) _BLOCK_TAGS = ["h1", "h2", "h3", "h4", "h5", "h6", "p", "li", "table", "pre", "blockquote", "dt", "dd", "figcaption", "caption", "address"] def html_to_text(html: str) -> str: """Main content of an HTML page as markdown-ish text (headings, paragraphs, lists, tables).""" from bs4 import BeautifulSoup try: soup = BeautifulSoup(html, "lxml") except Exception: soup = BeautifulSoup(html, "html.parser") title = soup.title.get_text(" ", strip=True) if soup.title else "" for t in soup(_DROP_TAGS): t.decompose() for t in soup.find_all(["header", "footer"]): if not t.find_parent(["article", "main"]): t.decompose() for t in soup.find_all(True): if t.attrs is None or t.name in ("html", "body", "article", "main"): continue ident = " ".join([t.get("id") or ""] + list(t.get("class") or [])) role = (t.get("role") or "").lower() hidden = t.get("aria-hidden") == "true" or "display:none" in (t.get("style") or "").replace(" ", "") if hidden or role in ("navigation", "banner", "contentinfo", "complementary", "search") or ( ident.strip() and _BOILER.search(ident)): t.decompose() arts = soup.find_all("article") root = (max(arts, key=lambda a: len(a.get_text())) if arts else None) or soup.find("main") or soup.body or soup out: list[str] = [] for el in root.find_all(_BLOCK_TAGS): if el.find_parent(_BLOCK_TAGS) is not None: # its text is emitted with the enclosing block continue if el.name == "table": rows = [[c.get_text(" ", strip=True) for c in tr.find_all(["th", "td"])] for tr in el.find_all("tr")] md = markdown_table(rows) if md: out.append(md) continue txt = el.get_text(" ", strip=True) txt = re.sub(r"\s+", " ", txt).strip() if not txt: continue if el.name in ("h1", "h2", "h3", "h4", "h5", "h6"): out.append("#" * int(el.name[1]) + " " + txt) elif el.name == "li": out.append("- " + txt) elif el.name == "pre": out.append(el.get_text().strip("\n")) else: out.append(txt) body = "\n\n".join(out) full = re.sub(r"\s+", " ", root.get_text(" ", strip=True)) if len(re.sub(r"\s", "", body)) < 0.3 * len(re.sub(r"\s", "", full)): body = root.get_text("\n", strip=True) # div soup without block tags if title and not re.search(r"^#\s", body, re.M) and title not in body[:500]: body = "# " + title + "\n\n" + body # join consecutive list items into one block body = re.sub(r"(^- .*)\n\n(?=- )", r"\1\n", body, flags=re.M) body = re.sub(r"(^- .*)\n\n(?=- )", r"\1\n", body, flags=re.M) return normalize_text(body) # ================================================================ questions from JSON + CLI def questions_from_json(spec: Mapping[str, Any]) -> dict: """Wire-format questions ({key: {"type", "instructions", "criteria"}}, as the HTTP API takes; a top-level {"questions": {...}} wrapper is accepted) -> public Choice / Score / Noul.""" from jevlike.types import Choice, Noul, Score spec = spec.get("questions", spec) if isinstance(spec, Mapping) else spec out = {} for key, q in spec.items(): t = q.get("type") if t == "choice": out[key] = Choice(q["instructions"], q["criteria"]) elif t == "score": c = q["criteria"] levels = [f"{k}: {d}" if d else k for k, d in c.items()] if isinstance(c, dict) else list(c) out[key] = Score(q["instructions"], levels) elif t == "noul": out[key] = Noul(q["instructions"]) else: raise ValueError(f"question {key!r}: unknown type {t!r}") return out def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Convert a document (PDF, image, HTML, text) to text and optionally " "answer SystemOne questions about it.") ap.add_argument("files", nargs="+", help="file(s); several files = pages of one document") ap.add_argument("--kind", choices=KINDS, default=None) ap.add_argument("--ocr", choices=OCR_ENGINES, default="tesseract") ap.add_argument("--lang", default="eng", help="tesseract language(s), e.g. eng+deu") ap.add_argument("--dpi", type=int, default=300) ap.add_argument("--min-chars", type=int, default=30) ap.add_argument("--no-tables", action="store_true") ap.add_argument("--no-preprocess", action="store_true") ap.add_argument("--keep-headers", action="store_true", help="do not strip repeated headers/footers") ap.add_argument("--max-pages", type=int, default=None) ap.add_argument("--json", action="store_true", help="print the IngestResult (and answers) as JSON") ap.add_argument("--questions", default=None, help="questions JSON (HTTP API format) to answer about the document") ap.add_argument("--checkpoint", default=None, help="checkpoint dir for --questions (e.g. checkpoints/jevlike-large-nf)") ap.add_argument("--device", default=None) args = ap.parse_args(argv) kw = dict(kind=args.kind, ocr=args.ocr, lang=args.lang, dpi=args.dpi, min_chars=args.min_chars, tables=not args.no_tables, preprocess=not args.no_preprocess, strip_headers=not args.keep_headers, max_pages=args.max_pages) src = args.files if len(args.files) > 1 else args.files[0] if args.questions: if not args.checkpoint: ap.error("--questions needs --checkpoint") from jevlike.predict import SystemOne qs = questions_from_json(json.loads(Path(args.questions).read_text())) model = SystemOne.load(args.checkpoint, device=args.device) res = predict_file(model, src, qs, ingest_kwargs=kw) print(json.dumps(res, indent=1, ensure_ascii=False, default=float)) return 0 r = to_text(src, **kw) if args.json: print(json.dumps(r.to_dict(), indent=1, ensure_ascii=False)) else: sys.stdout.write(r.text + "\n") info = ", ".join(f"{m}" for m in r.method_per_page) print(f"[{r.kind}: {len(r.pages)} page(s) [{info}], {len(r.text)} chars, {r.seconds:.2f}s]" + "".join(f"\n[warning] {w}" for w in r.warnings), file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())