nathan/AI-Prototype
1
0
Fork 0
Code Pull Requests Packages Projects Activity

early version of OCR viewer now integrated

Browse Source
This commit is contained in:
nathan
2026-05-13 10:03:38 -04:00
parent 77323821cb
commit c663627594
17 changed files with 5734 additions and 1517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
scripts/ocr_confidence.py Normal file
View File

@@ -0,0 +1,298 @@
"""
ocr_confidence.py
-----------------
Reads an existing olmOCR .md output file and scores each page for transcription
confidence using the DeepInfra API. Does NOT re-run OCR.
Requirements:
pip install requests
Usage:
python ocr_confidence.py --api_key "YOUR_KEY"
python ocr_confidence.py --api_key "YOUR_KEY" --input "path/to/file.md" --output "path/to/out.json"
"""
import argparse
import json
import sys
import time
import warnings
from pathlib import Path
import requests
# ---------------------------------------------------------------------------
# Load .env from project root (if present) — no external dependencies needed
# ---------------------------------------------------------------------------
def _load_dotenv() -> None:
env_path = Path(__file__).resolve().parent.parent / ".env"
if not env_path.exists():
return
import os
with open(env_path, encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
os.environ.setdefault(key.strip(), value.strip())
_load_dotenv()
# ---------------------------------------------------------------------------
# Configuration — mirrors ocr_wardiaries.py
# ---------------------------------------------------------------------------
DEEPINFRA_API_URL = "https://api.deepinfra.com/v1/openai/chat/completions"
# olmOCR is a vision model — use a chat LLM for text-based confidence review
MODEL = "google/gemma-3-27b-it"
MAX_RETRIES = 3
RETRY_DELAY = 5 # seconds between retries
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_INPUT = _PROJECT_ROOT / "Inputs" / "ocr-output" / "Calgary-Highlanders_War-Diary_Sep44_olmocr.md"
DEFAULT_OUTPUT = _PROJECT_ROOT / "Inputs" / "ocr-output" / "Calgary-Highlanders_War-Diary_Sep44_confidence.json"
CONFIDENCE_PROMPT = (
"You are reviewing an OCR transcription of a WWII war diary. Your job is to identify words that may have been misread by the OCR scanner — not to correct the soldier's original spelling or interpret abbreviations. Rate OCR accuracy only. Do not suggest what words "should" be. Respond in JSON only:
{"score": 7, "uncertain_words": ["Loon", "Fme"], "notes": "Possible OCR misread in line 3"}
)
ERROR_RESULT = {"score": 0, "uncertain_words": [], "notes": "API error"}
# ---------------------------------------------------------------------------
# Parse olmOCR markdown into {page_num: text} dict
# Same logic as parseOCRByPage() in p44-ocr-viewer.html
# ---------------------------------------------------------------------------
def parse_ocr_by_page(raw_text: str) -> dict[int, str]:
"""Split olmOCR markdown on '## Page N' headings into a page-keyed dict."""
pages: dict[int, str] = {}
lines = raw_text.split("\n")
page_num: int | None = None
buf: list[str] = []
for line in lines:
m_head = _PAGE_HEADING.match(line)
if m_head:
if page_num is not None:
pages[page_num] = "\n".join(buf)
page_num = int(m_head.group(1))
buf = [line] # keep the heading at the top
elif page_num is not None:
buf.append(line)
if page_num is not None:
pages[page_num] = "\n".join(buf)
return pages
import re as _re
_PAGE_HEADING = _re.compile(r"^## Page (\d+)\s*$")
# ---------------------------------------------------------------------------
# API call
# ---------------------------------------------------------------------------
def score_page(page_text: str, page_num: int, api_key: str) -> dict:
"""
Send one page's OCR text to DeepInfra for confidence scoring.
Returns a dict with keys: score, uncertain_words, notes.
On failure returns ERROR_RESULT.
"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
}
payload = {
"model": MODEL,
"messages": [
{
"role": "user",
"content": (
f"{CONFIDENCE_PROMPT}\n\n"
f"--- BEGIN OCR TEXT (page {page_num}) ---\n"
f"{page_text}\n"
f"--- END OCR TEXT ---"
),
}
],
"max_tokens": 256,
"temperature": 0.0,
}
for attempt in range(1, MAX_RETRIES + 1):
try:
response = requests.post(
DEEPINFRA_API_URL, headers=headers, json=payload, timeout=60
)
response.raise_for_status()
raw_content = response.json()["choices"][0]["message"]["content"].strip()
# Strip markdown code fences if the model wrapped the JSON
if raw_content.startswith("```"):
raw_content = _re.sub(r"^```[a-z]*\n?", "", raw_content)
raw_content = _re.sub(r"\n?```$", "", raw_content)
result = json.loads(raw_content)
# Validate expected keys; fill missing ones with defaults
score = int(result.get("score", 0))
uncertain_words = result.get("uncertain_words", [])
notes = result.get("notes", "")
if not isinstance(uncertain_words, list):
uncertain_words = []
return {"score": score, "uncertain_words": uncertain_words, "notes": notes}
except requests.exceptions.HTTPError as exc:
warnings.warn(f"Page {page_num}: HTTP error on attempt {attempt}/{MAX_RETRIES}: {exc}")
if attempt < MAX_RETRIES:
time.sleep(RETRY_DELAY)
except json.JSONDecodeError as exc:
warnings.warn(
f"Page {page_num}: Malformed JSON from API on attempt {attempt}/{MAX_RETRIES}: {exc}"
)
if attempt < MAX_RETRIES:
time.sleep(RETRY_DELAY)
except Exception as exc: # noqa: BLE001
warnings.warn(f"Page {page_num}: Unexpected error on attempt {attempt}/{MAX_RETRIES}: {exc}")
if attempt < MAX_RETRIES:
time.sleep(RETRY_DELAY)
warnings.warn(f"Page {page_num}: All {MAX_RETRIES} attempts failed — storing error result.")
return dict(ERROR_RESULT) # return a fresh copy
# ---------------------------------------------------------------------------
# Summary helpers
# ---------------------------------------------------------------------------
def print_summary(results: dict) -> None:
high = sum(1 for v in results.values() if v["score"] >= 8)
med = sum(1 for v in results.values() if 5 <= v["score"] <= 7)
low = sum(1 for v in results.values() if 1 <= v["score"] <= 4)
err = sum(1 for v in results.values() if v["score"] == 0)
total = len(results)
print("\n" + "=" * 50)
print(f"CONFIDENCE SCORING COMPLETE — {total} pages scored")
print("=" * 50)
print(f" High (8-10) : {high:>4} ({high/total*100:.1f}%)" if total else " High (8-10) : 0")
print(f" Medium (5-7) : {med:>4} ({med/total*100:.1f}%)" if total else " Medium (5-7) : 0")
print(f" Low (1-4) : {low:>4} ({low/total*100:.1f}%)" if total else " Low (1-4) : 0")
if err:
print(f" API errors : {err:>4}")
print("=" * 50)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
parser = argparse.ArgumentParser(
description="Score OCR confidence for each page of an olmOCR .md output file."
)
parser.add_argument(
"--api_key", default=None,
help="DeepInfra API key (defaults to DEEPINFRA_API_KEY env var)."
)
parser.add_argument(
"--input", default=str(DEFAULT_INPUT),
help=f"Path to olmOCR .md file. Default: {DEFAULT_INPUT}"
)
parser.add_argument(
"--output", default=str(DEFAULT_OUTPUT),
help=f"Path to write confidence JSON. Default: {DEFAULT_OUTPUT}"
)
parser.add_argument(
"--delay", type=float, default=0.5,
help="Seconds to pause between API calls (default: 0.5)."
)
args = parser.parse_args()
import os
api_key = args.api_key or os.environ.get("DEEPINFRA_API_KEY", "")
if not api_key:
print("ERROR: No API key provided. Use --api_key or set DEEPINFRA_API_KEY in .env", file=sys.stderr)
sys.exit(1)
input_path = Path(args.input)
output_path = Path(args.output)
# ── Read source file ──────────────────────────────────────────────────────
if not input_path.exists():
print(f"ERROR: Input file not found: {input_path}", file=sys.stderr)
sys.exit(1)
print(f"Reading OCR source: {input_path}")
raw_text = input_path.read_text(encoding="utf-8")
pages = parse_ocr_by_page(raw_text)
if not pages:
print("ERROR: No '## Page N' headings found in the input file.", file=sys.stderr)
sys.exit(1)
total_pages = len(pages)
sorted_pages = sorted(pages.keys())
print(f"Found {total_pages} pages (Page {sorted_pages[0]} – {sorted_pages[-1]}).")
# ── Load existing results (resume support) ────────────────────────────────
results: dict[str, dict] = {}
if output_path.exists():
try:
results = json.loads(output_path.read_text(encoding="utf-8"))
already_done = len(results)
print(f"Resuming — {already_done} page(s) already scored, skipping them.")
except (json.JSONDecodeError, OSError) as exc:
warnings.warn(f"Could not read existing output ({exc}); starting fresh.")
results = {}
output_path.parent.mkdir(parents=True, exist_ok=True)
# ── Score each page ───────────────────────────────────────────────────────
scored_this_run = 0
for page_num in sorted_pages:
page_key = str(page_num)
if page_key in results:
continue # already scored — skip
page_text = pages[page_num]
result = score_page(page_text, page_num, api_key)
results[page_key] = result
scored_this_run += 1
# Progress line
score_str = str(result["score"]) if result["score"] > 0 else "ERR"
print(f" Page {page_num}/{sorted_pages[-1]} — score: {score_str}")
# Write after every page so a crash loses minimal work
output_path.write_text(
json.dumps(results, indent=2, ensure_ascii=False),
encoding="utf-8"
)
if scored_this_run > 0 and page_num != sorted_pages[-1]:
time.sleep(args.delay)
# ── Final save & summary ──────────────────────────────────────────────────
output_path.write_text(
json.dumps(results, indent=2, ensure_ascii=False),
encoding="utf-8"
)
print(f"\nResults written to: {output_path}")
print_summary(results)
if __name__ == "__main__":
main()

162
scripts/ocr_wardiaries.py
View File

@@ -1,9 +1,12 @@
"""
ocr_wardiaries.py
-----------------
Converts war diary PDFs to Markdown using the olmOCR model hosted on DeepInfra.
Converts war diary PDFs to plain text using the olmOCR model hosted on DeepInfra.
Bypasses the olmOCR pipeline GPU check by calling the API directly.
Output format: plain text only — no markdown, no HTML, no tables.
Each diary entry is transcribed as prose with a blank line between entries.
Requirements:
pip install requests pillow
@@ -11,11 +14,11 @@ Poppler must be on PATH (pdftoppm command must work).
Usage:
python ocr_wardiaries.py --input_dir "C:/path/to/pdfs" --output_dir "C:/path/to/output" --api_key "YOUR_KEY"
python ocr_wardiaries.py --single "C:/path/to/diary.pdf" --output_dir "C:/path/to/output" --api_key "YOUR_KEY"
"""
import argparse
import base64
import json
import os
import subprocess
import sys
@@ -31,31 +34,71 @@ from PIL import Image
# ---------------------------------------------------------------------------
DEEPINFRA_API_URL = "https://api.deepinfra.com/v1/openai/chat/completions"
MODEL = "allenai/olmOCR-2-7B-1025"
DPI = 150 # Higher = better quality but slower/more expensive. 150 is good for typewritten text.
MAX_RETRIES = 3 # Retries per page on API error
RETRY_DELAY = 5 # Seconds between retries
MODEL = "allenai/olmOCR-2-7B-1025"
DPI = 150 # Higher = better quality but slower/more expensive. 150 is good for typewritten text.
MAX_RETRIES = 3 # Retries per page on API error
RETRY_DELAY = 5 # Seconds between retries
# ---------------------------------------------------------------------------
# OCR prompt
# ---------------------------------------------------------------------------
# Plain text only — no markdown, no HTML.
# War diary pages are tabular (Place / Date / Hour / Summary).
# We transcribe each row as prose to avoid HTML table output
# which causes downstream parsing problems.
SYSTEM_PROMPT = """\
You are transcribing a scanned WWII Canadian Army war diary page.
Output plain text only. No markdown. No HTML. No tables. No formatting symbols.
Rules:
- Transcribe ALL text visible on the page. Every word, every field, every stamp, every notation.
- Do not decide what is important. Do not skip anything. Transcribe everything.
- Do not correct spelling, grammar, or abbreviations. Transcribe exactly as written.
- If the page has a table with columns (Place / Date / Hour / Summary), transcribe each row \
as continuous prose in this format:
[DATE] [PLACE]: [SUMMARY TEXT]
Example: 4 Sep 44 France, NEUVILLE Les DIEPPE MR 2468 Sheet: The main topic for the morning...
- Separate multiple diary entries with a single blank line.
- Keep grid references exactly as written (e.g. MR 2468, GR 442891, MR 0675 Sheet).
- Keep military abbreviations exactly as written (e.g. Bn, Bde, HQ, OR, C.O., R de Mais, RHC).
- Weather notes go on their own line starting with: Weather:
- Transcribe page headers, form titles, stamps, handwritten notes, signatures, \
and margin notations — all of it.
- If a page is genuinely blank with no text at all, output only: [NO DIARY CONTENT]
- Do not add commentary, explanations, confidence notes, or summaries.
- Do not output page numbers or headings.
- Do not use asterisks, hashes, pipes, underscores, or any other formatting characters.\
"""
USER_PROMPT = (
"Transcribe every word on this war diary page exactly as written. "
"Plain text only — include all text visible on the page, nothing omitted."
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def pdf_to_images(pdf_path: Path, output_dir: Path, dpi: int = DPI) -> list[Path]:
def pdf_to_images(pdf_path: Path, output_dir: Path, dpi: int = DPI,
first_page: int = None, last_page: int = None) -> list[Path]:
"""Render each page of a PDF to a PNG image using pdftoppm (poppler)."""
prefix = output_dir / pdf_path.stem
cmd = [
"pdftoppm",
"-r", str(dpi),
"-png",
str(pdf_path),
str(prefix),
]
if first_page is not None:
cmd += ["-f", str(first_page)]
if last_page is not None:
cmd += ["-l", str(last_page)]
cmd += [str(pdf_path), str(prefix)]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"pdftoppm failed for {pdf_path.name}:\n{result.stderr}")
# pdftoppm names files like prefix-1.png, prefix-2.png etc (zero-padded)
images = sorted(output_dir.glob(f"{pdf_path.stem}-*.png"))
return images
@@ -67,7 +110,7 @@ def image_to_base64(image_path: Path) -> str:
def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
"""Send one page image to DeepInfra olmOCR and return the extracted text."""
"""Send one page image to DeepInfra olmOCR and return the extracted plain text."""
b64 = image_to_base64(image_path)
headers = {
@@ -78,6 +121,10 @@ def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
payload = {
"model": MODEL,
"messages": [
{
"role": "system",
"content": SYSTEM_PROMPT,
},
{
"role": "user",
"content": [
@@ -89,12 +136,7 @@ def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
},
{
"type": "text",
"text": (
"Below is a scanned page from a Canadian Army WWII war diary or supplementary document, "
"circa 1944-1945. Extract all text exactly as it appears, preserving layout, "
"column structure, dates, grid references, unit names, and abbreviations. "
"Output clean Markdown. Do not add commentary or summaries."
),
"text": USER_PROMPT,
},
],
}
@@ -105,7 +147,12 @@ def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
for attempt in range(1, MAX_RETRIES + 1):
try:
response = requests.post(DEEPINFRA_API_URL, headers=headers, json=payload, timeout=120)
response = requests.post(
DEEPINFRA_API_URL,
headers=headers,
json=payload,
timeout=120,
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]
@@ -123,8 +170,8 @@ def ocr_page(image_path: Path, api_key: str, page_num: int) -> str:
return f"[OCR FAILED - page {page_num} - Error: {e}]"
def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str) -> Path:
"""OCR a single PDF and write output to a Markdown file."""
def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str, max_pages: int = None) -> Path:
"""OCR a single PDF and write output to a plain text file."""
print(f"\n{'='*60}")
print(f"Processing: {pdf_path.name}")
print(f"{'='*60}")
@@ -133,7 +180,8 @@ def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str) -> Path:
# Skip if already done
if output_md.exists():
print(f" Already processed — skipping. Delete {output_md.name} to reprocess.")
print(f" Already processed — skipping.")
print(f" Delete {output_md.name} to reprocess.")
return output_md
with tempfile.TemporaryDirectory() as tmp_dir:
@@ -142,25 +190,33 @@ def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str) -> Path:
# Render PDF pages to images
print(f" Rendering PDF pages to images at {DPI} DPI...")
try:
images = pdf_to_images(pdf_path, tmp_path)
images = pdf_to_images(pdf_path, tmp_path, last_page=max_pages)
except RuntimeError as e:
print(f" ERROR: {e}")
return output_md
total_pages = len(images)
print(f" Found {total_pages} pages.")
print(f" Found {total_pages} pages to process.")
# File header — plain text, no markdown
all_text = [
f"# {pdf_path.stem}\n",
f"*OCR'd by olmOCR ({MODEL}) via DeepInfra*\n",
f"*Source: {pdf_path.name} — {total_pages} pages*\n",
"---\n",
f"# {pdf_path.stem}",
f"OCR by olmOCR ({MODEL}) via DeepInfra",
f"Source: {pdf_path.name} — {len(images)} page(s) processed",
"",
]
for i, image_path in enumerate(images, start=1):
print(f" Page {i}/{total_pages}...", end=" ", flush=True)
page_text = ocr_page(image_path, api_key, i)
all_text.append(f"\n---\n## Page {i}\n\n{page_text}\n")
# Each page is separated by a ## Page N marker so the viewer
# can split pages correctly. The content itself is plain text.
all_text.append(f"## Page {i}")
all_text.append("")
all_text.append(page_text.strip())
all_text.append("")
print("done")
# Small delay to avoid hammering the API
@@ -181,40 +237,68 @@ def ocr_pdf(pdf_path: Path, output_dir: Path, api_key: str) -> Path:
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(description="OCR war diary PDFs using olmOCR via DeepInfra.")
parser.add_argument("--input_dir", required=True, help="Folder containing PDF files to process.")
parser.add_argument("--output_dir", required=True, help="Folder to write Markdown output files.")
parser.add_argument("--api_key", required=True, help="Your DeepInfra API key.")
parser.add_argument("--single", default=None, help="Process a single PDF file instead of a folder.")
parser = argparse.ArgumentParser(
description="OCR war diary PDFs to plain text using olmOCR via DeepInfra."
)
parser.add_argument(
"--input_dir",
default=None,
help="Folder containing PDF files to process.",
)
parser.add_argument(
"--output_dir",
required=True,
help="Folder to write output files.",
)
parser.add_argument(
"--api_key",
required=True,
help="Your DeepInfra API key.",
)
parser.add_argument(
"--single",
default=None,
help="Process a single PDF file instead of a whole folder.",
)
parser.add_argument(
"--max_pages",
type=int,
default=None,
help="Only process the first N pages of each PDF (useful for testing).",
)
args = parser.parse_args()
input_dir = Path(args.input_dir)
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
if args.single:
pdfs = [Path(args.single)]
elif args.input_dir:
pdfs = sorted(Path(args.input_dir).glob("*.pdf"))
else:
pdfs = sorted(input_dir.glob("*.pdf"))
print("Error: provide --input_dir or --single")
sys.exit(1)
if not pdfs:
print(f"No PDF files found in {input_dir}")
print(f"No PDF files found.")
sys.exit(1)
print(f"Found {len(pdfs)} PDF(s) to process.")
print(f"Output directory: {output_dir}")
print(f"Model: {MODEL}")
print(f"DPI: {DPI}")
results = []
for pdf_path in pdfs:
output_md = ocr_pdf(pdf_path, output_dir, args.api_key)
output_md = ocr_pdf(pdf_path, output_dir, args.api_key, max_pages=args.max_pages)
results.append(output_md)
print(f"\n{'='*60}")
print(f"Complete. {len(results)} file(s) processed.")
print(f"Output files:")
print("Output files:")
for r in results:
print(f" {r}")
if __name__ == "__main__":
main()
main()