"""Quality checks on the parse — catching the damage that makes NO noise. The basis is a finding in `Hasil-Uji-MinerU.md`: the EOQ formula came out of the CPU pipeline with a digit changed, no error and no warning. Damage like that will never show up in a "finished" status — it has to be looked for. Everything here is a WARNING, never a failure: the document is still processed, but the warning is recorded in the manifest so quality cannot quietly degrade across hundreds of documents. """ from __future__ import annotations import re import unicodedata from collections import Counter from pathlib import Path from typing import Any # Digits inside latex. A technical formula almost always has a number or a divisor. _HAS_DIGIT = re.compile(r"\d") # Minimum length of a digit run before it is matched against the source. # # Measured on a real document: at a 2-digit threshold the genuine bug (55600) is # caught, but six false warnings come with it from short numbers — the PDF's text # layer is partly broken, so short numbers often fail to match despite being # correct. At 3 digits the bug is still caught and the noise disappears. MIN_DIGITS = 3 _ALL_DIGITS = re.compile(r"\d+") def _digit_runs(text: str, min_digits: int) -> set[str]: return {n for n in _ALL_DIGITS.findall(text) if len(n) >= min_digits} def _digits_from_latex(latex: str, min_digits: int = MIN_DIGITS) -> set[str]: """MinerU writes latex character-spaced ('5 5 6 0 0'), so spaces go first.""" return _digit_runs(re.sub(r"\s+", "", latex), min_digits) def _digits_per_page(pdf: Path, min_digits: int = MIN_DIGITS) -> dict[int, set[str]] | None: """Digit runs in the source PDF's own text layer, per page. Used as the reference. If the PDF cannot be read, return None and skip this check silently — not checking is better than warning falsely. """ try: from pypdf import PdfReader reader = PdfReader(str(pdf)) except Exception: return None result: dict[int, set[str]] = {} for i, page in enumerate(reader.pages): try: text = page.extract_text() or "" except Exception: text = "" # drop thousands separators & spaces so "$5,600" and "5 600" -> "5600" cleaned = re.sub(r"[, \s]", "", text) result[i] = _digit_runs(cleaned, min_digits) return result _WORD = re.compile(r"[A-Za-z][A-Za-z0-9]*") def vocabulary_per_page(pdf: Path) -> dict[int, frozenset[str]] | None: """Words in the source PDF's own text layer, per page. Used by `render.restore_word_boundaries` as the arbiter: MinerU writes formulas one character at a time and the word boundaries are lost there, while the PDF still holds them intact (`MOHH x Qty x PA`, not `MOHHxQtyxPAxUAxPty`). NFKC is required because formulas use a mathematical font — this page literally contains `𝑀𝑂𝐻𝐻` (U+1D440…) rather than ASCII, and without normalisation not a single word would match. Returns None if the PDF cannot be read; the caller then skips recovery silently, exactly as the digit check below does. """ try: from pypdf import PdfReader reader = PdfReader(str(pdf)) except Exception: return None result: dict[int, frozenset[str]] = {} for i, page in enumerate(reader.pages): try: text = page.extract_text() or "" except Exception: text = "" result[i] = frozenset(_WORD.findall(unicodedata.normalize("NFKC", text))) return result MIN_COVERAGE = 0.90 # share of MinerU's own words that must be found in the text layer MIN_WORDS_TO_JUDGE = 20 # below this there is no basis for a judgement def trustworthy_vocabulary( vocabulary: dict[int, frozenset[str]], items: list[dict[str, Any]], min_coverage: float = MIN_COVERAGE, ) -> dict[int, frozenset[str]]: """Drop pages whose text layer is not fit to arbitrate. `vocabulary_per_page` trusts the PDF text layer outright. That is safe when the layer is intact and dangerous when it is HALF there: a legitimate word starts looking unknown, and word-boundary recovery then splits it. The measure judges itself rather than relying on a constant tuned to one document: how many of MinerU's own prose words on that page are actually found in its text layer. A low score means the two read the same page differently — and a source that disagrees with the parser is not fit to arbitrate it. Measured on three real documents: 99.6% · 98.1% · 96.0%. Pages falling well below that lose their vocabulary, so recovery is skipped and the text is left as it is — damaged but honest, rather than repaired into something wrong. """ trusted: dict[int, frozenset[str]] = {} for page, words in vocabulary.items(): if not words: continue matched = total = 0 for x in items: if x.get("type") != "text" or x.get("page_idx") != page: continue for w in re.findall(r"[A-Za-z][A-Za-z0-9]*", x.get("text") or ""): if len(w) < 4: continue total += 1 matched += w in words # Too little prose to judge -> no basis for trusting it. if total < MIN_WORDS_TO_JUDGE or matched / total < min_coverage: continue trusted[page] = words return trusted def check_items( items: list[dict[str, Any]], min_latex_len: int = 8, source_pdf: Path | None = None, page_offset: int = 0, ) -> dict[str, Any]: """Return a summary plus the list of warnings for one document. `source_pdf` enables matching numbers against the original PDF — the only way to catch a number that changed silently (the EOQ bug: 5600 -> 55600). """ per_type = Counter(x.get("type") for x in items) pages = {x.get("page_idx") for x in items if x.get("page_idx") is not None} warnings: list[dict[str, Any]] = [] # 1. A page with no text at all -> it probably failed to read text_per_page = Counter( x.get("page_idx") for x in items if x.get("type") in {"text", "table", "equation"} and (x.get("text") or x.get("table_body")) ) for p in sorted(pages): if text_per_page.get(p, 0) == 0: warnings.append({"kind": "empty_page", "page": p}) # 2. Formula truncated / missing its numbers <- the EOQ bug for i, x in enumerate(items): if x.get("type") != "equation": continue latex = (x.get("text") or "").strip() if len(latex) < min_latex_len: warnings.append({"kind": "latex_too_short", "item": i, "page": x.get("page_idx"), "content": latex[:60]}) elif not _HAS_DIGIT.search(latex) and "\\frac" not in latex: warnings.append({"kind": "latex_without_digits", "item": i, "page": x.get("page_idx"), "content": latex[:60]}) # 3. ⭐ A number in a formula that is absent from the source PDF -> it changed # silently. This is the ONLY check that can catch the EOQ bug; every other # check in this file only inspects our own output, and a wrong number still # produces perfectly valid latex. source_digits = _digits_per_page(source_pdf) if source_pdf else None if source_digits is not None: for i, x in enumerate(items): if x.get("type") != "equation": continue p = x.get("page_idx", 0) + page_offset reference = source_digits.get(p) if not reference: # page with no text layer -> cannot be judged continue for number in sorted(_digits_from_latex(x.get("text") or "")): if number not in reference: warnings.append({ "kind": "digits_absent_from_source", "item": i, "page": p, "digits": number, "note": "a number in the formula was not found in the source PDF text", }) # 4. Table with no body for i, x in enumerate(items): if x.get("type") == "table" and not (x.get("table_body") or "").strip(): warnings.append({"kind": "empty_table", "item": i, "page": x.get("page_idx")}) # 5. Chart with no context whatsoever (the next stage can do nothing with it) for i, x in enumerate(items): if x.get("type") == "chart": has_any = (x.get("content") or "").strip() or (x.get("chart_caption") or []) if not has_any: warnings.append({"kind": "chart_without_context", "item": i, "page": x.get("page_idx")}) return { "items": len(items), "pages_detected": len(pages), "per_type": dict(per_type), "warnings": warnings, "warning_count": len(warnings), }