|
#!/usr/bin/env python3 |
|
""" |
|
nori — PDF/Image to Markdown converter using Claude's vision. |
|
Named after Nori, the dwarf scribe from LOTR. |
|
|
|
Uses macOS native APIs (PDFKit + CoreGraphics via JXA) to render |
|
PDF pages to images, then Claude Agent SDK to convert each page |
|
to faithful markdown, preserving diagrams as detailed descriptions. |
|
|
|
Supports resuming — if interrupted, re-run the same command |
|
and it picks up from where it left off. |
|
""" |
|
|
|
import sys |
|
import json |
|
import time |
|
import asyncio |
|
import hashlib |
|
import argparse |
|
import shutil |
|
import subprocess |
|
import tempfile |
|
import threading |
|
from pathlib import Path |
|
|
|
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, RateLimitEvent |
|
|
|
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tiff"} |
|
STATE_DIR = Path.home() / ".nori" |
|
|
|
JXA_PDF_TO_IMAGES = """ |
|
ObjC.import("Quartz"); |
|
ObjC.import("AppKit"); |
|
ObjC.import("CoreGraphics"); |
|
|
|
var pdfPath = "%PDF_PATH%"; |
|
var outputDir = "%OUTPUT_DIR%"; |
|
var scale = %SCALE%; |
|
|
|
var pdfURL = $.NSURL.fileURLWithPath(pdfPath); |
|
var pdfDoc = $.PDFDocument.alloc.initWithURL(pdfURL); |
|
var pageCount = pdfDoc.pageCount; |
|
|
|
for (var i = 0; i < pageCount; i++) { |
|
var page = pdfDoc.pageAtIndex(i); |
|
var bounds = page.boundsForBox($.kPDFDisplayBoxMediaBox); |
|
|
|
var w = Math.ceil(bounds.size.width * scale); |
|
var h = Math.ceil(bounds.size.height * scale); |
|
|
|
var colorSpace = $.CGColorSpaceCreateDeviceRGB(); |
|
var ctx = $.CGBitmapContextCreate( |
|
$.nil, w, h, 8, w * 4, colorSpace, |
|
$.kCGImageAlphaPremultipliedLast |
|
); |
|
|
|
$.CGContextSetRGBFillColor(ctx, 1, 1, 1, 1); |
|
$.CGContextFillRect(ctx, $.CGRectMake(0, 0, w, h)); |
|
$.CGContextScaleCTM(ctx, scale, scale); |
|
|
|
var nsctx = $.NSGraphicsContext.graphicsContextWithCGContextFlipped(ctx, false); |
|
$.NSGraphicsContext.setCurrentContext(nsctx); |
|
page.drawWithBox($.kPDFDisplayBoxMediaBox); |
|
|
|
var cgImage = $.CGBitmapContextCreateImage(ctx); |
|
var bitmapRep = $.NSBitmapImageRep.alloc.initWithCGImage(cgImage); |
|
var pngData = bitmapRep.representationUsingTypeProperties($.NSBitmapImageFileTypePNG, $()); |
|
|
|
var pageNum = String(i + 1).padStart(4, "0"); |
|
var outPath = outputDir + "/page_" + pageNum + ".png"; |
|
pngData.writeToFileAtomically(outPath, true); |
|
} |
|
|
|
pageCount; |
|
""" |
|
|
|
|
|
def pdf_to_images(pdf_path: str, output_dir: str, scale: float = 2.0) -> list[Path]: |
|
"""Convert PDF pages to PNGs using macOS native PDFKit via JXA.""" |
|
script = ( |
|
JXA_PDF_TO_IMAGES |
|
.replace("%PDF_PATH%", pdf_path) |
|
.replace("%OUTPUT_DIR%", output_dir) |
|
.replace("%SCALE%", str(scale)) |
|
) |
|
|
|
result = subprocess.run( |
|
["osascript", "-l", "JavaScript", "-e", script], |
|
capture_output=True, text=True, timeout=300, |
|
) |
|
|
|
if result.returncode != 0: |
|
print(f"Error rendering PDF: {result.stderr}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
page_count = int(result.stdout.strip()) |
|
images = sorted(Path(output_dir).glob("page_*.png")) |
|
|
|
if len(images) != page_count: |
|
print(f"Warning: expected {page_count} pages, found {len(images)} images") |
|
|
|
return images |
|
|
|
|
|
def collect_images(paths: list[str]) -> list[Path]: |
|
"""Collect and sort image files from paths (files or directories).""" |
|
images = [] |
|
for p in paths: |
|
path = Path(p).resolve() |
|
if path.is_dir(): |
|
images.extend( |
|
f for f in sorted(path.iterdir()) if f.suffix.lower() in IMAGE_EXTS |
|
) |
|
elif path.is_file() and path.suffix.lower() in IMAGE_EXTS: |
|
images.append(path) |
|
else: |
|
print(f"Warning: skipping {p}") |
|
return images |
|
|
|
|
|
# --- State management for resume support --- |
|
|
|
def get_job_id(image_paths: list[Path]) -> str: |
|
"""Generate a stable job ID from the list of image paths.""" |
|
content = "\n".join(str(p) for p in image_paths) |
|
return hashlib.sha256(content.encode()).hexdigest()[:12] |
|
|
|
|
|
def load_state(job_id: str) -> dict: |
|
"""Load saved state for a job.""" |
|
state_file = STATE_DIR / f"{job_id}.json" |
|
if state_file.exists(): |
|
return json.loads(state_file.read_text()) |
|
return {"completed_pages": {}, "total": 0} |
|
|
|
|
|
def save_state(job_id: str, state: dict): |
|
"""Save state for a job.""" |
|
STATE_DIR.mkdir(parents=True, exist_ok=True) |
|
state_file = STATE_DIR / f"{job_id}.json" |
|
state_file.write_text(json.dumps(state, indent=2)) |
|
|
|
|
|
def clear_state(job_id: str): |
|
"""Remove state file after successful completion.""" |
|
state_file = STATE_DIR / f"{job_id}.json" |
|
state_file.unlink(missing_ok=True) |
|
|
|
|
|
# --- Page conversion --- |
|
|
|
async def convert_single_page( |
|
image_path: Path, page_num: int, total: int, name: str, |
|
model: str | None = None, retries: int = 2, |
|
) -> str: |
|
"""Convert a single page image to markdown using Claude.""" |
|
img_dir = str(image_path.parent) |
|
prompt = f"""You are a document converter. Your ONLY job is to read ONE image file and convert it to markdown. |
|
|
|
Read this file and nothing else: {image_path} |
|
|
|
Do NOT read any other files. Do NOT explore the filesystem. ONLY read the file path above. |
|
|
|
This is page {page_num}/{total} of "{name}". |
|
|
|
Rules: |
|
- Convert all text content faithfully to markdown |
|
- For diagrams, charts, flowcharts, or figures: reproduce them as ASCII art |
|
inside a fenced code block. Preserve the structure, labels, arrows, and |
|
relationships as faithfully as possible. If ASCII art cannot capture it, |
|
fall back to a detailed text description in a blockquote. |
|
- For tables: reproduce them as markdown tables |
|
- For code snippets: use fenced code blocks with the appropriate language |
|
- For math equations: use LaTeX notation within $...$ or $$...$$ |
|
- Maintain the document's logical structure (headings, lists, paragraphs) |
|
- Output ONLY the markdown content, nothing else""" |
|
|
|
while True: |
|
try: |
|
result_parts = [] |
|
rate_limited = False |
|
async for message in query( |
|
prompt=prompt, |
|
options=ClaudeAgentOptions( |
|
cwd=img_dir, |
|
allowed_tools=["Read"], |
|
disallowed_tools=["Bash", "Write", "Edit", "Glob", "Grep"], |
|
permission_mode="acceptEdits", |
|
max_turns=3, |
|
setting_sources=[], |
|
**({"model": model} if model else {}), |
|
), |
|
): |
|
if isinstance(message, ResultMessage): |
|
result_parts.append(message.result) |
|
elif isinstance(message, RateLimitEvent): |
|
info = message.rate_limit_info |
|
if info.status == "rejected": |
|
resets_at = getattr(info, "resets_at", None) |
|
if resets_at: |
|
wait_secs = max(60, resets_at - int(time.time())) |
|
wait_mins = wait_secs // 60 |
|
print(f"\n Rate limited — waiting ~{wait_mins}m for reset...", flush=True) |
|
await asyncio.sleep(wait_secs + 10) # wait + small buffer |
|
rate_limited = True |
|
|
|
if rate_limited and not result_parts: |
|
continue # retry after waiting |
|
|
|
return "\n".join(result_parts) |
|
except Exception: |
|
# CLI exits with code 1 on rate limit rejection — wait and retry |
|
print(f"\n Page {page_num} failed — retrying in 30s...", flush=True) |
|
await asyncio.sleep(30) |
|
continue |
|
|
|
|
|
async def convert_images_to_markdown( |
|
image_paths: list[Path], name: str, workers: int = 2, |
|
model: str | None = None, |
|
) -> str: |
|
"""Convert page images to markdown in parallel, with resume support.""" |
|
total = len(image_paths) |
|
job_id = get_job_id(image_paths) |
|
state = load_state(job_id) |
|
state["total"] = total |
|
|
|
completed = state["completed_pages"] |
|
skip_count = len(completed) |
|
|
|
if skip_count > 0: |
|
print(f" Resuming — {skip_count}/{total} pages already done") |
|
|
|
# Find pages that still need conversion |
|
pending = [ |
|
(i, img_path) |
|
for i, img_path in enumerate(image_paths) |
|
if str(i) not in completed |
|
] |
|
|
|
if not pending: |
|
print(f" All {total} pages already converted") |
|
else: |
|
done_count = skip_count |
|
lock = threading.Lock() |
|
semaphore = asyncio.Semaphore(workers) |
|
|
|
async def process_page(i: int, img_path: Path): |
|
nonlocal done_count |
|
async with semaphore: |
|
page_num = i + 1 |
|
markdown = await convert_single_page(img_path, page_num, total, name, model=model) |
|
|
|
with lock: |
|
completed[str(i)] = markdown |
|
save_state(job_id, state) |
|
done_count += 1 |
|
print(f"\r Converted {done_count}/{total} pages ", end="", flush=True) |
|
|
|
# Stagger task starts to avoid thundering herd |
|
async def staggered_gather(): |
|
tasks = [] |
|
for idx, (i, img_path) in enumerate(pending): |
|
tasks.append(asyncio.create_task(process_page(i, img_path))) |
|
if idx < len(pending) - 1: |
|
await asyncio.sleep(0.5) # small delay between spawns |
|
return await asyncio.gather(*tasks, return_exceptions=True) |
|
|
|
results = await staggered_gather() |
|
|
|
errors = [(pending[j][0] + 1, r) for j, r in enumerate(results) if isinstance(r, Exception)] |
|
if errors: |
|
print() |
|
for page_num, err in errors: |
|
print(f" Error on page {page_num}: {err}") |
|
print(f" {len(completed)}/{total} pages saved. Re-run to retry failed pages.") |
|
sys.exit(1) |
|
|
|
print(f"\r Converted {total}/{total} pages — done ") |
|
|
|
# Assemble final markdown in order |
|
parts = [completed[str(i)] for i in range(total)] |
|
result = "\n\n---\n\n".join(parts) |
|
|
|
# Clean up state file on success |
|
clear_state(job_id) |
|
|
|
return result |
|
|
|
|
|
def main(): |
|
parser = argparse.ArgumentParser( |
|
prog="nori", |
|
description="Convert PDF or page images to Markdown using Claude's vision.", |
|
) |
|
parser.add_argument( |
|
"input", |
|
nargs="+", |
|
help="PDF file, image files, or directories containing images", |
|
) |
|
parser.add_argument( |
|
"-o", "--output", help="Output markdown file (default: <input_name>.md)" |
|
) |
|
parser.add_argument( |
|
"--scale", |
|
type=float, |
|
default=2.0, |
|
help="Scale factor for PDF rendering (default: 2.0)", |
|
) |
|
parser.add_argument( |
|
"-w", "--workers", |
|
type=int, |
|
default=2, |
|
help="Number of parallel workers (default: 2)", |
|
) |
|
parser.add_argument( |
|
"-m", "--model", |
|
default=None, |
|
help="Model to use (e.g., sonnet, opus). Default: auto", |
|
) |
|
parser.add_argument( |
|
"--clean", |
|
action="store_true", |
|
help="Clear saved state and start fresh", |
|
) |
|
args = parser.parse_args() |
|
|
|
# Determine if input is a PDF or images |
|
first_input = Path(args.input[0]) |
|
is_pdf = len(args.input) == 1 and first_input.suffix.lower() == ".pdf" |
|
|
|
if is_pdf: |
|
pdf_path = first_input.resolve() |
|
if not pdf_path.exists(): |
|
print(f"Error: File not found: {pdf_path}") |
|
sys.exit(1) |
|
|
|
doc_name = pdf_path.stem |
|
output_path = Path(args.output) if args.output else pdf_path.with_suffix(".md") |
|
|
|
print(f"nori — converting: {pdf_path.name}") |
|
print() |
|
|
|
# For PDF mode, use a persistent image cache dir so resume works |
|
img_cache = STATE_DIR / "images" / doc_name |
|
img_cache.mkdir(parents=True, exist_ok=True) |
|
|
|
# Only re-render if images don't exist yet |
|
existing = sorted(img_cache.glob("page_*.png")) |
|
if existing and not args.clean: |
|
images = existing |
|
print(f"[1/2] Using cached page images ({len(images)} pages)") |
|
else: |
|
print("[1/2] Rendering PDF pages to images (macOS native)...") |
|
images = pdf_to_images(str(pdf_path), str(img_cache), scale=args.scale) |
|
print(f" {len(images)} pages rendered") |
|
|
|
print() |
|
print("[2/2] Converting to markdown with Claude...") |
|
|
|
if args.clean: |
|
job_id = get_job_id(images) |
|
clear_state(job_id) |
|
|
|
markdown = asyncio.run(convert_images_to_markdown(images, doc_name, args.workers, model=args.model)) |
|
|
|
# Clean up image cache on success |
|
shutil.rmtree(img_cache, ignore_errors=True) |
|
else: |
|
images = collect_images(args.input) |
|
if not images: |
|
print("Error: No image files found.") |
|
sys.exit(1) |
|
|
|
doc_name = first_input.stem if first_input.is_file() else first_input.name |
|
output_path = Path(args.output) if args.output else Path(f"{doc_name}.md") |
|
|
|
print(f"nori — converting {len(images)} page images") |
|
print() |
|
print("Converting to markdown with Claude...") |
|
|
|
if args.clean: |
|
job_id = get_job_id(images) |
|
clear_state(job_id) |
|
|
|
markdown = asyncio.run(convert_images_to_markdown(images, doc_name, args.workers, model=args.model)) |
|
|
|
output_path.write_text(markdown, encoding="utf-8") |
|
print(f"\nDone! Output: {output_path}") |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |