Spaces:
Sleeping
Sleeping
| 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) |