app.py

import gradio as gr
from openai import OpenAI
import requests
import json
import httpx
import os
import logging
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from bs4 import BeautifulSoup
from googlesearch import search
from newsapi import NewsApiClient
import markdown
import re
import time

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class RaindropSearchBot:
    def __init__(self):
        self.openai_api_key = os.getenv('openaikey')
        self.raindrop_api_token = os.getenv('raindroptoken')
        self.newsapi_key = os.getenv('newsapikey')
        
        if not all([self.openai_api_key, self.raindrop_api_token, self.newsapi_key]):
            raise EnvironmentError(
                "Missing required environment variables. Please ensure all API keys are set."
            )
        
        # Updated OpenAI client initialization
        self.client = OpenAI(
            api_key=self.openai_api_key,
            http_client=httpx.Client(
                timeout=60.0,
                follow_redirects=True
            )
        )
        self.newsapi = NewsApiClient(api_key=self.newsapi_key)

    def get_google_results(self, query: str, num_results: int = 5) -> List[Dict]:
        """Get Google search results using googlesearch-python."""
        try:
            search_results = []
            for result in search(query, num_results=num_results, advanced=True):
                search_results.append({
                    'title': result.title,
                    'link': result.url,
                    'snippet': result.description
                })
            return search_results
                
        except Exception as e:
            logger.error(f"Google search error: {e}")
            return []

    def get_news_results(self, query: str, num_results: int = 5) -> List[Dict]:
        """Get news articles using NewsAPI."""
        try:
            news_results = self.newsapi.get_everything(
                q=query,
                language='en',
                sort_by='relevancy',
                page_size=num_results
            )
            
            return news_results.get('articles', [])
            
        except Exception as e:
            logger.error(f"News API error: {e}")
            return []

    def extract_content_from_url(self, url: str) -> Optional[str]:
        """Extract main content from a URL using BeautifulSoup."""
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
            }
            
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # Remove unwanted elements
            for element in soup(['script', 'style', 'nav', 'header', 'footer', 'iframe']):
                element.decompose()
            
            # Get title
            title = soup.title.string if soup.title else ''
            
            # Get main content
            # First try common content containers
            content_containers = soup.select('article, main, .content, .post-content, .entry-content')
            
            if content_containers:
                content = content_containers[0].get_text(separator='\n', strip=True)
            else:
                # Fallback to all paragraphs
                paragraphs = soup.find_all('p')
                content = '\n'.join(p.get_text(strip=True) for p in paragraphs)
            
            # Combine and clean
            full_content = f"{title}\n\n{content}"
            
            # Clean up the text
            full_content = re.sub(r'\n\s*\n', '\n\n', full_content)  # Remove extra newlines
            full_content = re.sub(r'\s+', ' ', full_content)  # Normalize whitespace
            
            return full_content if full_content.strip() else None
            
        except Exception as e:
            logger.error(f"Error extracting content from {url}: {e}")
            return None

    def get_content_and_summary(self, item: Dict, source_type: str) -> Dict:
        """Get content and generate summary for a single item."""
        try:
            # Get URL based on source type
            url = item.get('link') or item.get('url')
            if not url:
                logger.warning(f"No URL found in item from {source_type}")
                return item

            # For Raindrop items, use existing excerpt if available
            if source_type == 'raindrop' and item.get('excerpt'):
                content = item['excerpt']
            else:
                content = self.extract_content_from_url(url)

            if not content:
                logger.warning(f"No content extracted from {url}")
                item['detailed_summary'] = "Content extraction failed."
                return item

            # Generate summary focused on the query topic
            try:
                prompt = f"""
                Analyze this content and provide a detailed summary focusing on key points.
                
                Content: {content[:4000]}  # Limit content length for token constraints
                
                Requirements:
                1. Focus on the most important facts and findings
                2. Include specific data points and quotes if relevant
                3. Organize the information logically
                4. Keep the summary to 2-3 paragraphs
                5. Highlight any unique insights from this source
                """

                response = self.client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.3,
                    max_tokens=300
                )
                
                item['detailed_summary'] = response.choices[0].message.content
                item['processed_content'] = content[:1000]  # Store truncated content for later use
                
            except Exception as e:
                logger.error(f"Error generating summary: {e}")
                item['detailed_summary'] = "Summary generation failed."

            return item
            
        except Exception as e:
            logger.error(f"Error processing item from {source_type}: {e}")
            return item

    def search_raindrop(self, search_query: str) -> List[Dict]:
        """Search Raindrop.io with enhanced error handling and logging."""
        logger.info(f"Searching Raindrop with query: {search_query}")

        headers = {
            "Authorization": f"Bearer {self.raindrop_api_token}"
        }
        
        # Test API connection first
        try:
            test_response = requests.get(
                "https://api.raindrop.io/rest/v1/user",
                headers=headers
            )
            if test_response.status_code != 200:
                logger.error(f"API test failed: {test_response.status_code}")
                return []
        except Exception as e:
            logger.error(f"API connection error: {e}")
            return []

        # Perform search
        try:
            params = {
                "search": search_query,
                "perpage": 50,
                "sort": "-created",
                "page": 0
            }
            
            response = requests.get(
                "https://api.raindrop.io/rest/v1/raindrops/0",
                headers=headers,
                params=params
            )
            
            if response.status_code == 200:
                data = response.json()
                items = data.get("items", [])
                logger.info(f"Found {len(items)} results")
                return items
            else:
                logger.error(f"Search failed: {response.status_code}")
                return []
        except Exception as e:
            logger.error(f"Search error: {e}")
            return []

    def process_all_results(self, raindrop_results: List[Dict], 
                          google_results: List[Dict], 
                          news_results: List[Dict]) -> Tuple[List[Dict], List[Dict], List[Dict]]:
        """Process and enrich all results with content and summaries."""
        
        processed_raindrop = []
        for item in raindrop_results:
            processed_item = self.get_content_and_summary(item, 'raindrop')
            if processed_item.get('detailed_summary'):
                processed_raindrop.append(processed_item)

        processed_google = []
        for item in google_results:
            processed_item = self.get_content_and_summary(item, 'google')
            if processed_item.get('detailed_summary'):
                processed_google.append(processed_item)

        processed_news = []
        for item in news_results:
            processed_item = self.get_content_and_summary(item, 'news')
            if processed_item.get('detailed_summary'):
                processed_news.append(processed_item)

        return processed_raindrop, processed_google, processed_news

    def generate_essay_response(self, results: Tuple[List[Dict], List[Dict], List[Dict]], 
                              user_query: str) -> str:
        """Generate a structured essay-style response with references."""
        raindrop_results, google_results, news_results = results
        
        # Collect all content for analysis
        all_content = ""
        reference_map = {}
        ref_counter = 1
    
        def get_url(item):
            """Helper function to get URL from item regardless of field name"""
            if 'link' in item:
                return item['link']
            elif 'url' in item:
                return item['url']
            return None
    
        # Process Raindrop results
        for item in raindrop_results:
            url = get_url(item)
            if url and item.get('detailed_summary'):
                all_content += f"\n{item['detailed_summary']}\n"
                reference_map[url] = ref_counter
                ref_counter += 1
    
        # Process Google results
        for item in google_results:
            url = get_url(item)
            if url and item.get('detailed_summary'):
                all_content += f"\n{item['detailed_summary']}\n"
                reference_map[url] = ref_counter
                ref_counter += 1
    
        # Process News results
        for item in news_results:
            url = get_url(item)
            if url and item.get('detailed_summary'):
                all_content += f"\n{item['detailed_summary']}\n"
                reference_map[url] = ref_counter
                ref_counter += 1
    
        try:
            prompt = f"""
            Create a comprehensive essay-style analysis about: {user_query}
            
            Use this content as your source material:
            {all_content}
    
            Requirements:
            1. Structure the response in clear sections with markdown headers
            2. Include an introduction and conclusion
            3. Use reference numbers [n] to cite sources
            4. Make connections between different sources
            5. Highlight key findings and trends
            6. Address any contradictions or gaps
            7. Use markdown formatting for better readability
            
            Format the response as a proper academic essay with sections.
            """
    
            response = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0.5,
                max_tokens=1500
            )
            
            essay = response.choices[0].message.content
    
            # Replace reference placeholders with actual reference numbers
            for url, ref_num in reference_map.items():
                essay = essay.replace(f'[URL:{url}]', f'[{ref_num}]')
    
            return essay
    
        except Exception as e:
            logger.error(f"Error generating essay: {e}")
            return "Error generating analysis."


    def format_results(self, results: Tuple[List[Dict], List[Dict], List[Dict]], 
                      essay: str) -> str:
        """Format the essay and results with detailed summaries."""
        raindrop_results, google_results, news_results = results
        
        output = f"{essay}\n\n"
        output += "---\n\n"
        output += "# References and Detailed Summaries\n\n"
        
        ref_counter = 1

        # Format Raindrop results
        if raindrop_results:
            output += "## 🔍 Bookmarked Sources\n\n"
            for item in raindrop_results:
                output += f"### [{ref_counter}] {item.get('title', 'No Title')}\n"
                output += f"**Link**: {item.get('link')}\n"
                if item.get('tags'):
                    output += f"**Tags**: {', '.join(item['tags'])}\n"
                if item.get('created'):
                    output += f"**Created**: {item['created'][:10]}\n"
                output += "\n**Summary**:\n"
                output += f"{item.get('detailed_summary', 'No summary available.')}\n\n"
                ref_counter += 1

        # Format Google results
        if google_results:
            output += "## 🌐 Web Sources\n\n"
            for item in google_results:
                output += f"### [{ref_counter}] {item.get('title', 'No Title')}\n"
                output += f"**Link**: {item.get('link')}\n"
                output += "\n**Summary**:\n"
                output += f"{item.get('detailed_summary', 'No summary available.')}\n\n"
                ref_counter += 1

        # Format News results
        if news_results:
            output += "## 📰 Recent News\n\n"
            for item in news_results:
                output += f"### [{ref_counter}] {item.get('title', 'No Title')}\n"
                output += f"**Link**: {item.get('url')}\n"
                if item.get('source', {}).get('name'):
                    output += f"**Source**: {item['source']['name']}\n"
                if item.get('publishedAt'):
                    output += f"**Published**: {item['publishedAt'][:10]}\n"
                output += "\n**Summary**:\n"
                output += f"{item.get('detailed_summary', 'No summary available.')}\n\n"
                ref_counter += 1

        return output

    def process_request(self, user_request: str) -> str:
        """Process the user request with enhanced content collection and analysis."""
        try:
            logger.info(f"Processing request: {user_request}")
            
            # Generate search query
            search_query = self.generate_search_query(user_request)
            logger.info(f"Using search query: {search_query}")
            
            # Get results from all sources
            raindrop_results = self.search_raindrop(search_query)
            google_results = self.get_google_results(search_query)
            news_results = self.get_news_results(search_query)
            
            # Process all results to get content and summaries
            processed_results = self.process_all_results(
                raindrop_results, google_results, news_results
            )
            
            # Generate essay-style analysis
            essay = self.generate_essay_response(processed_results, user_request)
            
            # Format and return results
            return self.format_results(processed_results, essay)
            
        except Exception as e:
            logger.error(f"Error processing request: {e}", exc_info=True)
            return f"An error occurred while processing your request. Please try again."

    def generate_search_query(self, user_request: str) -> str:
        """Convert user request to optimized search terms."""
        logger.info(f"Generating search query for: {user_request}")
        
        prompt = f"""
        You are a search expert. Create a search query to find relevant documents in raindrop about:
        {user_request}
        
        Guidelines:
        - Focus on key concepts and synonyms
        - Use combination of keywords that would appear in titles or descriptions
        - Return only the search terms, no explanation
        - Include alternative phrasings
        - Keep it concise (max 6-8 key terms/phrases)
        - use the formatting authorised in raindrop search:
            o use " for exact search (ex: "artificial intelligence")
            o use - to exclude some terms (ex: -math)
            o use match:OR for alternatives (ex: apple banana match:OR)
            o use parenthesis for combinations ( ex: sugar (banana apple match:OR) match:AND )
            o use match:AND for inclusion of both cases systematically (ex: apple banana match:AND)
        Example elaborate request: ("artificial intelligence" AI match:OR) -"machine learning"
        
        Return only the search query terms.
        """

        try:
            response = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0.3,
                max_tokens=50
            )
            search_query = response.choices[0].message.content.strip()
            logger.info(f"Generated search query: {search_query}")
            return search_query
        except Exception as e:
            logger.error(f"Error generating search query: {e}")
            return user_request

# Initialize bot
bot = RaindropSearchBot()

# Create Gradio interface
def chatbot_interface(user_input: str) -> str:
    return bot.process_request(user_input)

def convert_to_markdown(output_text: str) -> gr.Markdown:
    try:
        # Create a new Gradio Markdown component with the output text
        output_textMarkdown = gr.Markdown(
            value=output_text,
            render=True,
            visible=True
        )
        return output_textMarkdown
    except Exception as e:
        logger.error(f"Error converting to markdown: {e}")
        # Return error message as markdown if conversion fails
        return gr.Markdown(
            value="Error converting content to markdown format. Please try again.",
            visible=True
        )

# Define and launch the interface
with gr.Blocks(title="Enhanced Search Assistant", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🔍 Enhanced Search Assistant
    Enter your search request in natural language, and I'll find and analyze information from multiple sources:
    - Your bookmarked content
    - Web search results
    - Recent news articles
    """)
    
    with gr.Row():
        input_text = gr.Textbox(
            label="What would you like to search for?",
            placeholder="Enter your search query here...",
            lines=2
        )
    
    with gr.Row():
        searchbutton = gr.Button("🔍 Search", variant="primary")
    
    with gr.Column():
        with gr.Accordion("Editable version", open=False):
            with gr.Column():
                output_text = gr.Textbox(
                    label="Analysis and Results - editable",
                    lines=20,
                    interactive=True
                )
                refreshbutton = gr.Button("Refresh", variant="primary")
        output_textMarkdown = gr.Markdown(
                label="Analysis and Results",
                height=600,
                max_height=800
            )
    
    searchbutton.click(
        fn=chatbot_interface,
        inputs=input_text,
        outputs=output_text
    ).then(
        fn=convert_to_markdown,
        inputs=output_text,
        outputs=output_textMarkdown)

    refreshbutton.click(
        fn=convert_to_markdown,
        inputs=output_text,
        outputs=output_textMarkdown)
    
# Launch the interface
if __name__ == "__main__":
    demo.launch(share=True)