"""Phase 2 — structured extraction of JoF PDFs with PyMuPDF (layout heuristics tuned on the JoF LaTeX template). Output: data/derived/parsed/.json (private: contains full text) with per-component confidence flags. """ from __future__ import annotations import json, re, unicodedata, statistics from pathlib import Path from dataclasses import dataclass, field, asdict import pandas as pd import pymupdf from .provenance import sha256_file, stamp, Ledger, now_iso from . import config PARSER_VERSION = "pymupdf-jof-heuristics-1.3" ROMAN = r"(?:I|II|III|IV|V|VI|VII|VIII|IX|X|XI|XII|XIII|XIV|XV)" SEC_RE = re.compile(rf"^{ROMAN}\.\s+\S") SUB_RE = re.compile(r"^[A-Z]\.(\d+\.)?\s+[A-Z0-9(“\"']") EQ_TAG = re.compile(r"\((\d{1,3}[a-z]?|[A-Z]\d{1,2}|IA\.?\d+)\)\s*$") TABLE_RE = re.compile(r"^Table\s+([IVXLC]+|[A-Z]?\d+|A\.?[IVX]+)(—Continued)?\s*$") FIG_RE = re.compile(r"^Figure\s+(\d+|[A-Z]\.?\d+)\.\s") CITE_RE = re.compile(r"\b([A-Z][A-Za-z'’\-]+(?:,? (?:and |& )?[A-Z][A-Za-z'’\-]+)*(?: et al\.)?)\s*\((\d{4}[a-z]?(?:,\s*\d{4}[a-z]?)*)\)") @dataclass class Line: page: int; x0: float; y0: float; x1: float; y1: float; font: str; size: float; text: str; bold: bool; italic: bool; mathfont: bool; bold_start: bool = False def norm(s: str) -> str: s = unicodedata.normalize("NFKC", s) s = "".join(ch for ch in s if ch.isprintable() or ch in "\t ") return s.replace("−", "-").replace("\xa0", " ") def page_lines(doc: pymupdf.Document, pno: int, rotated: list | None = None) -> list[Line]: page = doc[pno]; out = [] for b in page.get_text("dict")["blocks"]: if b["type"] != 0: continue for l in b["lines"]: spans = [s for s in l["spans"] if s["text"].strip()] if not spans: continue if abs(l.get("dir", (1, 0))[0]) < 0.5: # rotated (landscape table) line if rotated is not None: main = max(spans, key=lambda s: len(s["text"])) rotated.append(Line(pno + 1, *l["bbox"], main["font"], round(main["size"], 1), norm("".join(s["text"] for s in l["spans"])).strip(), "Bold" in main["font"], "Ital" in main["font"], False)) continue text = norm("".join(s["text"] for s in l["spans"])).strip() if not text: continue # dominant span by chars main = max(spans, key=lambda s: len(s["text"])) font = main["font"]; size = round(main["size"], 1) if "Times" in font and size < 5: continue # Wiley download watermark in margin x0, y0, x1, y1 = l["bbox"] if y1 - y0 > 3 * max(size, 1) and (x1 - x0) < 15: continue # vertical (rotated) text fragments out.append(Line(pno + 1, x0, y0, x1, y1, font, size, text, "Bold" in font, "Ital" in font, font.startswith(("MT", "RMT", "CM", "MS")), "Bold" in spans[0]["font"])) return out def is_header(l: Line) -> bool: return l.y0 < 56 def body_size(lines: list[Line]) -> float: sizes = [l.size for l in lines if "Schlbk-Roman" in l.font or "Roman" in l.font] try: return statistics.mode([s for s in sizes if s >= 9]) if any(s >= 9 for s in sizes) else 10.0 except statistics.StatisticsError: return 10.0 def parse_pdf(path: Path, doc_id: str, meta: dict) -> dict: doc = pymupdf.open(path) all_lines: list[Line] = []; rotated: list[Line] = [] for p in range(doc.page_count): all_lines += [l for l in page_lines(doc, p, rotated) if not is_header(l)] BS = body_size(all_lines) pw = doc[0].rect.width # ---------- front matter ---------- first = [l for l in all_lines if l.page <= 3] title_lines = [l for l in first if l.bold and l.size >= BS + 2.5] title = " ".join(l.text for l in title_lines).strip() or None title_page = title_lines[0].page if title_lines else 1 # authors: roman ~9pt lines between title and ABSTRACT abs_idx = next((i for i, l in enumerate(all_lines) if l.text.upper().startswith("ABSTRACT") and l.page <= 3), None) authors = None; abstract = None; abstract_conf = 0.0 if abs_idx is not None: t_end = all_lines.index(title_lines[-1]) if title_lines else 0 a_lines = [l for l in all_lines[t_end + 1:abs_idx] if l.size < BS and l.size >= BS - 1.5 and not l.bold] authors = " ".join(l.text for l in a_lines).replace("∗", "").strip() or None ab = [] for l in all_lines[abs_idx + 1:]: if l.size >= BS - 0.5 or l.page > title_page + 1: break if l.size <= BS - 1.5 and not l.bold: ab.append(l.text) abstract = " ".join(ab).strip() or None abstract_conf = 1.0 if abstract and 400 < len(abstract) < 2500 else (0.5 if abstract else 0.0) # ---------- structural pass ---------- sections = []; cur = {"number": None, "heading": "Introduction (unnumbered)", "level": 0, "page": 1, "paragraphs": [], "subsections": [], "_lines": [], "is_intro": True} equations, tables, figures, footnotes, refs_lines = [], [], [], [], [] in_refs = False; in_appendix = False; refs_page = None; appendix_page = None; supporting_page = None in_table_block = False; table_block_page = None fig_cap_y = {} for l in all_lines: if (l.bold or l.bold_start) and l.size <= BS - 1.5 and FIG_RE.match(l.text): fig_cap_y[l.page] = min(l.y0, fig_cap_y.get(l.page, 1e9)) i = 0; n = len(all_lines) body_left = statistics.median([l.x0 for l in all_lines if l.size == BS and not l.italic]) if any(l.size == BS for l in all_lines) else 61 while i < n: l = all_lines[i]; t = l.text centered = abs((l.x0 + l.x1) / 2 - pw / 2) < 40 # page number lines (digits only, near bottom) if re.fullmatch(r"\d{1,4}", t) and l.y0 > 600: i += 1; continue # references if l.bold and l.size >= BS - 0.5 and t.upper().strip() in ("REFERENCES", "REFERENCE"): in_refs = True; refs_page = l.page; in_table_block = False; i += 1; continue if l.bold and l.size >= BS - 0.5 and t.startswith("Supporting Information"): supporting_page = l.page; in_refs = False; i += 1; continue if in_refs: if l.bold and l.size >= BS - 0.5 and (t.lower().startswith("appendix") or SEC_RE.match(t)): in_refs = False # appendix after refs else: if l.size <= BS - 1.5 or l.size == BS: refs_lines.append(l) i += 1; continue # table caption if l.bold and l.size <= BS - 1.5 and TABLE_RE.match(t): m = TABLE_RE.match(t) if m.group(2): in_table_block = True; table_block_page = l.page; i += 1; continue ttl = []; j = i + 1 while j < n and all_lines[j].bold and all_lines[j].size >= BS - 0.5 and all_lines[j].page == l.page: ttl.append(all_lines[j].text); j += 1 notes = []; k = j while k < n and all_lines[k].page == l.page and all_lines[k].size <= BS - 1.5 and not all_lines[k].bold and len(notes) < 25: notes.append(all_lines[k].text); k += 1 if sum(len(x) for x in notes) > 2500: break tables.append({"label": f"Table {m.group(1)}", "title": " ".join(ttl) or None, "page": l.page, "notes_excerpt": " ".join(notes)[:1500], "rotated": l.x1 - l.x0 < 20, "section": cur["heading"]}) in_table_block = True; table_block_page = l.page; i = k; continue if in_table_block and l.page != table_block_page: in_table_block = False # figure caption if (l.bold or l.bold_start) and l.size <= BS - 1.5 and FIG_RE.match(t): cap = [t]; j = i + 1 while j < n and all_lines[j].page == l.page and all_lines[j].size <= BS - 1.5 and not FIG_RE.match(all_lines[j].text) and len(cap) < 15: cap.append(all_lines[j].text); j += 1 figures.append({"label": FIG_RE.match(t).group(0).strip().rstrip("."), "caption": " ".join(cap)[:1500], "page": l.page, "section": cur["heading"]}) i = j; continue # section heading (bold, body size, centered or starts with roman numeral) if l.bold and l.size >= BS - 0.5 and l.size < BS + 2 and (SEC_RE.match(t) or centered) and not TABLE_RE.match(t) and len(t) < 120: head = [t]; j = i + 1 while j < n and all_lines[j].bold and all_lines[j].size >= BS - 0.5 and all_lines[j].page == l.page and not SEC_RE.match(all_lines[j].text) and abs(all_lines[j].y0 - all_lines[j-1].y1) < 8: head.append(all_lines[j].text); j += 1 h = " ".join(head) m = re.match(rf"^({ROMAN})\.\s+(.*)$", h) is_app = h.lower().startswith("appendix") or (cur["heading"].lower().startswith("appendix") and m is None) if h.lower().startswith("appendix"): in_appendix = True; appendix_page = appendix_page or l.page sections.append(cur) cur = {"number": m.group(1) if m else None, "heading": m.group(2) if m else h, "level": 1, "page": l.page, "paragraphs": [], "subsections": [], "_lines": [], "is_appendix": in_appendix or is_app} i = j; continue # subsection heading (italic, body size, short, pattern "A. Title") if l.italic and l.size >= BS - 0.5 and SUB_RE.match(t) and len(t) < 110 and not t.endswith((",", ";")) and (i + 1 < n and (all_lines[i+1].size <= BS - 1.5 or not all_lines[i+1].italic or all_lines[i+1].y0 - l.y1 > 3)): cur["subsections"].append({"heading": t, "page": l.page, "para_index": len(cur["paragraphs"]) + (1 if cur["_lines"] else 0)}) cur["_lines"].append(("SUBHEAD", t)); i += 1; continue # footnotes: small (BS-2) lines in lower part preceded by tiny number, or small lines once a footnote started on this page if l.size <= BS - 1.5 and not in_table_block: if l.page in fig_cap_y and l.y0 < fig_cap_y[l.page]: i += 1; continue # text inside a figure if len(re.sub(r"[^A-Za-z]", "", t)) < 3 and not t.startswith("∗"): i += 1; continue # axis ticks / stray numbers if l.page <= 3 and abstract and l.y0 < 300 and l.size <= BS - 1.5 and l.x0 > body_left + 15: i += 1; continue # abstract already captured if l.size <= BS - 3.5 or re.match(r"^\d{1,3}\s", t) or (not footnotes) or footnotes[-1]["page"] != l.page or t.startswith("∗"): footnotes.append({"page": l.page, "text": t, "n_lines": 1}) else: footnotes[-1]["text"] = join_lines([footnotes[-1]["text"], t]); footnotes[-1]["n_lines"] += 1 i += 1; continue # equations: numbered tag at right margin if EQ_TAG.search(t) and l.x1 > pw * 0.8 and l.size >= BS - 0.5 and (len(t) < 6 or l.italic or l.mathfont or re.search(r"[=+−×∑∫]", t)): tag = EQ_TAG.search(t).group(1) band = [x for x in all_lines[max(0, i-6):i] if x.page == l.page and abs(x.y0 - l.y0) < 14 and x is not l] body = " ".join(x.text for x in band) + (" " + t[:EQ_TAG.search(t).start()].strip() if len(t) > 6 else "") equations.append({"tag": tag, "page": l.page, "text_excerpt": body.strip()[:300], "section": cur["heading"], "confidence": 0.9 if body.strip() else 0.5}) i += 1; continue if l.mathfont: cur["_lines"].append(("MATH", t)); i += 1; continue # body line if l.size >= BS - 0.5 and l.size <= BS + 0.5: cur["_lines"].append(("P_NEW" if l.x0 > body_left + 6 else "P", t, l.page)) i += 1 sections.append(cur) # rotated (landscape) tables for idx, l in enumerate(rotated): m = TABLE_RE.match(l.text) if l.bold and m and not m.group(2): ttl = [x.text for x in rotated[idx+1:idx+4] if x.bold and x.page == l.page and x.size >= BS - 0.5] notes = [x.text for x in rotated[idx+1:idx+30] if x.page == l.page and not x.bold and x.size <= BS - 1.5] tables.append({"label": f"Table {m.group(1)}", "title": " ".join(ttl) or None, "page": l.page, "notes_excerpt": " ".join(notes)[:1500], "rotated": True, "section": None}) tables.sort(key=lambda t: t["page"]) # ---------- paragraphs from lines ---------- def build_paras(sec): paras = []; buf = []; pg = sec["page"] for item in sec["_lines"]: kind = item[0] if kind == "SUBHEAD": if buf: paras.append({"text": join_lines(buf), "page": pg}); buf = [] paras.append({"text": item[1], "page": pg, "kind": "subheading"}); continue if kind == "MATH": if buf: buf.append(item[1]); continue continue if kind == "P_NEW" and buf: paras.append({"text": join_lines(buf), "page": pg}); buf = [] if not buf: pg = item[2] buf.append(item[1]) if buf: paras.append({"text": join_lines(buf), "page": pg}) sec["paragraphs"] = [p for p in paras if len(p["text"]) > 1] del sec["_lines"] sec["n_words"] = sum(len(p["text"].split()) for p in sec["paragraphs"] if p.get("kind") != "subheading") return sec sections = [build_paras(s) for s in sections] sections = [s for s in sections if s["paragraphs"] or not s.get("is_intro")] # ---------- references ---------- references = split_references(refs_lines) # ---------- citations in text ---------- body_text = "\n".join(p["text"] for s in sections for p in s["paragraphs"] if p.get("kind") != "subheading") cites = CITE_RE.findall(body_text) cite_strings = [f"{a} ({y})" for a, y in cites] # ---------- roles & confidence ---------- numbered = [s for s in sections if s.get("number")] roman_seq = [s["number"] for s in numbered] conf = { "title": 1.0 if title else 0.0, "authors": 1.0 if authors else 0.0, "abstract": abstract_conf, "sections": 1.0 if len(numbered) >= 3 and roman_seq == sorted(roman_seq, key=roman_val) else (0.6 if len(numbered) >= 2 else 0.2), "references": 1.0 if len(references) >= 10 else (0.5 if references else 0.0), "equations": 0.9 if equations else 0.7, "tables": 0.9 if tables else 0.7, "figures": 0.9, } conf["overall"] = round(sum(conf.values()) / len(conf), 3) out = { "doc_id": doc_id, "source_sha256": sha256_file(path), "source_relpath": meta.get("relpath"), "parser": PARSER_VERSION, "parsed_at": now_iso(), "pages": doc.page_count, "body_font_size": BS, "metadata": {"title": title, "authors": authors, "doi": meta.get("bib_doi"), "year": meta.get("bib_year"), "volume": meta.get("bib_volume"), "issue": meta.get("bib_issue"), "pages": meta.get("bib_pages"), "article_type": meta.get("article_type"), "bib_title": meta.get("bib_title"), "bib_authors": meta.get("bib_authors")}, "abstract": abstract, "sections": sections, "equations": equations, "tables": tables, "figures": figures, "footnotes": footnotes, "references": references, "citations_in_text": cite_strings, "markers": {"references_page": refs_page, "appendix_page": appendix_page, "supporting_information_page": supporting_page, "has_internet_appendix": bool(supporting_page) or "Internet Appendix" in body_text}, "stats": {"n_sections": len(numbered), "n_subsections": sum(len(s["subsections"]) for s in sections), "n_paragraphs": sum(len(s["paragraphs"]) for s in sections), "n_words_body": sum(s["n_words"] for s in sections if not s.get("is_appendix")), "n_words_appendix": sum(s["n_words"] for s in sections if s.get("is_appendix")), "n_equations": len(equations), "n_tables": len(tables), "n_figures": len(figures), "n_footnotes": len(footnotes), "n_references": len(references), "n_citations_in_text": len(cite_strings), "abstract_words": len(abstract.split()) if abstract else 0}, "confidence": conf, } doc.close() return out def join_lines(lines: list[str]) -> str: s = "" for x in lines: if s.endswith("-") and not s.endswith("--") and x[:1].islower(): s = s[:-1] + x else: s = (s + " " + x).strip() return s def split_references(lines: list[Line]) -> list[dict]: if not lines: return [] # hanging indent: entry starts at the page's left margin (odd/even pages have different margins) left_by_page = {} for l in lines: left_by_page.setdefault(l.page, []).append(l.x0) left_by_page = {k: min(v) for k, v in left_by_page.items()} entries = []; buf = []; pg = None lines = [l for l in lines if not re.match(r"^DOI:\s*10\.", l.text)] for l in lines: if l.x0 <= left_by_page[l.page] + 3 and buf: entries.append({"text": join_lines(buf), "page": pg}); buf = [] if not buf: pg = l.page buf.append(l.text) if buf: entries.append({"text": join_lines(buf), "page": pg}) out = [] for e in entries: m = re.match(r"^(.*?),\s*(\d{4}[a-z]?),\s*(.*)$", e["text"]) e["authors"] = m.group(1) if m else None; e["year"] = m.group(2) if m else None; e["rest"] = m.group(3) if m else None e["doi"] = (re.search(r"10\.\d{4,9}/\S+", e["text"]) or [None])[0] if "10." in e["text"] else None out.append(e) return out def roman_val(r: str) -> int: v = {"I": 1, "V": 5, "X": 10}; t = 0 for a, b in zip(r, r[1:] + " "): t += -v[a] if b != " " and v[a] < v[b] else v[a] return t def parse_corpus(limit: int | None = None, force: bool = False, only: str | None = None) -> dict: man = pd.read_parquet(config.manifests_dir() / "corpus_manifest.parquet") inc = man[man["decision"] == "include"] if only: inc = inc[inc["doc_id"] == only] if limit: inc = inc.head(limit) root = config.corpus_root(); outdir = config.derived_dir() / "parsed"; outdir.mkdir(parents=True, exist_ok=True) ledger = Ledger(config.derived_dir() / "parse_ledger.jsonl") done = ok = fail = 0; rows = [] for r in inc.itertuples(): outp = outdir / f"{r.doc_id}.json" if outp.exists() and not force: d = json.loads(outp.read_text()); rows.append(summary_row(d)); done += 1; continue try: d = parse_pdf(root / r.relpath, r.doc_id, r._asdict()) outp.write_text(json.dumps(d, ensure_ascii=False)); rows.append(summary_row(d)); ok += 1 ledger.append({"event": "parsed", "doc_id": r.doc_id, "confidence": d["confidence"]["overall"], "stats": d["stats"], **stamp()}) except Exception as e: fail += 1; ledger.append({"event": "parse_error", "doc_id": r.doc_id, "error": repr(e)[:300], **stamp()}) rows.append({"doc_id": r.doc_id, "error": repr(e)[:200]}) df = pd.DataFrame(rows); df.to_parquet(config.derived_dir() / "parse_summary.parquet", index=False); df.to_csv(config.derived_dir() / "parse_summary.csv", index=False) return {"parsed_new": ok, "cached": done, "failed": fail, "n": len(rows), "mean_confidence": float(df["confidence"].mean()) if "confidence" in df else None} def summary_row(d: dict) -> dict: return {"doc_id": d["doc_id"], "year": d["metadata"]["year"], "article_type": d["metadata"]["article_type"], "pages": d["pages"], "confidence": d["confidence"]["overall"], **{f"conf_{k}": v for k, v in d["confidence"].items() if k != "overall"}, **d["stats"], "has_abstract": bool(d["abstract"]), "refs_page": d["markers"]["references_page"], "has_internet_appendix": d["markers"]["has_internet_appendix"]}