From 5e0036f8b477bbd140ac7ea53b577339bac2a54e Mon Sep 17 00:00:00 2001 From: Aparup Ganguly Date: Thu, 20 Feb 2025 00:05:12 +0530 Subject: [PATCH 1/4] Implemented gemini 2.0 web extractor --- .../gemini-2.0-web-extractor.py | 210 ++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py diff --git a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py new file mode 100644 index 00000000..35d5ba8d --- /dev/null +++ b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py @@ -0,0 +1,210 @@ +import os +import json +import time +import requests +from dotenv import load_dotenv +from serpapi.google_search import GoogleSearch +from google import genai + +# ANSI color codes +class Colors: + CYAN = '\033[96m' + YELLOW = '\033[93m' + GREEN = '\033[92m' + RED = '\033[91m' + MAGENTA = '\033[95m' + BLUE = '\033[94m' + RESET = '\033[0m' + +# Load environment variables +load_dotenv() + +# Initialize clients +client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY")) +firecrawl_api_key = os.getenv("FIRECRAWL_API_KEY") +serp_api_key = os.getenv("SERP_API_KEY") + +# Add this debug print (remember to remove it before committing) +if not firecrawl_api_key: + print(f"{Colors.RED}Warning: FIRECRAWL_API_KEY not found in environment variables{Colors.RESET}") + +def search_google(query): + """Search Google using SerpAPI and return top results.""" + print(f"{Colors.YELLOW}Searching Google for '{query}'...{Colors.RESET}") + search = GoogleSearch({"q": query, "api_key": serp_api_key}) + results = search.get_dict().get("organic_results", []) + + print(f"{Colors.CYAN}Found {len(results)} search results{Colors.RESET}") + if results: + print("First result:", results[0]) + return results + +def select_urls_with_r1(company, objective, serp_results): + """ + Use Gemini 2.0 Flash to select URLs from SERP results. + Returns a list of URLs. + """ + try: + print(f"{Colors.CYAN}Processing {len(serp_results)} search results...{Colors.RESET}") + + serp_data = [{"title": r.get("title"), "link": r.get("link"), "snippet": r.get("snippet")} + for r in serp_results if r.get("link")] + + print(f"{Colors.CYAN}Prepared {len(serp_data)} valid results for processing{Colors.RESET}") + + prompt = ( + "You are a URL selector that always responds with valid JSON. " + f"Company: {company}\n" + f"Objective: {objective}\n" + f"SERP Results: {json.dumps(serp_data)}\n\n" + "Return a JSON object with a property 'selected_urls' that contains an array " + "of URLs most likely to help meet the objective. Add a /* to the end of the URL if you think it should search all of the pages in the site. " + "Do not return any social media links. For example: {\"selected_urls\": [\"https://example.com\", \"https://example2.com\"]}" + ) + + print(f"{Colors.CYAN}Calling Gemini API...{Colors.RESET}") + + response = client.models.generate_content( + model="gemini-2.0-flash", + contents=prompt + ) + + print(f"{Colors.CYAN}Gemini response: {response.text}{Colors.RESET}") + + try: + # Remove the markdown code block markers if they exist + cleaned_response = response.text.replace('```json\n', '').replace('\n```', '') + result = json.loads(cleaned_response) + + if isinstance(result, dict) and "selected_urls" in result: + urls = result["selected_urls"] + else: + urls = [] + except json.JSONDecodeError as e: + print(f"{Colors.RED}JSON parsing error: {e}{Colors.RESET}") + urls = [] + + if not urls: + print(f"{Colors.YELLOW}No valid URLs found.{Colors.RESET}") + return [] + + print(f"{Colors.CYAN}Selected URLs for extraction:{Colors.RESET}") + for url in urls: + print(f"- {url}") + + return urls + + except Exception as e: + print(f"{Colors.RED}Error selecting URLs: {e}{Colors.RESET}") + return [] + +def extract_company_info(urls, prompt, company, api_key): + if not api_key: + print(f"{Colors.RED}Error: Firecrawl API key is missing or invalid{Colors.RESET}") + return None + + print(f"{Colors.YELLOW}Using API key: {api_key[:8]}...{Colors.RESET}") # Only show first 8 chars for security + """Use requests to call Firecrawl's extract endpoint with selected URLs.""" + print(f"{Colors.YELLOW}Extracting structured data from the provided URLs using Firecrawl...{Colors.RESET}") + + headers = { + 'Content-Type': 'application/json', + 'Authorization': f'Bearer {api_key}' + } + + payload = { + "urls": urls, + "prompt": prompt + " for " + company, + "enableWebSearch": True + } + + try: + response = requests.post( + "https://api.firecrawl.dev/v1/extract", + headers=headers, + json=payload, + timeout=30 + ) + + data = response.json() + + if not data.get('success'): + print(f"{Colors.RED}API returned error: {data.get('error', 'No error message')}{Colors.RESET}") + return None + + extraction_id = data.get('id') + if not extraction_id: + print(f"{Colors.RED}No extraction ID found in response.{Colors.RESET}") + return None + + return poll_firecrawl_result(extraction_id, api_key) + + except requests.exceptions.RequestException as e: + print(f"{Colors.RED}Request failed: {e}{Colors.RESET}") + return None + except json.JSONDecodeError as e: + print(f"{Colors.RED}Failed to parse response: {e}{Colors.RESET}") + return None + except Exception as e: + print(f"{Colors.RED}Failed to extract data: {e}{Colors.RESET}") + return None + +def poll_firecrawl_result(extraction_id, api_key, interval=5, max_attempts=36): + """Poll Firecrawl API to get the extraction result.""" + url = f"https://api.firecrawl.dev/v1/extract/{extraction_id}" + headers = { + 'Authorization': f'Bearer {api_key}' + } + + for attempt in range(1, max_attempts + 1): + try: + response = requests.get(url, headers=headers, timeout=30) + response.raise_for_status() + data = response.json() + + if data.get('success') and data.get('data'): + print(f"{Colors.GREEN}Data successfully extracted:{Colors.RESET}") + print(json.dumps(data['data'], indent=2)) + return data['data'] + elif data.get('success') and not data.get('data'): + time.sleep(interval) + else: + print(f"{Colors.RED}API Error: {data.get('error', 'No error message provided')}{Colors.RESET}") + return None + + except requests.exceptions.RequestException: + return None + except json.JSONDecodeError: + return None + except Exception: + return None + + print(f"{Colors.RED}Max polling attempts reached. Extraction did not complete in time.{Colors.RESET}") + return None + +def main(): + company = input(f"{Colors.BLUE}Enter the company name: {Colors.RESET}") + objective = input(f"{Colors.BLUE}Enter what information you want about the company: {Colors.RESET}") + + # Make the search query more specific + serp_results = search_google(f"{company} company pricing website") + if not serp_results: + print(f"{Colors.RED}No search results found.{Colors.RESET}") + return + + # Use Gemini 2.0 Flash for URL selection + selected_urls = select_urls_with_r1(company, objective, serp_results) + + if not selected_urls: + print(f"{Colors.RED}No URLs were selected.{Colors.RESET}") + return + + data = extract_company_info(selected_urls, objective, company, firecrawl_api_key) + + if data: + print(f"{Colors.GREEN}Extraction completed successfully.{Colors.RESET}") + else: + print(f"{Colors.RED}Failed to extract the requested information. Try refining your prompt or choosing a different company.{Colors.RESET}") + +if __name__ == "__main__": + main() From 3a4ef05a705fb7c34e22c36d59e2256e6927e4ce Mon Sep 17 00:00:00 2001 From: Aparup Ganguly Date: Thu, 20 Feb 2025 01:30:59 +0530 Subject: [PATCH 2/4] Output imporvements --- .../gemini-2.0-web-extractor.py | 60 +++++++++++-------- 1 file changed, 36 insertions(+), 24 deletions(-) diff --git a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py index 35d5ba8d..69fd7e51 100644 --- a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py +++ b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py @@ -24,7 +24,6 @@ client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY")) firecrawl_api_key = os.getenv("FIRECRAWL_API_KEY") serp_api_key = os.getenv("SERP_API_KEY") -# Add this debug print (remember to remove it before committing) if not firecrawl_api_key: print(f"{Colors.RED}Warning: FIRECRAWL_API_KEY not found in environment variables{Colors.RESET}") @@ -34,12 +33,11 @@ def search_google(query): search = GoogleSearch({"q": query, "api_key": serp_api_key}) results = search.get_dict().get("organic_results", []) - print(f"{Colors.CYAN}Found {len(results)} search results{Colors.RESET}") if results: - print("First result:", results[0]) + print("Finding Results...") return results -def select_urls_with_r1(company, objective, serp_results): +def select_urls_with_gemini(company, objective, serp_results): """ Use Gemini 2.0 Flash to select URLs from SERP results. Returns a list of URLs. @@ -53,13 +51,12 @@ def select_urls_with_r1(company, objective, serp_results): print(f"{Colors.CYAN}Prepared {len(serp_data)} valid results for processing{Colors.RESET}") prompt = ( - "You are a URL selector that always responds with valid JSON. " + "You are a URL selector that always responds with valid JSON. You select URLs from the SERP results relevant to the company and objective. Your response must be a JSON object with a 'selected_urls' array property containing strings.\n\n" f"Company: {company}\n" f"Objective: {objective}\n" f"SERP Results: {json.dumps(serp_data)}\n\n" "Return a JSON object with a property 'selected_urls' that contains an array " - "of URLs most likely to help meet the objective. Add a /* to the end of the URL if you think it should search all of the pages in the site. " - "Do not return any social media links. For example: {\"selected_urls\": [\"https://example.com\", \"https://example2.com\"]}" + "of URLs most likely to help meet the objective. Add a /* to the end of the URL if you think it should search all of the pages in the site. Do not return any social media links. For example: {\"selected_urls\": [\"https://example.com\", \"https://example2.com\"]}" ) print(f"{Colors.CYAN}Calling Gemini API...{Colors.RESET}") @@ -67,35 +64,50 @@ def select_urls_with_r1(company, objective, serp_results): response = client.models.generate_content( model="gemini-2.0-flash", contents=prompt - ) + ) - print(f"{Colors.CYAN}Gemini response: {response.text}{Colors.RESET}") + # Get response text and clean it up + response_text = response.text.strip() + + # Remove markdown code block if present + if response_text.startswith('```'): + response_text = response_text.split('\n', 1)[1] # Remove first line + if response_text.endswith('```'): + response_text = response_text.rsplit('\n', 1)[0] # Remove last line + if response_text.startswith('json'): + response_text = response_text.split('\n', 1)[1] # Remove "json" line + + response_text = response_text.strip() try: - # Remove the markdown code block markers if they exist - cleaned_response = response.text.replace('```json\n', '').replace('\n```', '') - result = json.loads(cleaned_response) - + result = json.loads(response_text) if isinstance(result, dict) and "selected_urls" in result: urls = result["selected_urls"] else: - urls = [] - except json.JSONDecodeError as e: - print(f"{Colors.RED}JSON parsing error: {e}{Colors.RESET}") - urls = [] + urls = [line.strip() for line in response_text.split('\n') + if line.strip().startswith(('http://', 'https://'))] + except json.JSONDecodeError: + print(f"{Colors.YELLOW}Failed to parse JSON, falling back to text parsing{Colors.RESET}") + # If JSON parsing fails, fall back to text parsing + urls = [line.strip() for line in response_text.split('\n') + if line.strip().startswith(('http://', 'https://'))] - if not urls: + # Clean up URLs - remove wildcards and trailing slashes + cleaned_urls = [url.replace('/*', '').rstrip('/') for url in urls] + cleaned_urls = [url for url in cleaned_urls if url] + + if not cleaned_urls: print(f"{Colors.YELLOW}No valid URLs found.{Colors.RESET}") return [] print(f"{Colors.CYAN}Selected URLs for extraction:{Colors.RESET}") - for url in urls: + for url in cleaned_urls: print(f"- {url}") - return urls + return cleaned_urls except Exception as e: - print(f"{Colors.RED}Error selecting URLs: {e}{Colors.RESET}") + print(f"{Colors.RED}Error selecting URLs: {str(e)}{Colors.RESET}") return [] def extract_company_info(urls, prompt, company, api_key): @@ -103,7 +115,7 @@ def extract_company_info(urls, prompt, company, api_key): print(f"{Colors.RED}Error: Firecrawl API key is missing or invalid{Colors.RESET}") return None - print(f"{Colors.YELLOW}Using API key: {api_key[:8]}...{Colors.RESET}") # Only show first 8 chars for security + """Use requests to call Firecrawl's extract endpoint with selected URLs.""" print(f"{Colors.YELLOW}Extracting structured data from the provided URLs using Firecrawl...{Colors.RESET}") @@ -187,13 +199,13 @@ def main(): objective = input(f"{Colors.BLUE}Enter what information you want about the company: {Colors.RESET}") # Make the search query more specific - serp_results = search_google(f"{company} company pricing website") + serp_results = search_google(f"{company}") if not serp_results: print(f"{Colors.RED}No search results found.{Colors.RESET}") return # Use Gemini 2.0 Flash for URL selection - selected_urls = select_urls_with_r1(company, objective, serp_results) + selected_urls = select_urls_with_gemini(company, objective, serp_results) if not selected_urls: print(f"{Colors.RED}No URLs were selected.{Colors.RESET}") From 21b22d9f5cf2b6ce2dfa49126e8f6c1c85044d31 Mon Sep 17 00:00:00 2001 From: Aparup Ganguly Date: Thu, 20 Feb 2025 02:21:43 +0530 Subject: [PATCH 3/4] Output re-structured --- .../gemini-2.0-web-extractor.py | 85 +++++++++---------- 1 file changed, 38 insertions(+), 47 deletions(-) diff --git a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py index 69fd7e51..97675799 100644 --- a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py +++ b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py @@ -24,6 +24,7 @@ client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY")) firecrawl_api_key = os.getenv("FIRECRAWL_API_KEY") serp_api_key = os.getenv("SERP_API_KEY") +# Add this debug print (remember to remove it before committing) if not firecrawl_api_key: print(f"{Colors.RED}Warning: FIRECRAWL_API_KEY not found in environment variables{Colors.RESET}") @@ -31,11 +32,7 @@ def search_google(query): """Search Google using SerpAPI and return top results.""" print(f"{Colors.YELLOW}Searching Google for '{query}'...{Colors.RESET}") search = GoogleSearch({"q": query, "api_key": serp_api_key}) - results = search.get_dict().get("organic_results", []) - - if results: - print("Finding Results...") - return results + return search.get_dict().get("organic_results", []) def select_urls_with_gemini(company, objective, serp_results): """ @@ -43,61 +40,55 @@ def select_urls_with_gemini(company, objective, serp_results): Returns a list of URLs. """ try: - print(f"{Colors.CYAN}Processing {len(serp_results)} search results...{Colors.RESET}") - serp_data = [{"title": r.get("title"), "link": r.get("link"), "snippet": r.get("snippet")} for r in serp_results if r.get("link")] - - print(f"{Colors.CYAN}Prepared {len(serp_data)} valid results for processing{Colors.RESET}") prompt = ( - "You are a URL selector that always responds with valid JSON. You select URLs from the SERP results relevant to the company and objective. Your response must be a JSON object with a 'selected_urls' array property containing strings.\n\n" + "Task: Select relevant URLs from search results.\n\n" + "Instructions:\n" + "1. Analyze the search results for information about the specified company\n" + "2. Select URLs that are most likely to contain the requested information\n" + "3. Return ONLY a JSON object with the following structure: {\"selected_urls\": [\"url1\", \"url2\"]}\n" + "4. Do not include social media links\n\n" f"Company: {company}\n" - f"Objective: {objective}\n" - f"SERP Results: {json.dumps(serp_data)}\n\n" - "Return a JSON object with a property 'selected_urls' that contains an array " - "of URLs most likely to help meet the objective. Add a /* to the end of the URL if you think it should search all of the pages in the site. Do not return any social media links. For example: {\"selected_urls\": [\"https://example.com\", \"https://example2.com\"]}" + f"Information Needed: {objective}\n" + f"Search Results: {json.dumps(serp_data, indent=2)}\n\n" + "Response Format: {\"selected_urls\": [\"https://example.com\", \"https://example2.com\"]}" ) - print(f"{Colors.CYAN}Calling Gemini API...{Colors.RESET}") - response = client.models.generate_content( model="gemini-2.0-flash", contents=prompt - ) + ) - # Get response text and clean it up - response_text = response.text.strip() - - # Remove markdown code block if present - if response_text.startswith('```'): - response_text = response_text.split('\n', 1)[1] # Remove first line - if response_text.endswith('```'): - response_text = response_text.rsplit('\n', 1)[0] # Remove last line - if response_text.startswith('json'): - response_text = response_text.split('\n', 1)[1] # Remove "json" line - - response_text = response_text.strip() + # Clean the response text + cleaned_response = response.text.strip() + if cleaned_response.startswith('```'): + cleaned_response = cleaned_response.split('```')[1] + if cleaned_response.startswith('json'): + cleaned_response = cleaned_response[4:] + cleaned_response = cleaned_response.strip() try: - result = json.loads(response_text) + # Parse JSON response + result = json.loads(cleaned_response) if isinstance(result, dict) and "selected_urls" in result: urls = result["selected_urls"] else: - urls = [line.strip() for line in response_text.split('\n') + # Fallback to text parsing + urls = [line.strip() for line in cleaned_response.split('\n') if line.strip().startswith(('http://', 'https://'))] except json.JSONDecodeError: - print(f"{Colors.YELLOW}Failed to parse JSON, falling back to text parsing{Colors.RESET}") - # If JSON parsing fails, fall back to text parsing - urls = [line.strip() for line in response_text.split('\n') + # Fallback to text parsing + urls = [line.strip() for line in cleaned_response.split('\n') if line.strip().startswith(('http://', 'https://'))] - # Clean up URLs - remove wildcards and trailing slashes + # Clean up URLs cleaned_urls = [url.replace('/*', '').rstrip('/') for url in urls] cleaned_urls = [url for url in cleaned_urls if url] if not cleaned_urls: - print(f"{Colors.YELLOW}No valid URLs found.{Colors.RESET}") + print(f"{Colors.YELLOW}No valid URLs found in response.{Colors.RESET}") return [] print(f"{Colors.CYAN}Selected URLs for extraction:{Colors.RESET}") @@ -111,11 +102,6 @@ def select_urls_with_gemini(company, objective, serp_results): return [] def extract_company_info(urls, prompt, company, api_key): - if not api_key: - print(f"{Colors.RED}Error: Firecrawl API key is missing or invalid{Colors.RESET}") - return None - - """Use requests to call Firecrawl's extract endpoint with selected URLs.""" print(f"{Colors.YELLOW}Extracting structured data from the provided URLs using Firecrawl...{Colors.RESET}") @@ -161,13 +147,15 @@ def extract_company_info(urls, prompt, company, api_key): print(f"{Colors.RED}Failed to extract data: {e}{Colors.RESET}") return None -def poll_firecrawl_result(extraction_id, api_key, interval=5, max_attempts=36): +def poll_firecrawl_result(extraction_id, api_key, interval=10, max_attempts=60): """Poll Firecrawl API to get the extraction result.""" url = f"https://api.firecrawl.dev/v1/extract/{extraction_id}" headers = { 'Authorization': f'Bearer {api_key}' } + print(f"{Colors.YELLOW}Waiting for extraction to complete...{Colors.RESET}") + for attempt in range(1, max_attempts + 1): try: response = requests.get(url, headers=headers, timeout=30) @@ -179,16 +167,21 @@ def poll_firecrawl_result(extraction_id, api_key, interval=5, max_attempts=36): print(json.dumps(data['data'], indent=2)) return data['data'] elif data.get('success') and not data.get('data'): + if attempt % 6 == 0: # Show progress every minute + print(f"{Colors.YELLOW}Still processing... (attempt {attempt}/{max_attempts}){Colors.RESET}") time.sleep(interval) else: print(f"{Colors.RED}API Error: {data.get('error', 'No error message provided')}{Colors.RESET}") return None - except requests.exceptions.RequestException: + except requests.exceptions.RequestException as e: + print(f"{Colors.RED}Request error: {str(e)}{Colors.RESET}") return None - except json.JSONDecodeError: + except json.JSONDecodeError as e: + print(f"{Colors.RED}JSON parsing error: {str(e)}{Colors.RESET}") return None - except Exception: + except Exception as e: + print(f"{Colors.RED}Unexpected error: {str(e)}{Colors.RESET}") return None print(f"{Colors.RED}Max polling attempts reached. Extraction did not complete in time.{Colors.RESET}") @@ -198,13 +191,11 @@ def main(): company = input(f"{Colors.BLUE}Enter the company name: {Colors.RESET}") objective = input(f"{Colors.BLUE}Enter what information you want about the company: {Colors.RESET}") - # Make the search query more specific serp_results = search_google(f"{company}") if not serp_results: print(f"{Colors.RED}No search results found.{Colors.RESET}") return - # Use Gemini 2.0 Flash for URL selection selected_urls = select_urls_with_gemini(company, objective, serp_results) if not selected_urls: From 8caeab269129a7b40db9dda16c761d13fe71069d Mon Sep 17 00:00:00 2001 From: Aparup Ganguly Date: Thu, 20 Feb 2025 02:24:49 +0530 Subject: [PATCH 4/4] minor changes --- examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py index 97675799..937d6865 100644 --- a/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py +++ b/examples/gemini-2.0-web-extractor/gemini-2.0-web-extractor.py @@ -24,7 +24,7 @@ client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY")) firecrawl_api_key = os.getenv("FIRECRAWL_API_KEY") serp_api_key = os.getenv("SERP_API_KEY") -# Add this debug print (remember to remove it before committing) + if not firecrawl_api_key: print(f"{Colors.RED}Warning: FIRECRAWL_API_KEY not found in environment variables{Colors.RESET}") @@ -167,7 +167,7 @@ def poll_firecrawl_result(extraction_id, api_key, interval=10, max_attempts=60): print(json.dumps(data['data'], indent=2)) return data['data'] elif data.get('success') and not data.get('data'): - if attempt % 6 == 0: # Show progress every minute + if attempt % 6 == 0: print(f"{Colors.YELLOW}Still processing... (attempt {attempt}/{max_attempts}){Colors.RESET}") time.sleep(interval) else: