Add PDF→markdown batch converter and research-library workflow
convert.py walks pdfs/ (recursing topic subfolders), mirrors a .md tree into md/ via pymupdf4llm, idempotent on mtime. Detects no-text-layer PDFs (needs-ocr.txt) and falls back to plain per-page text when pymupdf4llm's layout pass returns near-empty despite a real text layer. Pin pymupdf4llm==0.3.4 (lightweight line; 1.27.x bundles an ML/OCR pipeline that fails on plain text PDFs). PDFs gitignored (copyrighted, large) — only generated markdown is committed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
159
convert.py
Normal file
159
convert.py
Normal file
@@ -0,0 +1,159 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Batch-convert a tree of text PDFs to markdown for whole-file LLM reading.
|
||||
|
||||
Walks a source directory (recursing into topic subfolders), converts each PDF to
|
||||
markdown with pymupdf4llm, and writes a mirrored .md tree under the output dir.
|
||||
|
||||
Design notes:
|
||||
- Idempotent: a PDF is skipped when its .md already exists and is newer.
|
||||
- No-text-layer PDFs (scans) yield ~no extractable text. Those are detected
|
||||
cheaply *before* the expensive markdown pass, logged to needs-ocr.txt, and
|
||||
left unwritten (no empty markdown) so they can go through an OCR tool later.
|
||||
- Output is optimized for a model reading the entire file: a small provenance
|
||||
header, then page-separated markdown that keeps headings / lists / tables.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pymupdf # bundled with pymupdf4llm
|
||||
import pymupdf4llm
|
||||
|
||||
# A scan (image-only PDF) extracts almost nothing. Real text PDFs comfortably
|
||||
# exceed this; the threshold only has to separate "basically empty" from "has a
|
||||
# text layer", so it is deliberately low to avoid false OCR flags.
|
||||
MIN_CHARS_PER_PAGE = 50
|
||||
|
||||
|
||||
def extractable_chars(pdf_path: Path) -> tuple[int, int]:
|
||||
"""Return (total_text_chars, page_count) using a cheap text extraction."""
|
||||
with pymupdf.open(pdf_path) as doc:
|
||||
pages = doc.page_count
|
||||
total = sum(len(page.get_text("text")) for page in doc)
|
||||
return total, pages
|
||||
|
||||
|
||||
def has_text_layer(total_chars: int, pages: int) -> bool:
|
||||
if pages == 0:
|
||||
return False
|
||||
return (total_chars / pages) >= MIN_CHARS_PER_PAGE
|
||||
|
||||
|
||||
def md_is_current(pdf_path: Path, md_path: Path) -> bool:
|
||||
"""True when md exists and is at least as new as the source PDF."""
|
||||
return md_path.exists() and md_path.stat().st_mtime >= pdf_path.stat().st_mtime
|
||||
|
||||
|
||||
def plain_text_markdown(pdf_path: Path) -> str:
|
||||
"""Fallback: raw per-page text with the same '-----' page separators.
|
||||
|
||||
Loses heading/list/table structure but never loses the text itself."""
|
||||
parts = []
|
||||
with pymupdf.open(pdf_path) as doc:
|
||||
for page in doc:
|
||||
parts.append(page.get_text("text").strip())
|
||||
return "\n\n-----\n\n".join(parts)
|
||||
|
||||
|
||||
def convert_one(pdf_path: Path, md_path: Path, raw_chars: int) -> str:
|
||||
"""Write markdown for one PDF. Returns the method used: 'pymupdf4llm' or 'plain'.
|
||||
|
||||
pymupdf4llm gives structured markdown ("-----" page separators preserve page
|
||||
boundaries for citation). On some PDFs its layout pass emits almost nothing
|
||||
even though a text layer exists; when its output is implausibly small versus
|
||||
the raw extractable text, fall back to plain text extraction.
|
||||
"""
|
||||
md_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
body = pymupdf4llm.to_markdown(str(pdf_path), write_images=False, show_progress=False)
|
||||
method = "pymupdf4llm"
|
||||
# Generous floor: only trips when pymupdf4llm essentially gave up.
|
||||
if len(body.strip()) < 0.2 * raw_chars:
|
||||
body = plain_text_markdown(pdf_path)
|
||||
method = "plain"
|
||||
header = f"# {pdf_path.stem}\n\n> Source: `{pdf_path.name}`\n\n---\n\n"
|
||||
md_path.write_text(header + body, encoding="utf-8")
|
||||
return method
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
||||
)
|
||||
parser.add_argument("--src", type=Path, default=Path("pdfs"), help="source dir of PDFs (default: pdfs)")
|
||||
parser.add_argument("--out", type=Path, default=Path("md"), help="output dir for markdown (default: md)")
|
||||
parser.add_argument(
|
||||
"--ocr-list", type=Path, default=Path("needs-ocr.txt"),
|
||||
help="file listing PDFs with no usable text layer (default: needs-ocr.txt)",
|
||||
)
|
||||
parser.add_argument("--force", action="store_true", help="reconvert even if md is current")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if not args.src.is_dir():
|
||||
print(f"error: source dir not found: {args.src}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
pdfs = sorted(args.src.rglob("*.pdf"))
|
||||
if not pdfs:
|
||||
print(f"No PDFs found under {args.src}")
|
||||
return 0
|
||||
|
||||
converted, skipped, failed, plain = 0, 0, 0, 0
|
||||
needs_ocr: list[Path] = []
|
||||
|
||||
for pdf in pdfs:
|
||||
rel = pdf.relative_to(args.src).with_suffix(".md")
|
||||
md_path = args.out / rel
|
||||
|
||||
if not args.force and md_is_current(pdf, md_path):
|
||||
print(f"skip {rel} (up to date)")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
try:
|
||||
total, pages = extractable_chars(pdf)
|
||||
except Exception as exc: # unreadable / corrupt PDF
|
||||
print(f"FAIL {rel} ({type(exc).__name__}: {exc})")
|
||||
failed += 1
|
||||
continue
|
||||
|
||||
if not has_text_layer(total, pages):
|
||||
print(f"ocr? {rel} ({total} chars / {pages} pages -> no text layer)")
|
||||
needs_ocr.append(pdf)
|
||||
continue
|
||||
|
||||
try:
|
||||
method = convert_one(pdf, md_path, total)
|
||||
except Exception as exc:
|
||||
print(f"FAIL {rel} ({type(exc).__name__}: {exc})")
|
||||
failed += 1
|
||||
continue
|
||||
|
||||
tag = " [plain-text fallback]" if method == "plain" else ""
|
||||
if method == "plain":
|
||||
plain += 1
|
||||
print(f"conv {rel} ({pages} pages){tag}")
|
||||
converted += 1
|
||||
|
||||
# Rewritten fresh each run: flagged PDFs are re-checked every time since they
|
||||
# never produce markdown to skip on.
|
||||
if needs_ocr:
|
||||
args.ocr_list.write_text("\n".join(str(p) for p in needs_ocr) + "\n", encoding="utf-8")
|
||||
elif args.ocr_list.exists():
|
||||
args.ocr_list.unlink()
|
||||
|
||||
print("\n" + "=" * 48)
|
||||
print(f"converted: {converted}" + (f" ({plain} via plain-text fallback)" if plain else ""))
|
||||
print(f"skipped: {skipped}")
|
||||
print(f"flagged for OCR: {len(needs_ocr)}")
|
||||
if failed:
|
||||
print(f"failed: {failed}")
|
||||
if needs_ocr:
|
||||
print(f"\nSee {args.ocr_list} -> run these through marker/docling (OCR).")
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user