"""OCR engine for the Togen OCR Tool (DVI-490, plan DVI-463). Wraps OCRmyPDF to turn an image-only / scanned PDF into a text-selectable ("searchable") PDF and emit the recognized text as a sidecar. Board directive (D1): keep this a clean seam so a cloud fallback for low-confidence pages can be added later without route/UI rework. The engine returns an `OcrResult` carrying the extracted text plus a per-page metadata list; a future cloud-fallback pass can inspect `pages[*].confidence` and re-OCR only the weak pages, then the route/UI stay unchanged. """ from __future__ import annotations import json import os import re import tempfile from dataclasses import dataclass, field from pathlib import Path # --- Backend registry (DVI-497) --------------------------------------------- # OCR backends are pluggable: each entry is a recognized engine the admin can # select. For now there is exactly one ("local" = Tesseract + region # heuristics, the implementation in this module). A future "Claude vision" # backend (Phase 3+) drops in here as a second entry with a matching dispatch # branch in ocr_pdf() and needs no route/UI rework. DEFAULT_OCR_BACKEND = "local" OCR_BACKENDS = { "local": { "id": "local", "label": "Local", "description": ( "On-device OCR (Tesseract) with region heuristics. Runs entirely " "on the Togen host; no data leaves the network." ), }, "claude": { "id": "claude", "label": "Claude Vision (Anthropic API)", "description": ( "Uses Claude's vision model to read text from scanned drawings. " "Requires an Anthropic API key configured in Assistant settings. " "Only available when the Assistant feature is enabled and accessible " "to users (or for admins)." ), }, } # Admin-writable config holding the selected backend. ocr_pdf() reads this. OCR_SETTINGS_FILE = Path( os.environ.get( "OCR_SETTINGS_FILE", Path(__file__).resolve().parent / "ocr_settings.json" ) ) def list_backends() -> list[dict]: """Return the registered backends as a list of public dicts (UI-ready).""" return [dict(b) for b in OCR_BACKENDS.values()] def load_backend() -> str: """Return the configured backend id, falling back to the default.""" if OCR_SETTINGS_FILE.is_file(): try: data = json.loads(OCR_SETTINGS_FILE.read_text()) if isinstance(data, dict) and data.get("backend") in OCR_BACKENDS: return data["backend"] except (json.JSONDecodeError, OSError): pass return DEFAULT_OCR_BACKEND def save_backend(backend: str) -> None: """Persist the selected backend id. Raises ValueError if unknown.""" if backend not in OCR_BACKENDS: raise ValueError(f"unknown OCR backend {backend!r}") OCR_SETTINGS_FILE.write_text(json.dumps({"backend": backend}, indent=2)) # OCR modes map to the mutually-exclusive OCRmyPDF flags of the same intent. # skip-text : OCR only pages with no existing text layer (default, safe) # redo-ocr : strip any prior OCR text layer and redo it (keeps vector text) # force-ocr : rasterize every page and OCR it (handles broken text layers) SKIP_TEXT = "skip-text" REDO_OCR = "redo-ocr" FORCE_OCR = "force-ocr" VALID_MODES = (SKIP_TEXT, REDO_OCR, FORCE_OCR) @dataclass class PageInfo: """Per-page OCR metadata. `confidence` is the seam for a cloud fallback: OCRmyPDF/Tesseract does not surface a per-page confidence through the public API, so it stays None here until a richer engine populates it.""" number: int confidence: float | None = None @dataclass class OcrResult: text: str = "" pages: list[PageInfo] = field(default_factory=list) class OcrEngineUnavailable(RuntimeError): """Raised when the ocrmypdf package (or its host tools) are not installed.""" def ocr_pdf(in_path, out_path, *, mode: str = SKIP_TEXT, backend: str | None = None) -> OcrResult: """OCR `in_path` into a searchable PDF at `out_path`. Dispatches to the configured backend (defaults to the admin-selected one in OCR_SETTINGS_FILE). Returns an OcrResult whose `text` is the recognized sidecar text and whose `pages` carries per-page metadata. Raises OcrEngineUnavailable if the backend's host tools are not installed. """ if backend is None: backend = load_backend() if backend not in OCR_BACKENDS: backend = DEFAULT_OCR_BACKEND if backend == "claude": return _ocr_pdf_claude(in_path, out_path, mode=mode) return _ocr_pdf_local(in_path, out_path, mode=mode) # --- Mixed-format OCR (DVI-646) --------------------------------------------- # The OCR Tool's Projects tab lets users OCR files that already live in the # shared Projects tree, not just uploaded PDFs. Those files may be images or # Office documents, so ocr_file() normalizes the input to a PDF and then runs # the same searchable-PDF pipeline. Image inputs are inherently image-only, so # they always force OCR regardless of the requested mode. IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".bmp", ".gif", ".webp"} OFFICE_EXTS = {".doc", ".docx", ".odt", ".rtf", ".xls", ".xlsx", ".ods", ".ppt", ".pptx", ".odp"} PROCESSABLE_EXTS = {".pdf"} | IMAGE_EXTS | OFFICE_EXTS def is_processable(filename) -> bool: """True if the filename's extension is something ocr_file() can handle.""" return Path(str(filename)).suffix.lower() in PROCESSABLE_EXTS def ocr_file(in_path, out_path, *, mode: str = SKIP_TEXT, backend: str | None = None) -> OcrResult: """OCR any supported file into a searchable PDF at `out_path`. PDFs go straight to ocr_pdf(). Images are wrapped into a one-page PDF first (and forced through OCR, since an image has no text layer). Office documents are converted to PDF via LibreOffice if it is installed on the host; otherwise OcrEngineUnavailable is raised with guidance. Unknown extensions raise ValueError.""" in_path = Path(in_path) out_path = Path(out_path) ext = in_path.suffix.lower() if ext == ".pdf": return ocr_pdf(in_path, out_path, mode=mode, backend=backend) if ext in IMAGE_EXTS: with tempfile.TemporaryDirectory() as td: pdf_in = Path(td) / "input.pdf" _image_to_pdf(in_path, pdf_in) return ocr_pdf(pdf_in, out_path, mode=FORCE_OCR, backend=backend) if ext in OFFICE_EXTS: with tempfile.TemporaryDirectory() as td: pdf_in = _office_to_pdf(in_path, Path(td)) if pdf_in is None: raise OcrEngineUnavailable( "Converting Office documents to PDF requires LibreOffice " "(soffice) on the Togen host. Convert the file to PDF first, " "or ask an admin to install LibreOffice." ) return ocr_pdf(pdf_in, out_path, mode=mode, backend=backend) raise ValueError(f"Unsupported file type for OCR: {ext or '(none)'}") def _image_to_pdf(img_path, pdf_out) -> None: """Wrap a raster image into a single-page PDF using PyMuPDF.""" try: import fitz # noqa: PLC0415 — same PyMuPDF dep used by previews/region OCR except ImportError as exc: raise OcrEngineUnavailable( "PyMuPDF (fitz) is required to OCR image files." ) from exc doc = fitz.open(str(img_path)) try: pdf_bytes = doc.convert_to_pdf() finally: doc.close() Path(pdf_out).write_bytes(pdf_bytes) def _office_to_pdf(src, work_dir): """Convert an Office document to PDF via headless LibreOffice. Returns the output Path, or None if LibreOffice is unavailable / conversion failed.""" import shutil as _shutil # noqa: PLC0415 import subprocess # noqa: PLC0415 soffice = _shutil.which("soffice") or _shutil.which("libreoffice") if not soffice: return None try: subprocess.run( [soffice, "--headless", "--convert-to", "pdf", "--outdir", str(work_dir), str(src)], check=True, capture_output=True, timeout=180, ) except (subprocess.SubprocessError, OSError): return None out = Path(work_dir) / (Path(src).stem + ".pdf") return out if out.exists() else None def _ocr_pdf_local(in_path, out_path, *, mode: str = SKIP_TEXT) -> OcrResult: """Local (Tesseract/OCRmyPDF) backend implementation.""" if mode not in VALID_MODES: raise ValueError(f"mode must be one of {VALID_MODES}, got {mode!r}") try: import ocrmypdf # noqa: PLC0415 — host package installed by ST0 (DVI-489) except ImportError as exc: # pragma: no cover - environment dependent raise OcrEngineUnavailable( "ocrmypdf is not installed. The OCR host packages " "(ocrmypdf, tesseract, ghostscript, qpdf) are provisioned by the " "Linux Automation Engineer subtask (DVI-489)." ) from exc out_path = Path(out_path) sidecar_path = out_path.with_suffix(".txt") flags = { SKIP_TEXT: {"skip_text": True}, REDO_OCR: {"redo_ocr": True}, FORCE_OCR: {"force_ocr": True}, }[mode] ocrmypdf.ocr( str(in_path), str(out_path), sidecar=str(sidecar_path), progress_bar=False, **flags, ) text = "" if sidecar_path.exists(): text = sidecar_path.read_text(encoding="utf-8", errors="replace") # OCRmyPDF separates pages in the sidecar with form feeds (\f). Build the # per-page seam from that split so a fallback engine can attach confidence. page_texts = text.split("\f") if text else [] pages = [PageInfo(number=i + 1) for i in range(len(page_texts))] return OcrResult(text=text, pages=pages) def _ocr_pdf_claude(in_path, out_path, *, mode: str = SKIP_TEXT) -> OcrResult: """Claude-vision OCR backend: rasterize each PDF page and send to Claude. Requires the org-level Anthropic API key configured in assistant_settings.json (via the Assistant admin settings). Falls back gracefully when the key is absent or the anthropic package is not installed. """ try: from assistant import extract_fields, AssistantUnavailable, AssistantError except ImportError as exc: raise OcrEngineUnavailable( "assistant module not found — cannot use Claude-vision backend." ) from exc try: import fitz # PyMuPDF — used to rasterize PDF pages except ImportError as exc: raise OcrEngineUnavailable( "PyMuPDF (fitz) is not installed — required for Claude-vision OCR rasterization." ) from exc in_path = Path(in_path) out_path = Path(out_path) try: doc = fitz.open(str(in_path)) except Exception as exc: raise OcrEngineUnavailable(f"Failed to open PDF for Claude-vision OCR: {exc}") from exc all_text: list[str] = [] pages: list[PageInfo] = [] try: for page_num in range(len(doc)): page = doc.load_page(page_num) # Render at 150 DPI; good enough for text extraction, keeps tokens low mat = fitz.Matrix(150 / 72, 150 / 72) pix = page.get_pixmap(matrix=mat) img_bytes = pix.tobytes("png") try: result = extract_fields( img_bytes, ["full_text"], ) page_text = result.get("full_text", "") except (AssistantUnavailable, AssistantError) as exc: raise OcrEngineUnavailable( f"Claude-vision OCR failed on page {page_num + 1}: {exc}" ) from exc all_text.append(page_text) pages.append(PageInfo(number=page_num + 1)) finally: doc.close() # Claude-vision doesn't produce a searchable PDF; copy input to output # (gives downstream code a valid PDF file to work with) import shutil shutil.copy2(str(in_path), str(out_path)) combined = "\f".join(all_text) return OcrResult(text=combined, pages=pages) # --- Title-block field discovery (DVI-498) ---------------------------------- # Phase 2 (P2-A): read "CUSTOMER:", "JOB NO:", etc. from a drawing's title block # (bottom-right of the page; the area below the QR zone on carrier sheets) and # map them to Create-Carrier dialog fields so the form can be pre-filled. The # values are suggestions only — they stay user-editable, and we never write them # to the take-off CSV (extra data belongs in project_metadata.json). # Label -> Create-dialog field. Aliases are matched case-insensitively and the # longest alias wins, so more specific labels ("structure id") beat broader ones # ("structure"). Anchored to a ":" or "#" so plain prose does not trigger. _FIELD_ALIASES: list[tuple[str, str]] = [ ("customer_name", "customer name"), ("customer_name", "customer"), ("customer_name", "general contractor"), ("customer_name", "contractor name"), ("customer_name", "contractor"), ("customer_name", "client"), ("customer_name", "sold to"), ("project_name", "project name"), ("project_name", "project title"), ("project_name", "project"), ("job_number", "job number"), ("job_number", "job no"), ("job_number", "job #"), ("product_code", "product code"), ("product_code", "catalog no"), ("product_code", "model no"), ("product_code", "product"), ("part_ref", "part reference"), ("part_ref", "part ref"), ("part_ref", "part number"), ("part_ref", "part no"), ("part_ref", "piece mark"), ("part_ref", "mark"), ("weight", "unit weight"), ("weight", "weight"), ("structure_id", "structure id"), ("structure_id", "assembly id"), ("structure_id", "structure"), ("structure_id", "assembly"), ("spec_type", "spec./type"), ("spec_type", "spec type"), ("spec_type", "specification"), ("spec_type", "spec"), ("structure_size", "structure size"), ("structure_size", "size"), ("station", "station"), ("station", "sta"), ("production_date", "production date"), ("production_date", "date"), ] # How much of page 1 to scan for labeled fields, as fractions of page # width/height (PDF user space origin is bottom-left, so "bottom" = small y). # We scan the whole page: real carrier-sheet/drawing title blocks are anchored # at the far left and span the full width (DVI-504 board smoke test showed the # "Customer Name"/"Project Name" labels sit at ~3% from the left edge), so a # bottom-right-only window missed them entirely. The label-anchored parser in # _parse_labeled_fields is the real filter, so scanning wider only helps recall. _TITLEBLOCK_X_FRAC = 0.0 _TITLEBLOCK_Y_FRAC = 1.0 def extract_title_block_fields(pdf_path, *, backend: str | None = None) -> dict[str, str]: """Read labeled metadata from a PDF's title block and map it to Create dialog fields. Returns a dict of {field: value} for whatever was found (possibly empty). Never raises for a malformed/unreadable PDF — extraction is best-effort prefill, so failures degrade to an empty result. Strategy: prefer the embedded text layer (vector drawings / carrier sheets), restricted to the title-block region for region heuristics. If that yields nothing (image-only/scanned PDF), fall back to OCR via the selected backend. """ text = "" try: text = _titleblock_text_embedded(pdf_path) except Exception: # noqa: BLE001 — prefill must never break the caller text = "" fields = _parse_labeled_fields(text) if fields: return fields try: ocr_text = _titleblock_text_ocr(pdf_path, backend) except Exception: # noqa: BLE001 ocr_text = "" return _parse_labeled_fields(ocr_text) def extract_title_block_text(pdf_path, *, backend: str | None = None) -> str: """Return the raw title-block text (embedded layer, else OCR fallback) that feeds extract_title_block_fields. Exposed for callers that need to scan the text directly rather than rely on labeled "LABEL: value" pairs — e.g. the Project Builder tranId discovery (DVI-661). Best-effort: never raises.""" try: text = _titleblock_text_embedded(pdf_path) except Exception: # noqa: BLE001 — discovery must never break the caller text = "" if text and text.strip(): return text try: return _titleblock_text_ocr(pdf_path, backend) or "" except Exception: # noqa: BLE001 return "" def _titleblock_text_embedded(pdf_path) -> str: """Extract embedded text from the title-block region of page 1, preserving rough line structure. Returns "" if the page has no usable text layer.""" from pypdf import PdfReader # noqa: PLC0415 reader = PdfReader(str(pdf_path)) if not reader.pages: return "" page = reader.pages[0] box = page.mediabox width = float(box.width) or 1.0 height = float(box.height) or 1.0 x_min = width * _TITLEBLOCK_X_FRAC y_max = height * _TITLEBLOCK_Y_FRAC chunks: list[tuple[float, float, str, float]] = [] def _visitor(text, cm, tm, font_dict, font_size): # pypdf visitor signature if not text or not text.strip(): return x, y = tm[4], tm[5] if x >= x_min and y <= y_max: chunks.append((y, x, text, float(font_size or 0.0))) page.extract_text(visitor_text=_visitor) if not chunks: return "" # Group chunks into visual rows by y (top-to-bottom = descending y), then # order each row left-to-right by x. The grouping tolerance is proportional # to font size rather than a fixed value: in a two-column title block the # label and its value can sit ~15% of a line-height apart (label and value # have slightly different baselines, especially in an OCR'd text layer), # while distinct rows are a full line-height or more apart. A fixed 3pt # tolerance split "Customer Name:" from its value (DVI-504); 0.4x font size # pairs them yet still separates adjacent rows, and degrades to ~3pt for # ordinary small body fonts so vector-PDF parsing is unchanged. chunks.sort(key=lambda c: (-c[0], c[1])) lines: list[str] = [] cur: list[tuple[float, str]] = [] cur_y: float | None = None cur_fs = 0.0 for y, x, t, fs in chunks: tol = max(3.0, 0.4 * max(fs, cur_fs)) if cur_y is None or abs(y - cur_y) <= tol: cur.append((x, t)) cur_y = y if cur_y is None else cur_y cur_fs = max(cur_fs, fs) else: lines.append(" ".join(t for _x, t in sorted(cur))) cur = [(x, t)] cur_y = y cur_fs = fs if cur: lines.append(" ".join(t for _x, t in sorted(cur))) return "\n".join(lines) def _titleblock_text_ocr(pdf_path, backend: str | None) -> str: """OCR fallback for scanned/image-only PDFs: OCR page 1 via the configured backend, then re-read the *coordinate-aware* text from the resulting searchable PDF. Returns "" if OCR is unavailable. We deliberately do NOT use the linear OCR sidecar text here: title blocks are laid out as two columns (labels left, values right), and Tesseract's page segmentation emits the whole label column and then the whole value column, so "Customer Name:" and its value land on different lines and never pair up (DVI-504). Reading the OCR'd PDF's embedded text layer restores per-word positions, so _titleblock_text_embedded groups each label with its value by row and the same-line "LABEL: value" parser works.""" try: from pypdf import PdfReader, PdfWriter # noqa: PLC0415 except ImportError: return "" with tempfile.TemporaryDirectory() as td: td_path = Path(td) # Limit OCR to the first page to keep prefill fast on large drawing sets. try: reader = PdfReader(str(pdf_path)) if not reader.pages: return "" writer = PdfWriter() writer.add_page(reader.pages[0]) in_path = td_path / "page1.pdf" with open(in_path, "wb") as fh: writer.write(fh) except Exception: # noqa: BLE001 in_path = Path(pdf_path) out_path = td_path / "ocr.pdf" try: ocr_pdf(in_path, out_path, mode=FORCE_OCR, backend=backend) except OcrEngineUnavailable: return "" try: return _titleblock_text_embedded(out_path) except Exception: # noqa: BLE001 — prefill must never break the caller return "" def extract_region(pdf_path, page_no: int, frac_rect, *, raw: bool = False, backend: str | None = None): """OCR a user-selected sub-region of one page (DVI-645). `frac_rect` is (x0, y0, x1, y1) as fractions (0..1) of the page's rendered width/height — the front-end maps the on-screen selection to fractions so it never needs to know the server render zoom. Strategy mirrors extract_title_block_fields: prefer embedded text inside the rectangle (instant, works for vector drawings / carrier sheets); if the region has no text layer (scanned/image PDF), render just the clip and OCR it. Never raises — region OCR is convenience and degrades to empty output. Returns the raw recognized text (str) when `raw` is True — used to fill one targeted field. Otherwise returns a {field: value} dict parsed from "LABEL: value" lines — used for the "OCR this region → best guess" flow. """ text = _region_text(pdf_path, page_no, frac_rect, backend) if raw: return _clean_region_text(text) return _parse_labeled_fields(text) def _region_text(pdf_path, page_no: int, frac_rect, backend: str | None) -> str: """Embedded-text-in-rect → OCR-the-clip fallback. Returns "" on any failure.""" try: import fitz # noqa: PLC0415 — PyMuPDF, same dep used by the preview render except Exception: # noqa: BLE001 return "" try: fx0, fy0, fx1, fy1 = frac_rect except (TypeError, ValueError): return "" try: doc = fitz.open(str(pdf_path)) except Exception: # noqa: BLE001 return "" try: if page_no < 0 or page_no >= doc.page_count: return "" page = doc.load_page(page_no) r = page.rect clip = fitz.Rect( r.x0 + fx0 * r.width, r.y0 + fy0 * r.height, r.x0 + fx1 * r.width, r.y0 + fy1 * r.height, ) try: embedded = (page.get_text("text", clip=clip) or "").strip() except Exception: # noqa: BLE001 embedded = "" if embedded: return embedded return _region_ocr_text(page, clip, backend) finally: doc.close() def _region_ocr_text(page, clip, backend: str | None) -> str: """Render the clip to an image-only PDF and OCR it, then read the coordinate-aware text from the result (the linear sidecar jumbles two-column title blocks — see _titleblock_text_ocr). Returns "" if OCR is unavailable.""" try: import fitz # noqa: PLC0415 except Exception: # noqa: BLE001 return "" try: pix = page.get_pixmap(matrix=fitz.Matrix(3.0, 3.0), clip=clip) except Exception: # noqa: BLE001 return "" with tempfile.TemporaryDirectory() as td: td_path = Path(td) in_path = td_path / "region.pdf" out_path = td_path / "region_ocr.pdf" try: img_doc = fitz.open(stream=pix.tobytes("png"), filetype="png") try: pdf_bytes = img_doc.convert_to_pdf() finally: img_doc.close() in_path.write_bytes(pdf_bytes) except Exception: # noqa: BLE001 return "" try: ocr_pdf(in_path, out_path, mode=FORCE_OCR, backend=backend) except OcrEngineUnavailable: return "" except Exception: # noqa: BLE001 return "" try: od = fitz.open(str(out_path)) try: return (od.load_page(0).get_text("text") or "").strip() finally: od.close() except Exception: # noqa: BLE001 return "" def _clean_region_text(text: str) -> str: """Collapse a region's recognized text into a single trimmed line, suitable for dropping straight into one form field.""" if not text: return "" return re.sub(r"\s+", " ", text).strip() def _parse_labeled_fields(text: str) -> dict[str, str]: """Parse "LABEL: value" pairs out of title-block text into Create fields. Splits each line into column segments on runs of 2+ spaces/tabs (title blocks are grid-like), then matches the longest known label that a segment starts with. First non-empty value per field wins.""" if not text: return {} pairs = sorted(_FIELD_ALIASES, key=lambda fa: len(fa[1]), reverse=True) out: dict[str, str] = {} for raw_line in text.splitlines(): segs = [s.strip() for s in re.split(r"\s{2,}|\t+", raw_line)] for i, seg in enumerate(segs): if not seg or (":" not in seg and "#" not in seg): continue for fieldname, alias in pairs: if fieldname in out: continue m = re.match(re.escape(alias) + r"\s*[:#]+\s*(.*)$", seg, re.IGNORECASE) if not m: continue val = _clean_value(m.group(1)) if not val: # Label cell with no inline value: take the next column on # this row as the value (two-column title blocks put labels # and values in separate cells, e.g. "Customer Name:" then # "Blankenberger Brothers, Inc."). Skip if the next cell is # itself a labeled cell so we never steal another field. nxt = segs[i + 1] if i + 1 < len(segs) else "" if nxt and ":" not in nxt and "#" not in nxt: val = _clean_value(nxt) if val: out[fieldname] = val break return out def _clean_value(val: str) -> str: """Tidy a captured value: collapse whitespace, strip stray separators, cap length. Drops values that are obviously empty placeholders.""" val = re.sub(r"\s+", " ", val).strip(" \t:#-_/|").strip() if val.lower() in ("", "n/a", "na", "none", "tbd", "-"): return "" return val[:120]