| """Deterministic Kerala SSLC Physics textbook ingestion. |
| |
| The official PDF is the only source of educational text in the generated |
| artifacts. Native extraction is attempted first. Pages with too little text |
| are reported for selective OCR instead of silently fabricating content. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import os |
| import re |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any, Iterable |
|
|
| from pypdf import PdfReader |
|
|
|
|
| PROJECT_ROOT = Path(__file__).resolve().parents[2] |
| OUTPUT_ROOT = PROJECT_ROOT / "data" / "curriculum" / "kerala-sslc" / "physics" |
| EXTRACTION_VERSION = "docdoe-textbook-v1.1" |
| DISCOVERY_HINTS = ( |
| "sslc", |
| "standard-10", |
| "standard 10", |
| "class-10", |
| "class 10", |
| "physics", |
| "scert", |
| "textbook", |
| ) |
| SKIP_DIRS = {".git", ".next", "node_modules", ".venv", "generated-videos", "uploads"} |
|
|
|
|
| @dataclass(frozen=True) |
| class ChapterCandidate: |
| number: str |
| title: str |
| start_page: int |
|
|
|
|
| def sha256_file(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def stable_hash(*parts: object, length: int = 20) -> str: |
| raw = "|".join(str(part) for part in parts) |
| return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:length] |
|
|
|
|
| def normalized_text(value: str) -> str: |
| return re.sub(r"\s+", " ", value).strip() |
|
|
|
|
| def source_roots() -> list[Path]: |
| roots = [PROJECT_ROOT] |
| configured = os.getenv("TEXTBOOK_SOURCE_DIRS", "") |
| for raw in configured.split(os.pathsep): |
| if raw.strip(): |
| roots.append(Path(raw.strip()).expanduser().resolve()) |
| return list(dict.fromkeys(roots)) |
|
|
|
|
| def discover_pdf_candidates(roots: Iterable[Path] | None = None) -> list[Path]: |
| candidates: list[Path] = [] |
| for root in roots or source_roots(): |
| if not root.exists(): |
| continue |
| for path in root.rglob("*.pdf"): |
| if any(part.lower() in SKIP_DIRS for part in path.parts): |
| continue |
| hint = str(path).lower().replace("_", " ") |
| if any(term in hint for term in DISCOVERY_HINTS): |
| candidates.append(path.resolve()) |
| return sorted(set(candidates)) |
|
|
|
|
| def read_native_pages(path: Path) -> tuple[list[str], dict[str, Any]]: |
| reader = PdfReader(str(path)) |
| metadata = {str(key).lstrip("/"): str(value) for key, value in (reader.metadata or {}).items()} |
| pages = [(page.extract_text() or "").replace("\x00", "") for page in reader.pages] |
| return pages, metadata |
|
|
|
|
| def read_native_opening(path: Path, count: int = 6) -> list[str]: |
| reader = PdfReader(str(path)) |
| return [ |
| (reader.pages[index].extract_text() or "").replace("\x00", "") |
| for index in range(min(count, len(reader.pages))) |
| ] |
|
|
|
|
| def is_kerala_sslc_physics(pages: list[str]) -> bool: |
| opening = normalized_text("\n".join(pages[:6])).lower() |
| return ( |
| "physics" in opening |
| and "state council of educational research and training" in opening |
| and ("standard" in opening or "sslc" in opening) |
| and "government of kerala" in opening |
| ) |
|
|
|
|
| def detect_part(pages: list[str]) -> str | None: |
| cover = normalized_text("\n".join(pages[:3])) |
| match = re.search(r"\bPart\s+([12])\b", cover, flags=re.IGNORECASE) |
| return f"Part {match.group(1)}" if match else None |
|
|
|
|
| def parse_contents_chapters(page_text: str) -> list[ChapterCandidate]: |
| candidates: list[ChapterCandidate] = [] |
| for raw_line in page_text.splitlines(): |
| line = normalized_text(raw_line) |
| match = re.match(r"^(\d{1,2})\s+(.+?)(?:[-.\s]{3,})\s*(\d{1,3})$", line) |
| if not match: |
| match = re.match(r"^(\d{1,2})\s+([A-Za-z][A-Za-z :,&-]{4,}?)\s+(\d{1,3})$", line) |
| if not match: |
| continue |
| number, title, start = match.groups() |
| title = title.strip(" -.\t") |
| if title and int(start) > 0: |
| candidates.append(ChapterCandidate(number=number, title=title, start_page=int(start))) |
| return candidates |
|
|
|
|
| def detect_chapters(pages: list[str]) -> tuple[list[ChapterCandidate], list[str]]: |
| warnings: list[str] = [] |
| candidates: list[ChapterCandidate] = [] |
| for page in pages[:12]: |
| if "contents" in page.lower(): |
| candidates.extend(parse_contents_chapters(page)) |
| unique: dict[tuple[str, int], ChapterCandidate] = {} |
| for candidate in candidates: |
| unique[(candidate.number, candidate.start_page)] = candidate |
| ordered = sorted(unique.values(), key=lambda item: item.start_page) |
| if not ordered: |
| warnings.append("No chapter table-of-contents entries were detected.") |
| for previous, current in zip(ordered, ordered[1:]): |
| if current.start_page <= previous.start_page: |
| warnings.append( |
| f"Chapter boundary is not increasing: {previous.title} -> {current.title}.", |
| ) |
| return ordered, warnings |
|
|
|
|
| def find_chapter_pdf_page(pages: list[str], title: str) -> int | None: |
| needle = normalized_text(title).lower() |
| needle_tokens = set(re.findall(r"[a-z]+", needle)) |
| for index, page in enumerate(pages[5:], start=6): |
| haystack = normalized_text(page).lower() |
| haystack_tokens = set(re.findall(r"[a-z]+", haystack[:1400])) |
| overlap = len(needle_tokens & haystack_tokens) / max(1, len(needle_tokens)) |
| if needle in haystack or overlap >= 0.8: |
| return index |
| return None |
|
|
|
|
| def classify_block(text: str) -> str: |
| value = normalized_text(text) |
| lower = value.lower() |
| if re.match(r"^fig\.?\s*\d", lower): |
| return "caption" |
| if "let's assess" in lower or value.endswith("?"): |
| return "question" |
| if lower.startswith(("let's do", "activity", "experiment")): |
| return "activity" |
| if re.search(r"\b(is called|is known as|refers to|are those|is the)\b", lower): |
| return "definition" |
| if "=" in value or re.search(r"[λ∝×÷]\s*", value): |
| return "formula" |
| if lower.startswith(("summary", "remember", "in brief")): |
| return "summary" |
| if len(value) <= 90 and not value.endswith((".", "?", "!")): |
| return "heading" |
| return "paragraph" |
|
|
|
|
| def page_blocks( |
| source_id: str, |
| chapter_id: str, |
| page_number: int, |
| pdf_page_number: int, |
| page_text: str, |
| ) -> list[dict[str, Any]]: |
| raw_blocks = [normalized_text(part) for part in re.split(r"\n\s*\n|\n(?=[A-Z•])", page_text)] |
| blocks: list[dict[str, Any]] = [] |
| for index, raw in enumerate(block for block in raw_blocks if block): |
| block_type = classify_block(raw) |
| source_hash = stable_hash(source_id, page_number, block_type, raw, length=64) |
| blocks.append( |
| { |
| "blockId": f"blk-{stable_hash(chapter_id, page_number, index, raw)}", |
| "chapterId": chapter_id, |
| "pageNumber": page_number, |
| "pdfPageNumber": pdf_page_number, |
| "blockType": block_type, |
| "rawText": raw, |
| "normalizedText": normalized_text(raw), |
| "confidence": 0.92 if len(raw) >= 20 else 0.72, |
| "sourceHash": source_hash, |
| } |
| ) |
| return blocks |
|
|
|
|
| def figures_from_page( |
| source_id: str, |
| chapter_id: str, |
| page_number: int, |
| pdf_page_number: int, |
| page_text: str, |
| ) -> list[dict[str, Any]]: |
| figures: list[dict[str, Any]] = [] |
| for index, match in enumerate(re.finditer(r"Fig\.?\s*(\d+\.\d+)(?:\s*\(([a-z])\))?", page_text, re.IGNORECASE)): |
| label = f"Fig. {match.group(1)}{f' ({match.group(2)})' if match.group(2) else ''}" |
| figures.append( |
| { |
| "figureId": f"fig-{stable_hash(chapter_id, page_number, label, index)}", |
| "chapterId": chapter_id, |
| "pageNumber": page_number, |
| "pdfPageNumber": pdf_page_number, |
| "caption": label, |
| "sourceHash": stable_hash(source_id, page_number, label, length=64), |
| "confidence": 0.82, |
| "requiresRegionRender": True, |
| } |
| ) |
| return figures |
|
|
|
|
| def build_source(path: Path) -> tuple[dict[str, Any], list[str], dict[str, Any]]: |
| pages, metadata = read_native_pages(path) |
| if not is_kerala_sslc_physics(pages): |
| raise ValueError(f"Not a verified Kerala SSLC Physics textbook: {path}") |
| file_hash = sha256_file(path) |
| part = detect_part(pages) |
| part_slug = (part or "part-unknown").lower().replace(" ", "-") |
| source_id = f"scert-sslc-physics-{part_slug}-{file_hash[:12]}" |
| source = { |
| "sourceId": source_id, |
| "filePath": str(path.relative_to(PROJECT_ROOT)).replace("\\", "/") if path.is_relative_to(PROJECT_ROOT) else str(path), |
| "fileName": path.name, |
| "sha256": file_hash, |
| "board": "Kerala SCERT", |
| "classLevel": "SSLC", |
| "subject": "Physics", |
| "medium": "English", |
| "part": part, |
| "pageCount": len(pages), |
| "extractionVersion": EXTRACTION_VERSION, |
| "metadata": metadata, |
| } |
| return source, pages, metadata |
|
|
|
|
| def ingest_textbook(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: |
| source, pages, _metadata = build_source(path) |
| chapter_candidates, warnings = detect_chapters(pages) |
| physical_starts = [find_chapter_pdf_page(pages, candidate.title) for candidate in chapter_candidates] |
| chapters: list[dict[str, Any]] = [] |
| assigned_pages: set[int] = set() |
| for index, candidate in enumerate(chapter_candidates): |
| physical_start = physical_starts[index] |
| if physical_start is None: |
| warnings.append(f"Could not locate the physical PDF start for chapter: {candidate.title}.") |
| continue |
| next_physical = physical_starts[index + 1] if index + 1 < len(physical_starts) else None |
| physical_end = next_physical - 1 if next_physical is not None else len(pages) |
| page_offset = candidate.start_page - physical_start |
| end_page = physical_end + page_offset |
| part_number = re.search(r"\d+", source.get("part") or "") |
| chapter_id = f"phy-p{part_number.group(0) if part_number else 'x'}-c{candidate.number}" |
| blocks: list[dict[str, Any]] = [] |
| figures: list[dict[str, Any]] = [] |
| low_text_pages: list[int] = [] |
| for pdf_page_number in range(physical_start, physical_end + 1): |
| assigned_pages.add(pdf_page_number) |
| page_number = pdf_page_number + page_offset |
| page_text = pages[pdf_page_number - 1] |
| if len(normalized_text(page_text)) < 80: |
| low_text_pages.append(page_number) |
| blocks.extend(page_blocks(source["sourceId"], chapter_id, page_number, pdf_page_number, page_text)) |
| figures.extend(figures_from_page(source["sourceId"], chapter_id, page_number, pdf_page_number, page_text)) |
| headings = [block for block in blocks if block["blockType"] == "heading"] |
| chapter_warnings = [f"Page {page} needs selective OCR review." for page in low_text_pages] |
| chapters.append( |
| { |
| "chapterId": chapter_id, |
| "sourceId": source["sourceId"], |
| "chapterNumber": candidate.number, |
| "title": candidate.title, |
| "startPage": candidate.start_page, |
| "endPage": end_page, |
| "pdfStartPage": physical_start, |
| "pdfEndPage": physical_end, |
| "learningOutcomes": [], |
| "sections": headings, |
| "exercises": [block for block in blocks if block["blockType"] == "question"], |
| "figures": figures, |
| "blocks": blocks, |
| "extractionWarnings": chapter_warnings, |
| } |
| ) |
| unassigned = [page for page in range(1, len(pages) + 1) if page not in assigned_pages] |
| extracted = { |
| "schemaVersion": 1, |
| "source": source, |
| "pages": [ |
| { |
| "pdfPageNumber": index + 1, |
| "text": text, |
| "nativeTextLength": len(normalized_text(text)), |
| "needsOcr": len(normalized_text(text)) < 80, |
| } |
| for index, text in enumerate(pages) |
| ], |
| "chapters": chapters, |
| "warnings": warnings, |
| } |
| report = { |
| "sourceId": source["sourceId"], |
| "textbooksDiscovered": 1, |
| "chaptersDiscovered": len(chapters), |
| "chapterPages": [ |
| { |
| "chapterId": chapter["chapterId"], |
| "printedStart": chapter["startPage"], |
| "printedEnd": chapter["endPage"], |
| "pdfStart": chapter["pdfStartPage"], |
| "pdfEnd": chapter["pdfEndPage"], |
| } |
| for chapter in chapters |
| ], |
| "unassignedPages": unassigned, |
| "duplicateChapterCandidates": [], |
| "lowConfidenceSections": [ |
| block["blockId"] |
| for chapter in chapters |
| for block in chapter["blocks"] |
| if block["confidence"] < 0.8 |
| ], |
| "equationExtractionWarnings": [ |
| f"Review formula block {block['blockId']} on page {block['pageNumber']}" |
| for chapter in chapters |
| for block in chapter["blocks"] |
| if block["blockType"] == "formula" and len(block["normalizedText"]) > 220 |
| ], |
| "warnings": warnings, |
| } |
| return extracted, report |
|
|
|
|
| def write_json_if_changed(path: Path, payload: Any) -> bool: |
| rendered = json.dumps(payload, ensure_ascii=False, indent=2) + "\n" |
| if path.exists() and path.read_text(encoding="utf-8") == rendered: |
| return False |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(rendered, encoding="utf-8") |
| return True |
|
|
|
|
| def build_manifest(extractions: list[dict[str, Any]]) -> dict[str, Any]: |
| sources = [item["source"] for item in extractions] |
| chapters: list[dict[str, Any]] = [] |
| for item in extractions: |
| for chapter in item["chapters"]: |
| blocks = chapter["blocks"] |
| chapters.append( |
| { |
| "chapterId": chapter["chapterId"], |
| "chapterNumber": chapter["chapterNumber"], |
| "title": chapter["title"], |
| "sourceId": chapter["sourceId"], |
| "sourcePages": {"start": chapter["startPage"], "end": chapter["endPage"]}, |
| "pdfPages": {"start": chapter["pdfStartPage"], "end": chapter["pdfEndPage"]}, |
| "sections": [block["normalizedText"] for block in chapter["sections"]], |
| "learningOutcomes": [], |
| "figures": chapter["figures"], |
| "experiments": [block for block in blocks if block["blockType"] in {"activity", "experiment"}], |
| "formulas": [block for block in blocks if block["blockType"] == "formula"], |
| "workedExamples": [block for block in blocks if block["blockType"] == "example"], |
| "exerciseQuestionCount": len(chapter["exercises"]), |
| "extractionWarnings": chapter["extractionWarnings"], |
| } |
| ) |
| return { |
| "schemaVersion": 1, |
| "board": "Kerala SCERT", |
| "classLevel": "SSLC", |
| "subject": "Physics", |
| "medium": "English", |
| "textbookSources": sources, |
| "chapters": chapters, |
| } |
|
|
|
|
| def main() -> int: |
| verified_by_hash: dict[str, Path] = {} |
| rejected: list[str] = [] |
| for candidate in discover_pdf_candidates(): |
| try: |
| pages = read_native_opening(candidate) |
| if is_kerala_sslc_physics(pages): |
| verified_by_hash.setdefault(sha256_file(candidate), candidate) |
| except Exception as exc: |
| rejected.append(f"{candidate}: {exc}") |
| verified = sorted(verified_by_hash.values()) |
| if not verified: |
| raise SystemExit("No verified Kerala SSLC Physics textbook PDFs were discovered.") |
|
|
| extractions: list[dict[str, Any]] = [] |
| reports: list[dict[str, Any]] = [] |
| changed = 0 |
| for path in verified: |
| extraction, report = ingest_textbook(path) |
| extractions.append(extraction) |
| reports.append(report) |
| cache_path = OUTPUT_ROOT / "textbooks" / f"{extraction['source']['sha256']}.json" |
| changed += int(write_json_if_changed(cache_path, extraction)) |
| changed += int(write_json_if_changed(OUTPUT_ROOT / "chapter-manifest.json", build_manifest(extractions))) |
| changed += int( |
| write_json_if_changed( |
| OUTPUT_ROOT / "ingestion-report.json", |
| { |
| "schemaVersion": 1, |
| "verifiedTextbooks": [str(path) for path in verified], |
| "rejectedCandidates": rejected, |
| "reports": reports, |
| }, |
| ) |
| ) |
| print( |
| json.dumps( |
| { |
| "verifiedTextbooks": len(verified), |
| "chapters": sum(len(item["chapters"]) for item in extractions), |
| "filesChanged": changed, |
| "output": str(OUTPUT_ROOT), |
| }, |
| indent=2, |
| ) |
| ) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
