Spaces:
Sleeping
Sleeping
| # import gradio as gr | |
| # import torch | |
| # import re | |
| # import os | |
| # import pypdf | |
| # import fitz # PyMuPDF | |
| # import cv2 | |
| # import numpy as np | |
| # import base64 | |
| # import json | |
| # import time | |
| # import nest_asyncio | |
| # import spacy | |
| # from transformers import ( | |
| # T5Tokenizer, | |
| # T5ForConditionalGeneration, | |
| # pipeline, | |
| # AutoModelForQuestionAnswering, | |
| # AutoTokenizer | |
| # ) | |
| # from transformers.pipelines import QuestionAnsweringPipeline | |
| # from peft import PeftModel | |
| # from sentence_transformers import CrossEncoder | |
| # from ultralytics import YOLO | |
| # # ========================================== | |
| # # 0. SPACY & VISION SETUP | |
| # # ========================================== | |
| # print("🧠 Booting Syntactic Parser...") | |
| # try: | |
| # nlp = spacy.load("en_core_web_sm") | |
| # except OSError: | |
| # os.system("python -m spacy download en_core_web_sm") | |
| # nlp = spacy.load("en_core_web_sm") | |
| # # ========================================== | |
| # # 1. HARDWARE & MODEL SETUP | |
| # # ========================================== | |
| # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # # 🔥 FIXED: Defining the missing torch_dtype | |
| # torch_dtype = torch.float16 if device == "cuda" else torch.float32 | |
| # print(f"🚀 Booting Edugenius Multimodal on: {device.upper()}") | |
| # # --- Text Models --- | |
| # model_id = "t5-base" | |
| # adapter_path = "." | |
| # print("⏳ Loading T5 Generator...") | |
| # base_model = T5ForConditionalGeneration.from_pretrained(model_id, torch_dtype=torch_dtype) | |
| # model = PeftModel.from_pretrained(base_model, adapter_path) | |
| # # tokenizer = T5Tokenizer.from_pretrained(adapter_path) | |
| # tokenizer = T5Tokenizer.from_pretrained("t5-base") | |
| # model.to(device).eval() | |
| # # --- Validation Models --- | |
| # print("⏳ Loading Validation Guardrails...") | |
| # validator = CrossEncoder('cross-encoder/ms-marco-TinyBERT-L-2-v2', device=device) | |
| # # qa_validator = pipeline("question-answering", model="deepset/minilm-uncased-squad2", device=0 if device == "cuda" else -1) | |
| # # qa_validator = pipeline("question-answering", model="deepset/minilm-uncased-squad2", device=0 if device == "cuda" else -1, trust_remote_code=True) | |
| # # --- FIXED: Explicit Round-Trip QA Construction --- | |
| # qa_model = AutoModelForQuestionAnswering.from_pretrained("deepset/minilm-uncased-squad2") | |
| # qa_tokenizer = AutoTokenizer.from_pretrained("deepset/minilm-uncased-squad2") | |
| # # qa_validator = QuestionAnsweringPipeline(model=qa_model, tokenizer=qa_tokenizer, device=0 if device == "cuda" else -1) | |
| # # --- FIXED: The Modern, Cross-Platform Way to Load QA --- | |
| # print("🕵️ Booting Round-Trip QA Validator...") | |
| # qa_validator = pipeline( | |
| # "question-answering", | |
| # model=qa_model, | |
| # tokenizer=qa_tokenizer, | |
| # device=0 if device == "cuda" else -1 | |
| # ) | |
| # # --- Vision Model (YOLO) --- | |
| # print("⏳ Loading YOLO Figure Detector...") | |
| # YOLO_WEIGHTS = 'best.pt' | |
| # vision_model = YOLO(YOLO_WEIGHTS) | |
| # print("✅ ALL MODELS LOADED AND READY.") | |
| # # ========================================== | |
| # # 2. UTILITY FUNCTIONS | |
| # # ========================================== | |
| # def get_paragraph_subject(paragraph_text): | |
| # doc = nlp(paragraph_text[:200]) | |
| # for chunk in doc.noun_chunks: | |
| # if chunk.root.dep_ in ('nsubj', 'nsubjpass') and chunk.root.pos_ in ('PROPN', 'NOUN'): | |
| # return chunk.text | |
| # return "The subject" | |
| # def chunk_text_by_sentence(text): | |
| # raw_sentences = text.replace('\n', ' ').split('.') | |
| # sentences = [s.strip() + '.' for s in raw_sentences if len(s.strip()) > 15] | |
| # chunks = [] | |
| # for i in range(len(sentences)): | |
| # if i == 0: | |
| # chunks.append(sentences[i]) | |
| # else: | |
| # chunks.append(sentences[i-1] + " " + sentences[i]) | |
| # return chunks | |
| # def process_pdf_text_clean(pdf_path): | |
| # """Extracts and cleans PDF text, removing page numbers and fixing wraps.""" | |
| # doc = fitz.open(pdf_path) | |
| # full_text = "" | |
| # for page in doc: | |
| # text = page.get_text("text") | |
| # if not text: continue | |
| # lines = text.split('\n') | |
| # cleaned_lines = [] | |
| # for line in lines: | |
| # line = line.strip() | |
| # if not line or line.isdigit(): continue | |
| # cleaned_lines.append(line) | |
| # page_text = "" | |
| # for line in cleaned_lines: | |
| # page_text += line | |
| # if not re.search(r'[.!?:]$', line): | |
| # page_text += " " | |
| # else: | |
| # page_text += "\n\n" | |
| # full_text += page_text + "\n\n" | |
| # doc.close() | |
| # return full_text | |
| # def detect_figures(pdf_path): | |
| # """Extracts only 'figure' class objects as Base64 strings.""" | |
| # doc = fitz.open(pdf_path) | |
| # visuals = [] | |
| # fig_idx = 1 | |
| # mat = fitz.Matrix(2.0, 2.0) | |
| # for page_num, page in enumerate(doc): | |
| # pix = page.get_pixmap(matrix=mat) | |
| # img = np.frombuffer(pix.samples, dtype=np.uint8).reshape((pix.h, pix.w, pix.n)) | |
| # img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| # results = vision_model.predict(img, conf=0.3, verbose=False) | |
| # for box in results[0].boxes: | |
| # cls = int(box.cls[0]) | |
| # if vision_model.names[cls] == 'figure': | |
| # x1, y1, x2, y2 = map(int, box.xyxy[0]) | |
| # crop = img[y1:y2, x1:x2] | |
| # _, buffer = cv2.imencode('.png', crop) | |
| # b64 = base64.b64encode(buffer).decode('utf-8') | |
| # visuals.append({ | |
| # "id": f"FIG_{fig_idx}", | |
| # "page": page_num + 1, | |
| # "base64": b64 | |
| # }) | |
| # fig_idx += 1 | |
| # doc.close() | |
| # return visuals | |
| # # ========================================== | |
| # # 3. CORE EXTRACTION ENGINE | |
| # # ========================================== | |
| # def extract_flashcards(text): | |
| # if not text or len(text.strip()) < 10: | |
| # yield "Please provide more text.", [] | |
| # return | |
| # accepted_cards = [] | |
| # accepted_markdown = "### ✅ Accepted Flashcards\n\n" | |
| # rejected_markdown = "### 🛑 Blocked by Guardrails (Debug)\n\n" | |
| # seen_answers = set() | |
| # paragraphs = text.replace('\r', '').split('\n\n') | |
| # for paragraph in paragraphs: | |
| # if len(paragraph.strip()) < 20: continue | |
| # current_subject = get_paragraph_subject(paragraph) | |
| # chunks = chunk_text_by_sentence(paragraph) | |
| # for chunk in chunks: | |
| # grounded_chunk = f"{current_subject}: {chunk}" | |
| # prompt = f"task: generate Understand flashcard context: {grounded_chunk}" | |
| # inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(device) | |
| # with torch.no_grad(): | |
| # outputs = model.generate(**inputs, max_new_tokens=128, num_beams=4, early_stopping=True) | |
| # raw_output = tokenizer.decode(outputs[0], skip_special_tokens=True) | |
| # if "QS:" in raw_output and "| AN:" in raw_output: | |
| # try: | |
| # parts = raw_output.split("| AN:") | |
| # q = parts[0].replace("QS:", "").strip() | |
| # a = parts[1].strip() | |
| # if a.lower() in seen_answers or len(q) < 15 or len(a.split()) > 12: continue | |
| # seen_answers.add(a.lower()) | |
| # # --- NER & Syntax Check --- | |
| # ans_doc = nlp(a) | |
| # q_lower = q.lower() | |
| # first_word = q_lower.split()[0] if q_lower.split() else "" | |
| # ents = [ent.label_ for ent in ans_doc.ents] | |
| # blocked_reason = None | |
| # if first_word == "when" and not any(e in ents for e in ["DATE", "TIME", "CARDINAL"]): | |
| # blocked_reason = "NER: Expected Date" | |
| # elif first_word == "who" and not any(e in ents for e in ["PERSON", "ORG", "NORP"]): | |
| # blocked_reason = "NER: Expected Person" | |
| # elif first_word == "where" and not any(e in ents for e in ["GPE", "LOC"]): | |
| # blocked_reason = "NER: Expected Location" | |
| # if blocked_reason: | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: {blocked_reason})*\n\n---\n\n" | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # continue | |
| # # --- Semantic Check --- | |
| # score = validator.predict([[q, grounded_chunk]])[0] | |
| # if score < -4.0: | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: CE Score {score:.2f})*\n\n---\n\n" | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # continue | |
| # # --- Round-Trip QA Check --- | |
| # qa_res = qa_validator(question=q, context=grounded_chunk) | |
| # qa_a = qa_res['answer'].lower() | |
| # overlap = 1.0 if (a.lower() in qa_a or qa_a in a.lower()) else 0.0 | |
| # telemetry = f"*(**CE Score:** {score:.2f} | **QA Match:** {overlap:.0%} | **Subject:** \"{current_subject}\" | **Source:** \"{chunk.strip()}\")*" | |
| # if overlap < 0.5: | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: QA Mismatch. Predicted: {qa_a})*\n{telemetry}\n\n---\n\n" | |
| # else: | |
| # accepted_markdown += f"**Q:** {q}\n**A:** {a}\n{telemetry}\n\n---\n\n" | |
| # accepted_cards.append({"question": q, "answer": a, "subject": current_subject}) | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # except Exception: continue | |
| # # ========================================== | |
| # # 4. FINAL GRADIO INTERFACE | |
| # # ========================================== | |
| # def run_pipeline(pdf_file): | |
| # if pdf_file is None: return "Upload a PDF.", None | |
| # # 1. Detect Visuals (Figures Only) | |
| # yield "🔍 Step 1/3: Scanning for Figures...", None | |
| # visuals = detect_figures(pdf_file.name) | |
| # # 2. Extract Text | |
| # yield "📝 Step 2/3: Cleaning Text and Analyzing Subject...", None | |
| # clean_text = process_pdf_text_clean(pdf_file.name) | |
| # # 3. Extract Flashcards | |
| # final_cards = [] | |
| # md = "" | |
| # for md_update, card_list in extract_flashcards(clean_text): | |
| # md = md_update | |
| # final_cards = card_list | |
| # yield md, None | |
| # # 4. 🔥 REFORMAT FOR BACKEND SCHEMA WITH UUID | |
| # import uuid | |
| # backend_mcqs = [] | |
| # # Add Figures as Flashcards | |
| # for i, fig in enumerate(visuals): | |
| # backend_mcqs.append({ | |
| # "question": "Figure", | |
| # "answer": f"Visual from page {fig['page']}", | |
| # "question_type": "FLASHCARD", | |
| # "figure1": fig['base64'], | |
| # "documentIndex": i + 1, | |
| # "questionId": str(uuid.uuid4()), # 🔥 Secure Unique ID | |
| # "options": [], | |
| # "predicted_subject": None, | |
| # "predicted_concept": None | |
| # }) | |
| # # Add Text Flashcards | |
| # start_idx = len(backend_mcqs) + 1 | |
| # for i, card in enumerate(final_cards): | |
| # backend_mcqs.append({ | |
| # "question": card["question"], | |
| # "answer": card["answer"], | |
| # "question_type": "FLASHCARD", | |
| # "figure1": None, | |
| # "documentIndex": start_idx + i, | |
| # "questionId": str(uuid.uuid4()), # 🔥 Secure Unique ID | |
| # "options": [], | |
| # "predicted_subject": {"label": card["subject"], "confidence": 1.0}, | |
| # "predicted_concept": None | |
| # }) | |
| # # Wrap in the final structure | |
| # generated_id = str(uuid.uuid4()) | |
| # data_pack = [{ | |
| # "document": "Edugenius Generated", | |
| # "id": generated_id, | |
| # "metadata": { | |
| # "answerFound": True, | |
| # "card_count": len(backend_mcqs), | |
| # "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| # "file_name": os.path.basename(pdf_file.name), | |
| # "generatedQAId": generated_id, | |
| # "mcqs": backend_mcqs | |
| # } | |
| # }] | |
| # output_path = "edugenius_pack.json" | |
| # with open(output_path, "w") as f: | |
| # json.dump(data_pack, f, indent=2) | |
| # yield md + "\n\n✅ **Processing Complete! Backend-compatible JSON Ready.**", output_path | |
| # with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| # gr.Markdown("# 🚀 Edugenius Multimodal Study-Pack Generator") | |
| # with gr.Row(): | |
| # file_in = gr.File(label="Upload Textbook (PDF)") | |
| # btn = gr.Button("Generate Study Pack", variant="primary") | |
| # with gr.Row(): | |
| # md_out = gr.Markdown(label="Real-time Extraction") | |
| # json_out = gr.File(label="Download Full Data (JSON)") | |
| # btn.click(run_pipeline, inputs=file_in, outputs=[md_out, json_out]) | |
| # demo.launch(share=True) | |
| # import gradio as gr | |
| # import torch | |
| # import re | |
| # import os | |
| # import fitz # PyMuPDF | |
| # import cv2 | |
| # import numpy as np | |
| # import base64 | |
| # import json | |
| # import time | |
| # import uuid | |
| # import spacy | |
| # from transformers import ( | |
| # T5Tokenizer, | |
| # T5ForConditionalGeneration, | |
| # ) | |
| # from peft import PeftModel | |
| # from sentence_transformers import CrossEncoder | |
| # from ultralytics import YOLO | |
| # # ========================================== | |
| # # 0. SPACY SETUP | |
| # # ========================================== | |
| # print("🧠 Booting Syntactic Parser...") | |
| # try: | |
| # nlp = spacy.load("en_core_web_sm") | |
| # except OSError: | |
| # os.system("python -m spacy download en_core_web_sm") | |
| # nlp = spacy.load("en_core_web_sm") | |
| # # # ========================================== | |
| # # # 1. HARDWARE & MODEL SETUP | |
| # # # ========================================== | |
| # # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # # torch_dtype = torch.float16 if device == "cuda" else torch.float32 | |
| # # print(f"🚀 Booting Edugenius Multimodal on: {device.upper()}") | |
| # # # --- Text Models --- | |
| # # model_id = "t5-base" | |
| # # adapter_path = "." | |
| # # print("⏳ Loading T5 Generator...") | |
| # # base_model = T5ForConditionalGeneration.from_pretrained(model_id, torch_dtype=torch_dtype) | |
| # # try: | |
| # # model = PeftModel.from_pretrained(base_model, adapter_path) | |
| # # except Exception as e: | |
| # # print(f"⚠️ Could not load PeftModel adapter from {adapter_path}. Using base model. Error: {e}") | |
| # # model = base_model | |
| # # tokenizer = T5Tokenizer.from_pretrained("t5-base") | |
| # # model.to(device).eval() | |
| # import multiprocessing | |
| # # ========================================== | |
| # # 1. HARDWARE & MODEL SETUP | |
| # # ========================================== | |
| # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # torch_dtype = torch.float16 if device == "cuda" else torch.float32 | |
| # # 🔥 FIX 1: WAKE UP THE CPU | |
| # # Force PyTorch to use ALL available CPU cores instead of defaulting to 1 | |
| # num_cores = multiprocessing.cpu_count() | |
| # torch.set_num_threads(num_cores) | |
| # os.environ["OMP_NUM_THREADS"] = str(num_cores) | |
| # print(f"🚀 Booting Edugenius Multimodal on: {device.upper()} with {num_cores} Threads") | |
| # # --- Text Models --- | |
| # model_id = "t5-base" | |
| # adapter_path = "." | |
| # print("⏳ Loading T5 Generator...") | |
| # base_model = T5ForConditionalGeneration.from_pretrained(model_id, torch_dtype=torch_dtype) | |
| # try: | |
| # print("⏳ Loading and Merging PEFT Adapter...") | |
| # peft_model = PeftModel.from_pretrained(base_model, adapter_path) | |
| # # 🔥 FIX 2: MERGE AND UNLOAD | |
| # # This bakes the LoRA weights into the base model. | |
| # # It removes the massive CPU calculation overhead during generation. | |
| # model = peft_model.merge_and_unload() | |
| # print("✅ Adapter merged successfully!") | |
| # except Exception as e: | |
| # print(f"⚠️ Could not load or merge PeftModel. Using base model. Error: {e}") | |
| # model = base_model | |
| # tokenizer = T5Tokenizer.from_pretrained("t5-base") | |
| # model.to(device).eval() | |
| # # --- Validation Models --- | |
| # print("⏳ Loading Validation Guardrails (CrossEncoder Only)...") | |
| # validator = CrossEncoder('cross-encoder/ms-marco-TinyBERT-L-2-v2', device=device) | |
| # # --- Vision Model (YOLO) --- | |
| # print("⏳ Loading YOLO Figure Detector...") | |
| # YOLO_WEIGHTS = 'best.pt' | |
| # vision_model = YOLO(YOLO_WEIGHTS) | |
| # print("✅ ALL MODELS LOADED AND READY.") | |
| # # ========================================== | |
| # # 2. UTILITY FUNCTIONS | |
| # # ========================================== | |
| # def get_paragraph_subject(paragraph_text): | |
| # doc = nlp(paragraph_text[:200]) | |
| # for chunk in doc.noun_chunks: | |
| # if chunk.root.dep_ in ('nsubj', 'nsubjpass') and chunk.root.pos_ in ('PROPN', 'NOUN'): | |
| # return chunk.text | |
| # return "The subject" | |
| # # def chunk_text_by_sentence(text): | |
| # # raw_sentences = text.replace('\n', ' ').split('.') | |
| # # sentences = [s.strip() + '.' for s in raw_sentences if len(s.strip()) > 15] | |
| # # chunks = [] | |
| # # for i in range(len(sentences)): | |
| # # if i == 0: | |
| # # chunks.append(sentences[i]) | |
| # # else: | |
| # # chunks.append(sentences[i-1] + " " + sentences[i]) | |
| # # return chunks | |
| # # def chunk_text_by_sentence(text): | |
| # # doc = nlp(text) | |
| # # # Only keep actual sentences (more than 5 words) | |
| # # sentences = [sent.text.strip() for sent in doc.sents if len(sent.text.split()) > 5] | |
| # # chunks = [] | |
| # # # Group 2 sentences together for better context | |
| # # for i in range(0, len(sentences), 2): | |
| # # chunk = " ".join(sentences[i:i+2]) | |
| # # chunks.append(chunk) | |
| # # return chunks | |
| # def chunk_text_by_sentence(text): | |
| # doc = nlp(text) | |
| # # Only keep actual sentences (more than 5 words) | |
| # sentences = [sent.text.strip() for sent in doc.sents if len(sent.text.split()) > 5] | |
| # chunks = [] | |
| # # SLIDING WINDOW: Overlap sentences to force the model to generate more questions | |
| # # from the same paragraph without using slow beam search. | |
| # for i in range(len(sentences)): | |
| # if i == 0: | |
| # # Chunk 1: Sentences 0 and 1 | |
| # chunk = " ".join(sentences[0:2]) | |
| # else: | |
| # # Chunk 2+: Sentences i-1 and i (This overlaps!) | |
| # chunk = " ".join(sentences[i-1:i+1]) | |
| # if chunk not in chunks: | |
| # chunks.append(chunk) | |
| # # Also add single sentences as their own chunks to guarantee granular detail extraction | |
| # for sent in sentences: | |
| # if sent not in chunks: | |
| # chunks.append(sent) | |
| # return chunks | |
| # # def process_pdf_text_clean(pdf_path): | |
| # # print(f"\n[DEBUG] Starting text extraction for {pdf_path}") | |
| # # doc = fitz.open(pdf_path) | |
| # # full_text = "" | |
| # # for page in doc: | |
| # # text = page.get_text("text") | |
| # # if not text: continue | |
| # # lines = text.split('\n') | |
| # # cleaned_lines = [] | |
| # # for line in lines: | |
| # # line = line.strip() | |
| # # if not line or line.isdigit(): continue | |
| # # cleaned_lines.append(line) | |
| # # page_text = "" | |
| # # for line in cleaned_lines: | |
| # # page_text += line | |
| # # if not re.search(r'[.!?:]$', line): | |
| # # page_text += " " | |
| # # else: | |
| # # page_text += "\n\n" | |
| # # full_text += page_text + "\n\n" | |
| # # doc.close() | |
| # # print(f"[DEBUG] Text extraction complete. Total length: {len(full_text)} characters.") | |
| # # return full_text | |
| # def process_pdf_text_clean(pdf_path): | |
| # print(f"\n[DEBUG] Starting clean text extraction for {pdf_path}") | |
| # doc = fitz.open(pdf_path) | |
| # full_text = "" | |
| # for page in doc: | |
| # text = page.get_text("text") | |
| # if not text: continue | |
| # lines = text.split('\n') | |
| # for line in lines: | |
| # line = line.strip() | |
| # # 🔥 THE FIX: Skip numbers, diagram labels (under 4 words), and figure text | |
| # word_count = len(line.split()) | |
| # if word_count < 4 or line.isdigit() or line.lower().startswith("figure") or "Reprint" in line: | |
| # continue | |
| # full_text += line + " " | |
| # doc.close() | |
| # # Clean up extra spaces | |
| # clean_result = re.sub(r'\s+', ' ', full_text).strip() | |
| # print(f"[DEBUG] Text cleaned. Length: {len(clean_result)} chars.") | |
| # return clean_result | |
| # def detect_figures(pdf_path): | |
| # print(f"\n[DEBUG] Starting YOLO figure detection for {pdf_path}") | |
| # doc = fitz.open(pdf_path) | |
| # visuals = [] | |
| # fig_idx = 1 | |
| # mat = fitz.Matrix(1.0, 1.0) | |
| # for page_num, page in enumerate(doc): | |
| # pix = page.get_pixmap(matrix=mat) | |
| # img = np.frombuffer(pix.samples, dtype=np.uint8).reshape((pix.h, pix.w, pix.n)) | |
| # img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| # results = vision_model.predict(img, conf=0.3, verbose=False) | |
| # for box in results[0].boxes: | |
| # cls = int(box.cls[0]) | |
| # if vision_model.names[cls] == 'figure': | |
| # x1, y1, x2, y2 = map(int, box.xyxy[0]) | |
| # crop = img[y1:y2, x1:x2] | |
| # _, buffer = cv2.imencode('.png', crop) | |
| # b64 = base64.b64encode(buffer).decode('utf-8') | |
| # visuals.append({ | |
| # "id": f"FIG_{fig_idx}", | |
| # "page": page_num + 1, | |
| # "base64": b64 | |
| # }) | |
| # fig_idx += 1 | |
| # print(f"[DEBUG] Found figure on page {page_num + 1}") | |
| # doc.close() | |
| # print(f"[DEBUG] Figure detection complete. Total figures: {len(visuals)}") | |
| # return visuals | |
| # # ========================================== | |
| # # 3. CORE EXTRACTION ENGINE | |
| # # ========================================== | |
| # def extract_flashcards(text): | |
| # if not text or len(text.strip()) < 10: | |
| # yield "Please provide more text.", [] | |
| # return | |
| # accepted_cards = [] | |
| # accepted_markdown = "### ✅ Accepted Flashcards\n\n" | |
| # rejected_markdown = "### 🛑 Blocked by Guardrails (Debug)\n\n" | |
| # seen_answers = set() | |
| # paragraphs = text.replace('\r', '').split('\n\n') | |
| # print(f"\n[DEBUG] Starting extraction loop on {len(paragraphs)} paragraphs.") | |
| # for p_idx, paragraph in enumerate(paragraphs): | |
| # if len(paragraph.strip()) < 20: continue | |
| # current_subject = get_paragraph_subject(paragraph) | |
| # chunks = chunk_text_by_sentence(paragraph) | |
| # for c_idx, chunk in enumerate(chunks): | |
| # grounded_chunk = f"{current_subject}: {chunk}" | |
| # print(f"\n[DEBUG] --- Processing P{p_idx}-C{c_idx} ---") | |
| # print(f"[DEBUG] Chunk context: {grounded_chunk[:100]}...") | |
| # prompt = f"task: generate Understand flashcard context: {grounded_chunk}" | |
| # inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(device) | |
| # with torch.no_grad(): | |
| # outputs = model.generate(**inputs, max_new_tokens=128, num_beams=1) | |
| # raw_output = tokenizer.decode(outputs[0], skip_special_tokens=True) | |
| # print(f"[GEN] Raw T5 Output: {raw_output}") | |
| # if "QS:" in raw_output and "| AN:" in raw_output: | |
| # try: | |
| # parts = raw_output.split("| AN:") | |
| # q = parts[0].replace("QS:", "").strip() | |
| # a = parts[1].strip() | |
| # print(f"[GEN] Parsed Q: {q}") | |
| # print(f"[GEN] Parsed A: {a}") | |
| # if a.lower() in seen_answers or len(q) < 15 or len(a.split()) > 12: | |
| # print("[REJECT] Blocked by basic length or duplicate checks.") | |
| # continue | |
| # seen_answers.add(a.lower()) | |
| # # --- NER & Syntax Check --- | |
| # ans_doc = nlp(a) | |
| # q_lower = q.lower() | |
| # first_word = q_lower.split()[0] if q_lower.split() else "" | |
| # ents = [ent.label_ for ent in ans_doc.ents] | |
| # blocked_reason = None | |
| # if first_word == "when" and not any(e in ents for e in ["DATE", "TIME", "CARDINAL"]): | |
| # blocked_reason = "NER: Expected Date" | |
| # elif first_word == "who" and not any(e in ents for e in ["PERSON", "ORG", "NORP"]): | |
| # blocked_reason = "NER: Expected Person" | |
| # elif first_word == "where" and not any(e in ents for e in ["GPE", "LOC"]): | |
| # blocked_reason = "NER: Expected Location" | |
| # if blocked_reason: | |
| # print(f"[REJECT] {blocked_reason}") | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: {blocked_reason})*\n\n---\n\n" | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # continue | |
| # # --- Semantic Check --- | |
| # score = validator.predict([[q, grounded_chunk]])[0] | |
| # if score < -4.0: | |
| # print(f"[REJECT] Semantic score too low: {score:.2f}") | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: CE Score {score:.2f})*\n\n---\n\n" | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # continue | |
| # print(f"[ACCEPT] Semantic score passed: {score:.2f}") | |
| # telemetry = f"*(**CE Score:** {score:.2f} | **Subject:** \"{current_subject}\" | **Source:** \"{chunk.strip()}\")*" | |
| # accepted_markdown += f"**Q:** {q}\n**A:** {a}\n{telemetry}\n\n---\n\n" | |
| # accepted_cards.append({"question": q, "answer": a, "subject": current_subject}) | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # except Exception as e: | |
| # print(f"[ERROR] Exception during output parsing: {e}") | |
| # continue | |
| # # ========================================== | |
| # # 4. FINAL GRADIO INTERFACE | |
| # # ========================================== | |
| # def run_pipeline(pdf_file): | |
| # if pdf_file is None: | |
| # print("[ERROR] No PDF file uploaded.") | |
| # return "Upload a PDF.", None | |
| # print(f"\n[START] Processing started for {pdf_file.name}") | |
| # # 1. Detect Visuals | |
| # yield "🔍 Step 1/3: Scanning for Figures...", None | |
| # visuals = detect_figures(pdf_file.name) | |
| # # 2. Extract Text | |
| # yield "📝 Step 2/3: Cleaning Text and Analyzing Subject...", None | |
| # clean_text = process_pdf_text_clean(pdf_file.name) | |
| # # 3. Extract Flashcards | |
| # final_cards = [] | |
| # md = "" | |
| # for md_update, card_list in extract_flashcards(clean_text): | |
| # md = md_update | |
| # final_cards = card_list | |
| # yield md, None | |
| # # 4. REFORMAT FOR BACKEND SCHEMA WITH UUID | |
| # print("\n[DEBUG] Compiling final backend JSON package...") | |
| # backend_mcqs = [] | |
| # for i, fig in enumerate(visuals): | |
| # backend_mcqs.append({ | |
| # "question": "Figure", | |
| # "answer": f"Visual from page {fig['page']}", | |
| # "question_type": "FLASHCARD", | |
| # "figure1": fig['base64'], | |
| # "documentIndex": i + 1, | |
| # "questionId": str(uuid.uuid4()), | |
| # "options": [], | |
| # "predicted_subject": None, | |
| # "predicted_concept": None | |
| # }) | |
| # start_idx = len(backend_mcqs) + 1 | |
| # for i, card in enumerate(final_cards): | |
| # backend_mcqs.append({ | |
| # "question": card["question"], | |
| # "answer": card["answer"], | |
| # "question_type": "FLASHCARD", | |
| # "figure1": None, | |
| # "documentIndex": start_idx + i, | |
| # "questionId": str(uuid.uuid4()), | |
| # "options": [], | |
| # "predicted_subject": {"label": card["subject"], "confidence": 1.0}, | |
| # "predicted_concept": None | |
| # }) | |
| # generated_id = str(uuid.uuid4()) | |
| # data_pack = [{ | |
| # "document": "Edugenius Generated", | |
| # "id": generated_id, | |
| # "metadata": { | |
| # "answerFound": True, | |
| # "card_count": len(backend_mcqs), | |
| # "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| # "file_name": os.path.basename(pdf_file.name), | |
| # "generatedQAId": generated_id, | |
| # "mcqs": backend_mcqs | |
| # } | |
| # }] | |
| # output_path = "edugenius_pack.json" | |
| # with open(output_path, "w") as f: | |
| # json.dump(data_pack, f, indent=2) | |
| # print(f"[SUCCESS] Pipeline complete. Generated {len(backend_mcqs)} total items.") | |
| # yield md + "\n\n✅ **Processing Complete! Backend-compatible JSON Ready.**", output_path | |
| # with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| # gr.Markdown("# 🚀 Edugenius Multimodal Study-Pack Generator (CPU Optimized)") | |
| # with gr.Row(): | |
| # file_in = gr.File(label="Upload Textbook (PDF)") | |
| # btn = gr.Button("Generate Study Pack", variant="primary") | |
| # with gr.Row(): | |
| # md_out = gr.Markdown(label="Real-time Extraction") | |
| # json_out = gr.File(label="Download Full Data (JSON)") | |
| # btn.click(run_pipeline, inputs=file_in, outputs=[md_out, json_out]) | |
| # demo.launch(share=True) | |
| # import gradio as gr | |
| # import torch | |
| # import re | |
| # import os | |
| # import fitz # PyMuPDF | |
| # import cv2 | |
| # import numpy as np | |
| # import base64 | |
| # import json | |
| # import time | |
| # import uuid | |
| # import spacy | |
| # import multiprocessing | |
| # from transformers import ( | |
| # T5Tokenizer, | |
| # T5ForConditionalGeneration, | |
| # ) | |
| # from peft import PeftModel | |
| # from sentence_transformers import CrossEncoder | |
| # from ultralytics import YOLO | |
| # # ========================================== | |
| # # 0. SPACY SETUP | |
| # # ========================================== | |
| # print("🧠 Booting Syntactic Parser...") | |
| # try: | |
| # nlp = spacy.load("en_core_web_sm") | |
| # except OSError: | |
| # os.system("python -m spacy download en_core_web_sm") | |
| # nlp = spacy.load("en_core_web_sm") | |
| # # ========================================== | |
| # # 1. HARDWARE & MODEL SETUP | |
| # # ========================================== | |
| # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # torch_dtype = torch.float16 if device == "cuda" else torch.float32 | |
| # # 🔥 FIX 1: WAKE UP THE CPU | |
| # # Force PyTorch to use ALL available CPU cores instead of defaulting to 1 | |
| # num_cores = multiprocessing.cpu_count() | |
| # torch.set_num_threads(num_cores) | |
| # os.environ["OMP_NUM_THREADS"] = str(num_cores) | |
| # print(f"🚀 Booting Edugenius Multimodal on: {device.upper()} with {num_cores} Threads") | |
| # # --- Text Models --- | |
| # model_id = "t5-base" | |
| # adapter_path = "." | |
| # print("⏳ Loading T5 Generator...") | |
| # base_model = T5ForConditionalGeneration.from_pretrained(model_id, torch_dtype=torch_dtype) | |
| # try: | |
| # print("⏳ Loading and Merging PEFT Adapter...") | |
| # peft_model = PeftModel.from_pretrained(base_model, adapter_path) | |
| # # 🔥 FIX 2: MERGE AND UNLOAD | |
| # # This bakes the LoRA weights into the base model. | |
| # # It removes the massive CPU calculation overhead during generation. | |
| # model = peft_model.merge_and_unload() | |
| # print("✅ Adapter merged successfully!") | |
| # except Exception as e: | |
| # print(f"⚠️ Could not load or merge PeftModel. Using base model. Error: {e}") | |
| # model = base_model | |
| # tokenizer = T5Tokenizer.from_pretrained("t5-base") | |
| # model.to(device).eval() | |
| # # --- Validation Models --- | |
| # print("⏳ Loading Validation Guardrails (CrossEncoder Only)...") | |
| # validator = CrossEncoder('cross-encoder/ms-marco-TinyBERT-L-2-v2', device=device) | |
| # # --- Vision Model (YOLO) --- | |
| # print("⏳ Loading YOLO Figure Detector...") | |
| # YOLO_WEIGHTS = 'best.pt' | |
| # vision_model = YOLO(YOLO_WEIGHTS) | |
| # print("✅ ALL MODELS LOADED AND READY.") | |
| # # ========================================== | |
| # # 2. UTILITY FUNCTIONS | |
| # # ========================================== | |
| # def get_paragraph_subject(paragraph_text): | |
| # doc = nlp(paragraph_text[:200]) | |
| # for chunk in doc.noun_chunks: | |
| # if chunk.root.dep_ in ('nsubj', 'nsubjpass') and chunk.root.pos_ in ('PROPN', 'NOUN'): | |
| # return chunk.text | |
| # return "The subject" | |
| # # 🔥 FIX 4: MICRO-CHUNKING | |
| # def chunk_text_by_sentence(text): | |
| # doc = nlp(text) | |
| # chunks = [] | |
| # for sent in doc.sents: | |
| # s_text = sent.text.strip() | |
| # if len(s_text.split()) < 5: continue | |
| # # 1. Add the full sentence for broad context | |
| # chunks.append(s_text) | |
| # # 2. MICRO-CHUNKING: Split by clauses so the model is forced | |
| # # to look at isolated facts (like measurements or specific functions) | |
| # sub_clauses = re.split(r'(?:, which |, and |; |\()', s_text) | |
| # for clause in sub_clauses: | |
| # # Clean up trailing parentheses if we split on an open one | |
| # clause = clause.replace(')', '').strip() | |
| # if len(clause.split()) >= 4: | |
| # chunks.append(clause) | |
| # # Remove duplicates while preserving order | |
| # return list(dict.fromkeys(chunks)) | |
| # # 🔥 FIX 3: DIAGRAM LABEL CLEANER | |
| # def process_pdf_text_clean(pdf_path): | |
| # print(f"\n[DEBUG] Starting clean text extraction for {pdf_path}") | |
| # doc = fitz.open(pdf_path) | |
| # full_text = "" | |
| # for page in doc: | |
| # text = page.get_text("text") | |
| # if not text: continue | |
| # lines = text.split('\n') | |
| # for line in lines: | |
| # line = line.strip() | |
| # # Skip numbers, diagram labels (under 4 words), and figure text | |
| # word_count = len(line.split()) | |
| # if word_count < 4 or line.isdigit() or line.lower().startswith("figure") or "Reprint" in line: | |
| # continue | |
| # full_text += line + " " | |
| # doc.close() | |
| # # Clean up extra spaces | |
| # clean_result = re.sub(r'\s+', ' ', full_text).strip() | |
| # print(f"[DEBUG] Text cleaned. Length: {len(clean_result)} chars.") | |
| # return clean_result | |
| # def detect_figures(pdf_path): | |
| # print(f"\n[DEBUG] Starting YOLO figure detection for {pdf_path}") | |
| # doc = fitz.open(pdf_path) | |
| # visuals = [] | |
| # fig_idx = 1 | |
| # mat = fitz.Matrix(1.0, 1.0) | |
| # for page_num, page in enumerate(doc): | |
| # pix = page.get_pixmap(matrix=mat) | |
| # img = np.frombuffer(pix.samples, dtype=np.uint8).reshape((pix.h, pix.w, pix.n)) | |
| # img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| # results = vision_model.predict(img, conf=0.3, verbose=False) | |
| # for box in results[0].boxes: | |
| # cls = int(box.cls[0]) | |
| # if vision_model.names[cls] == 'figure': | |
| # x1, y1, x2, y2 = map(int, box.xyxy[0]) | |
| # crop = img[y1:y2, x1:x2] | |
| # _, buffer = cv2.imencode('.png', crop) | |
| # b64 = base64.b64encode(buffer).decode('utf-8') | |
| # visuals.append({ | |
| # "id": f"FIG_{fig_idx}", | |
| # "page": page_num + 1, | |
| # "base64": b64 | |
| # }) | |
| # fig_idx += 1 | |
| # print(f"[DEBUG] Found figure on page {page_num + 1}") | |
| # doc.close() | |
| # print(f"[DEBUG] Figure detection complete. Total figures: {len(visuals)}") | |
| # return visuals | |
| # # ========================================== | |
| # # 3. CORE EXTRACTION ENGINE | |
| # # ========================================== | |
| # def extract_flashcards(text): | |
| # if not text or len(text.strip()) < 10: | |
| # yield "Please provide more text.", [] | |
| # return | |
| # accepted_cards = [] | |
| # accepted_markdown = "### ✅ Accepted Flashcards\n\n" | |
| # rejected_markdown = "### 🛑 Blocked by Guardrails (Debug)\n\n" | |
| # seen_answers = set() | |
| # paragraphs = text.replace('\r', '').split('\n\n') | |
| # print(f"\n[DEBUG] Starting extraction loop on {len(paragraphs)} paragraphs.") | |
| # for p_idx, paragraph in enumerate(paragraphs): | |
| # if len(paragraph.strip()) < 20: continue | |
| # current_subject = get_paragraph_subject(paragraph) | |
| # chunks = chunk_text_by_sentence(paragraph) | |
| # for c_idx, chunk in enumerate(chunks): | |
| # grounded_chunk = f"{current_subject}: {chunk}" | |
| # print(f"\n[DEBUG] --- Processing P{p_idx}-C{c_idx} ---") | |
| # print(f"[DEBUG] Chunk context: {grounded_chunk[:100]}...") | |
| # prompt = f"task: generate Understand flashcard context: {grounded_chunk}" | |
| # inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(device) | |
| # with torch.no_grad(): | |
| # # 🔥 FIX 6: BALANCE SPEED AND QUALITY WITH BEAMS=2 | |
| # outputs = model.generate(**inputs, max_new_tokens=128, num_beams=2, early_stopping=True) | |
| # raw_output = tokenizer.decode(outputs[0], skip_special_tokens=True) | |
| # print(f"[GEN] Raw T5 Output: {raw_output}") | |
| # if "QS:" in raw_output and "| AN:" in raw_output: | |
| # try: | |
| # parts = raw_output.split("| AN:") | |
| # q = parts[0].replace("QS:", "").strip() | |
| # a = parts[1].strip() | |
| # q_lower = q.lower() | |
| # a_lower = a.lower() | |
| # print(f"[GEN] Parsed Q: {q}") | |
| # print(f"[GEN] Parsed A: {a}") | |
| # if a_lower in seen_answers or len(q) < 15 or len(a.split()) > 15 or len(a) < 3: | |
| # print("[REJECT] Failed basic length/duplicate checks.") | |
| # continue | |
| # # 🔥 FIX 5: ADVANCED CIRCULAR Q/A CHECK | |
| # stop_words = {'what', 'is', 'are', 'the', 'a', 'an', 'of', 'in', 'to', 'for'} | |
| # q_words = set([w for w in re.findall(r'\b\w+\b', q_lower) if w not in stop_words]) | |
| # a_words = set([w for w in re.findall(r'\b\w+\b', a_lower) if w not in stop_words]) | |
| # if a_words.issubset(q_words) or q_words.issubset(a_words): | |
| # print("[REJECT] Answer matches question too closely (Circular).") | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: Circular Answer)*\n\n---\n\n" | |
| # # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # continue | |
| # seen_answers.add(a_lower) | |
| # # --- NER & Syntax Check --- | |
| # ans_doc = nlp(a) | |
| # first_word = q_lower.split()[0] if q_lower.split() else "" | |
| # ents = [ent.label_ for ent in ans_doc.ents] | |
| # blocked_reason = None | |
| # if first_word == "when" and not any(e in ents for e in ["DATE", "TIME", "CARDINAL"]): | |
| # blocked_reason = "NER: Expected Date" | |
| # elif first_word == "who" and not any(e in ents for e in ["PERSON", "ORG", "NORP"]): | |
| # blocked_reason = "NER: Expected Person" | |
| # elif first_word == "where" and not any(e in ents for e in ["GPE", "LOC"]): | |
| # blocked_reason = "NER: Expected Location" | |
| # if blocked_reason: | |
| # print(f"[REJECT] {blocked_reason}") | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: {blocked_reason})*\n\n---\n\n" | |
| # # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # continue | |
| # # --- Semantic Check --- | |
| # score = validator.predict([[q, grounded_chunk]])[0] | |
| # if score < -4.0: | |
| # print(f"[REJECT] Semantic score too low: {score:.2f}") | |
| # rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: CE Score {score:.2f})*\n\n---\n\n" | |
| # # yield accepted_markdown + "\n\n" + rejected_markdown, [] | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # continue | |
| # print(f"[ACCEPT] Flashcard passed all checks! (CE Score: {score:.2f})") | |
| # # telemetry = f"*(**CE Score:** {score:.2f} | **Subject:** \"{current_subject}\")*" | |
| # telemetry = f"*(**CE Score:** {score:.2f} | **Subject:** \"{current_subject}\" | **Source:** \"{chunk.strip()}\")*" | |
| # accepted_markdown += f"**Q:** {q}\n**A:** {a}\n{telemetry}\n\n---\n\n" | |
| # accepted_cards.append({"question": q, "answer": a, "subject": current_subject}) | |
| # yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| # except Exception as e: | |
| # print(f"[ERROR] Parsing exception: {e}") | |
| # continue | |
| # # ========================================== | |
| # # 4. FINAL GRADIO INTERFACE | |
| # # ========================================== | |
| # def run_pipeline(pdf_file): | |
| # if pdf_file is None: | |
| # print("[ERROR] No PDF file uploaded.") | |
| # return "Upload a PDF.", None | |
| # print(f"\n[START] Processing started for {pdf_file.name}") | |
| # # 1. Detect Visuals | |
| # yield "🔍 Step 1/3: Scanning for Figures...", None | |
| # visuals = detect_figures(pdf_file.name) | |
| # # 2. Extract Text | |
| # yield "📝 Step 2/3: Cleaning Text and Analyzing Subject...", None | |
| # clean_text = process_pdf_text_clean(pdf_file.name) | |
| # # 3. Extract Flashcards | |
| # final_cards = [] | |
| # md = "" | |
| # for md_update, card_list in extract_flashcards(clean_text): | |
| # md = md_update | |
| # final_cards = card_list | |
| # yield md, None | |
| # # 4. REFORMAT FOR BACKEND SCHEMA WITH UUID | |
| # print("\n[DEBUG] Compiling final backend JSON package...") | |
| # backend_mcqs = [] | |
| # for i, fig in enumerate(visuals): | |
| # backend_mcqs.append({ | |
| # "question": "Figure", | |
| # "answer": f"Visual from page {fig['page']}", | |
| # "question_type": "FLASHCARD", | |
| # "figure1": fig['base64'], | |
| # "documentIndex": i + 1, | |
| # "questionId": str(uuid.uuid4()), | |
| # "options": [], | |
| # "predicted_subject": None, | |
| # "predicted_concept": None | |
| # }) | |
| # start_idx = len(backend_mcqs) + 1 | |
| # for i, card in enumerate(final_cards): | |
| # backend_mcqs.append({ | |
| # "question": card["question"], | |
| # "answer": card["answer"], | |
| # "question_type": "FLASHCARD", | |
| # "figure1": None, | |
| # "documentIndex": start_idx + i, | |
| # "questionId": str(uuid.uuid4()), | |
| # "options": [], | |
| # "predicted_subject": {"label": card["subject"], "confidence": 1.0}, | |
| # "predicted_concept": None | |
| # }) | |
| # generated_id = str(uuid.uuid4()) | |
| # data_pack = [{ | |
| # "document": "Edugenius Generated", | |
| # "id": generated_id, | |
| # "metadata": { | |
| # "answerFound": True, | |
| # "card_count": len(backend_mcqs), | |
| # "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| # "file_name": os.path.basename(pdf_file.name), | |
| # "generatedQAId": generated_id, | |
| # "mcqs": backend_mcqs | |
| # } | |
| # }] | |
| # output_path = "edugenius_pack.json" | |
| # with open(output_path, "w") as f: | |
| # json.dump(data_pack, f, indent=2) | |
| # print(f"[SUCCESS] Pipeline complete. Generated {len(backend_mcqs)} total items.") | |
| # yield md + "\n\n✅ **Processing Complete! Backend-compatible JSON Ready.**", output_path | |
| # with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| # gr.Markdown("# 🚀 Edugenius Multimodal Study-Pack Generator (CPU Optimized)") | |
| # with gr.Row(): | |
| # file_in = gr.File(label="Upload Textbook (PDF)") | |
| # btn = gr.Button("Generate Study Pack", variant="primary") | |
| # with gr.Row(): | |
| # md_out = gr.Markdown(label="Real-time Extraction") | |
| # json_out = gr.File(label="Download Full Data (JSON)") | |
| # btn.click(run_pipeline, inputs=file_in, outputs=[md_out, json_out]) | |
| # demo.launch(share=True) | |
| import gradio as gr | |
| import torch | |
| import re | |
| import os | |
| import fitz # PyMuPDF | |
| import cv2 | |
| import numpy as np | |
| import base64 | |
| import json | |
| import time | |
| import uuid | |
| import spacy | |
| import multiprocessing | |
| from transformers import ( | |
| T5Tokenizer, | |
| T5ForConditionalGeneration, | |
| ) | |
| from peft import PeftModel | |
| from sentence_transformers import CrossEncoder | |
| from ultralytics import YOLO | |
| # ========================================== | |
| # 0. SPACY & COREF SETUP | |
| # ========================================== | |
| print("🧠 Booting Syntactic Parser...") | |
| try: | |
| nlp = spacy.load("en_core_web_sm") | |
| except OSError: | |
| os.system("python -m spacy download en_core_web_sm") | |
| nlp = spacy.load("en_core_web_sm") | |
| print("🧠 Booting Coreference Resolution...") | |
| try: | |
| from fastcoref import FCoref | |
| from fastcoref.modeling import FCorefModel | |
| # --- MONKEY PATCH --- | |
| # Fixes compatibility issues with modern transformers (>=4.38.0) | |
| # Force overwrite in case a previous notebook run left it as a set() | |
| FCorefModel.all_tied_weights_keys = {} | |
| coref_device = 'cuda:0' if torch.cuda.is_available() else 'cpu' | |
| coref_model = FCoref(device=coref_device) | |
| USE_NEURAL_COREF = True | |
| print("✅ FastCoref loaded successfully.") | |
| except Exception as e: | |
| print(f"⚠️ Neural coref failed to load. Falling back to heuristic tracking. Error: {e}") | |
| USE_NEURAL_COREF = False | |
| # ========================================== | |
| # 1. HARDWARE & MODEL SETUP | |
| # ========================================== | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| torch_dtype = torch.float16 if device == "cuda" else torch.float32 | |
| # Force PyTorch to use ALL available CPU cores instead of defaulting to 1 | |
| num_cores = multiprocessing.cpu_count() | |
| torch.set_num_threads(num_cores) | |
| os.environ["OMP_NUM_THREADS"] = str(num_cores) | |
| print(f"🚀 Booting Multimodal Generator on: {device.upper()} with {num_cores} Threads") | |
| # --- Text Models --- | |
| model_id = "t5-base" | |
| adapter_path = "." | |
| print("⏳ Loading T5 Generator...") | |
| base_model = T5ForConditionalGeneration.from_pretrained(model_id, torch_dtype=torch_dtype) | |
| try: | |
| print("⏳ Loading and Merging PEFT Adapter...") | |
| peft_model = PeftModel.from_pretrained(base_model, adapter_path) | |
| model = peft_model.merge_and_unload() | |
| print("✅ Adapter merged successfully!") | |
| except Exception as e: | |
| print(f"⚠️ Could not load or merge PeftModel. Using base model. Error: {e}") | |
| model = base_model | |
| tokenizer = T5Tokenizer.from_pretrained("t5-base") | |
| model.to(device).eval() | |
| # --- Validation Models --- | |
| print("⏳ Loading Validation Guardrails (CrossEncoder Only)...") | |
| validator = CrossEncoder('cross-encoder/ms-marco-TinyBERT-L-2-v2', device=device) | |
| # --- Vision Model (YOLO) --- | |
| print("⏳ Loading YOLO Figure Detector...") | |
| YOLO_WEIGHTS = 'best.pt' | |
| vision_model = YOLO(YOLO_WEIGHTS) | |
| print("✅ ALL MODELS LOADED AND READY.") | |
| # ========================================== | |
| # 2. UTILITY FUNCTIONS | |
| # ========================================== | |
| def resolve_paragraph_coref(paragraph): | |
| """Resolves pronouns across the entire paragraph before chunking.""" | |
| if len(paragraph.split()) < 5: | |
| return paragraph | |
| if USE_NEURAL_COREF: | |
| try: | |
| # fastcoref processes lists of strings | |
| preds = coref_model.predict(texts=[paragraph]) | |
| # get_resolved_text returns the text with pronouns swapped for nouns | |
| resolved_text = preds[0].get_resolved_text() | |
| return resolved_text | |
| except Exception as e: | |
| print(f"[DEBUG] Neural Coref failed on a paragraph, falling back. Error: {e}") | |
| pass | |
| # Smart Heuristic Fallback: Tracks the actual grammatical subject sentence-to-sentence | |
| doc = nlp(paragraph) | |
| resolved_sentences = [] | |
| last_subject = None | |
| for sent in doc.sents: | |
| text = sent.text.strip() | |
| if not text: continue | |
| first_token = sent[0] | |
| # If sentence starts with a pronoun, replace with the last known valid subject | |
| if first_token.pos_ == "PRON" and last_subject: | |
| text = re.sub(r'^' + re.escape(first_token.text), last_subject, text, count=1, flags=re.IGNORECASE) | |
| # Update the active subject for the NEXT sentence | |
| for token in sent: | |
| if "subj" in token.dep_ and token.pos_ in ["NOUN", "PROPN"]: | |
| # Extract the noun phrase | |
| last_subject = "".join([w.text_with_ws for w in token.subtree]).strip() | |
| break | |
| resolved_sentences.append(text) | |
| return " ".join(resolved_sentences) | |
| def get_sentence_sub_subject(chunk_text, main_subject): | |
| """Extracts prominent noun chunks, ignoring generic stop-words.""" | |
| doc = nlp(chunk_text) | |
| # Priority 1: Check for explicit Named Entities first | |
| for ent in doc.ents: | |
| if ent.label_ not in ['DATE', 'TIME', 'PERCENT', 'MONEY', 'CARDINAL', 'ORDINAL', 'QUANTITY']: | |
| if main_subject.lower() not in ent.text.lower() and len(ent.text) > 2: | |
| return ent.text.strip('.,;()') | |
| # Priority 2: Filtered Noun Chunks | |
| candidates = [] | |
| stop_nouns = {'presence', 'years', 'conditions', 'majority', 'most', 'some', 'part', 'type', 'number', 'numbers', 'example', 'organisms', 'surface', 'them'} | |
| for chunk in doc.noun_chunks: | |
| root = chunk.root | |
| if root.pos_ in ['PRON', 'DET']: continue | |
| if main_subject.lower() in chunk.text.lower(): continue | |
| if root.text.lower() in stop_nouns: continue | |
| if root.dep_ in ('attr', 'dobj', 'pobj', 'appos', 'nsubjpass'): | |
| clean_text = chunk.text.strip('.,;()') | |
| if len(clean_text) > 2: | |
| candidates.append(clean_text) | |
| return candidates[0] if candidates else None | |
| def chunk_text_by_blocks(text, block_size=2): | |
| """ | |
| Groups sentences into larger contextual blocks rather than | |
| forcing the model to evaluate every single sentence or clause. | |
| """ | |
| doc = nlp(text) | |
| # Filter out junk/tiny sentences | |
| valid_sentences = [sent.text.strip() for sent in doc.sents if len(sent.text.split()) >= 5] | |
| chunks = [] | |
| # Group sentences into blocks of 'block_size' | |
| for i in range(0, len(valid_sentences), block_size): | |
| block = " ".join(valid_sentences[i:i + block_size]) | |
| chunks.append(block) | |
| return list(dict.fromkeys(chunks)) | |
| def extract_structured_text(pdf_path): | |
| print(f"\n[DEBUG] Starting structured text extraction for {pdf_path}") | |
| doc = fitz.open(pdf_path) | |
| structured_blocks = [] | |
| current_header = "General Context" | |
| for page_num, page in enumerate(doc): | |
| page_dict = page.get_text("dict") | |
| blocks = page_dict.get("blocks", []) | |
| for block in blocks: | |
| if block.get("type") != 0: | |
| continue | |
| block_text = "" | |
| is_bold = False | |
| max_font_size = 0 | |
| for line in block.get("lines", []): | |
| for span in line.get("spans", []): | |
| text = span.get("text", "").strip() | |
| if not text: continue | |
| block_text += text + " " | |
| if span.get("size", 0) > max_font_size: | |
| max_font_size = span.get("size") | |
| if "bold" in span.get("font", "").lower(): | |
| is_bold = True | |
| block_text = block_text.strip() | |
| word_count = len(block_text.split()) | |
| # --- CLEANING HEURISTICS --- | |
| if word_count < 4 and block_text.replace(".", "").isdigit(): | |
| continue | |
| if block_text.lower().startswith("figure") or "Reprint" in block_text: | |
| continue | |
| # --- HEADER DETECTION --- | |
| if word_count < 10 and (max_font_size > 11.5 or is_bold): | |
| clean_header = re.sub(r'^\d+(\.\d+)*\s*', '', block_text).strip() | |
| # NEW: Ignore chapters, summaries, and long all-caps text (running headers) | |
| lower_header = clean_header.lower() | |
| if lower_header.startswith("chapter") or "summary" in lower_header or "exercises" in lower_header: | |
| continue | |
| if clean_header.isupper() and word_count > 3: | |
| continue | |
| current_header = clean_header | |
| print(f"[DEBUG] Found New Header: {current_header}") | |
| else: | |
| if word_count >= 10: | |
| structured_blocks.append({ | |
| "header": current_header, | |
| "text": block_text | |
| }) | |
| doc.close() | |
| return structured_blocks | |
| def detect_figures(pdf_path): | |
| print(f"\n[DEBUG] Starting YOLO figure detection for {pdf_path}") | |
| doc = fitz.open(pdf_path) | |
| visuals = [] | |
| fig_idx = 1 | |
| mat = fitz.Matrix(1.0, 1.0) | |
| for page_num, page in enumerate(doc): | |
| pix = page.get_pixmap(matrix=mat) | |
| img = np.frombuffer(pix.samples, dtype=np.uint8).reshape((pix.h, pix.w, pix.n)) | |
| img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| results = vision_model.predict(img, conf=0.3, verbose=False) | |
| for box in results[0].boxes: | |
| cls = int(box.cls[0]) | |
| if vision_model.names[cls] == 'figure': | |
| x1, y1, x2, y2 = map(int, box.xyxy[0]) | |
| crop = img[y1:y2, x1:x2] | |
| _, buffer = cv2.imencode('.png', crop) | |
| b64 = base64.b64encode(buffer).decode('utf-8') | |
| visuals.append({ | |
| "id": f"FIG_{fig_idx}", | |
| "page": page_num + 1, | |
| "base64": b64 | |
| }) | |
| fig_idx += 1 | |
| print(f"[DEBUG] Found figure on page {page_num + 1}") | |
| doc.close() | |
| print(f"[DEBUG] Figure detection complete. Total figures: {len(visuals)}") | |
| return visuals | |
| # ========================================== | |
| # 3. CORE EXTRACTION ENGINE | |
| # ========================================== | |
| def extract_flashcards(structured_blocks): | |
| if not structured_blocks: | |
| yield "Please provide more text.", [] | |
| return | |
| accepted_cards = [] | |
| accepted_markdown = "### ✅ Accepted Flashcards\n\n" | |
| rejected_markdown = "### 🛑 Blocked by Guardrails (Debug)\n\n" | |
| seen_answers = set() | |
| print(f"\n[DEBUG] Starting extraction loop on {len(structured_blocks)} structured blocks.") | |
| for b_idx, block in enumerate(structured_blocks): | |
| current_subject = block["header"] | |
| raw_paragraph = block["text"] | |
| # 1. Resolve coreferences on the ENTIRE paragraph | |
| resolved_paragraph = resolve_paragraph_coref(raw_paragraph) | |
| # 2. Chunk the resolved text into 2-sentence blocks | |
| chunks = chunk_text_by_blocks(resolved_paragraph, block_size=2) | |
| for c_idx, chunk in enumerate(chunks): | |
| # 3. Extract sub-subject for metadata | |
| sub_subject = get_sentence_sub_subject(chunk, current_subject) | |
| # 4. Format prompt | |
| grounded_chunk = f"{current_subject}: {chunk}" | |
| print(f"\n[DEBUG] --- Processing B{b_idx}-C{c_idx} ---") | |
| print(f"[DEBUG] Prompt Context: {grounded_chunk[:100]}...") | |
| prompt = f"task: generate Understand flashcard context: {grounded_chunk}" | |
| inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(device) | |
| with torch.no_grad(): | |
| outputs = model.generate(**inputs, max_new_tokens=128, num_beams=2, early_stopping=True) | |
| raw_output = tokenizer.decode(outputs[0], skip_special_tokens=True) | |
| print(f"[GEN] Raw T5 Output: {raw_output}") | |
| if "QS:" in raw_output and "| AN:" in raw_output: | |
| try: | |
| parts = raw_output.split("| AN:") | |
| q = parts[0].replace("QS:", "").strip() | |
| a = parts[1].strip() | |
| q_lower = q.lower() | |
| a_lower = a.lower() | |
| if a_lower in seen_answers or len(q) < 15 or len(a.split()) > 15 or len(a) < 3: | |
| print("[REJECT] Failed basic length/duplicate checks.") | |
| continue | |
| # --- ADVANCED CIRCULAR Q/A CHECK --- | |
| stop_words = {'what', 'is', 'are', 'the', 'a', 'an', 'of', 'in', 'to', 'for'} | |
| q_words = set([w for w in re.findall(r'\b\w+\b', q_lower) if w not in stop_words]) | |
| a_words = set([w for w in re.findall(r'\b\w+\b', a_lower) if w not in stop_words]) | |
| if a_words.issubset(q_words) or q_words.issubset(a_words): | |
| print("[REJECT] Answer matches question too closely (Circular).") | |
| rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: Circular Answer)*\n\n---\n\n" | |
| yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| continue | |
| seen_answers.add(a_lower) | |
| # --- NER & Syntax Check --- | |
| ans_doc = nlp(a) | |
| first_word = q_lower.split()[0] if q_lower.split() else "" | |
| ents = [ent.label_ for ent in ans_doc.ents] | |
| blocked_reason = None | |
| if first_word == "when" and not any(e in ents for e in ["DATE", "TIME", "CARDINAL"]): | |
| blocked_reason = "NER: Expected Date" | |
| if blocked_reason: | |
| print(f"[REJECT] {blocked_reason}") | |
| rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: {blocked_reason})*\n\n---\n\n" | |
| yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| continue | |
| # --- SUBJECT INJECTION (Restored and Improved) --- | |
| # We only inject if it's a real topic, not the default "General Context" | |
| if current_subject and current_subject != "General Context" and current_subject.lower() not in q_lower: | |
| if q_lower.startswith(("what", "how", "why", "when", "where", "who", "which")): | |
| q = f"Regarding {current_subject}, {q[0].lower()}{q[1:]}" | |
| else: | |
| q = f"For {current_subject}, {q[0].lower()}{q[1:]}" | |
| print(f"[REFINED] New Question: {q}") | |
| # --- Semantic Check --- | |
| # CRITICAL FIX: Evaluate against the grounded_chunk so the CE model | |
| # knows what subject is being talked about and doesn't falsely block it. | |
| score = validator.predict([[q, grounded_chunk]])[0] | |
| if score < -3.5: | |
| print(f"[REJECT] Semantic score too low: {score:.2f}") | |
| rejected_markdown += f"~~**Q:** {q}~~\n*(Blocked: CE Score {score:.2f})*\n\n---\n\n" | |
| yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| continue | |
| print(f"[ACCEPT] Flashcard passed all checks! (CE Score: {score:.2f})") | |
| telemetry_sub = f" | **Sub-Subject:** \"{sub_subject}\"" if sub_subject else "" | |
| telemetry = f"*(**CE Score:** {score:.2f} | **Topic Header:** \"{current_subject}\"{telemetry_sub} | **Source:** \"{chunk.strip()}\")*" | |
| accepted_markdown += f"**Q:** {q}\n**A:** {a}\n{telemetry}\n\n---\n\n" | |
| accepted_cards.append({ | |
| "question": q, | |
| "answer": a, | |
| "subject": current_subject, | |
| "sub_subject": sub_subject | |
| }) | |
| yield accepted_markdown + "\n\n" + rejected_markdown, accepted_cards | |
| except Exception as e: | |
| print(f"[ERROR] Parsing exception: {e}") | |
| continue | |
| # ========================================== | |
| # 4. FINAL GRADIO INTERFACE | |
| # ========================================== | |
| def run_pipeline(pdf_file): | |
| if pdf_file is None: | |
| print("[ERROR] No PDF file uploaded.") | |
| return "Upload a PDF.", None | |
| print(f"\n[START] Processing started for {pdf_file.name}") | |
| # 1. Detect Visuals | |
| yield "🔍 Step 1/3: Scanning for Figures...", None | |
| visuals = detect_figures(pdf_file.name) | |
| # 2. Extract Text | |
| yield "📝 Step 2/3: Analyzing Document Layout and Context...", None | |
| structured_data = extract_structured_text(pdf_file.name) | |
| # 3. Extract Flashcards | |
| final_cards = [] | |
| md = "" | |
| for md_update, card_list in extract_flashcards(structured_data): | |
| md = md_update | |
| final_cards = card_list | |
| yield md, None | |
| # 4. REFORMAT FOR BACKEND SCHEMA WITH UUID | |
| print("\n[DEBUG] Compiling final backend JSON package...") | |
| backend_mcqs = [] | |
| for i, fig in enumerate(visuals): | |
| backend_mcqs.append({ | |
| "question": "Figure", | |
| "answer": f"Visual from page {fig['page']}", | |
| "question_type": "FLASHCARD", | |
| "figure1": fig['base64'], | |
| "documentIndex": i + 1, | |
| "questionId": str(uuid.uuid4()), | |
| "options": [], | |
| "predicted_subject": None, | |
| "predicted_concept": None | |
| }) | |
| start_idx = len(backend_mcqs) + 1 | |
| for i, card in enumerate(final_cards): | |
| backend_mcqs.append({ | |
| "question": card["question"], | |
| "answer": card["answer"], | |
| "question_type": "FLASHCARD", | |
| "figure1": None, | |
| "documentIndex": start_idx + i, | |
| "questionId": str(uuid.uuid4()), | |
| "options": [], | |
| "predicted_subject": {"label": card["subject"], "confidence": 1.0}, | |
| "predicted_concept": {"label": card["sub_subject"], "confidence": 1.0} if card["sub_subject"] else None | |
| }) | |
| generated_id = str(uuid.uuid4()) | |
| data_pack = [{ | |
| "document": "AI Generated", | |
| "id": generated_id, | |
| "metadata": { | |
| "answerFound": True, | |
| "card_count": len(backend_mcqs), | |
| "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| "file_name": os.path.basename(pdf_file.name), | |
| "generatedQAId": generated_id, | |
| "mcqs": backend_mcqs | |
| } | |
| }] | |
| output_path = "study_pack.json" | |
| with open(output_path, "w") as f: | |
| json.dump(data_pack, f, indent=2) | |
| print(f"[SUCCESS] Pipeline complete. Generated {len(backend_mcqs)} total items.") | |
| yield md + "\n\n✅ **Processing Complete! Backend-compatible JSON Ready.**", output_path | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# 🚀 Multimodal Study-Pack Generator") | |
| with gr.Row(): | |
| file_in = gr.File(label="Upload Textbook (PDF)") | |
| btn = gr.Button("Generate Study Pack", variant="primary") | |
| with gr.Row(): | |
| md_out = gr.Markdown(label="Real-time Extraction") | |
| json_out = gr.File(label="Download Full Data (JSON)") | |
| btn.click(run_pipeline, inputs=file_in, outputs=[md_out, json_out]) | |
| if __name__ == "__main__": | |
| demo.launch(share=True) |