Shreyas094's picture
Update app.py
96b8cb4 verified
raw
history blame
13.9 kB
import fitz # PyMuPDF
import gradio as gr
import requests
from bs4 import BeautifulSoup
import urllib.parse
import random
import os
from dotenv import load_dotenv
import shutil
import tempfile
load_dotenv() # Load environment variables from .env file
# Now replace the hard-coded token with the environment variable
HUGGINGFACE_API_TOKEN = os.getenv("HUGGINGFACE_TOKEN")
def clear_cache():
try:
# Clear Gradio cache
cache_dir = tempfile.gettempdir()
shutil.rmtree(os.path.join(cache_dir, "gradio"), ignore_errors=True)
# Clear any custom cache you might have
# For example, if you're caching PDF files or search results:
if os.path.exists("output_summary.pdf"):
os.remove("output_summary.pdf")
# Add any other cache clearing operations here
print("Cache cleared successfully.")
return "Cache cleared successfully."
except Exception as e:
print(f"Error clearing cache: {e}")
return f"Error clearing cache: {e}"
_useragent_list = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.59 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.59 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36",
]
# Function to extract visible text from HTML content of a webpage
def extract_text_from_webpage(html):
print("Extracting text from webpage...")
soup = BeautifulSoup(html, 'html.parser')
for script in soup(["script", "style"]):
script.extract() # Remove scripts and styles
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = '\n'.join(chunk for chunk in chunks if chunk)
print(f"Extracted text length: {len(text)}")
return text
# Function to perform a Google search and retrieve results
def google_search(term, num_results=5, lang="en", timeout=5, safe="active", ssl_verify=None):
"""Performs a Google search and returns the results."""
print(f"Searching for term: {term}")
escaped_term = urllib.parse.quote_plus(term)
start = 0
all_results = []
max_chars_per_page = 8000 # Limit the number of characters from each webpage to stay under the token limit
with requests.Session() as session:
while start < num_results:
print(f"Fetching search results starting from: {start}")
try:
# Choose a random user agent
user_agent = random.choice(_useragent_list)
headers = {
'User-Agent': user_agent
}
print(f"Using User-Agent: {headers['User-Agent']}")
resp = session.get(
url="https://www.google.com/search",
headers=headers,
params={
"q": term,
"num": num_results - start,
"hl": lang,
"start": start,
"safe": safe,
},
timeout=timeout,
verify=ssl_verify,
)
resp.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error fetching search results: {e}")
break
soup = BeautifulSoup(resp.text, "html.parser")
result_block = soup.find_all("div", attrs={"class": "g"})
if not result_block:
print("No more results found.")
break
for result in result_block:
link = result.find("a", href=True)
if link:
link = link["href"]
print(f"Found link: {link}")
try:
webpage = session.get(link, headers=headers, timeout=timeout)
webpage.raise_for_status()
visible_text = extract_text_from_webpage(webpage.text)
if len(visible_text) > max_chars_per_page:
visible_text = visible_text[:max_chars_per_page] + "..."
all_results.append({"link": link, "text": visible_text})
except requests.exceptions.RequestException as e:
print(f"Error fetching or processing {link}: {e}")
all_results.append({"link": link, "text": None})
else:
print("No link found in result.")
all_results.append({"link": None, "text": None})
start += len(result_block)
print(f"Total results fetched: {len(all_results)}")
return all_results
# Function to format the prompt for the Hugging Face API
def format_prompt(query, search_results, instructions):
formatted_results = ""
for result in search_results:
link = result["link"]
text = result["text"]
if link:
formatted_results += f"URL: {link}\nContent: {text}\n{'-' * 80}\n"
else:
formatted_results += "No link found.\n" + '-' * 80 + '\n'
prompt = f"{instructions}User Query: {query}\n\nWeb Search Results:\n{formatted_results}\n\nAssistant:"
return prompt
# Function to generate text using Hugging Face API
def generate_text(input_text, temperature=0.7, repetition_penalty=1.0, top_p=0.9):
print("Generating text using Hugging Face API...")
endpoint = "https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.3"
headers = {
"Authorization": f"Bearer {HUGGINGFACE_API_TOKEN}", # Use the environment variable
"Content-Type": "application/json"
}
data = {
"inputs": input_text,
"parameters": {
"max_new_tokens": 8000, # Adjust as needed
"temperature": temperature,
"repetition_penalty": repetition_penalty,
"top_p": top_p
}
}
try:
response = requests.post(endpoint, headers=headers, json=data)
response.raise_for_status()
# Check if response is JSON
try:
json_data = response.json()
except ValueError:
print("Response is not JSON.")
return None
# Extract generated text from response JSON
if isinstance(json_data, list):
# Handle list response (if applicable for your use case)
generated_text = json_data[0].get("generated_text") if json_data else None
elif isinstance(json_data, dict):
# Handle dictionary response
generated_text = json_data.get("generated_text")
else:
print("Unexpected response format.")
return None
if generated_text is not None:
print("Text generation complete using Hugging Face API.")
print(f"Generated text: {generated_text}") # Debugging line
return generated_text
else:
print("Generated text not found in response.")
return None
except requests.exceptions.RequestException as e:
print(f"Error generating text using Hugging Face API: {e}")
return None
# Function to read and extract text from a PDF
def read_pdf(file_obj):
with fitz.open(file_obj.name) as document:
text = ""
for page_num in range(document.page_count):
page = document.load_page(page_num)
text += page.get_text()
return text
# Function to format the prompt with instructions for text generation
def format_prompt_with_instructions(text, instructions):
prompt = f"{instructions}{text}\n\nAssistant:"
return prompt
# Function to save text to a PDF
def save_text_to_pdf(text, output_path):
print(f"Saving text to PDF at {output_path}...")
doc = fitz.open() # Create a new PDF document
page = doc.new_page() # Create a new page
# Set the page margins
margin = 50 # 50 points margin
page_width = page.rect.width
page_height = page.rect.height
text_width = page_width - 2 * margin
text_height = page_height - 2 * margin
# Define font size and line spacing
font_size = 9
line_spacing = 1 * font_size
fontname = "times-roman" # Use a supported font name
# Process the text to handle line breaks and paragraphs
paragraphs = text.split("\n") # Split text into paragraphs
y_position = margin
for paragraph in paragraphs:
words = paragraph.split()
current_line = ""
for word in words:
word = str(word) # Ensure word is treated as string
# Calculate the length of the current line plus the new word
current_line_length = fitz.get_text_length(current_line + " " + word, fontsize=font_size, fontname=fontname)
if current_line_length <= text_width:
current_line += " " + word
else:
page.insert_text(fitz.Point(margin, y_position), current_line.strip(), fontsize=font_size, fontname=fontname)
y_position += line_spacing
if y_position + line_spacing > page_height - margin:
page = doc.new_page() # Add a new page if text exceeds page height
y_position = margin
current_line = word
# Add the last line of the paragraph
page.insert_text(fitz.Point(margin, y_position), current_line.strip(), fontsize=font_size, fontname=fontname)
y_position += line_spacing
# Add extra space for new paragraph
y_position += line_spacing
if y_position + line_spacing > page_height - margin:
page = doc.new_page() # Add a new page if text exceeds page height
y_position = margin
doc.save(output_path) # Save the PDF to the specified path
print("PDF saved successfully.")
# Integrated function to perform web scraping, formatting, and text generation
def scrape_and_display(query, num_results, instructions, web_search=True, temperature=0.7, repetition_penalty=1.0, top_p=0.9):
print(f"Scraping and displaying results for query: {query} with num_results: {num_results}")
if web_search:
search_results = google_search(query, num_results)
formatted_prompt = format_prompt(query, search_results, instructions)
generated_summary = generate_text(formatted_prompt, temperature=temperature, repetition_penalty=repetition_penalty, top_p=top_p)
else:
formatted_prompt = format_prompt_with_instructions(query, instructions)
generated_summary = generate_text(formatted_prompt, temperature=temperature, repetition_penalty=repetition_penalty, top_p=top_p)
print("Scraping and display complete.")
if generated_summary:
# Extract and return text starting from "Assistant:"
assistant_index = generated_summary.find("Assistant:")
if assistant_index != -1:
generated_summary = generated_summary[assistant_index:]
else:
generated_summary = "Assistant: No response generated."
print(f"Generated summary: {generated_summary}") # Debugging line
return generated_summary
# Main Gradio interface function
def gradio_interface(query, use_pdf, pdf, num_results, instructions, temperature, repetition_penalty, top_p, clear_cache_flag):
if clear_cache_flag:
return clear_cache()
if use_pdf and pdf is not None:
pdf_text = read_pdf(pdf)
generated_summary = scrape_and_display(pdf_text, num_results=0, instructions=instructions, web_search=False, temperature=temperature, repetition_penalty=repetition_penalty, top_p=top_p)
else:
generated_summary = scrape_and_display(query, num_results=num_results, instructions=instructions, web_search=True, temperature=temperature, repetition_penalty=repetition_penalty, top_p=top_p)
# Save the generated summary to a PDF
output_pdf_path = "output_summary.pdf"
save_text_to_pdf(generated_summary, output_pdf_path)
return generated_summary, output_pdf_path
# Deploy Gradio Interface
gr.Interface(
fn=gradio_interface,
inputs=[
gr.Textbox(label="Query"),
gr.Checkbox(label="Use PDF"),
gr.File(label="Upload PDF"),
gr.Slider(minimum=1, maximum=20, value=5, step=1, label="Number of Results"),
gr.Textbox(label="Instructions"),
gr.Slider(minimum=0.1, maximum=1.0, value=0.7, step=0.1, label="Temperature"),
gr.Slider(minimum=0.1, maximum=2.0, value=1.0, step=0.1, label="Repetition Penalty"),
gr.Slider(minimum=0.1, maximum=1.0, value=0.9, step=0.1, label="Top p"),
gr.Checkbox(label="Clear Cache", visible=False) # Hidden checkbox for clear cache functionality
],
outputs=["text", gr.File(label="Generated PDF")],
title="Financial Analyst AI Assistant",
description="Enter your query about a company's financials to get valuable insights. Optionally, upload a PDF for analysis. Please instruct me for curating your output template. For web search, you can modify my search results but it's advisable to restrict them to 10. You can also adjust parameters like Temperature, Repetition Penalty, and Top_P. It's advisable to set repetition penalty at 1 and other two parameters at 0.1.",
allow_flagging="never"
).launch(share=True)