""" text_layer.py - deterministic PDF text-layer extraction (PyMuPDF, no LLM). Most CAD-produced drawing sets carry a real vector text layer. We extract it once per job and feed it to the extractor (grounding), the grounding guard (rescue tier), and the wave-5b verifier (text oracle + high-DPI evidence crops). Pages below TEXT_LAYER_MIN_CHARS of text are treated as having no text layer (scanned/raster sheets stay vision-only). If PyMuPDF is unavailable the module degrades gracefully: every public function returns empty/None, equivalent to TEXT_LAYER_ENABLED=false. """ import re from typing import Dict, List, Optional, Tuple from backend import config try: # PyMuPDF >= 1.24 prefers the pymupdf name; fitz works everywhere. import pymupdf as fitz except ImportError: # pragma: no cover - older PyMuPDF try: import fitz except ImportError: # pragma: no cover - PyMuPDF not installed fitz = None _warned_unavailable = False # Word token normalization for evidence matching: lowercase alphanumeric only. _TOKEN_RE = re.compile(r"[^a-z0-9]+") # Fuzzy match floor: fraction of needle tokens that must align with the page's # word sequence for a bbox to count as a confident evidence location. _FUZZY_MIN_RATIO = 0.6 def _fitz_or_none(): """Return the fitz module, logging once if PyMuPDF is missing.""" global _warned_unavailable if fitz is None and not _warned_unavailable: print("[TextLayer] PyMuPDF not available - text-layer grounding disabled") _warned_unavailable = True return fitz def extract_text_layers(pdf_path: str) -> Dict[int, Dict]: """ Extract the text layer of every page. Returns {1-based page_number: {"text": str, "words": [{"text", "bbox": (x0,y0,x1,y1)}, ...], "has_text_layer": bool}}. Returns {} when disabled or unavailable. """ if not config.TEXT_LAYER_ENABLED: return {} f = _fitz_or_none() if f is None: return {} try: doc = f.open(pdf_path) except Exception as exc: print(f"[TextLayer] could not open {pdf_path}: {exc}") return {} layers: Dict[int, Dict] = {} try: for index in range(doc.page_count): page = doc[index] text = page.get_text("text") or "" words = [ {"text": w[4], "bbox": (w[0], w[1], w[2], w[3])} for w in (page.get_text("words") or []) ] has_text_layer = len(text.strip()) >= config.TEXT_LAYER_MIN_CHARS if not has_text_layer: print(f"[TextLayer] Page {index + 1}: {len(text.strip())} chars " f"(< TEXT_LAYER_MIN_CHARS={config.TEXT_LAYER_MIN_CHARS}) - " f"vision-only") layers[index + 1] = { "text": text, "words": words, "has_text_layer": has_text_layer, } finally: doc.close() return layers def attach_text_layers( pdf_path: str, pages: List[Dict], text_dir: Optional[str] = None, ) -> Dict[int, List[Dict]]: """ Attach page["text_layer"] (text or None) to each converted page dict and return the runner-local {page_number: words} map (kept off page dicts - those get serialized). When text_dir is set, dump one .txt per page there (plain file writes; ProjectMemory is a closed registry). """ layers = extract_text_layers(pdf_path) page_words: Dict[int, List[Dict]] = {} for page in pages: layer = layers.get(page["page_number"]) or {} page["text_layer"] = layer.get("text") if layer.get("has_text_layer") else None page_words[page["page_number"]] = layer.get("words") or [] if text_dir and layers: import os os.makedirs(text_dir, exist_ok=True) for page_number, layer in layers.items(): if not layer.get("has_text_layer"): continue with open(os.path.join(text_dir, f"page-{page_number:03d}.txt"), "w", encoding="utf-8") as fh: fh.write(layer.get("text") or "") return page_words def _tokens(text: str) -> List[str]: return [t for t in _TOKEN_RE.split(text.lower()) if t] def _union_bbox(boxes: List[Tuple[float, float, float, float]]): return ( min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes), ) def find_evidence_bbox( words: List[Dict], needle: str, ) -> Optional[Tuple[float, float, float, float]]: """ Best-effort fuzzy substring match of an evidence source_text against the page's word sequence. Returns the union bbox of the matched words, or None when nothing aligns confidently. Exact contiguous token runs win; otherwise the best-scoring window with >= _FUZZY_MIN_RATIO token alignment is accepted (vision quotes imperfectly but the value is real page text). """ if not words or not needle: return None needle_tokens = _tokens(str(needle)) if not needle_tokens: return None page_tokens = [_tokens(w.get("text") or "") for w in words] # Flatten multi-token words, remembering which word each token came from. flat: List[Tuple[str, int]] = [] for word_index, parts in enumerate(page_tokens): for part in parts: flat.append((part, word_index)) if not flat: return None n = len(needle_tokens) best_span = None best_score = 0.0 for start in range(0, len(flat)): window = flat[start:start + n] if not window: break score = sum(1 for i, tok in enumerate(needle_tokens) if i < len(window) and window[i][0] == tok) / n if score > best_score: best_score = score best_span = window if best_score == 1.0: break if best_span is None or best_score < _FUZZY_MIN_RATIO: return None word_indexes = {word_index for _, word_index in best_span} return _union_bbox([words[i]["bbox"] for i in sorted(word_indexes)]) def render_crop( pdf_path: str, page_number: int, bbox: Tuple[float, float, float, float], dpi: Optional[int] = None, margin_pts: Optional[float] = None, ) -> Optional[bytes]: """ Render a clip of one page around bbox (+ margin, clamped to the page) at the given DPI and return JPEG bytes, or None on any failure. """ f = _fitz_or_none() if f is None: return None dpi = dpi or config.VERIFY_CROP_DPI margin_pts = config.VERIFY_CROP_MARGIN_PTS if margin_pts is None else margin_pts try: doc = f.open(pdf_path) try: page = doc[page_number - 1] rect = f.Rect( bbox[0] - margin_pts, bbox[1] - margin_pts, bbox[2] + margin_pts, bbox[3] + margin_pts, ) & page.rect if rect.is_empty: return None pix = page.get_pixmap(clip=rect, dpi=dpi) return pix.tobytes("jpeg") finally: doc.close() except Exception as exc: print(f"[TextLayer] render_crop failed on page {page_number}: {exc}") return None def coverage_gaps(pages: List[Dict], sheets: List[Dict]) -> List[int]: """ Page numbers that have a text layer but whose extraction failed or returned 0 objects - the silent extraction-loss signal. Logs one [TextLayer] line per gap. """ by_page = {s.get("page_number"): s for s in sheets or []} gaps: List[int] = [] for page in pages: text = page.get("text_layer") if not text: continue sheet = by_page.get(page["page_number"]) extracted = len(sheet.get("assertions") or []) if sheet else 0 if extracted == 0: gaps.append(page["page_number"]) print(f"[TextLayer] Page {page['page_number']}: text layer present " f"({len(text)} chars) but no objects extracted — possible " f"extraction gap") return gaps