Spaces:
Sleeping
Sleeping
| """ | |
| Automated Topic Extraction Engine. | |
| Foolproof two-stage architecture: | |
| 1. Font-metric & structure scanner: isolates true section headers from math/exercises. | |
| 2. LLM Consolidation & Normalization: turns raw headers into 4-8 clean curriculum topics. | |
| """ | |
| import re | |
| import json | |
| from typing import List, Dict, Any, Optional | |
| import fitz # PyMuPDF | |
| from backend.quiz_engine import quiz_engine | |
| EXCLUDE_KEYWORDS = [ | |
| "exercise", "exercises", "try these", "think, discuss", "what have we discussed", | |
| "what you have learnt", "solution", "example", "ncert", "republished", "chapter", | |
| "activity", "do this", "summary", "notes", "revision", "formulae", "question", | |
| "questions", "uestions", "additional exercise", "let's recall" | |
| ] | |
| def _is_noise(text: str) -> bool: | |
| """Disqualifies math equations, tables, and textbook exercise noise.""" | |
| t = text.strip() | |
| # Must have at least 3 letters (English or Devanagari Indic script) | |
| if len(re.findall(r"[a-zA-Z\u0900-\u097F]", t)) < 3: | |
| return True | |
| # Cannot contain math operators, equation symbols, or fill-in blanks | |
| if re.search(r"[=<>\+\*÷_]|(?:\s*=\s*)", t): | |
| return True | |
| lower = t.lower() | |
| if lower in ("uestions", "question", "questions", "additional"): | |
| return True | |
| if re.search(r"^activity\b", lower): | |
| return True | |
| # Check noise keywords | |
| hindi_noise = ["अभ्यास", "प्रश्नावली", "क्रियाकलाप", "उत्तरमाला", "हमने क्या सीखा", "सारांश", "प्रश्न", "प्रश्नोत्तरी"] | |
| if any(k in lower for k in EXCLUDE_KEYWORDS) or any(k in t for k in hindi_noise): | |
| return True | |
| return False | |
| def _is_chapter_banner(text: str, chap: str) -> bool: | |
| """Detects decorative multi-font title banners on opening pages.""" | |
| t = text.lower().strip() | |
| c = chap.lower().strip() | |
| if not t or not c: | |
| return False | |
| if t == c or t in c or c in t: | |
| return True | |
| words = [w for w in re.findall(r"[a-zA-Z\u0900-\u097F]+", t) if len(w) > 1] | |
| if not words: | |
| return True | |
| common_banner = {"chapter", "hapter", "unit", "part", "पाठ", "अध्याय"} | |
| matches = sum(1 for w in words if (w in c or w in common_banner)) | |
| return (matches / len(words)) >= 0.5 | |
| def _clean_title(text: str) -> str: | |
| """Strips section numbers like 2.3 or 2.3.1 and normalizes casing.""" | |
| t = text.strip() | |
| t = re.sub(r"^(\d+(\.\d+)*|[A-Z]\.|\b[IVXLCDM]+\.)\s*", "", t) | |
| t = re.sub(r"^(section|unit|topic|part)\s+[\dIVXLCDM]+[:\.\-]?\s*", "", t, flags=re.IGNORECASE) | |
| t = re.sub(r"\s+", " ", t).strip() | |
| if t.isupper() and len(t) > 3: | |
| t = t.title() | |
| return t.rstrip(":").rstrip("?").strip() | |
| _clean_topic_name = _clean_title | |
| FORBIDDEN_PROMPT_EXAMPLES = { | |
| "balanced and unbalanced forces", | |
| "first law of motion", | |
| "second law of motion", | |
| "third law of motion", | |
| "introduction", | |
| "core concepts", | |
| "core subject lesson", | |
| "topic name", | |
| "<topic name>", | |
| "topic title", | |
| "<topic title>", | |
| "कवि परिचय", | |
| "मुख्य पाठ" | |
| } | |
| def _is_prompt_leak(topic_name: str, doc_text: str) -> bool: | |
| """Returns True if the topic matches a known prompt format example but does not exist in the document.""" | |
| t_clean = topic_name.strip().lower() | |
| if t_clean in FORBIDDEN_PROMPT_EXAMPLES: | |
| words = [w for w in re.findall(r"[\w\u0900-\u097F]{3,}", t_clean) if w not in {"the", "and", "for", "with"}] | |
| matches = sum(1 for w in words if w in doc_text) | |
| if matches == 0: | |
| return True | |
| return False | |
| def _is_topic_grounded(topic_name: str, doc_text: str, headings: List[Dict[str, Any]]) -> bool: | |
| """Verifies that a topic has lexical grounding in the document text or candidate headings.""" | |
| t_clean = topic_name.strip().lower() | |
| if not t_clean or t_clean in {"<topic name>", "topic name", "<topic title>", "topic"}: | |
| return False | |
| # Check if exact topic or significant words appear in candidate headings | |
| for h in headings: | |
| h_clean = h.get("topic", "").strip().lower() | |
| if t_clean in h_clean or h_clean in t_clean: | |
| return True | |
| t_words = set(re.findall(r"[\w\u0900-\u097F]{3,}", t_clean)) | |
| h_words = set(re.findall(r"[\w\u0900-\u097F]{3,}", h_clean)) | |
| if t_words and (t_words & h_words): | |
| return True | |
| # Check if significant words appear in the document text | |
| words = [w for w in re.findall(r"[\w\u0900-\u097F]{3,}", t_clean) if w not in {"the", "and", "for", "with", "from", "part", "unit"}] | |
| if not words: | |
| return False | |
| matches = sum(1 for w in words if w in doc_text) | |
| return (matches / len(words)) >= 0.5 | |
| class TopicExtractor: | |
| """ | |
| Extracts authentic curriculum topics from any textbook PDF. | |
| """ | |
| def _scan_font_headings(doc: fitz.Document, chapter_name: str) -> List[Dict[str, Any]]: | |
| """ | |
| Scans PDF text spans using font-size metrics to isolate true headings. | |
| Works across all languages and layouts without relying on numbering schemes. | |
| """ | |
| # 1. Determine base body font size across document | |
| char_counts: Dict[float, int] = {} | |
| for page in doc: | |
| d = page.get_text("dict") | |
| for b in d.get("blocks", []): | |
| for line in b.get("lines", []): | |
| for s in line.get("spans", []): | |
| sz = round(s["size"], 1) | |
| char_counts[sz] = char_counts.get(sz, 0) + len(s["text"].strip()) | |
| if not char_counts: | |
| return [{"topic": chapter_name, "page": 1}] | |
| body_sz = max(char_counts.items(), key=lambda x: x[1])[0] | |
| from backend.ingest import detect_legacy_font, decode_legacy_text | |
| legacy_font = detect_legacy_font(doc) | |
| # 2. Extract heading-sized text lines grouped by block | |
| raw_candidates = [] | |
| for pno in range(len(doc)): | |
| page = doc[pno] | |
| d = page.get_text("dict") | |
| for b in d.get("blocks", []): | |
| curr_lines = [] | |
| max_sz_block = 0.0 | |
| bbox = b.get("bbox", (0, 0, 0, 0)) | |
| for line in b.get("lines", []): | |
| line_spans = [s for s in line.get("spans", []) if s["text"].strip()] | |
| if not line_spans: | |
| continue | |
| line_str = " ".join(s["text"].strip() for s in line_spans) | |
| if legacy_font: | |
| line_str = decode_legacy_text(line_str, legacy_font) | |
| max_sz = max(s["size"] for s in line_spans) | |
| # Universal font hierarchy: distinct header font (>= 1.15 body) or bold | |
| is_large = max_sz >= body_sz * 1.15 | |
| is_bold = any("bold" in s.get("font", "").lower() for s in line_spans) and (max_sz >= body_sz * 1.05) | |
| if is_large or is_bold: | |
| curr_lines.append(line_str) | |
| max_sz_block = max(max_sz_block, max_sz) | |
| else: | |
| if curr_lines: | |
| raw_candidates.append({ | |
| "page": pno + 1, | |
| "text": " ".join(curr_lines).strip(), | |
| "bbox": bbox, | |
| "size": max_sz_block | |
| }) | |
| curr_lines = [] | |
| max_sz_block = 0.0 | |
| if curr_lines: | |
| raw_candidates.append({ | |
| "page": pno + 1, | |
| "text": " ".join(curr_lines).strip(), | |
| "bbox": bbox, | |
| "size": max_sz_block | |
| }) | |
| # 3. Merge vertically adjacent heading blocks on the same page (multi-line titles) | |
| merged = [] | |
| for cand in raw_candidates: | |
| if not merged: | |
| merged.append(cand) | |
| continue | |
| prev = merged[-1] | |
| vert_gap = cand["bbox"][1] - prev["bbox"][3] | |
| if cand["page"] == prev["page"] and 0 <= vert_gap <= 20: | |
| prev["text"] = prev["text"] + " " + cand["text"] | |
| prev["bbox"] = (min(prev["bbox"][0], cand["bbox"][0]), prev["bbox"][1], max(prev["bbox"][2], cand["bbox"][2]), cand["bbox"][3]) | |
| else: | |
| merged.append(cand) | |
| # 4. Filter noise, opening page banners, and deduplicate | |
| headings = [] | |
| seen = set() | |
| for m in merged: | |
| txt = m["text"].strip() | |
| p = m["page"] | |
| if _is_noise(txt): | |
| continue | |
| if p <= 2 and _is_chapter_banner(txt, chapter_name): | |
| continue | |
| cleaned = _clean_title(txt) | |
| if cleaned and len(cleaned) > 3 and cleaned.lower() not in seen and cleaned.lower() != chapter_name.lower(): | |
| seen.add(cleaned.lower()) | |
| headings.append({ | |
| "topic": cleaned, | |
| "raw": txt, | |
| "page": p | |
| }) | |
| return headings | |
| def extract_topics_from_doc(cls, doc: fitz.Document, chapter_name: str) -> List[Dict[str, Any]]: | |
| """ | |
| Extracts and normalizes topics using font analysis + LLM synthesis. | |
| Strict grounding: rejects prompt example leaks and hallucinated topics. | |
| """ | |
| raw_headings = cls._scan_font_headings(doc, chapter_name) | |
| if not raw_headings: | |
| return [{"topic": chapter_name, "page": 1}] | |
| # Prepare document sample text for grounding verification | |
| from backend.ingest import detect_legacy_font, decode_legacy_text | |
| legacy_font = detect_legacy_font(doc) | |
| doc_sample_parts = [] | |
| for pno in range(min(len(doc), 10)): | |
| txt = doc[pno].get_text().strip().lower() | |
| if legacy_font: | |
| txt = decode_legacy_text(txt, legacy_font).lower() | |
| doc_sample_parts.append(txt) | |
| doc_sample_text = " ".join(doc_sample_parts) | |
| # If LLM client is available and we have candidate headings, extract curriculum topics | |
| if quiz_engine.client and len(raw_headings) >= 3: | |
| try: | |
| headers_list = [f"Page {h['page']}: {h['raw']}" for h in raw_headings] | |
| is_indic = bool(re.search(r"[\u0900-\u097F]", chapter_name + "".join([h["raw"] for h in raw_headings]))) | |
| lang_rule = ( | |
| "CRITICAL: The content is in Hindi. Return all topic titles in authentic Hindi (Devanagari script), " | |
| "e.g. 'पाठ का सारांश', 'मुख्य प्रसंग'. Do NOT translate into English." | |
| if is_indic | |
| else "Return clean, conceptual topic titles in English (e.g. 'Multiplication of Fractions')." | |
| ) | |
| prompt = ( | |
| f"You are a master textbook curriculum analyzer. Below are section headings extracted from a " | |
| f"textbook chapter titled '{chapter_name}'.\n" | |
| f"Extract all distinct, substantive study topics/sections covered in this chapter.\n\n" | |
| f"Extracted Headings:\n{json.dumps(headers_list, indent=2, ensure_ascii=False)}\n\n" | |
| "RULES:\n" | |
| f"1. {lang_rule}\n" | |
| "2. Preserve every distinct substantive lesson topic, law, concept, story section, or thematic unit. " | |
| "Do NOT merge distinct curriculum concepts into one generic heading unless they are near-identical duplicates.\n" | |
| "3. Remove all section numbering (no '9.1', '9.2', '9.4.1').\n" | |
| "4. Provide the starting page number for each topic based on the headings.\n" | |
| "5. Output typically 4 to 12 distinct topics reflecting the full coverage of the chapter.\n" | |
| "6. Output ONLY a valid JSON array of objects with keys 'topic' and 'page'. No markdown, no prose.\n" | |
| 'Format:\n[{"topic": "<Topic Name>", "page": 1}, {"topic": "<Topic Name>", "page": 3}]' | |
| ) | |
| resp = quiz_engine.client.chat_completion( | |
| messages=[{"role": "user", "content": prompt}], | |
| max_tokens=600, | |
| temperature=0.2 | |
| ) | |
| raw_text = resp.choices[0].message.content.strip() | |
| # Robust JSON array extraction | |
| cleaned = re.sub(r"^```(?:json)?", "", raw_text, flags=re.MULTILINE) | |
| cleaned = re.sub(r"```$", "", cleaned, flags=re.MULTILINE).strip() | |
| m = re.search(r"\[.*\]", cleaned, re.DOTALL) | |
| if m: | |
| parsed = json.loads(m.group(0)) | |
| if isinstance(parsed, list) and len(parsed) >= 2: | |
| # 1. Anti-Leak Guard: Check if the model echoed prompt format examples | |
| leaks = [ | |
| item.get("topic", "").strip() for item in parsed | |
| if isinstance(item, dict) and _is_prompt_leak(item.get("topic", ""), doc_sample_text) | |
| ] | |
| if leaks: | |
| raise ValueError( | |
| f"Topic extraction failed for '{chapter_name}': The AI model echoed prompt example topics " | |
| f"({', '.join(leaks[:2])}) that do not exist in the uploaded PDF. " | |
| "The document content could not be verified for authentic curriculum topics." | |
| ) | |
| # 2. Grounding Filter: Retain only topics verified in document content | |
| grounded = [ | |
| {"topic": item["topic"].strip(), "page": int(item.get("page", 1))} | |
| for item in parsed | |
| if isinstance(item, dict) and "topic" in item | |
| and _is_topic_grounded(item["topic"].strip(), doc_sample_text, raw_headings) | |
| ] | |
| if len(grounded) >= 2: | |
| return grounded | |
| except ValueError: | |
| raise | |
| except Exception: | |
| pass # Gracefully fall back to font-scanned headings | |
| # Fallback if font scanning yielded fewer than 3 headings (e.g. flat styling or scanned PDF) | |
| if len(raw_headings) < 3 and quiz_engine.client and len(doc) > 0: | |
| try: | |
| sample_texts = [] | |
| step = max(1, len(doc) // 6) | |
| for pno in range(0, len(doc), step): | |
| txt = doc[pno].get_text().strip()[:400] | |
| if legacy_font: | |
| txt = decode_legacy_text(txt, legacy_font) | |
| if txt: | |
| sample_texts.append(f"Page {pno + 1}:\n{txt}") | |
| if sample_texts: | |
| prompt = ( | |
| f"You are a school curriculum expert. Below are text samples from different pages of a " | |
| f"textbook chapter titled '{chapter_name}'.\n" | |
| f"Identify all primary study topics covered across this chapter.\n\n" | |
| f"{json.dumps(sample_texts, indent=2, ensure_ascii=False)}\n\n" | |
| "Output ONLY a valid JSON array of objects with keys 'topic' and 'page'." | |
| ) | |
| resp = quiz_engine.client.chat_completion( | |
| messages=[{"role": "user", "content": prompt}], | |
| max_tokens=500, | |
| temperature=0.2 | |
| ) | |
| raw_text = resp.choices[0].message.content.strip() | |
| cleaned = re.sub(r"^```(?:json)?", "", raw_text, flags=re.MULTILINE) | |
| cleaned = re.sub(r"```$", "", cleaned, flags=re.MULTILINE).strip() | |
| m = re.search(r"\[.*\]", cleaned, re.DOTALL) | |
| if m: | |
| parsed = json.loads(m.group(0)) | |
| if isinstance(parsed, list) and len(parsed) >= 2: | |
| leaks = [ | |
| item.get("topic", "").strip() for item in parsed | |
| if isinstance(item, dict) and _is_prompt_leak(item.get("topic", ""), doc_sample_text) | |
| ] | |
| if leaks: | |
| raise ValueError( | |
| f"Topic extraction failed for '{chapter_name}': The AI model echoed prompt example topics " | |
| f"({', '.join(leaks[:2])}) that do not exist in the uploaded PDF." | |
| ) | |
| grounded = [ | |
| {"topic": item["topic"].strip(), "page": int(item.get("page", 1))} | |
| for item in parsed | |
| if isinstance(item, dict) and "topic" in item | |
| and _is_topic_grounded(item["topic"].strip(), doc_sample_text, raw_headings) | |
| ] | |
| if len(grounded) >= 2: | |
| return grounded | |
| except ValueError: | |
| raise | |
| except Exception: | |
| pass | |
| # Fallback: Deduplicate verified font-scanned headings | |
| grounded_raw = [ | |
| h for h in raw_headings | |
| if _is_topic_grounded(h["topic"], doc_sample_text, raw_headings) | |
| ] | |
| if len(grounded_raw) >= 2: | |
| return grounded_raw | |
| # If even raw headings cannot be verified against the document content | |
| raise ValueError( | |
| f"Topic extraction failed for '{chapter_name}': Could not verify authentic curriculum topics " | |
| "from the uploaded document. Please ensure the PDF is clear and readable." | |
| ) | |
| def assign_topic_to_chunks(chunks: List[Any], topics: List[Dict[str, Any]], chapter_name: str) -> None: | |
| """ | |
| Assigns the appropriate subtopic to each chunk based on page boundaries. | |
| """ | |
| if not topics: | |
| for chunk in chunks: | |
| chunk.metadata["topic"] = chapter_name | |
| return | |
| sorted_topics = sorted(topics, key=lambda x: x["page"]) | |
| for chunk in chunks: | |
| chunk_page = chunk.metadata.get("page", 1) | |
| assigned_topic = sorted_topics[0]["topic"] | |
| for t in sorted_topics: | |
| if t["page"] <= chunk_page: | |
| assigned_topic = t["topic"] | |
| else: | |
| break | |
| chunk.metadata["topic"] = assigned_topic | |
| topic_extractor = TopicExtractor() | |