Upload 6 files

ans_generator1.py

@@ -0,0 +1,44 @@
+from transformers import pipeline
+import PyPDF2
+import os
+
+UPLOAD_DIR = "uploaded_pdfs"
+
+class AnswerGenerator:
+    def __init__(self):
+        # ✅ Default FLAN-T5 model for question answering
+        self.qa_pipeline = pipeline("question-answering", model="google/flan-t5-base")
+
+#---------------------------------------------------------------
+# updated the modal
+ #self.qa_pipeline = pipeline(
+           # "question-answering",
+          #  model="tiiuae/falcon-7b-instruct",    # <-- Updated model here
+          #  tokenizer="tiiuae/falcon-7b-instruct"  # <-- Explicitly specifying tokenizer)
+#-----------------------------------------------------------------
+    def extract_pdf_text(self, token):
+        pdf_path = os.path.join(UPLOAD_DIR, f"{token}.pdf")
+        if not os.path.exists(pdf_path):
+            raise FileNotFoundError("PDF not found for given token")
+
+        with open(pdf_path, "rb") as f:
+            reader = PyPDF2.PdfReader(f)
+            return [page.extract_text() or "" for page in reader.pages]  # List of page texts
+
+    def generate_answers(self, token, questions):
+        pages = self.extract_pdf_text(token)
+        full_text = "\n".join(pages)  # Merge pages as context
+        results = []
+
+        for question in questions:
+            try:
+                # ✅ Default FLAN-T5 logic
+                result = self.qa_pipeline(question=question, context=full_text)
+                results.append({"question": question, "answer": result["answer"]})
+
+
+
+            except Exception as e:
+                results.append({"question": question, "answer": "Error", "error": str(e)})
+
+        return results

app.py

@@ -0,0 +1,116 @@
+import gradio as gr
+import uuid
+import sqlite3
+import json
+import re
+import PyPDF2
+import io
+import numpy as np
+from transformers import pipeline
+from sklearn.metrics.pairwise import cosine_similarity
+
+from database1 import create_db
+from first1 import pdf_query
+from q_generator1 import QGenerator
+from ans_generator1 import AnswerGenerator
+
+# Models and tools
+qgen = QGenerator()
+ansgen = AnswerGenerator()
+from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
+
+tokenizer = AutoTokenizer.from_pretrained("google/flan-t5-base", use_fast=False)
+model = AutoModelForSeq2SeqLM.from_pretrained("google/flan-t5-base")
+qa_model = pipeline("text2text-generation", model=model, tokenizer=tokenizer)
+
+
+# ✅ Upload PDF and store to SQLite
+def upload_pdf(file):
+    try:
+        filename = file.name
+        token = str(uuid.uuid4())
+
+        pdf_reader = PyPDF2.PdfReader(file)
+        text = "".join([page.extract_text() or "" for page in pdf_reader.pages])
+        chunks = [text[i:i+500] for i in range(0, len(text), 500)]
+
+        create_db(token, chunks, filename, text)
+        return f"✅ Uploaded and stored: {filename} (Token: {token})"
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
+
+# ✅ Generate Q&A from filename
+def generate_qa(filename):
+    try:
+        with sqlite3.connect("my_database.db") as conn:
+            cursor = conn.cursor()
+            cursor.execute("SELECT chunk_data FROM token_data WHERE filename = ?", (filename,))
+            row = cursor.fetchone()
+
+        if not row:
+            return "❌ No data found for this filename."
+
+        chunks = json.loads(row[0])
+        qa_pairs = []
+
+        for chunk in chunks:
+            questions = qgen.generate(chunk)
+            if not questions:
+                continue
+            question = questions[0]
+            prompt = f"Context: {chunk}\n\nQuestion: {question}\n\nAnswer:"
+            result = qa_model(prompt, max_length=256, do_sample=False)
+            answer = result[0]["generated_text"].strip()
+            qa_pairs.append(f"Q: {question}\nA: {answer}")
+        return "\n\n".join(qa_pairs)
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
+
+# ✅ Ask a question using token (semantic similarity)
+def ask_question(token, question):
+    try:
+        with sqlite3.connect("my_database.db") as conn:
+            cursor = conn.cursor()
+            cursor.execute("SELECT chunk_data FROM token_data WHERE token_id = ?", (token,))
+            row = cursor.fetchone()
+        if not row:
+            return "❌ Token not found."
+
+        chunks = json.loads(row[0])
+        processor = pdf_query()
+        model = processor.model
+        chunk_embeddings = model.encode(chunks)
+        q_embedding = model.encode([question])
+        scores = cosine_similarity(q_embedding, chunk_embeddings)[0]
+        top_index = int(np.argmax(scores))
+        top_score = float(scores[top_index])
+        best_text = re.sub(r'\s+', ' ', chunks[top_index].strip())
+
+        if top_score >= 0.5:
+            return f"Q: {question}\nA: {best_text}\nScore: {round(top_score, 3)}"
+        else:
+            return "⚠️ No relevant answer found (score too low)."
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
+
+# ✅ Gradio UI
+with gr.Blocks() as demo:
+    gr.Markdown("# 📄 PDF QA System")
+
+    with gr.Tab("1. Upload PDF"):
+        file = gr.File(label="Upload a PDF")
+        upload_out = gr.Textbox(label="Upload Result")
+        file.change(fn=upload_pdf, inputs=file, outputs=upload_out)
+
+    with gr.Tab("2. Generate Q&A"):
+        fname = gr.Textbox(label="Enter uploaded filename")
+        qa_result = gr.Textbox(label="Q&A Output", lines=10)
+        gr.Button("Generate Q&A").click(fn=generate_qa, inputs=fname, outputs=qa_result)
+
+    with gr.Tab("3. Ask a Question"):
+        token_box = gr.Textbox(label="Enter Token ID")
+        question_box = gr.Textbox(label="Your Question")
+        answer_result = gr.Textbox(label="Answer", lines=5)
+        gr.Button("Ask").click(fn=ask_question, inputs=[token_box, question_box], outputs=answer_result)
+
+demo.launch()

database1.py

@@ -0,0 +1,46 @@
+import sqlite3
+import json
+
+class create_db:
+    def __init__(self, token, chunk_json1, filename, full_content):
+        conn = sqlite3.connect('my_database.db')
+        cursor = conn.cursor()
+
+        # Only store into this table
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS token_data (
+                token_id TEXT PRIMARY KEY,
+                chunk_data TEXT,
+                filename TEXT,
+                full_content TEXT
+            )
+        """)
+
+        chunk_json = json.dumps(chunk_json1)
+
+        try:
+            cursor.execute(
+                "INSERT INTO token_data (token_id, chunk_data, filename, full_content) VALUES (?, ?, ?, ?)",
+                (token, chunk_json, filename, full_content)
+            )
+            conn.commit()
+            print({"message": "PDF uploaded and stored successfully"})
+        except sqlite3.IntegrityError:
+            print({"error": "Token already exists."})
+
+        conn.close()
+
+    @staticmethod
+    def get_all_filenames():
+        conn = sqlite3.connect('my_database.db')
+        cursor = conn.cursor()
+        cursor.execute("SELECT filename FROM token_data")
+        rows = cursor.fetchall()
+        conn.close()
+
+        if rows:
+            return {
+                "pdfs": [{"filename": row[0]} for row in rows]
+            }
+        else:
+            return {"pdfs": []}

first1.py

@@ -0,0 +1,57 @@
+import PyPDF2
+from sentence_transformers import SentenceTransformer
+from sklearn.metrics.pairwise import cosine_similarity
+import numpy as np
+import re
+
+
+class pdf_query:
+    def __init__(self):
+        self.model = SentenceTransformer("all-MiniLM-L6-v2")
+        self.read = None
+
+    def file(self, file):
+        self.read = PyPDF2.PdfReader(file)
+
+    def extract_text(self):
+        text = ""
+        for page in self.read.pages:
+            content = page.extract_text()
+            if content:
+                text += content + "\n"
+        return text.strip()
+
+    def split_into_chunks(self, text, chunk_size=300):
+        # Split using punctuation for better sentence boundaries
+        sentences = re.split(r'(?<=[.!?])\s+', text)
+        chunks = []
+        current_chunk = ""
+
+        for sentence in sentences:
+            if len(current_chunk) + len(sentence) <= chunk_size:
+                current_chunk += sentence + " "
+            else:
+                chunks.append(current_chunk.strip())
+                current_chunk = sentence + " "
+        if current_chunk:
+            chunks.append(current_chunk.strip())
+
+        return chunks
+
+    def creat_model(self,chunks):
+        model = SentenceTransformer("all-MiniLM-L6-v2")
+        chunk_embeddings = model.encode(chunks)
+        return model,chunk_embeddings
+
+    def answer_question(self,question, chunks, chunk_embeddings,model,threshold=0.6):
+        q_embedding = model.encode([question])  # same model as above
+        scores = cosine_similarity(q_embedding, chunk_embeddings)
+        best_score = np.max(scores)
+        best_chunk_index = np.argmax(scores)
+        if best_score >= threshold:
+            best_chunk = chunks[best_chunk_index]
+            # Clean the answer
+            cleaned_answer = re.sub(r'\s+', ' ', best_chunk.strip())
+            return  cleaned_answer
+        else:
+            return {"answer": "Answer not found in PDF"}

q_generator1.py

@@ -0,0 +1,34 @@
+from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, pipeline
+
+class QGenerator:
+    def __init__(self):
+        tokenizer = AutoTokenizer.from_pretrained("valhalla/t5-small-qg-hl", use_fast=False)
+        model = AutoModelForSeq2SeqLM.from_pretrained("valhalla/t5-small-qg-hl")
+        self.qg = pipeline("text2text-generation", model=model, tokenizer=tokenizer)
+
+    def split_sentences(self, text):
+        # Simple sentence splitting (for better results, use nltk or spacy)
+        return [s.strip() for s in text.split('.') if s.strip()]
+
+    def chunk_text(self, text, chunk_size=512):
+        return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
+
+    def generate(self, text, max_questions=5):
+        questions = []
+        sentences = self.split_sentences(text)
+
+        for sentence in sentences:
+            if len(questions) >= max_questions:
+                break
+
+            input_text = f"generate question: {sentence} </s>"
+            try:
+                result = self.qg(input_text, max_length=64, num_return_sequences=1)[0]
+                question = result["generated_text"]
+                if question and question not in questions:
+                    questions.append(question)
+            except Exception as e:
+                print("Error generating question:", e)
+                continue
+
+        return questions

requirements.txt

@@ -0,0 +1,8 @@
+gradio
+transformers
+torch
+sentence-transformers
+PyPDF2
+scikit-learn
+numpy
+uuid