{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "e549bafd-78b1-4a83-80b4-2cb597efff79",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"import json\n",
"from google.oauth2 import service_account\n",
"from googleapiclient.discovery import build\n",
"from googleapiclient.http import MediaIoBaseDownload\n",
"import openai\n",
"from dotenv import load_dotenv, dotenv_values\n",
"import io"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "04361235-7896-4439-9d04-1400e043528b",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"True"
]
},
"execution_count": 2,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"load_dotenv()"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "9ae411c5-c84b-4bfd-b089-69b5c5ba70ae",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"OPENAI_API_KEY\n",
"ANTHROPIC_API_KEY\n",
"GOOGLE_SERVICE_ACCOUNT_FILE\n"
]
}
],
"source": [
"config = dotenv_values(\".env\")\n",
"for key in config.keys():\n",
" print(key)"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "7622a0e4-64a6-4848-b588-bd65d56c55e0",
"metadata": {},
"outputs": [],
"source": [
"from openai import OpenAI\n",
"openai.api_key = os.getenv('OPENAI_API_KEY')\n",
"openai = OpenAI(api_key = openai.api_key)"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "2ead7c59-1f11-478a-bb69-2928ddc38901",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Hello! How can I assist you today?\n"
]
}
],
"source": [
"response = openai.chat.completions.create(\n",
" model = \"gpt-4o-mini\",\n",
" messages = [\n",
" {\"role\":\"system\", \"content\":\"you are a helpful assistant\"},\n",
" {\"role\":\"user\", \"content\":\"hi\"}\n",
" ])\n",
"\n",
"reply = response.choices[0].message.content\n",
"print(reply)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "b3bcbada-6a72-4cc8-a166-d2e596cd1fc4",
"metadata": {},
"outputs": [],
"source": [
"service_account_file_path = os.getenv(\"GOOGLE_SERVICE_ACCOUNT_FILE\")"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "81c59a6d-0831-4bff-b3fc-fbe3d4cc1e31",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"openai activated\n",
"service_account_file_path activated\n"
]
}
],
"source": [
"#troubleshoot\n",
"if openai is None:\n",
" print(\"openai not activated\")\n",
"else: \n",
" print (\"openai activated\")\n",
"\n",
"if service_account_file_path is None:\n",
" print(\"service_account_file_path not activated\")\n",
"else: \n",
" print (\"service_account_file_path activated\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a70f32aa-9e43-4175-8cca-d6af723aef91",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 8,
"id": "0efe325d-badd-4533-affe-47d572ef128e",
"metadata": {},
"outputs": [],
"source": [
"class GPTDriveIntegration:\n",
" def __init__(self):\n",
" # Initialize Google Drive API\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE'),\n",
" scopes=['https://www.googleapis.com/auth/drive.readonly']\n",
" )\n",
" self.drive_service = build('drive', 'v3', credentials=self.credentials)\n",
" \n",
" # Initialize OpenAI\n",
" openai.api_key = os.getenv('OPENAI_API_KEY')\n",
" \n",
" def search_files(self, query, file_types=None):\n",
" \"\"\"Search for files in Google Drive\"\"\"\n",
" search_query = f\"name contains '{query}'\"\n",
" \n",
" if file_types:\n",
" type_queries = []\n",
" for file_type in file_types:\n",
" if file_type.lower() == 'pdf':\n",
" type_queries.append(\"mimeType='application/pdf'\")\n",
" elif file_type.lower() in ['doc', 'docx']:\n",
" type_queries.append(\"mimeType contains 'document'\")\n",
" elif file_type.lower() in ['xls', 'xlsx']:\n",
" type_queries.append(\"mimeType contains 'spreadsheet'\")\n",
" \n",
" if type_queries:\n",
" search_query += f\" and ({' or '.join(type_queries)})\"\n",
" \n",
" results = self.drive_service.files().list(\n",
" q=search_query,\n",
" fields=\"files(id, name, mimeType, size)\"\n",
" ).execute()\n",
" \n",
" return results.get('files', [])\n",
" \n",
" def get_file_content(self, file_id, mime_type):\n",
" \"\"\"Download and extract text content from file\"\"\"\n",
" try:\n",
" if 'text' in mime_type or 'document' in mime_type:\n",
" # For Google Docs, export as plain text\n",
" if 'document' in mime_type:\n",
" request = self.drive_service.files().export_media(\n",
" fileId=file_id, mimeType='text/plain'\n",
" )\n",
" else:\n",
" request = self.drive_service.files().get_media(fileId=file_id)\n",
" \n",
" file_content = io.BytesIO()\n",
" downloader = MediaIoBaseDownload(file_content, request)\n",
" done = False\n",
" while done is False:\n",
" status, done = downloader.next_chunk()\n",
" \n",
" return file_content.getvalue().decode('utf-8')\n",
" \n",
" elif 'spreadsheet' in mime_type:\n",
" # For Google Sheets, export as CSV\n",
" request = self.drive_service.files().export_media(\n",
" fileId=file_id, mimeType='text/csv'\n",
" )\n",
" file_content = io.BytesIO()\n",
" downloader = MediaIoBaseDownload(file_content, request)\n",
" done = False\n",
" while done is False:\n",
" status, done = downloader.next_chunk()\n",
" \n",
" return file_content.getvalue().decode('utf-8')\n",
" \n",
" elif mime_type == 'application/pdf':\n",
" # For PDF files, download binary content and extract text\n",
" request = self.drive_service.files().get_media(fileId=file_id)\n",
" file_content = io.BytesIO()\n",
" downloader = MediaIoBaseDownload(file_content, request)\n",
" done = False\n",
" while done is False:\n",
" status, done = downloader.next_chunk()\n",
" \n",
" # Extract text from PDF using PyPDF2 or pdfplumber\n",
" file_content.seek(0) # Reset buffer position\n",
" \n",
" # Option 1: Using PyPDF2\n",
" try:\n",
" import PyPDF2\n",
" pdf_reader = PyPDF2.PdfReader(file_content)\n",
" text = \"\"\n",
" for page in pdf_reader.pages:\n",
" text += page.extract_text() + \"\\n\"\n",
" return text\n",
" except ImportError:\n",
" pass\n",
" \n",
" # Option 2: Using pdfplumber (better for complex PDFs)\n",
" try:\n",
" import pdfplumber\n",
" text = \"\"\n",
" with pdfplumber.open(file_content) as pdf:\n",
" for page in pdf.pages:\n",
" page_text = page.extract_text()\n",
" if page_text:\n",
" text += page_text + \"\\n\"\n",
" return text\n",
" except ImportError:\n",
" pass\n",
" \n",
" # Option 3: Using pymupdf (fitz) - fastest option\n",
" try:\n",
" import fitz # pymupdf\n",
" pdf_document = fitz.open(stream=file_content.read(), filetype=\"pdf\")\n",
" text = \"\"\n",
" for page_num in range(pdf_document.page_count):\n",
" page = pdf_document[page_num]\n",
" text += page.get_text() + \"\\n\"\n",
" pdf_document.close()\n",
" return text\n",
" except ImportError:\n",
" pass\n",
" \n",
" return \"PDF text extraction requires PyPDF2, pdfplumber, or pymupdf library\"\n",
" \n",
" else:\n",
" return \"File type not supported for text extraction\"\n",
" \n",
" except Exception as e:\n",
" return f\"Error reading file: {str(e)}\"\n",
" \n",
" def query_gpt_with_context(self, user_query, file_contents):\n",
" \"\"\"Send query to GPT with file context\"\"\"\n",
" context = \"\\n\\n\".join([\n",
" f\"File: {content['name']}\\nContent: {content['text'][:2000]}...\"\n",
" for content in file_contents\n",
" ])\n",
" \n",
" messages = [\n",
" {\n",
" \"role\": \"system\", \n",
" \"content\": \"\"\"\n",
" You are an AI assistant that can analyze documents from Google Drive. \n",
" Use the provided file contents to answer user questions.\"\"\"\n",
" },\n",
" {\n",
" \"role\": \"user\", \n",
" \"content\": f\"Context from Google Drive files:\\n{context}\\n\\nUser Question: {user_query}\"\n",
" }\n",
" ]\n",
" \n",
" response = openai.chat.completions.create(\n",
" model=\"gpt-4o-mini\",\n",
" messages=messages,\n",
" max_tokens=1000\n",
" )\n",
" \n",
" return response.choices[0].message.content\n",
" \n",
" def process_query(self, user_query, search_terms=None):\n",
" \"\"\"Main function to process user queries\"\"\"\n",
" # Extract search terms from query if not provided\n",
" if not search_terms:\n",
" search_terms = user_query.split()[:3] # Simple extraction\n",
" \n",
" # Search for relevant files\n",
" files = []\n",
" for term in search_terms:\n",
" files.extend(self.search_files(term))\n",
" \n",
" # Remove duplicates\n",
" unique_files = {f['id']: f for f in files}.values()\n",
" \n",
" # Get content from top 3 most relevant files\n",
" file_contents = []\n",
" for file in list(unique_files)[:3]:\n",
" content = self.get_file_content(file['id'], file['mimeType'])\n",
" file_contents.append({\n",
" 'name': file['name'],\n",
" 'text': content\n",
" })\n",
" \n",
" # Query GPT with context\n",
" if file_contents:\n",
" response = self.query_gpt_with_context(user_query, file_contents)\n",
" return {\n",
" 'answer': response,\n",
" 'sources': [f['name'] for f in file_contents]\n",
" }\n",
" else:\n",
" return {\n",
" 'answer': \"No relevant files found in your Google Drive.\",\n",
" 'sources': []\n",
" }"
]
},
{
"cell_type": "code",
"execution_count": 9,
"id": "3c2c1ccf-9ade-482d-a170-978e97bc1c08",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Answer: The transmission of nerves is called \"nerve conduction.\" This process involves the propagation of electrical impulses along the nerve fibers, allowing for communication between different parts of the body.\n",
"Sources: ['Neuro Note Dr Clement.docx']\n"
]
}
],
"source": [
"if __name__ == \"__main__\":\n",
" integration = GPTDriveIntegration()\n",
" \n",
" # Test query\n",
" result = integration.process_query(\n",
" \"The transmission of nerves is called?\",\n",
" search_terms=[\"nerves\", \"Dr Clement\"]\n",
" )\n",
" \n",
" print(\"Answer:\", result['answer'])\n",
" print(\"Sources:\", result['sources'])"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "120e7c93-b38a-4c89-8e76-5b0170d22548",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "61e976a6-1fe4-45de-a63d-3cf849eedbe1",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 11,
"id": "c514b8af-1af4-4497-9044-139a71aedd36",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"* Running on local URL: http://127.0.0.1:7860\n",
"* Running on public URL: https://5320171f6af3e2eef9.gradio.live\n",
"\n",
"This share link expires in 1 week. For free permanent hosting and GPU upgrades, run `gradio deploy` from the terminal in the working directory to deploy to Hugging Face Spaces (https://huggingface.co/spaces)\n"
]
},
{
"data": {
"text/html": [
""
],
"text/plain": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Keyboard interruption in main thread... closing server.\n",
"Killing tunnel 127.0.0.1:7860 <> https://5320171f6af3e2eef9.gradio.live\n"
]
}
],
"source": [
"gpt_drive = GPTDriveIntegration()\n",
"\n",
"def process_user_query(query, search_terms_input):\n",
" \"\"\"Process user query and return formatted response\"\"\"\n",
" if not query.strip():\n",
" return \"Please enter a question.\", \"\"\n",
" \n",
" # Parse search terms if provided\n",
" search_terms = None\n",
" # if search_terms_input.strip():\n",
" # search_terms = [term.strip() for term in search_terms_input.split(',')]\n",
" \n",
" # Process the query\n",
" result = gpt_drive.process_query(query, search_terms)\n",
" \n",
" # Format the response\n",
" answer = result['answer']\n",
" sources = result['sources']\n",
" \n",
" sources_text = \"\"\n",
" if sources:\n",
" sources_text = \"**Sources used:**\\n\" + \"\\n\".join([f\"• {source}\" for source in sources])\n",
" \n",
" return answer, sources_text\n",
"\n",
"def check_setup():\n",
" \"\"\"Check if the APIs are properly configured\"\"\"\n",
" status_messages = []\n",
" \n",
" # Check Google Drive API\n",
" if gpt_drive.drive_initialized:\n",
" status_messages.append(\"✅ Google Drive API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ Google Drive API: {getattr(gpt_drive, 'drive_error', 'Not configured')}\")\n",
" \n",
" # Check OpenAI API\n",
" if gpt_drive.openai_initialized:\n",
" status_messages.append(\"✅ OpenAI API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ OpenAI API: {getattr(gpt_drive, 'openai_error', 'Not configured')}\")\n",
" \n",
" return \"\\n\".join(status_messages)\n",
"\n",
"# Create Gradio interface\n",
"with gr.Blocks(title=\"Augusta's Anatomy Reading Assistant\", theme=gr.themes.Soft()) as app:\n",
" gr.Markdown(\"# 🤖 Augusta's Anatomy bot\")\n",
" gr.Markdown(\"Ask questions about your anatomy books using AI!\")\n",
" \n",
" with gr.Row():\n",
" with gr.Column(scale=2):\n",
" # Main query interface\n",
" with gr.Group():\n",
" gr.Markdown(\"### Ask a Question\")\n",
" query_input = gr.Textbox(\n",
" label=\"Your Question\",\n",
" placeholder=\"Ask me any question about your anatomy books?\",\n",
" lines=3\n",
" )\n",
" \n",
" search_terms_input = gr.Textbox(\n",
" label=\"Search Terms (optional)\",\n",
" placeholder=\"Enter comma-separated terms to search for specific files\",\n",
" lines=1\n",
" )\n",
" \n",
" submit_btn = gr.Button(\"Search & Ask\", variant=\"primary\", size=\"lg\")\n",
" \n",
" # Results section\n",
" with gr.Group():\n",
" gr.Markdown(\"### Answer\")\n",
" answer_output = gr.Textbox(\n",
" label=\"AI Response\",\n",
" lines=10,\n",
" interactive=False\n",
" )\n",
" \n",
" sources_output = gr.Textbox(\n",
" label=\"Sources\",\n",
" lines=3,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Column(scale=1):\n",
" # Status and setup info\n",
" with gr.Group():\n",
" gr.Markdown(\"### System Status\")\n",
" status_btn = gr.Button(\"Check Status\", size=\"sm\")\n",
" status_output = gr.Textbox(\n",
" label=\"API Status\",\n",
" lines=4,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Group():\n",
" gr.Markdown(\"### Setup Instructions\")\n",
" gr.Markdown(\"\"\"\n",
" **Important Notes:**\n",
" 1.Only documents shared with it, it can answer\n",
" \n",
" **File Types Supported:**\n",
" - Google Docs\n",
" - Google Sheets \n",
" - PDF files\n",
" - Text files\n",
" \n",
" **Tips:**\n",
" - Use specific search terms for better results\n",
" - The system searches the top 3 most relevant files\n",
" - Ask clear, specific questions for better answers\n",
" \"\"\")\n",
" \n",
" # Event handlers\n",
" submit_btn.click(\n",
" fn=process_user_query,\n",
" inputs=[query_input, search_terms_input],\n",
" outputs=[answer_output, sources_output]\n",
" )\n",
" \n",
" status_btn.click(\n",
" fn=check_setup,\n",
" outputs=status_output\n",
" )\n",
" \n",
" # Example queries\n",
" with gr.Row():\n",
" gr.Examples(\n",
" examples=[\n",
" [\"What is morbid Anatomy?\", \"morbid, Anatomy\"],\n",
" [\"The transmission of nerves from one neuron to another is as a result of what?\", \"neuron, nerves, Dr Clement\"],\n",
" ],\n",
" inputs=[query_input, search_terms_input],\n",
" )\n",
"\n",
"# Launch the app\n",
"if __name__ == \"__main__\":\n",
" app.launch(\n",
" share=True,debug =True)"
]
},
{
"cell_type": "code",
"execution_count": 42,
"id": "593c4f63-cdda-46e8-aad6-9f9d7db66615",
"metadata": {},
"outputs": [],
"source": [
"#Gradio\n",
"# !pip install gradio"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "859060a9-56f9-49e9-bb41-04ccf7b10d3f",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "d893999c-4f3f-4306-8b0b-072e3fbb8236",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 10,
"id": "01b70311-ff22-4c88-bf22-64dc441afaab",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"C:\\Users\\Uche Buzz\\anaconda3\\envs\\RAG\\Lib\\site-packages\\tqdm\\auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"* Running on local URL: http://127.0.0.1:7860\n",
"* Running on public URL: https://8fac475cac6193423b.gradio.live\n",
"\n",
"This share link expires in 1 week. For free permanent hosting and GPU upgrades, run `gradio deploy` from the terminal in the working directory to deploy to Hugging Face Spaces (https://huggingface.co/spaces)\n"
]
},
{
"data": {
"text/html": [
""
],
"text/plain": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Keyboard interruption in main thread... closing server.\n",
"Killing tunnel 127.0.0.1:7860 <> https://8fac475cac6193423b.gradio.live\n"
]
}
],
"source": [
"import gradio as gr\n",
"import os\n",
"import io\n",
"import openai\n",
"from google.oauth2 import service_account\n",
"from googleapiclient.discovery import build\n",
"from googleapiclient.http import MediaIoBaseDownload\n",
"\n",
"class GPTDriveIntegration:\n",
" def __init__(self):\n",
" # Initialize Google Drive API\n",
" try:\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE'),\n",
" scopes=['https://www.googleapis.com/auth/drive.readonly']\n",
" )\n",
" self.drive_service = build('drive', 'v3', credentials=self.credentials)\n",
" self.drive_initialized = True\n",
" except Exception as e:\n",
" self.drive_initialized = False\n",
" self.drive_error = str(e)\n",
" \n",
" # Initialize OpenAI\n",
" try:\n",
" openai.api_key = os.getenv('OPENAI_API_KEY')\n",
" self.openai_initialized = True\n",
" except Exception as e:\n",
" self.openai_initialized = False\n",
" self.openai_error = str(e)\n",
" \n",
" def search_files(self, query, file_types=None):\n",
" \"\"\"Search for files in Google Drive\"\"\"\n",
" if not self.drive_initialized:\n",
" return []\n",
" \n",
" search_query = f\"name contains '{query}'\"\n",
" \n",
" if file_types:\n",
" type_queries = []\n",
" for file_type in file_types:\n",
" if file_type.lower() == 'pdf':\n",
" type_queries.append(\"mimeType='application/pdf'\")\n",
" elif file_type.lower() in ['doc', 'docx']:\n",
" type_queries.append(\"mimeType contains 'document'\")\n",
" elif file_type.lower() in ['xls', 'xlsx']:\n",
" type_queries.append(\"mimeType contains 'spreadsheet'\")\n",
" \n",
" if type_queries:\n",
" search_query += f\" and ({' or '.join(type_queries)})\"\n",
" \n",
" try:\n",
" results = self.drive_service.files().list(\n",
" q=search_query,\n",
" fields=\"files(id, name, mimeType, size)\"\n",
" ).execute()\n",
" \n",
" return results.get('files', [])\n",
" except Exception as e:\n",
" return []\n",
" \n",
" def get_file_content(self, file_id, mime_type):\n",
" \"\"\"Download and extract text content from file\"\"\"\n",
" try:\n",
" if 'text' in mime_type or 'document' in mime_type:\n",
" # For Google Docs, export as plain text\n",
" if 'document' in mime_type:\n",
" request = self.drive_service.files().export_media(\n",
" fileId=file_id, mimeType='text/plain'\n",
" )\n",
" else:\n",
" request = self.drive_service.files().get_media(fileId=file_id)\n",
" \n",
" file_content = io.BytesIO()\n",
" downloader = MediaIoBaseDownload(file_content, request)\n",
" done = False\n",
" while done is False:\n",
" status, done = downloader.next_chunk()\n",
" \n",
" return file_content.getvalue().decode('utf-8')\n",
" \n",
" elif 'spreadsheet' in mime_type:\n",
" # For Google Sheets, export as CSV\n",
" request = self.drive_service.files().export_media(\n",
" fileId=file_id, mimeType='text/csv'\n",
" )\n",
" file_content = io.BytesIO()\n",
" downloader = MediaIoBaseDownload(file_content, request)\n",
" done = False\n",
" while done is False:\n",
" status, done = downloader.next_chunk()\n",
" \n",
" return file_content.getvalue().decode('utf-8')\n",
" \n",
" else:\n",
" return \"File type not supported for text extraction\"\n",
" \n",
" except Exception as e:\n",
" return f\"Error reading file: {str(e)}\"\n",
" \n",
" def query_gpt_with_context(self, user_query, file_contents):\n",
" \"\"\"Send query to GPT with file context\"\"\"\n",
" context = \"\\n\\n\".join([\n",
" f\"File: {content['name']}\\nContent: {content['text'][:2000]}...\"\n",
" for content in file_contents\n",
" ])\n",
" \n",
" messages = [\n",
" {\n",
" \"role\": \"system\", \n",
" \"content\": \"\"\"\n",
" You are an AI assistant that can analyze anatomy documents from Google Drive. \n",
" Use the provided file contents to answer user questions. Always answer straightforwardly. Don't Hallucinate.\n",
" Always end with 'Do you have any other question?'\n",
" \n",
" \"\"\"\n",
" },\n",
" {\n",
" \"role\": \"user\", \n",
" \"content\": f\"Context from Google Drive files:\\n{context}\\n\\nUser Question: {user_query}\"\n",
" }\n",
" ]\n",
" \n",
" try:\n",
" response = openai.chat.completions.create(\n",
" model=\"gpt-4o-mini\",\n",
" messages=messages,\n",
" max_tokens=1000\n",
" )\n",
" \n",
" return response.choices[0].message.content\n",
" except Exception as e:\n",
" return f\"Error querying OpenAI: {str(e)}\"\n",
" \n",
" def process_query(self, user_query, search_terms=None):\n",
" \"\"\"Main function to process user queries\"\"\"\n",
" # Check if services are initialized\n",
" if not self.drive_initialized:\n",
" return {\n",
" 'answer': f\"Google Drive API not initialized: {getattr(self, 'drive_error', 'Unknown error')}\",\n",
" 'sources': []\n",
" }\n",
" \n",
" if not self.openai_initialized:\n",
" return {\n",
" 'answer': f\"OpenAI API not initialized: {getattr(self, 'openai_error', 'Unknown error')}\",\n",
" 'sources': []\n",
" }\n",
" \n",
" # Extract search terms from query if not provided\n",
" if not search_terms:\n",
" search_terms = user_query.split()[:3] # Simple extraction\n",
" \n",
" # Search for relevant files\n",
" files = []\n",
" for term in search_terms:\n",
" files.extend(self.search_files(term))\n",
" \n",
" # Remove duplicates\n",
" unique_files = {f['id']: f for f in files}.values()\n",
" \n",
" # Get content from top 3 most relevant files\n",
" file_contents = []\n",
" for file in list(unique_files)[:3]:\n",
" content = self.get_file_content(file['id'], file['mimeType'])\n",
" file_contents.append({\n",
" 'name': file['name'],\n",
" 'text': content\n",
" })\n",
" \n",
" # Query GPT with context\n",
" if file_contents:\n",
" response = self.query_gpt_with_context(user_query, file_contents)\n",
" return {\n",
" 'answer': response,\n",
" 'sources': [f['name'] for f in file_contents]\n",
" }\n",
" else:\n",
" return {\n",
" 'answer': \"No relevant files found in your Google Drive.\",\n",
" 'sources': []\n",
" }\n",
"\n",
"# Initialize the GPT Drive Integration\n",
"gpt_drive = GPTDriveIntegration()\n",
"\n",
"def process_user_query(query, search_terms_input):\n",
" \"\"\"Process user query and return formatted response\"\"\"\n",
" if not query.strip():\n",
" return \"Please enter a question.\", \"\"\n",
" \n",
" # Parse search terms if provided\n",
" search_terms = None\n",
" if search_terms_input.strip():\n",
" search_terms = [term.strip() for term in search_terms_input.split(',')]\n",
" \n",
" # Process the query\n",
" result = gpt_drive.process_query(query, search_terms)\n",
" \n",
" # Format the response\n",
" answer = result['answer']\n",
" sources = result['sources']\n",
" \n",
" sources_text = \"\"\n",
" if sources:\n",
" sources_text = \"**Sources used:**\\n\" + \"\\n\".join([f\"• {source}\" for source in sources])\n",
" \n",
" return answer, sources_text\n",
"\n",
"def check_setup():\n",
" \"\"\"Check if the APIs are properly configured\"\"\"\n",
" status_messages = []\n",
" \n",
" # Check Google Drive API\n",
" if gpt_drive.drive_initialized:\n",
" status_messages.append(\"✅ Google Drive API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ Google Drive API: {getattr(gpt_drive, 'drive_error', 'Not configured')}\")\n",
" \n",
" # Check OpenAI API\n",
" if gpt_drive.openai_initialized:\n",
" status_messages.append(\"✅ OpenAI API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ OpenAI API: {getattr(gpt_drive, 'openai_error', 'Not configured')}\")\n",
" \n",
" return \"\\n\".join(status_messages)\n",
"\n",
"# Create Gradio interface\n",
"with gr.Blocks(title=\"Augusta's Anatomy Reading Assistant\", theme=gr.themes.Soft()) as app:\n",
" gr.Markdown(\"# 🤖 Augusta's Anatomy bot\")\n",
" gr.Markdown(\"Ask questions about your anatomy books using AI!\")\n",
" \n",
" with gr.Row():\n",
" with gr.Column(scale=2):\n",
" # Main query interface\n",
" with gr.Group():\n",
" gr.Markdown(\"### Ask a Question\")\n",
" query_input = gr.Textbox(\n",
" label=\"Your Question\",\n",
" placeholder=\"Ask me any question about your anatomy books?\",\n",
" lines=3\n",
" )\n",
" \n",
" search_terms_input = gr.Textbox(\n",
" label=\"Search Terms (optional)\",\n",
" placeholder=\"Enter comma-separated terms to search for specific files\",\n",
" lines=1\n",
" )\n",
" \n",
" submit_btn = gr.Button(\"Search & Ask\", variant=\"primary\", size=\"lg\")\n",
" \n",
" # Results section\n",
" with gr.Group():\n",
" gr.Markdown(\"### Answer\")\n",
" answer_output = gr.Textbox(\n",
" label=\"AI Response\",\n",
" lines=10,\n",
" interactive=False\n",
" )\n",
" \n",
" sources_output = gr.Textbox(\n",
" label=\"Sources\",\n",
" lines=3,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Column(scale=1):\n",
" # Status and setup info\n",
" with gr.Group():\n",
" gr.Markdown(\"### System Status\")\n",
" status_btn = gr.Button(\"Check Status\", size=\"sm\")\n",
" status_output = gr.Textbox(\n",
" label=\"API Status\",\n",
" lines=4,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Group():\n",
" gr.Markdown(\"### Setup Instructions\")\n",
" gr.Markdown(\"\"\"\n",
" **Important Notes:**\n",
" 1.Only documents shared with it, it can answer\n",
" \n",
" **File Types Supported:**\n",
" - Google Docs\n",
" - Google Sheets \n",
" - PDF files\n",
" - Text files\n",
" \n",
" **Tips:**\n",
" - Use specific search terms for better results\n",
" - The system searches the top 3 most relevant files\n",
" - Ask clear, specific questions for better answers\n",
" \"\"\")\n",
" \n",
" # Event handlers\n",
" submit_btn.click(\n",
" fn=process_user_query,\n",
" inputs=[query_input, search_terms_input],\n",
" outputs=[answer_output, sources_output]\n",
" )\n",
" \n",
" status_btn.click(\n",
" fn=check_setup,\n",
" outputs=status_output\n",
" )\n",
" \n",
" # Example queries\n",
" with gr.Row():\n",
" gr.Examples(\n",
" examples=[\n",
" [\"What is morbid Anatomy?\", \"morbid, Anatomy\"],\n",
" [\"The transmission of nerves from one neuron to another is as a result of what?\", \"neuron, nerves, Dr Clement\"],\n",
" ],\n",
" inputs=[query_input, search_terms_input],\n",
" )\n",
"\n",
"# Launch the app\n",
"if __name__ == \"__main__\":\n",
" app.launch(\n",
" share=True,\n",
" debug=True\n",
" )"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a6fd9c8c-e8e9-4b1c-8d21-d7ea2c42303a",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "95c85a4f-470f-4858-8172-78d7f9d04d1a",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "f237c114-7924-4074-9c11-c7ba13833293",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "a2f9c7ae-35bb-4261-adc4-f24c6ab5c8a1",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "aaa446b0-a14e-4c98-b466-f503a521f052",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "5770d683-7bfc-49e9-a3c4-a67060e27620",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "2ef7744c-1346-4692-ac28-013ae0e8c4ac",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "08643ace-f94f-4a4a-bbc1-3289b7fc29a1",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "70849fd3-6f4e-4c57-b780-bda53db83e14",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "c18d5d20-439a-45a7-a847-25f659fd4a10",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 11,
"id": "634b9787-d493-46e5-8114-0851c6172ed6",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"🔍 Starting Google Drive API Diagnostic...\n",
"==================================================\n",
"\n",
"1️⃣ Checking service account file...\n",
"✅ Service account file is valid\n",
" 📧 Service account email: rag-base@rag-system-463320.iam.gserviceaccount.com\n",
" 🏗️ Project ID: rag-system-463320\n",
"\n",
"2️⃣ Checking credentials...\n",
"✅ Credentials created successfully\n",
"\n",
"3️⃣ Testing API connection...\n",
"✅ Successfully connected to Google Drive API\n",
" 👤 Connected as: rag-base@rag-system-463320.iam.gserviceaccount.com\n",
"\n",
"4️⃣ Checking basic permissions...\n",
"✅ Can access Drive API - found 0 files in test query\n",
"⚠️ No files found - this might mean:\n",
" • Service account has no shared files\n",
" • No files are shared with the service account\n",
"\n",
"5️⃣ Testing access to 'Blue berry' folder...\n",
"❌ Folder 'Blue berry' not found or not accessible\n",
"💡 Possible solutions:\n",
" • Make sure the folder exists in Google Drive\n",
" • Share the folder with your service account email\n",
" • Check folder name spelling (case sensitive)\n",
"\n",
"✅ All basic checks passed!\n"
]
}
],
"source": [
"import json\n",
"from google.auth.exceptions import RefreshError\n",
"from googleapiclient.errors import HttpError\n",
"\n",
"class GPTDriveTroubleshooter:\n",
" def __init__(self, service_account_file_path):\n",
" self.service_account_file = service_account_file_path\n",
" self.credentials = None\n",
" self.drive_service = None\n",
" \n",
" def run_full_diagnostic(self):\n",
" \"\"\"Run complete diagnostic check\"\"\"\n",
" print(\"🔍 Starting Google Drive API Diagnostic...\")\n",
" print(\"=\" * 50)\n",
" \n",
" # Step 1: Check service account file\n",
" if not self.check_service_account_file():\n",
" return False\n",
" \n",
" # Step 2: Check credentials\n",
" if not self.check_credentials():\n",
" return False\n",
" \n",
" # Step 3: Test API connection\n",
" if not self.test_api_connection():\n",
" return False\n",
" \n",
" # Step 4: Check permissions\n",
" if not self.check_basic_permissions():\n",
" return False\n",
" \n",
" # Step 5: Test folder access\n",
" self.test_folder_access()\n",
" \n",
" print(\"\\n✅ All basic checks passed!\")\n",
" return True\n",
" \n",
" def check_service_account_file(self):\n",
" \"\"\"Check if service account file exists and is valid\"\"\"\n",
" print(\"\\n1️⃣ Checking service account file...\")\n",
" \n",
" if not os.path.exists(self.service_account_file):\n",
" print(f\"❌ Service account file not found: {self.service_account_file}\")\n",
" print(\"💡 Make sure you've downloaded the JSON key file from Google Cloud Console\")\n",
" return False\n",
" \n",
" try:\n",
" with open(self.service_account_file, 'r') as f:\n",
" service_account_info = json.load(f)\n",
" \n",
" required_fields = ['type', 'project_id', 'private_key_id', 'private_key', 'client_email']\n",
" missing_fields = [field for field in required_fields if field not in service_account_info]\n",
" \n",
" if missing_fields:\n",
" print(f\"❌ Service account file missing required fields: {missing_fields}\")\n",
" return False\n",
" \n",
" print(f\"✅ Service account file is valid\")\n",
" print(f\" 📧 Service account email: {service_account_info['client_email']}\")\n",
" print(f\" 🏗️ Project ID: {service_account_info['project_id']}\")\n",
" \n",
" return True\n",
" \n",
" except json.JSONDecodeError:\n",
" print(\"❌ Service account file is not valid JSON\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Error reading service account file: {e}\")\n",
" return False\n",
" \n",
" def check_credentials(self):\n",
" \"\"\"Check if credentials can be created\"\"\"\n",
" print(\"\\n2️⃣ Checking credentials...\")\n",
" \n",
" try:\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" self.service_account_file,\n",
" scopes=['https://www.googleapis.com/auth/drive.readonly']\n",
" )\n",
" print(\"✅ Credentials created successfully\")\n",
" return True\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Failed to create credentials: {e}\")\n",
" print(\"💡 Check if your service account key file is corrupted\")\n",
" return False\n",
" \n",
" def test_api_connection(self):\n",
" \"\"\"Test basic API connection\"\"\"\n",
" print(\"\\n3️⃣ Testing API connection...\")\n",
" \n",
" try:\n",
" self.drive_service = build('drive', 'v3', credentials=self.credentials)\n",
" \n",
" # Try a simple API call\n",
" about = self.drive_service.about().get(fields=\"user, storageQuota\").execute()\n",
" print(\"✅ Successfully connected to Google Drive API\")\n",
" print(f\" 👤 Connected as: {about.get('user', {}).get('emailAddress', 'Unknown')}\")\n",
" \n",
" return True\n",
" \n",
" except HttpError as e:\n",
" print(f\"❌ HTTP Error connecting to API: {e}\")\n",
" if e.resp.status == 403:\n",
" print(\"💡 This is likely a permissions issue - check if Drive API is enabled\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Failed to connect to API: {e}\")\n",
" return False\n",
" \n",
" def check_basic_permissions(self):\n",
" \"\"\"Check basic file listing permissions\"\"\"\n",
" print(\"\\n4️⃣ Checking basic permissions...\")\n",
" \n",
" try:\n",
" # Try to list files (this should work with readonly access)\n",
" results = self.drive_service.files().list(\n",
" pageSize=1,\n",
" fields=\"files(id, name)\"\n",
" ).execute()\n",
" \n",
" files = results.get('files', [])\n",
" print(f\"✅ Can access Drive API - found {len(files)} files in test query\")\n",
" \n",
" if len(files) == 0:\n",
" print(\"⚠️ No files found - this might mean:\")\n",
" print(\" • Service account has no shared files\")\n",
" print(\" • No files are shared with the service account\")\n",
" \n",
" return True\n",
" \n",
" except HttpError as e:\n",
" print(f\"❌ Permission error: {e}\")\n",
" if e.resp.status == 403:\n",
" print(\"💡 Common causes:\")\n",
" print(\" • Google Drive API not enabled in Google Cloud Console\")\n",
" print(\" • Service account doesn't have proper permissions\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Error checking permissions: {e}\")\n",
" return False\n",
" \n",
" def test_folder_access(self, folder_name=\"Blue berry\"):\n",
" \"\"\"Test access to specific folder\"\"\"\n",
" print(f\"\\n5️⃣ Testing access to '{folder_name}' folder...\")\n",
" \n",
" try:\n",
" # Search for the folder\n",
" query = f\"name='{folder_name}' and mimeType='application/vnd.google-apps.folder' and trashed=false\"\n",
" results = self.drive_service.files().list(\n",
" q=query,\n",
" fields=\"files(id, name, owners, permissions)\"\n",
" ).execute()\n",
" \n",
" folders = results.get('files', [])\n",
" \n",
" if not folders:\n",
" print(f\"❌ Folder '{folder_name}' not found or not accessible\")\n",
" print(\"💡 Possible solutions:\")\n",
" print(\" • Make sure the folder exists in Google Drive\")\n",
" print(\" • Share the folder with your service account email\")\n",
" print(\" • Check folder name spelling (case sensitive)\")\n",
" return False\n",
" \n",
" folder = folders[0]\n",
" print(f\"✅ Found folder '{folder_name}'\")\n",
" print(f\" 📁 Folder ID: {folder['id']}\")\n",
" \n",
" # Test listing files in the folder\n",
" files_in_folder = self.drive_service.files().list(\n",
" q=f\"'{folder['id']}' in parents and trashed=false\",\n",
" fields=\"files(id, name, mimeType)\"\n",
" ).execute()\n",
" \n",
" files = files_in_folder.get('files', [])\n",
" print(f\" 📄 Contains {len(files)} files\")\n",
" \n",
" if files:\n",
" print(\" 📝 Sample files:\")\n",
" for file in files[:3]: # Show first 3 files\n",
" print(f\" • {file['name']} ({file['mimeType']})\")\n",
" \n",
" return True\n",
" \n",
" except HttpError as e:\n",
" print(f\"❌ HTTP Error accessing folder: {e}\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Error accessing folder: {e}\")\n",
" return False\n",
" \n",
" def check_file_permissions(self, file_id):\n",
" \"\"\"Check permissions for a specific file\"\"\"\n",
" print(f\"\\n🔍 Checking permissions for file ID: {file_id}\")\n",
" \n",
" try:\n",
" file_info = self.drive_service.files().get(\n",
" fileId=file_id,\n",
" fields=\"id, name, mimeType, owners, permissions, capabilities\"\n",
" ).execute()\n",
" \n",
" print(f\"✅ File: {file_info['name']}\")\n",
" print(f\" 🔗 Type: {file_info['mimeType']}\")\n",
" print(f\" 👤 Owner: {file_info.get('owners', [{}])[0].get('emailAddress', 'Unknown')}\")\n",
" \n",
" capabilities = file_info.get('capabilities', {})\n",
" print(f\" 📖 Can read: {capabilities.get('canDownload', False)}\")\n",
" print(f\" 📤 Can export: {capabilities.get('canExport', False)}\")\n",
" \n",
" except HttpError as e:\n",
" print(f\"❌ Cannot access file: {e}\")\n",
" except Exception as e:\n",
" print(f\"❌ Error: {e}\")\n",
" \n",
" def get_sharing_instructions(self):\n",
" \"\"\"Provide step-by-step sharing instructions\"\"\"\n",
" print(\"\\n📋 HOW TO SHARE FOLDER WITH SERVICE ACCOUNT:\")\n",
" print(\"=\" * 50)\n",
" \n",
" if self.credentials:\n",
" service_email = self.credentials.service_account_email\n",
" print(f\"1. Copy this service account email: {service_email}\")\n",
" else:\n",
" print(\"1. Find your service account email in the JSON key file (client_email field)\")\n",
" \n",
" print(\"2. Open Google Drive in your browser\")\n",
" print(\"3. Right-click on your 'Blue berry' folder\")\n",
" print(\"4. Select 'Share'\")\n",
" print(\"5. Paste the service account email\")\n",
" print(\"6. Set permission to 'Viewer' or 'Editor'\")\n",
" print(\"7. Click 'Send' (you can uncheck 'Notify people')\")\n",
" print(\"8. Wait a few minutes for permissions to propagate\")\n",
" \n",
" def run_connection_test(self):\n",
" \"\"\"Quick connection test\"\"\"\n",
" print(\"🚀 Quick Connection Test\")\n",
" print(\"-\" * 30)\n",
" \n",
" try:\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" self.service_account_file,\n",
" scopes=['https://www.googleapis.com/auth/drive.readonly']\n",
" )\n",
" \n",
" service = build('drive', 'v3', credentials=self.credentials)\n",
" \n",
" # Test basic query\n",
" results = service.files().list(pageSize=5).execute()\n",
" files = results.get('files', [])\n",
" \n",
" print(f\"✅ Connected! Found {len(files)} accessible files\")\n",
" print(f\"📧 Service account: {self.credentials.service_account_email}\")\n",
" \n",
" return True\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Connection failed: {e}\")\n",
" return False\n",
"\n",
"# Usage Examples\n",
"def troubleshoot_drive_access():\n",
" \"\"\"Main troubleshooting function\"\"\"\n",
" service_account_file = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE')\n",
" \n",
" if not service_account_file:\n",
" print(\"❌ GOOGLE_SERVICE_ACCOUNT_FILE environment variable not set\")\n",
" print(\"💡 Add this to your .env file:\")\n",
" print(\"GOOGLE_SERVICE_ACCOUNT_FILE=path/to/your/service-account-key.json\")\n",
" return\n",
" \n",
" troubleshooter = GPTDriveTroubleshooter(service_account_file)\n",
" \n",
" # Run full diagnostic\n",
" success = troubleshooter.run_full_diagnostic()\n",
" \n",
" if not success:\n",
" print(\"\\n\" + \"=\"*50)\n",
" troubleshooter.get_sharing_instructions()\n",
" \n",
" return success\n",
"\n",
"# Quick test function\n",
"def quick_test():\n",
" \"\"\"Quick test to verify everything works\"\"\"\n",
" service_account_file = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE')\n",
" troubleshooter = GPTDriveTroubleshooter(service_account_file)\n",
" return troubleshooter.run_connection_test()\n",
"\n",
"if __name__ == \"__main__\":\n",
" # Uncomment the test you want to run:\n",
" \n",
" # Full diagnostic (recommended for first-time setup)\n",
" troubleshoot_drive_access()\n",
" \n",
" # Quick test (for regular checks)\n",
" # quick_test()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "6b336789-d402-455d-a631-26e987d79ed6",
"metadata": {},
"outputs": [],
"source": [
"troubleshooter = GPTDriveTroubleshooter(GOOGLE_SERVICE_ACCOUNT_FILE)\n",
"troubleshooter.test_folder_access(\"Blue berry\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e2786c67-215e-46b0-b3ca-2c3176bd8971",
"metadata": {},
"outputs": [],
"source": [
"# Add this test function to verify access\n",
"def test_folder_access():\n",
" integration = GPTDriveIntegration()\n",
" folder_id = integration.find_folder_by_name(\"Blue berry\")\n",
" if folder_id:\n",
" print(\"✅ Successfully found 'Blue berry' folder!\")\n",
" \n",
" # List files in the folder\n",
" results = integration.drive_service.files().list(\n",
" q=f\"'{folder_id}' in parents and trashed=false\",\n",
" fields=\"files(id, name, mimeType)\"\n",
" ).execute()\n",
" \n",
" files = results.get('files', [])\n",
" print(f\"Found {len(files)} files in the folder:\")\n",
" for file in files[:5]: # Show first 5 files\n",
" print(f\" - {file['name']} ({file['mimeType']})\")\n",
" else:\n",
" print(\"❌ Could not access 'Blue berry' folder\")\n",
" print(\"Make sure you've shared the folder with your service account\")\n",
"\n",
"# Run this test first\n",
"test_folder_access()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c3dede9f-5e01-436d-a7b7-905e1646baf9",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "b933c8b6-add6-40fc-827f-e5e07447ac00",
"metadata": {},
"outputs": [],
"source": [
"#Troubleshooting"
]
},
{
"cell_type": "code",
"execution_count": 32,
"id": "b63a1f4b-1315-44fe-ba84-93fb1c2655cc",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"🔍 Searching for 'Blue Berry' folder...\n",
"Total folders found: 2\n",
"\n",
"✅ Found 1 'Blue Berry' folder(s):\n",
"📁 Anatomy Books \n",
" Full Path: Anatomy Books \n",
" Folder ID: 125NdiCL7moQsNYWRuHojVM7If88Bxh0q\n",
"\n"
]
}
],
"source": [
"import os\n",
"from google.oauth2 import service_account\n",
"from googleapiclient.discovery import build\n",
"\n",
"class DrivefolderLister:\n",
" def __init__(self):\n",
" # Initialize Google Drive API\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE'),\n",
" scopes=['https://www.googleapis.com/auth/drive.readonly']\n",
" )\n",
" self.drive_service = build('drive', 'v3', credentials=self.credentials)\n",
" \n",
" def get_all_folders(self):\n",
" \"\"\"Get ALL folders in the Google Drive\"\"\"\n",
" try:\n",
" # Query to get all folders (not trashed)\n",
" query = \"mimeType='application/vnd.google-apps.folder' and trashed=false\"\n",
" \n",
" all_folders = []\n",
" page_token = None\n",
" \n",
" while True:\n",
" results = self.drive_service.files().list(\n",
" q=query,\n",
" fields=\"nextPageToken, files(id, name, parents)\",\n",
" pageSize=1000, # Maximum allowed\n",
" pageToken=page_token\n",
" ).execute()\n",
" \n",
" folders = results.get('files', [])\n",
" all_folders.extend(folders)\n",
" \n",
" page_token = results.get('nextPageToken')\n",
" if not page_token:\n",
" break\n",
" \n",
" print(f\"Total folders found: {len(all_folders)}\")\n",
" return all_folders\n",
" \n",
" except Exception as e:\n",
" print(f\"Error retrieving folders: {e}\")\n",
" return []\n",
" \n",
" def build_folder_tree(self, folders):\n",
" \"\"\"Build a tree structure from the flat list of folders\"\"\"\n",
" # Create a dictionary for quick lookup\n",
" folder_dict = {folder['id']: folder for folder in folders}\n",
" \n",
" # Add root folder\n",
" folder_dict['root'] = {'id': 'root', 'name': 'My Drive', 'parents': []}\n",
" \n",
" # Build the tree structure\n",
" tree = {}\n",
" \n",
" def build_path(folder_id, visited=None):\n",
" if visited is None:\n",
" visited = set()\n",
" \n",
" if folder_id in visited: # Prevent infinite loops\n",
" return \"CIRCULAR_REFERENCE\"\n",
" \n",
" visited.add(folder_id)\n",
" \n",
" if folder_id not in folder_dict:\n",
" return \"UNKNOWN\"\n",
" \n",
" folder = folder_dict[folder_id]\n",
" \n",
" if folder_id == 'root':\n",
" return \"My Drive\"\n",
" \n",
" parents = folder.get('parents', [])\n",
" if not parents:\n",
" return folder['name']\n",
" \n",
" parent_path = build_path(parents[0], visited.copy())\n",
" return f\"{parent_path}/{folder['name']}\"\n",
" \n",
" # Build paths for all folders\n",
" folder_paths = []\n",
" for folder in folders:\n",
" path = build_path(folder['id'])\n",
" folder_paths.append({\n",
" 'name': folder['name'],\n",
" 'id': folder['id'],\n",
" 'path': path\n",
" })\n",
" \n",
" return folder_paths\n",
" \n",
" def search_folder_by_name(self, folder_name, folders_with_paths):\n",
" \"\"\"Search for folders by name (case-insensitive)\"\"\"\n",
" matches = []\n",
" search_name = folder_name.lower()\n",
" \n",
" for folder in folders_with_paths:\n",
" if search_name in folder['name'].lower():\n",
" matches.append(folder)\n",
" \n",
" return matches\n",
" \n",
" def display_all_folders(self, folders_with_paths, search_term=None):\n",
" \"\"\"Display all folders in a readable format\"\"\"\n",
" if search_term:\n",
" print(f\"\\n=== Searching for folders containing '{search_term}' ===\")\n",
" matches = self.search_folder_by_name(search_term, folders_with_paths)\n",
" if matches:\n",
" print(f\"Found {len(matches)} matching folders:\")\n",
" for folder in matches:\n",
" print(f\"📁 {folder['name']}\")\n",
" print(f\" Path: {folder['path']}\")\n",
" print(f\" ID: {folder['id']}\")\n",
" print()\n",
" else:\n",
" print(f\"No folders found containing '{search_term}'\")\n",
" else:\n",
" print(f\"\\n=== All {len(folders_with_paths)} folders in your Google Drive ===\")\n",
" \n",
" # Sort by path for better readability\n",
" sorted_folders = sorted(folders_with_paths, key=lambda x: x['path'])\n",
" \n",
" for folder in sorted_folders:\n",
" print(f\"📁 {folder['path']}\")\n",
" # print(f\" ID: {folder['id']}\") # Uncomment if you need IDs\n",
" \n",
" def find_blue_berry_folder(self):\n",
" \"\"\"Specifically look for the Anatomy Books folder\"\"\"\n",
" print(\"🔍 Searching for 'Blue Berry' folder...\")\n",
" \n",
" folders = self.get_all_folders()\n",
" if not folders:\n",
" print(\"No folders found or error occurred.\")\n",
" return\n",
" \n",
" folders_with_paths = self.build_folder_tree(folders)\n",
" \n",
" # Search for Anatomy Books specifically\n",
" blue_berry_matches = self.search_folder_by_name(\"Anatomy Books\", folders_with_paths)\n",
" \n",
" if blue_berry_matches:\n",
" print(f\"\\n✅ Found {len(blue_berry_matches)} 'Blue Berry' folder(s):\")\n",
" for folder in blue_berry_matches:\n",
" print(f\"📁 {folder['name']}\")\n",
" print(f\" Full Path: {folder['path']}\")\n",
" print(f\" Folder ID: {folder['id']}\")\n",
" print()\n",
" else:\n",
" print(\"\\n❌ No 'Blue Berry' folder found.\")\n",
" print(\"Let me show you all folders to help you locate it:\")\n",
" \n",
" # Show all folders if Blue Berry not found\n",
" self.display_all_folders(folders_with_paths)\n",
" \n",
" def interactive_folder_search(self):\n",
" \"\"\"Interactive search for any folder\"\"\"\n",
" print(\"📂 Google Drive Folder Explorer\")\n",
" print(\"=\" * 40)\n",
" \n",
" folders = self.get_all_folders()\n",
" if not folders:\n",
" print(\"No folders found or error occurred.\")\n",
" return\n",
" \n",
" folders_with_paths = self.build_folder_tree(folders)\n",
" \n",
" while True:\n",
" print(\"\\nOptions:\")\n",
" print(\"1. Search for a specific folder\")\n",
" print(\"2. Show all folders\")\n",
" print(\"3. Find 'Blue Berry' folder\")\n",
" print(\"4. Exit\")\n",
" \n",
" choice = input(\"\\nEnter your choice (1-4): \").strip()\n",
" \n",
" if choice == '1':\n",
" search_term = input(\"Enter folder name to search: \").strip()\n",
" if search_term:\n",
" self.display_all_folders(folders_with_paths, search_term)\n",
" \n",
" elif choice == '2':\n",
" self.display_all_folders(folders_with_paths)\n",
" \n",
" elif choice == '3':\n",
" blue_berry_matches = self.search_folder_by_name(\"blue berry\", folders_with_paths)\n",
" if blue_berry_matches:\n",
" print(f\"\\n✅ Found 'Blue Berry' folder(s):\")\n",
" for folder in blue_berry_matches:\n",
" print(f\"📁 {folder['name']} -> {folder['path']}\")\n",
" else:\n",
" print(\"\\n❌ 'Blue Berry' folder not found in your Drive.\")\n",
" \n",
" elif choice == '4':\n",
" print(\"Goodbye!\")\n",
" break\n",
" \n",
" else:\n",
" print(\"Invalid choice. Please try again.\")\n",
"\n",
"def main():\n",
" \"\"\"Main function to run the folder lister\"\"\"\n",
" try:\n",
" lister = DrivefolderLister()\n",
" \n",
" # Quick search for Blue Berry folder\n",
" lister.find_blue_berry_folder()\n",
" \n",
" # Uncomment the line below for interactive mode\n",
" # lister.interactive_folder_search()\n",
" \n",
" except Exception as e:\n",
" print(f\"Error initializing Drive connection: {e}\")\n",
" print(\"Make sure your GOOGLE_SERVICE_ACCOUNT_FILE environment variable is set correctly.\")\n",
"\n",
"if __name__ == \"__main__\":\n",
" main()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "29a61039-9043-44d5-94b1-4847350b2200",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 31,
"id": "7350d687-f76e-46f4-a30b-0518e5b8236e",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"🚀 Google Drive API Diagnostic Tool\n",
"==================================================\n",
"🔧 Testing Environment Variables...\n",
"----------------------------------------\n",
"✅ GOOGLE_SERVICE_ACCOUNT_FILE: C:/Users/Uche Buzz/myprojects/RAG/rag-system-463320-f292991d0516.json\n",
"✅ Service account file exists\n",
"✅ Service account file is valid JSON\n",
"✅ All required fields present in service account file\n",
"📧 Service account email: rag-base@rag-system-463320.iam.gserviceaccount.com\n",
"✅ OPENAI_API_KEY: ************************************************************************************************************************************************************K1OkRRkA\n",
"\n",
"🔑 Testing Credentials Loading...\n",
"----------------------------------------\n",
"\n",
"Testing scope set 1: ['https://www.googleapis.com/auth/drive.readonly']\n",
"✅ Credentials loaded successfully with scopes: ['https://www.googleapis.com/auth/drive.readonly']\n",
"\n",
"🔧 Testing Drive Service Creation...\n",
"----------------------------------------\n",
"✅ Google Drive service created successfully\n",
"\n",
"📡 Testing Basic API Call...\n",
"----------------------------------------\n",
"✅ API call successful!\n",
"📧 Connected as: rag-base@rag-system-463320.iam.gserviceaccount.com\n",
"👤 Display name: rag-base@rag-system-463320.iam.gserviceaccount.com\n",
"\n",
"🏠 Testing Root Folder Access...\n",
"----------------------------------------\n",
"✅ Root folder accessible\n",
"📁 Root folder name: My Drive\n",
"📄 Items in root folder: 0\n",
"\n",
"📁 Testing File Listing...\n",
"----------------------------------------\n",
"Test 1: Listing all files (including documents)...\n",
"✅ Found 7 files total\n",
"📄 First few files:\n",
" - Neuro Note Dr Clement.docx (application/vnd.openxmlformats-officedocument.wordprocessingml.document)\n",
" - Morbid Anatomy - ANA 3107.pdf (application/pdf)\n",
" - ANT & POST TRIANGLES OF THE NECK.pdf (application/pdf)\n",
" - Anatomy Books (application/vnd.google-apps.folder)\n",
" - Blue Berry (application/vnd.google-apps.folder)\n",
"\n",
"Test 2: Listing folders only...\n",
"✅ Found 2 folders\n",
"📁 Folders found:\n",
" - Anatomy Books (ID: 125NdiCL7moQsNYWRuHojVM7If88Bxh0q)\n",
" - Blue Berry (ID: 1AYaS0yt_srFlgdE4mSNlqA6FLm10rdt4)\n",
"\n",
"✅ All diagnostic tests passed!\n",
"🎉 Your Google Drive API connection is working correctly!\n"
]
}
],
"source": [
"import os\n",
"from google.oauth2 import service_account\n",
"from googleapiclient.discovery import build\n",
"from googleapiclient.errors import HttpError\n",
"import json\n",
"\n",
"class DriveConnectionDiagnostic:\n",
" def __init__(self):\n",
" self.credentials = None\n",
" self.drive_service = None\n",
" self.setup_success = False\n",
" \n",
" def test_environment_variables(self):\n",
" \"\"\"Test if environment variables are set correctly\"\"\"\n",
" print(\"🔧 Testing Environment Variables...\")\n",
" print(\"-\" * 40)\n",
" \n",
" service_account_file = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE')\n",
" openai_key = os.getenv('OPENAI_API_KEY')\n",
" \n",
" if service_account_file:\n",
" print(f\"✅ GOOGLE_SERVICE_ACCOUNT_FILE: {service_account_file}\")\n",
" \n",
" # Check if file exists\n",
" if os.path.exists(service_account_file):\n",
" print(f\"✅ Service account file exists\")\n",
" \n",
" # Check if file is readable\n",
" try:\n",
" with open(service_account_file, 'r') as f:\n",
" service_data = json.load(f)\n",
" print(f\"✅ Service account file is valid JSON\")\n",
" \n",
" # Check required fields\n",
" required_fields = ['type', 'project_id', 'private_key_id', 'private_key', 'client_email']\n",
" missing_fields = [field for field in required_fields if field not in service_data]\n",
" \n",
" if missing_fields:\n",
" print(f\"❌ Missing required fields in service account file: {missing_fields}\")\n",
" return False\n",
" else:\n",
" print(f\"✅ All required fields present in service account file\")\n",
" print(f\"📧 Service account email: {service_data.get('client_email')}\")\n",
" \n",
" except json.JSONDecodeError:\n",
" print(f\"❌ Service account file is not valid JSON\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Error reading service account file: {e}\")\n",
" return False\n",
" \n",
" else:\n",
" print(f\"❌ Service account file does not exist at: {service_account_file}\")\n",
" return False\n",
" else:\n",
" print(f\"❌ GOOGLE_SERVICE_ACCOUNT_FILE environment variable not set\")\n",
" return False\n",
" \n",
" if openai_key:\n",
" print(f\"✅ OPENAI_API_KEY: {'*' * (len(openai_key) - 8) + openai_key[-8:]}\")\n",
" else:\n",
" print(f\"⚠️ OPENAI_API_KEY not set (not required for folder listing)\")\n",
" \n",
" return True\n",
" \n",
" def test_credentials_loading(self):\n",
" \"\"\"Test if credentials can be loaded\"\"\"\n",
" print(\"\\n🔑 Testing Credentials Loading...\")\n",
" print(\"-\" * 40)\n",
" \n",
" try:\n",
" service_account_file = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE')\n",
" \n",
" # Try different scope combinations\n",
" scopes_to_test = [\n",
" ['https://www.googleapis.com/auth/drive.readonly'],\n",
" ['https://www.googleapis.com/auth/drive'],\n",
" ['https://www.googleapis.com/auth/drive.metadata.readonly'],\n",
" ['https://www.googleapis.com/auth/drive.file']\n",
" ]\n",
" \n",
" for i, scopes in enumerate(scopes_to_test):\n",
" try:\n",
" print(f\"\\nTesting scope set {i+1}: {scopes}\")\n",
" self.credentials = service_account.Credentials.from_service_account_file(\n",
" service_account_file,\n",
" scopes=scopes\n",
" )\n",
" print(f\"✅ Credentials loaded successfully with scopes: {scopes}\")\n",
" return True\n",
" except Exception as e:\n",
" print(f\"❌ Failed to load credentials with scopes {scopes}: {e}\")\n",
" \n",
" return False\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Error loading credentials: {e}\")\n",
" return False\n",
" \n",
" def test_drive_service_creation(self):\n",
" \"\"\"Test if Drive service can be created\"\"\"\n",
" print(\"\\n🔧 Testing Drive Service Creation...\")\n",
" print(\"-\" * 40)\n",
" \n",
" try:\n",
" if not self.credentials:\n",
" print(\"❌ No credentials available\")\n",
" return False\n",
" \n",
" self.drive_service = build('drive', 'v3', credentials=self.credentials)\n",
" print(\"✅ Google Drive service created successfully\")\n",
" return True\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Error creating Drive service: {e}\")\n",
" return False\n",
" \n",
" def test_basic_api_call(self):\n",
" \"\"\"Test basic API functionality\"\"\"\n",
" print(\"\\n📡 Testing Basic API Call...\")\n",
" print(\"-\" * 40)\n",
" \n",
" try:\n",
" if not self.drive_service:\n",
" print(\"❌ No Drive service available\")\n",
" return False\n",
" \n",
" # Test getting user info\n",
" about = self.drive_service.about().get(fields=\"user\").execute()\n",
" user_info = about.get('user', {})\n",
" print(f\"✅ API call successful!\")\n",
" print(f\"📧 Connected as: {user_info.get('emailAddress', 'Unknown')}\")\n",
" print(f\"👤 Display name: {user_info.get('displayName', 'Unknown')}\")\n",
" \n",
" return True\n",
" \n",
" except HttpError as e:\n",
" print(f\"❌ HTTP Error: {e}\")\n",
" if e.resp.status == 403:\n",
" print(\" This might be a permissions issue. Check:\")\n",
" print(\" 1. Is the service account enabled?\")\n",
" print(\" 2. Does it have the right permissions?\")\n",
" print(\" 3. Is the Google Drive API enabled in your project?\")\n",
" return False\n",
" except Exception as e:\n",
" print(f\"❌ Error making API call: {e}\")\n",
" return False\n",
" \n",
" def test_file_listing(self):\n",
" \"\"\"Test different file listing approaches\"\"\"\n",
" print(\"\\n📁 Testing File Listing...\")\n",
" print(\"-\" * 40)\n",
" \n",
" if not self.drive_service:\n",
" print(\"❌ No Drive service available\")\n",
" return False\n",
" \n",
" # Test 1: List any files (not just folders)\n",
" try:\n",
" print(\"Test 1: Listing all files (including documents)...\")\n",
" results = self.drive_service.files().list(\n",
" q=\"trashed=false\",\n",
" fields=\"files(id, name, mimeType)\",\n",
" pageSize=10\n",
" ).execute()\n",
" \n",
" files = results.get('files', [])\n",
" print(f\"✅ Found {len(files)} files total\")\n",
" \n",
" if files:\n",
" print(\"📄 First few files:\")\n",
" for file in files[:5]:\n",
" print(f\" - {file['name']} ({file['mimeType']})\")\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Error listing files: {e}\")\n",
" return False\n",
" \n",
" # Test 2: List only folders\n",
" try:\n",
" print(f\"\\nTest 2: Listing folders only...\")\n",
" results = self.drive_service.files().list(\n",
" q=\"mimeType='application/vnd.google-apps.folder' and trashed=false\",\n",
" fields=\"files(id, name, parents)\",\n",
" pageSize=10\n",
" ).execute()\n",
" \n",
" folders = results.get('files', [])\n",
" print(f\"✅ Found {len(folders)} folders\")\n",
" \n",
" if folders:\n",
" print(\"📁 Folders found:\")\n",
" for folder in folders:\n",
" print(f\" - {folder['name']} (ID: {folder['id']})\")\n",
" else:\n",
" print(\"⚠️ No folders found - this might indicate:\")\n",
" print(\" 1. Your Google Drive is empty\")\n",
" print(\" 2. The service account doesn't have access to your personal Drive\")\n",
" print(\" 3. You need to share folders with the service account\")\n",
" \n",
" return len(folders) > 0\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Error listing folders: {e}\")\n",
" return False\n",
" \n",
" def test_root_access(self):\n",
" \"\"\"Test access to root folder\"\"\"\n",
" print(\"\\n🏠 Testing Root Folder Access...\")\n",
" print(\"-\" * 40)\n",
" \n",
" try:\n",
" if not self.drive_service:\n",
" print(\"❌ No Drive service available\")\n",
" return False\n",
" \n",
" # Try to get root folder info\n",
" root_info = self.drive_service.files().get(\n",
" fileId='root',\n",
" fields=\"id, name, mimeType\"\n",
" ).execute()\n",
" \n",
" print(f\"✅ Root folder accessible\")\n",
" print(f\"📁 Root folder name: {root_info.get('name', 'My Drive')}\")\n",
" \n",
" # Try to list contents of root\n",
" results = self.drive_service.files().list(\n",
" q=\"'root' in parents and trashed=false\",\n",
" fields=\"files(id, name, mimeType)\",\n",
" pageSize=10\n",
" ).execute()\n",
" \n",
" root_contents = results.get('files', [])\n",
" print(f\"📄 Items in root folder: {len(root_contents)}\")\n",
" \n",
" if root_contents:\n",
" print(\"🔍 Root folder contents:\")\n",
" for item in root_contents[:5]:\n",
" item_type = \"📁\" if \"folder\" in item['mimeType'] else \"📄\"\n",
" print(f\" {item_type} {item['name']}\")\n",
" \n",
" return True\n",
" \n",
" except Exception as e:\n",
" print(f\"❌ Error accessing root folder: {e}\")\n",
" return False\n",
" \n",
" def run_full_diagnostic(self):\n",
" \"\"\"Run complete diagnostic\"\"\"\n",
" print(\"🚀 Google Drive API Diagnostic Tool\")\n",
" print(\"=\" * 50)\n",
" \n",
" # Step 1: Environment variables\n",
" if not self.test_environment_variables():\n",
" print(\"\\n❌ Environment setup failed. Please check your service account file.\")\n",
" return False\n",
" \n",
" # Step 2: Credentials\n",
" if not self.test_credentials_loading():\n",
" print(\"\\n❌ Credentials loading failed.\")\n",
" return False\n",
" \n",
" # Step 3: Service creation\n",
" if not self.test_drive_service_creation():\n",
" print(\"\\n❌ Drive service creation failed.\")\n",
" return False\n",
" \n",
" # Step 4: Basic API call\n",
" if not self.test_basic_api_call():\n",
" print(\"\\n❌ Basic API call failed.\")\n",
" return False\n",
" \n",
" # Step 5: Root access\n",
" if not self.test_root_access():\n",
" print(\"\\n❌ Root folder access failed.\")\n",
" return False\n",
" \n",
" # Step 6: File listing\n",
" if not self.test_file_listing():\n",
" print(\"\\n❌ File listing failed or no folders found.\")\n",
" print(\"\\n🔧 SOLUTION SUGGESTIONS:\")\n",
" print(\"1. If you're using a service account, you need to SHARE folders with it\")\n",
" print(\"2. Share your 'Blue Berry' folder with the service account email\")\n",
" print(\"3. Or consider using OAuth2 instead of service account for personal Drive access\")\n",
" return False\n",
" \n",
" print(\"\\n✅ All diagnostic tests passed!\")\n",
" print(\"🎉 Your Google Drive API connection is working correctly!\")\n",
" return True\n",
" \n",
" def show_service_account_sharing_instructions(self):\n",
" \"\"\"Show instructions for sharing with service account\"\"\"\n",
" print(\"\\n📋 SERVICE ACCOUNT SHARING INSTRUCTIONS:\")\n",
" print(\"=\" * 50)\n",
" print(\"If you're using a service account, you need to share folders with it:\")\n",
" print()\n",
" print(\"1. Open Google Drive in your browser\")\n",
" print(\"2. Find the folder you want to access (e.g., 'Blue Berry')\")\n",
" print(\"3. Right-click the folder → 'Share'\")\n",
" print(\"4. Add the service account email address\")\n",
" print(\"5. Give it 'Viewer' or 'Editor' permissions\")\n",
" print(\"6. Click 'Send'\")\n",
" print()\n",
" \n",
" if self.credentials:\n",
" try:\n",
" service_account_file = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE')\n",
" with open(service_account_file, 'r') as f:\n",
" service_data = json.load(f)\n",
" email = service_data.get('client_email')\n",
" print(f\"📧 Your service account email: {email}\")\n",
" print(\" ^ Share your folders with this email address\")\n",
" except:\n",
" print(\"❌ Could not read service account email\")\n",
"\n",
"def main():\n",
" diagnostic = DriveConnectionDiagnostic()\n",
" success = diagnostic.run_full_diagnostic()\n",
" \n",
" if not success:\n",
" diagnostic.show_service_account_sharing_instructions()\n",
"\n",
"if __name__ == \"__main__\":\n",
" main()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e3374b17-d2fa-4c80-a70b-76b8bf94c702",
"metadata": {},
"outputs": [],
"source": [
"#flask"
]
},
{
"cell_type": "code",
"execution_count": 30,
"id": "c8ac4c93-8b36-49d2-b8e1-efb7d1c4b104",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
" * Serving Flask app '__main__'\n",
" * Debug mode: on\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
"WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.\n",
" * Running on http://127.0.0.1:5000\n",
"Press CTRL+C to quit\n",
" * Restarting with stat\n"
]
},
{
"ename": "SystemExit",
"evalue": "1",
"output_type": "error",
"traceback": [
"An exception has occurred, use %tb to see the full traceback.\n",
"\u001b[31mSystemExit\u001b[39m\u001b[31m:\u001b[39m 1\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
"C:\\Users\\Uche Buzz\\anaconda3\\envs\\RAG\\Lib\\site-packages\\IPython\\core\\interactiveshell.py:3680: UserWarning: To exit: use 'exit', 'quit', or Ctrl-D.\n",
" warn(\"To exit: use 'exit', 'quit', or Ctrl-D.\", stacklevel=1)\n"
]
}
],
"source": [
"from flask import Flask, request, jsonify, render_template_string\n",
"\n",
"app = Flask(__name__)\n",
"integration = GPTDriveIntegration()\n",
"\n",
"HTML_TEMPLATE = \"\"\"\n",
"\n",
"\n",
"\n",
" GPT-Drive Integration\n",
" \n",
"\n",
"\n",
" GPT-Google Drive Integration
\n",
" \n",
" \n",
"\n",
" \n",
"\n",
"\n",
"\"\"\"\n",
"\n",
"@app.route('/')\n",
"def home():\n",
" return render_template_string(HTML_TEMPLATE)\n",
"\n",
"@app.route('/query', methods=['POST'])\n",
"def query():\n",
" data = request.json\n",
" result = integration.process_query(data['query'])\n",
" return jsonify(result)\n",
"\n",
"if __name__ == '__main__':\n",
" app.run(debug=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "caef8af5-73f9-4627-9203-0e9c8d098db0",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "ff7582bd-f80c-46c4-a51b-b20fd7493b0e",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "96ea63b3-3cf3-41c1-b160-72d677ab0949",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "a8a64f0d-c9b5-4ce7-a887-e1b60de4dac3",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 12,
"id": "abbc0f6c-5350-4fd9-8711-de68d4aa739d",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": null,
"id": "02ae258f-bef0-469b-a696-1edf6a7b5921",
"metadata": {},
"outputs": [],
"source": []
},
{
"cell_type": "code",
"execution_count": 13,
"id": "09a0b418-eb0e-486e-bf05-bc7b43645784",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"* Running on local URL: http://127.0.0.1:7860\n",
"* Running on public URL: https://a9ef60dd3438b2ef0e.gradio.live\n",
"\n",
"This share link expires in 1 week. For free permanent hosting and GPU upgrades, run `gradio deploy` from the terminal in the working directory to deploy to Hugging Face Spaces (https://huggingface.co/spaces)\n"
]
},
{
"data": {
"text/html": [
""
],
"text/plain": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Keyboard interruption in main thread... closing server.\n",
"Killing tunnel 127.0.0.1:7860 <> https://a9ef60dd3438b2ef0e.gradio.live\n"
]
}
],
"source": [
"gpt_drive = GPTDriveIntegration()\n",
"\n",
"def process_user_query(query, search_terms_input):\n",
" \"\"\"Process user query and return formatted response\"\"\"\n",
" if not query.strip():\n",
" return \"Please enter a question.\", \"\"\n",
" \n",
" # Parse search terms if provided\n",
" search_terms = None\n",
" # if search_terms_input.strip():\n",
" # search_terms = [term.strip() for term in search_terms_input.split(',')]\n",
" \n",
" # Process the query\n",
" result = gpt_drive.process_query(query, search_terms)\n",
" \n",
" # Format the response\n",
" answer = result['answer']\n",
" sources = result['sources']\n",
" \n",
" sources_text = \"\"\n",
" if sources:\n",
" sources_text = \"**Sources used:**\\n\" + \"\\n\".join([f\"• {source}\" for source in sources])\n",
" \n",
" return answer, sources_text\n",
"\n",
"def check_setup():\n",
" \"\"\"Check if the APIs are properly configured\"\"\"\n",
" status_messages = []\n",
" \n",
" # Check Google Drive API\n",
" if gpt_drive.drive_initialized:\n",
" status_messages.append(\"✅ Google Drive API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ Google Drive API: {getattr(gpt_drive, 'drive_error', 'Not configured')}\")\n",
" \n",
" # Check OpenAI API\n",
" if gpt_drive.openai_initialized:\n",
" status_messages.append(\"✅ OpenAI API: Connected\")\n",
" else:\n",
" status_messages.append(f\"❌ OpenAI API: {getattr(gpt_drive, 'openai_error', 'Not configured')}\")\n",
" \n",
" return \"\\n\".join(status_messages)\n",
"\n",
"# Create Gradio interface\n",
"with gr.Blocks(title=\"Augusta's Anatomy Reading Assistant\", theme=gr.themes.Soft()) as app:\n",
" gr.Markdown(\"# 🤖 Augusta's Anatomy bot\")\n",
" gr.Markdown(\"Ask questions about your anatomy books using AI!\")\n",
" \n",
" with gr.Row():\n",
" with gr.Column(scale=2):\n",
" # Main query interface\n",
" with gr.Group():\n",
" gr.Markdown(\"### Ask a Question\")\n",
" query_input = gr.Textbox(\n",
" label=\"Your Question\",\n",
" placeholder=\"Ask me any question about your anatomy books?\",\n",
" lines=3\n",
" )\n",
" \n",
" search_terms_input = gr.Textbox(\n",
" label=\"Search Terms (optional)\",\n",
" placeholder=\"Enter comma-separated terms to search for specific files\",\n",
" lines=1\n",
" )\n",
" \n",
" submit_btn = gr.Button(\"Search & Ask\", variant=\"primary\", size=\"lg\")\n",
" \n",
" # Results section\n",
" with gr.Group():\n",
" gr.Markdown(\"### Answer\")\n",
" answer_output = gr.Textbox(\n",
" label=\"AI Response\",\n",
" lines=10,\n",
" interactive=False\n",
" )\n",
" \n",
" sources_output = gr.Textbox(\n",
" label=\"Sources\",\n",
" lines=3,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Column(scale=1):\n",
" # Status and setup info\n",
" with gr.Group():\n",
" gr.Markdown(\"### System Status\")\n",
" status_btn = gr.Button(\"Check Status\", size=\"sm\")\n",
" status_output = gr.Textbox(\n",
" label=\"API Status\",\n",
" lines=4,\n",
" interactive=False\n",
" )\n",
" \n",
" with gr.Group():\n",
" gr.Markdown(\"### Setup Instructions\")\n",
" gr.Markdown(\"\"\"\n",
" **Important Notes:**\n",
" 1.Only documents shared with it, it can answer\n",
" \n",
" **File Types Supported:**\n",
" - Google Docs\n",
" - Google Sheets \n",
" - PDF files\n",
" - Text files\n",
" \n",
" **Tips:**\n",
" - Use specific search terms for better results\n",
" - The system searches the top 3 most relevant files\n",
" - Ask clear, specific questions for better answers\n",
" \"\"\")\n",
" \n",
" # Event handlers\n",
" submit_btn.click(\n",
" fn=process_user_query,\n",
" inputs=[query_input, search_terms_input],\n",
" outputs=[answer_output, sources_output]\n",
" )\n",
" \n",
" status_btn.click(\n",
" fn=check_setup,\n",
" outputs=status_output\n",
" )\n",
" \n",
" # Example queries\n",
" with gr.Row():\n",
" gr.Examples(\n",
" examples=[\n",
" [\"What is morbid Anatomy?\", \"morbid, Anatomy\"],\n",
" [\"The transmission of nerves from one neuron to another is as a result of what?\", \"neuron, nerves, Dr Clement\"],\n",
" ],\n",
" inputs=[query_input, search_terms_input],\n",
" )\n",
"\n",
"# Launch the app\n",
"if __name__ == \"__main__\":\n",
" app.launch(\n",
" share=True,debug =True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "35d0a347-2cc6-4f4e-bb70-cc47067eaf53",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.13.5"
}
},
"nbformat": 4,
"nbformat_minor": 5
}