import gradio as gr from openai import OpenAI import requests import json import os import logging from typing import Dict, List from datetime import datetime # Set up logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'raindrop_search_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'), logging.StreamHandler() ] ) logger = logging.getLogger('RaindropSearchBot') OPENAI_API_KEY = os.getenv('openaikey') RAINDROP_TOKEN = os.getenv('raindroptoken') if not OPENAI_API_KEY or not RAINDROP_TOKEN: logger.error("Missing environment variables") raise EnvironmentError( "Missing required environment variables. Please ensure 'openaikey' and 'raindroptoken' are set." ) class RaindropSearchBot: def __init__(self): self.openai_api_key = OPENAI_API_KEY self.raindrop_api_token = RAINDROP_TOKEN self.client = OpenAI(api_key=self.openai_api_key) logger.info("RaindropSearchBot initialized") def generate_search_query(self, user_request: str) -> str: """Convert user request to a tailored search query using OpenAI.""" logger.info(f"Generating search query for request: {user_request}") prompt = f""" Convert this request into a simple search query for finding bookmarks about this topic. Use quotes for exact phrases and OR between alternatives. Only return the search query itself, no explanations. Keep it concise and focused. Request: {user_request} """ try: response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.3, # Lower temperature for more focused results max_tokens=100 # Reduced tokens since we only need the query ) 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: {str(e)}") # Fallback to a simple search if query generation fails return user_request def search_raindrop(self, search_query: str) -> List[Dict]: """Search Raindrop.io with the generated query.""" logger.info(f"Original search query: {search_query}") # Extract the actual search terms from between the backticks try: if "```" in search_query: search_terms = search_query.split("```")[1].strip() else: search_terms = search_query.strip() # Clean up the search terms search_terms = search_terms.replace('\n', ' ').strip() logger.info(f"Cleaned search terms: {search_terms}") except Exception as e: logger.error(f"Error processing search query: {e}") search_terms = search_query headers = { "Authorization": f"Bearer {self.raindrop_api_token}", "Content-Type": "application/json" } # Test the API connection first try: test_response = requests.get( "https://api.raindrop.io/rest/v1/user", headers=headers ) logger.info(f"API test response status: {test_response.status_code}") if test_response.status_code != 200: logger.error(f"API test failed: {test_response.text}") return [] # Search parameters params = { "search": search_terms, "page": 0, "perpage": 50, "sort": "-created" } logger.info(f"Making request to Raindrop API with params: {params}") # Make the search request response = requests.get( "https://api.raindrop.io/rest/v1/raindrops/0", headers=headers, params=params ) logger.info(f"Search response status code: {response.status_code}") if response.status_code == 200: data = response.json() items = data.get("items", []) logger.info(f"Found {len(items)} results") logger.debug(f"Response data: {json.dumps(data, indent=2)}") return items else: logger.error(f"Search failed with status {response.status_code}: {response.text}") return [] except Exception as e: logger.error(f"Error during Raindrop search: {str(e)}", exc_info=True) return [] def generate_summary(self, item: Dict) -> str: logger.info(f"Generating summary for item: {item.get('title', 'No Title')}") content = f""" Title: {item.get('title', 'No Title')} Description: {item.get('excerpt', '')} Tags: {', '.join(item.get('tags', []))} Type: {item.get('type', 'unknown')} """ prompt = f""" Please provide a brief, informative summary of this bookmarked content: {content} Keep the summary concise but include key points and relevance. """ try: response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=150 ) summary = response.choices[0].message.content logger.info(f"Generated summary: {summary}") return summary except Exception as e: logger.error(f"Error generating summary: {str(e)}") return "Summary generation failed" def format_results(self, results: List[Dict]) -> str: logger.info(f"Formatting {len(results)} results") if not results: logger.warning("No results to format") return "No results found." formatted_output = "🔍 Search Results:\n\n" for idx, item in enumerate(results, 1): logger.debug(f"Formatting item {idx}: {item.get('title', 'No Title')}") formatted_output += f"{idx}. {item.get('title', 'No Title')}\n" formatted_output += f" Link: {item.get('link', 'No Link')}\n" if item.get('tags'): formatted_output += f" Tags: {', '.join(item['tags'])}\n" formatted_output += f" Type: {item.get('type', 'unknown')} | Created: {item.get('created', 'unknown')}\n" summary = self.generate_summary(item) formatted_output += f" Summary: {summary}\n\n" logger.info("Results formatting completed") return formatted_output def process_request(self, user_request: str) -> str: logger.info(f"Processing new request: {user_request}") try: search_query = self.generate_search_query(user_request) logger.info(f"Using search query: {search_query}") results = self.search_raindrop(search_query) logger.info(f"Found {len(results)} results") formatted_results = self.format_results(results) logger.info("Request processing completed") return formatted_results except Exception as e: logger.error(f"Error processing request: {str(e)}", exc_info=True) return f"An error occurred: {str(e)}\nPlease check the logs for more details." # Initialize the bot bot = RaindropSearchBot() # Create Gradio interface def chatbot_interface(user_input: str) -> str: logger.info(f"New interface request received: {user_input}") result = bot.process_request(user_input) logger.info("Interface request completed") return result # Define and launch the Gradio interface with gr.Blocks(title="Raindrop.io Link Search Assistant", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🔍 Raindrop.io Link Search Assistant Enter your search request in natural language, and I'll find and summarize relevant bookmarked links from your Raindrop.io collection. """) with gr.Row(): input_text = gr.Textbox( label="What would you like to search for?", placeholder="Ex: Give me the list of all the links related to blockchain in Hong Kong", lines=2 ) with gr.Row(): search_button = gr.Button("🔍 Search", variant="primary") with gr.Row(): output_text = gr.Textbox( label="Search Results", lines=15, interactive=False ) search_button.click( fn=chatbot_interface, inputs=input_text, outputs=output_text ) gr.Markdown(""" ### How to use: 1. Enter your search request in natural language 2. Click the Search button 3. View the results and summaries from your Raindrop.io bookmarks The assistant will: - Convert your request into an optimized search query - Search across all your collections - Generate summaries for each result - Present the findings with relevant metadata """) if __name__ == "__main__": logger.info("Starting Gradio interface") demo.launch(share=True)