# 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)