import streamlit as st import requests import PyPDF2 from typing import Optional, Dict, List import json from langchain.text_splitter import RecursiveCharacterTextSplitter from concurrent.futures import ThreadPoolExecutor import xml.etree.ElementTree as ET import re from datetime import datetime import time from dotenv import load_dotenv import os import pandas as pd # Load environment variables load_dotenv() PERPLEXITY_API_KEY = os.getenv("PERPLEXITY_API_KEY") PERPLEXITY_API_URL = "https://api.perplexity.ai/chat/completions" SAPLING_API_KEY = os.getenv("SAPLING_API_KEY") def call_perplexity_api(prompt: str) -> str: """Call Perplexity AI with a prompt, return the text response if successful.""" headers = { "Authorization": f"Bearer {PERPLEXITY_API_KEY}", "Content-Type": "application/json", } payload = { "model": "llama-3.1-sonar-small-128k-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, } try: response = requests.post(PERPLEXITY_API_URL, headers=headers, json=payload) response.raise_for_status() return response.json()["choices"][0]["message"]["content"] except Exception as e: st.error(f"API Error: {str(e)}") return "" def extract_text_from_pdf(pdf_file): """Extract text content from a PDF file.""" pdf_reader = PyPDF2.PdfReader(pdf_file) text = "" for page in pdf_reader.pages: text += page.extract_text() + "\n" return text def analyze_paper(text: str, category: str) -> str: """Generate a prompt and get analysis for a specific category.""" prompts = { "Journal": "In which journal was this research published:", "Journal Quality": "What is the quality or impact factor of the journal in which this research was published:", "No Of Citations": "How many times has this research paper been cited:", "Date Of Publications": "When was this research paper published:", "Title": "What is the title of this research paper:", "Abstract": "Provide a summarized version of the abstract of this paper:", "Author Keywords": "What keywords were provided by the authors for this research paper:", "Theories Used in The Paper": "What theories are utilized or referenced in this research paper:", "Context Used In the Paper": "What is the specific context or scenario used in this research:", "Methods and Material Used in This Paper": "What methods and materials are used in conducting this research:", "Antecedents and Problems": "What antecedents and problems are identified in this research:", "Decision and Frameworks To Solve the Problem": "What decision-making frameworks or solutions are proposed in this research:", "Outcomes": "What are the outcomes or results of this research:", "Study Findings": "What are the detailed findings of this research study:", "Conclusions": "What conclusions are drawn from this research:", "TSC ADO": "Provide details about the TSC ADO (Theory-Specific Constructs Applied in this research):" } if category in prompts: prompt = f"{prompts[category]}\n\nPaper text: {text[:5000]}" # Limit text to avoid token limits else: prompt = f"Analyze the following text for the category '{category}':\n\nPaper text: {text[:5000]}" return call_perplexity_api(prompt) class ResearchAssistant: def __init__(self, perplexity_key: str): self.perplexity_key = perplexity_key def chat_with_pdf(self, pdf_text: str, query: str) -> Dict: chunks = self._split_text(pdf_text) relevant_chunks = self._get_relevant_chunks(chunks, query) prompt = f"Context from PDF:\n\n{relevant_chunks}\n\nQuestion: {query}" response_text = call_perplexity_api(prompt) return {"choices": [{"message": {"content": response_text}}]} def generate_literature_review(self, topic: str) -> Dict: try: # Search arXiv for papers papers = self._search_arxiv(topic) if not papers: return {"error": "No papers found on the topic"} # Format paper information papers_summary = "\n\n".join( [ f"Paper: {p['title']}\nAuthors: {', '.join(p['authors'])}\nSummary: {p['summary']}" for p in papers ] ) prompt = f"""Generate a comprehensive literature review on '{topic}'. Based on these papers: {papers_summary} Structure the review as follows: 1. Introduction and Background 2. Current Research Trends 3. Key Findings and Themes 4. Research Gaps 5. Future Directions""" response_text = call_perplexity_api(prompt) return {"choices": [{"message": {"content": response_text}}]} except Exception as e: return {"error": f"Literature review generation failed: {str(e)}"} def ai_writer(self, outline: str, references: List[str]) -> Dict: prompt = f"""Write a research paper following this structure: Outline: {outline} References to incorporate: {json.dumps(references)} Instructions: - Follow academic writing style - Include appropriate citations - Maintain logical flow - Include introduction and conclusion""" response_text = call_perplexity_api(prompt) return {"choices": [{"message": {"content": response_text}}]} def refine_response(self, response: str, column: str) -> str: prompt = f"""Refine the following response to fit the '{column}' column in a research paper CSV format: Response: {response} Ensure the response is clear, concise, and fits the context of the column.""" refined_response = call_perplexity_api(prompt) return refined_response def paraphrase(self, text: str) -> Dict: prompt = f"""Paraphrase the following text while: - Maintaining academic tone - Preserving key meaning - Improving clarity Text: {text}""" response_text = call_perplexity_api(prompt) return {"choices": [{"message": {"content": response_text}}]} def generate_citation(self, paper_info: Dict, style: str = "APA") -> Dict: prompt = f"""Generate a {style} citation for: Title: {paper_info['title']} Authors: {', '.join(paper_info['authors'])} Year: {paper_info['year']} Follow exact {style} format guidelines.""" response_text = call_perplexity_api(prompt) return {"citation": response_text} def detect_ai_content(self, text: str) -> Dict: prompt = f"""You are an AI content detector. Analyze the text for: 1. Writing style consistency 2. Language patterns 3. Contextual coherence 4. Common AI patterns Provide a clear analysis with confidence level. Text: {text}""" response = requests.post( "https://api.sapling.ai/api/v1/aidetect", json={"key": SAPLING_API_KEY, "text": text}, ) st.info( "A score from 0 to 1 will be returned, with 0 indicating the maximum confidence that the text is human-written, and 1 indicating the maximum confidence that the text is AI-generated." ) if response.status_code == 200: return {"choices": [{"message": {"content": response.json()}}]} else: return { "error": f"Sapling API Error: {response.status_code} - {response.text}" } def _split_text(self, text: str) -> List[str]: splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, separators=["\n\n", "\n", ". ", " ", ""] ) return splitter.split_text(text) def _get_relevant_chunks(self, chunks: List[str], query: str) -> str: # Simple keyword-based relevance scoring query_words = set(query.lower().split()) scored_chunks = [] for chunk in chunks: chunk_words = set(chunk.lower().split()) score = len(query_words.intersection(chunk_words)) scored_chunks.append((score, chunk)) scored_chunks.sort(reverse=True) return "\n\n".join(chunk for _, chunk in scored_chunks[:3]) def _search_arxiv(self, topic: str) -> List[Dict]: try: query = "+AND+".join(topic.split()) url = f"http://export.arxiv.org/api/query?search_query=all:{query}&start=0&max_results=5" response = requests.get(url, timeout=10) response.raise_for_status() return self._parse_arxiv_response(response.text) except Exception as e: print(f"arXiv search failed: {str(e)}") return [] def _parse_arxiv_response(self, response_text: str) -> List[Dict]: try: root = ET.fromstring(response_text) papers = [] for entry in root.findall("{http://www.w3.org/2005/Atom}entry"): paper = { "id": entry.find("{http://www.w3.org/2005/Atom}id").text, "title": entry.find( "{http://www.w3.org/2005/Atom}title" ).text.strip(), "summary": entry.find( "{http://www.w3.org/2005/Atom}summary" ).text.strip(), "authors": [ author.find("{http://www.w3.org/2005/Atom}name").text.strip() for author in entry.findall( "{http://www.w3.org/2005/Atom}author" ) ], "published": entry.find( "{http://www.w3.org/2005/Atom}published" ).text[:10], } papers.append(paper) return papers except Exception as e: print(f"arXiv response parsing failed: {str(e)}") return [] def main(): # st.set_page_config(page_title="Research Assistant", layout="wide") st.title("Research Copilot") if not PERPLEXITY_API_KEY: st.warning("Perplexity API key not found in environment variables.") return assistant = ResearchAssistant(PERPLEXITY_API_KEY) tabs = st.tabs( [ "Chat with PDF", "Literature Review", "AI Writer", "Extract Data", "Paraphraser", "Citation Generator", "AI Detector", ] ) with tabs[0]: # Chat with PDF st.header("Chat with PDF") # File uploader with clear button col1, col2 = st.columns([3, 1]) with col1: uploaded_file = st.file_uploader("Upload PDF", type="pdf", key="pdf_chat") with col2: if st.button("Clear PDF"): st.session_state.pop("pdf_text", None) st.rerun() if uploaded_file: if "pdf_text" not in st.session_state: with st.spinner("Processing PDF..."): reader = PyPDF2.PdfReader(uploaded_file) st.session_state.pdf_text = "" for page in reader.pages: st.session_state.pdf_text += page.extract_text() st.success("PDF processed successfully!") query = st.text_input("Ask a question about the PDF") if query: with st.spinner("Analyzing..."): response = assistant.chat_with_pdf(st.session_state.pdf_text, query) if "error" in response: st.error(response["error"]) else: st.write(response["choices"][0]["message"]["content"]) with tabs[1]: # Literature Review st.header("Literature Review") topic = st.text_input("Enter research topic") if st.button("Generate Review") and topic: with st.spinner("Generating literature review..."): review = assistant.generate_literature_review(topic) if "error" in review: st.error(review["error"]) else: st.write(review["choices"][0]["message"]["content"]) with tabs[2]: # AI Writer st.header("AI Writer") outline = st.text_area("Enter paper outline") references = st.text_area("Enter references (one per line)") if st.button("Generate Paper") and outline: with st.spinner("Writing paper..."): paper = assistant.ai_writer(outline, references.split("\n")) if "error" in paper: st.error(paper["error"]) else: st.write(paper["choices"][0]["message"]["content"]) with tabs[3]: # Extract Data st.header("Extract Data") uploaded_files = st.file_uploader( "Upload multiple PDF files", type="pdf", accept_multiple_files=True ) if 'categories' not in st.session_state: st.session_state.categories = [ "Journal", "Journal Quality", "No Of Citations", "Date Of Publications", "Title", "Abstract", "Author Keywords", "Theories Used in The Paper", "Context Used In the Paper", "Methods and Material Used in This Paper", "Antecedents and Problems", "Decision and Frameworks To Solve the Problem", "Outcomes", "Study Findings", "Conclusions", "TSC ADO" ] # Display current categories st.write("### Current Categories") st.write(st.session_state.categories) # Input to add new category new_category = st.text_input("Add a new category") if st.button("Add Category"): if new_category.strip(): # Check if input is not empty if new_category not in st.session_state.categories: # Avoid duplicates st.session_state.categories.append(new_category) st.success(f"Category '{new_category}' added!") else: st.warning(f"Category '{new_category}' already exists!") else: st.error("Category cannot be empty!") # Button to add the category # Display updated categories st.write("### Updated Categories") st.write(st.session_state.categories) if uploaded_files: if st.button("Process Papers"): # Initialize progress bar progress_bar = st.progress(0) status_text = st.empty() # Initialize results dictionary results = [] # Define categories # categories = [ # "Summarized Abstract", # "Results", # "Summarized Introduction", # "Methods Used", # "Literature Survey", # "Limitations", # "Contributions", # "Practical Implications", # "Objectives", # "Findings", # "Future Research", # "Dependent Variables", # "Independent Variables", # "Dataset", # "Problem Statement", # "Challenges", # "Applications", # ] # # Display current categories # st.write("### Current Categories") # st.write(categories) # # Input to add new category # new_category = st.text_input("Add a new category") # # Button to add the category # if st.button("Add Category"): # if new_category.strip(): # Check if input is not empty # if new_category not in categories: # Avoid duplicates # categories.append(new_category) # st.success(f"Category '{new_category}' added!") # else: # st.warning(f"Category '{new_category}' already exists!") # else: # st.error("Category cannot be empty!") # # Display updated categories # st.write("### Updated Categories") # st.write(categories) # Process each file for i, file in enumerate(uploaded_files): status_text.text(f"Processing {file.name}...") # Extract text from PDF text = extract_text_from_pdf(file) # Initialize paper results paper_results = {"Filename": file.name} # Analyze each category for j, category in enumerate(st.session_state.categories): status_text.text(f"Processing {file.name} - {category}") paper_results[category] = analyze_paper(text, category) # Update progress progress = (i * len(st.session_state.categories) + j + 1) / ( len(uploaded_files) * len(st.session_state.categories) ) progress_bar.progress(progress) # Add small delay to avoid API rate limits time.sleep(1) results.append(paper_results) # Create DataFrame df = pd.DataFrame(results) # Convert DataFrame to CSV csv = df.to_csv(index=False) # Create download button st.download_button( label="Download Results as CSV", data=csv, file_name="research_papers_analysis.csv", mime="text/csv", ) # Display results in the app st.subheader("Analysis Results") st.dataframe(df) status_text.text("Processing complete!") progress_bar.progress(1.0) with tabs[4]: # Paraphraser st.header("Paraphraser") text = st.text_area("Enter text to paraphrase") if st.button("Paraphrase") and text: with st.spinner("Paraphrasing..."): result = assistant.paraphrase(text) if "error" in result: st.error(result["error"]) else: st.write(result["choices"][0]["message"]["content"]) with tabs[5]: # Citation Generator st.header("Citation Generator") col1, col2 = st.columns(2) with col1: title = st.text_input("Paper Title") authors = st.text_input("Authors (comma-separated)") with col2: year = st.text_input("Year") style = st.selectbox("Citation Style", ["APA", "MLA", "Chicago"]) if st.button("Generate Citation") and title: with st.spinner("Generating citation..."): citation = assistant.generate_citation( { "title": title, "authors": [a.strip() for a in authors.split(",")], "year": year, }, style, ) if "error" in citation: st.error(citation["error"]) else: st.code(citation["citation"], language="text") with tabs[6]: # AI Detector st.header("AI Detector") text = st.text_area("Enter text to analyze") if st.button("Detect AI Content") and text: with st.spinner("Analyzing..."): result = assistant.detect_ai_content(text) if "error" in result: st.error(result["error"]) else: st.write(result["choices"][0]["message"]["content"]) if __name__ == "__main__": main()