Spaces:
Configuration error
Configuration error
| """ | |
| ingest.py — Build the search index for the Digital KP 2030 assistant. | |
| Run this ONCE on your own machine, then commit the `index/` folder to the Space. | |
| The app itself never runs OCR, so the Space starts fast and needs no system packages. | |
| python ingest.py | |
| Why OCR matters here | |
| -------------------- | |
| The published Digital KP 2030 file is a design-heavy, image-based document. Its | |
| embedded text layer is almost empty (~90-120 usable characters on most roadmap | |
| pages) and parts of it are mojibake from custom font encodings. A plain | |
| PyPDF/pdfplumber pipeline therefore retrieves almost nothing — which is the most | |
| likely reason the official website assistant fails. This script extracts real | |
| text first and falls back to OCR for any page that comes back thin or garbled. | |
| """ | |
| from __future__ import annotations | |
| import io | |
| import json | |
| import os | |
| import re | |
| import sys | |
| import zipfile | |
| from pathlib import Path | |
| import numpy as np | |
| # -------------------------------------------------------------------------- | |
| # Configuration | |
| # -------------------------------------------------------------------------- | |
| ROOT = Path(__file__).parent | |
| PDF_PATH = ROOT / "data" / "digital_kp_2030.pdf" | |
| INDEX_DIR = ROOT / "index" | |
| EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2" | |
| CHUNK_CHARS = 1100 # target chunk size in characters | |
| CHUNK_OVERLAP = 180 # overlap so sentences are not cut mid-idea | |
| MIN_PAGE_CHARS = 220 # below this, a page is treated as "no real text" -> OCR | |
| MOJIBAKE_RATIO = 0.10 # >10% control characters -> broken encoding -> OCR | |
| OCR_LANGS = "eng" # add "+urd" once you install the Urdu tesseract data | |
| OCR_DPI = 220 | |
| # -------------------------------------------------------------------------- | |
| # Text quality helpers | |
| # -------------------------------------------------------------------------- | |
| def is_unusable(text: str) -> bool: | |
| """True when a page's extracted text is too thin or too corrupted to use.""" | |
| if not text or len(text.strip()) < MIN_PAGE_CHARS: | |
| return True | |
| control = sum(1 for c in text if ord(c) < 32 and c not in "\r\n\t") | |
| return control / max(len(text), 1) > MOJIBAKE_RATIO | |
| def clean(text: str) -> str: | |
| """Normalise whitespace and strip control characters left by odd encodings.""" | |
| text = "".join(c for c in text if ord(c) >= 32 or c in "\n\t") | |
| text = text.replace("\r", "\n") | |
| text = re.sub(r"[ \t]+", " ", text) | |
| text = re.sub(r"\n{3,}", "\n\n", text) | |
| return text.strip() | |
| # -------------------------------------------------------------------------- | |
| # Page extraction | |
| # -------------------------------------------------------------------------- | |
| def ocr_image(image) -> str: | |
| """OCR a PIL image. Imported lazily so the app never needs these packages.""" | |
| import pytesseract | |
| return pytesseract.image_to_string(image, lang=OCR_LANGS) | |
| def pages_from_zip(path: Path) -> list[tuple[int, str]]: | |
| """ | |
| Some distributions of this document are actually a ZIP of page images | |
| (1.jpeg ... N.jpeg) with a .pdf extension. Handle that case explicitly. | |
| """ | |
| from PIL import Image | |
| out: list[tuple[int, str]] = [] | |
| with zipfile.ZipFile(path) as z: | |
| names = [n for n in z.namelist() if n.lower().endswith((".jpg", ".jpeg", ".png"))] | |
| names.sort(key=lambda n: int(re.sub(r"\D", "", Path(n).stem) or 0)) | |
| for name in names: | |
| page_no = int(re.sub(r"\D", "", Path(name).stem) or 0) | |
| sidecar = f"{Path(name).stem}.txt" | |
| text = "" | |
| if sidecar in z.namelist(): | |
| text = clean(z.read(sidecar).decode("utf-8", errors="replace")) | |
| if is_unusable(text): | |
| print(f" page {page_no:>3}: OCR (image archive)") | |
| text = clean(ocr_image(Image.open(io.BytesIO(z.read(name))))) | |
| else: | |
| print(f" page {page_no:>3}: text layer ({len(text)} chars)") | |
| if text: | |
| out.append((page_no, text)) | |
| return out | |
| def pages_from_pdf(path: Path) -> list[tuple[int, str]]: | |
| """Extract text page by page, falling back to OCR wherever the layer is bad.""" | |
| import pdfplumber | |
| out: list[tuple[int, str]] = [] | |
| ocr_pages: list[int] = [] | |
| with pdfplumber.open(path) as pdf: | |
| for i, page in enumerate(pdf.pages, start=1): | |
| text = clean(page.extract_text() or "") | |
| if is_unusable(text): | |
| ocr_pages.append(i) | |
| out.append((i, "")) # placeholder, filled in below | |
| print(f" page {i:>3}: queued for OCR") | |
| else: | |
| out.append((i, text)) | |
| print(f" page {i:>3}: text layer ({len(text)} chars)") | |
| if ocr_pages: | |
| from pdf2image import convert_from_path | |
| print(f"\nRunning OCR on {len(ocr_pages)} page(s) at {OCR_DPI} DPI...") | |
| for n in ocr_pages: | |
| image = convert_from_path(str(path), dpi=OCR_DPI, first_page=n, last_page=n)[0] | |
| text = clean(ocr_image(image)) | |
| out[n - 1] = (n, text) | |
| print(f" page {n:>3}: OCR ({len(text)} chars)") | |
| return [(n, t) for n, t in out if t] | |
| def load_pages(path: Path) -> list[tuple[int, str]]: | |
| if not path.exists(): | |
| sys.exit(f"ERROR: {path} not found. Put the PDF there first — see README.md.") | |
| if zipfile.is_zipfile(path): | |
| print("Detected an image archive rather than a true PDF.\n") | |
| return pages_from_zip(path) | |
| return pages_from_pdf(path) | |
| # -------------------------------------------------------------------------- | |
| # Chunking | |
| # -------------------------------------------------------------------------- | |
| def split_page(text: str) -> list[str]: | |
| """ | |
| Split one page into overlapping chunks on paragraph then sentence boundaries. | |
| Deliberately dependency-free so there is nothing to debug on deployment. | |
| """ | |
| paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()] | |
| if not paragraphs: | |
| return [] | |
| chunks: list[str] = [] | |
| buffer = "" | |
| for para in paragraphs: | |
| if len(buffer) + len(para) + 2 <= CHUNK_CHARS: | |
| buffer = f"{buffer}\n\n{para}" if buffer else para | |
| continue | |
| if buffer: | |
| chunks.append(buffer) | |
| while len(para) > CHUNK_CHARS: | |
| cut = para.rfind(". ", 0, CHUNK_CHARS) | |
| cut = cut + 1 if cut > CHUNK_CHARS // 2 else CHUNK_CHARS | |
| chunks.append(para[:cut].strip()) | |
| para = para[max(0, cut - CHUNK_OVERLAP):].strip() | |
| buffer = para | |
| if buffer: | |
| chunks.append(buffer) | |
| # add tail-overlap between neighbouring chunks | |
| joined: list[str] = [] | |
| for i, c in enumerate(chunks): | |
| if i > 0: | |
| joined.append((chunks[i - 1][-CHUNK_OVERLAP:] + " " + c).strip()) | |
| else: | |
| joined.append(c) | |
| return [c for c in joined if len(c) > 60] | |
| def build_chunks(pages: list[tuple[int, str]]) -> list[dict]: | |
| records: list[dict] = [] | |
| for page_no, text in pages: | |
| for j, chunk in enumerate(split_page(text)): | |
| records.append({ | |
| "id": f"p{page_no}-{j}", | |
| "page": page_no, | |
| "text": chunk, | |
| }) | |
| return records | |
| # -------------------------------------------------------------------------- | |
| # Embedding + index | |
| # -------------------------------------------------------------------------- | |
| def main() -> None: | |
| print(f"Reading {PDF_PATH.name}\n") | |
| pages = load_pages(PDF_PATH) | |
| print(f"\nExtracted text from {len(pages)} page(s).") | |
| records = build_chunks(pages) | |
| if not records: | |
| sys.exit("ERROR: no chunks produced. Check that OCR ran correctly.") | |
| print(f"Built {len(records)} chunks.\n") | |
| from sentence_transformers import SentenceTransformer | |
| import faiss | |
| print(f"Embedding with {EMBED_MODEL} (first run downloads ~90 MB)...") | |
| model = SentenceTransformer(EMBED_MODEL) | |
| vectors = model.encode( | |
| [r["text"] for r in records], | |
| batch_size=32, | |
| show_progress_bar=True, | |
| normalize_embeddings=True, # so inner product == cosine similarity | |
| ).astype("float32") | |
| index = faiss.IndexFlatIP(vectors.shape[1]) | |
| index.add(vectors) | |
| INDEX_DIR.mkdir(exist_ok=True) | |
| faiss.write_index(index, str(INDEX_DIR / "faiss.index")) | |
| (INDEX_DIR / "chunks.json").write_text( | |
| json.dumps(records, ensure_ascii=False), encoding="utf-8" | |
| ) | |
| (INDEX_DIR / "meta.json").write_text(json.dumps({ | |
| "embed_model": EMBED_MODEL, | |
| "chunks": len(records), | |
| "pages": len(pages), | |
| "dim": int(vectors.shape[1]), | |
| "source": PDF_PATH.name, | |
| }, indent=2), encoding="utf-8") | |
| print(f"\nDone. Index written to {INDEX_DIR}/") | |
| print(f" {len(records)} chunks · {len(pages)} pages · dim {vectors.shape[1]}") | |
| if __name__ == "__main__": | |
| main() | |