import os import requests from bs4 import BeautifulSoup from typing import List, Dict, Optional from dotenv import load_dotenv import json from datetime import datetime import time import random from urllib.parse import quote_plus # Load environment variables load_dotenv() HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36' } try: from selenium import webdriver from selenium.webdriver.chrome.options import Options as ChromeOptions SELENIUM_AVAILABLE = True except ImportError: SELENIUM_AVAILABLE = False webdriver = None try: from scrapy.selector import Selector as ScrapySelector SCRAPY_AVAILABLE = True except ImportError: SCRAPY_AVAILABLE = False ScrapySelector = None # Add nodriver as primary stealth option try: import nodriver as uc import asyncio NODRIVER_AVAILABLE = True except ImportError: NODRIVER_AVAILABLE = False uc = None class WebSearcher: def __init__(self): self.google_api_key = os.getenv('GOOGLE_API_KEY') self.google_cx = os.getenv('GOOGLE_CX') # Custom Search Engine ID # Use free DuckDuckGo search if Google keys are missing or USE_FREE_SEARCH=1 self.use_duckduckgo = False # Disabled – we only want Google & Bing if not self.google_api_key or not self.google_cx: raise ValueError("GOOGLE_API_KEY and GOOGLE_CX environment variables must be set") self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36' }) def search(self, query: str, location: str = "us", search_type: str = "organic", max_results: int = 15) -> List[ Dict]: """ Perform a web search and return aggregated results. The default priority when GOOGLE_ONLY is not set is: 1. Google HTML scrape 2. Bing HTML scrape Both engines are **always** queried so you get a blended result set. DuckDuckGo is disabled by default. Set the environment variable `GOOGLE_ONLY=1` if you want to bypass DuckDuckGo and Bing completely and fetch *only* Google results (scraped HTML, no API key required). Args: query: Search phrase location: Two-letter country code (used only for Google-API path) search_type: "organic" or "news" (news only applies to Google API) max_results: number of unique results requested """ google_only = os.getenv("GOOGLE_ONLY", "0") == "1" results: List[Dict] = [] seen: set[str] = set() # Helper to add unique results def _add_unique(batch): nonlocal results, seen for r in batch: if r.get("url") and r["url"] not in seen: results.append(r); seen.add(r["url"]) if len(results) == max_results: return True # reached limit return False if google_only: # Respect explicit override – only Google HTML scrape _add_unique(self._search_google_html(query, max_results)) else: # Always get Google *and* Bing results (in that order) _add_unique(self._search_google_html(query, max_results)) _add_unique(self._search_bing_html(query, max_results)) # Return up to max_results merged & deduplicated return results[:max_results] def _fetch_with_selenium(self, url: str) -> Optional[str]: """Headless-browser fallback to get full HTML for pages that block requests.""" if not SELENIUM_AVAILABLE: return None try: from webdriver_manager.chrome import ChromeDriverManager options = ChromeOptions() options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--no-sandbox') options.add_argument(f'user-agent={HEADERS["User-Agent"]}') driver = webdriver.Chrome(ChromeDriverManager().install(), options=options) driver.get(url) html = driver.page_source driver.quit() soup = BeautifulSoup(html, 'html.parser') for element in soup.find_all(['script', 'style', 'nav', 'header', 'footer']): element.decompose() text = ' '.join(soup.stripped_strings) text = ' '.join(text.split()) return text if text else None except Exception: return None def _fetch_with_nodriver(self, url: str) -> Optional[str]: """Use nodriver for maximum stealth and success rate against anti-bot systems.""" if not NODRIVER_AVAILABLE: return None try: # Create a new event loop for this thread if none exists try: loop = asyncio.get_event_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # Run the async nodriver fetch return loop.run_until_complete(self._async_fetch_with_nodriver(url)) except Exception: return None async def _async_fetch_with_nodriver(self, url: str) -> Optional[str]: """Async helper for nodriver fetching.""" browser = None try: browser = await uc.start(headless=True, browser_args=['--headless=new']) page = await browser.get(url) # Wait for content to load await asyncio.sleep(3) # Get page content content = await page.get_content() # Clean up the text if content: soup = BeautifulSoup(content, 'html.parser') # Remove script, style, nav, header, footer elements for element in soup.find_all(['script', 'style', 'nav', 'header', 'footer']): element.decompose() # Try to find main content areas first main_content = (soup.find('article') or soup.find('main') or soup.find('div', {'id': ['content', 'main-content']}) or soup.find('div', {'class': ['content', 'main-content']})) if main_content: text = ' '.join(main_content.stripped_strings) else: text = ' '.join(soup.stripped_strings) # Clean up whitespace text = ' '.join(text.split()) return text if text else None return None except Exception: return None finally: if browser: try: browser.stop() except Exception: pass def fetch_page_content(self, url: str) -> Optional[str]: """Fetch and extract main content from a webpage. Tries requests first, then Selenium.""" try: time.sleep(random.uniform(0.5, 1.5)) response = self.session.get(url, timeout=10) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') for element in soup.find_all(['script', 'style', 'nav', 'header', 'footer']): element.decompose() main_content = None if soup.find('article'): main_content = soup.find('article') elif soup.find('main'): main_content = soup.find('main') elif soup.find('div', {'id': ['content', 'main-content', 'article-content']}): main_content = soup.find('div', {'id': ['content', 'main-content', 'article-content']}) if main_content: text = ' '.join(main_content.stripped_strings) text = ' '.join(text.split()) return text # Fallback to whole cleaned page text text = ' '.join(soup.stripped_strings) text = ' '.join(text.split()) return text if text else None except Exception as e: # Try nodriver fallback first (best stealth) alt = self._fetch_with_nodriver(url) if alt: return alt # Try selenium fallback alt = self._fetch_with_selenium(url) if alt: return alt # Final fallback – Scrapy (if available) try: from scrapy_fetcher import fetch_with_scrapy html_map = fetch_with_scrapy([url]) if url in html_map and html_map[url]: return html_map[url] except Exception: pass print(f"⚠️ Error fetching page content: {str(e)}") return None def _search_bing_html(self, query: str, max_results: int = 15) -> List[Dict]: """Scrape Bing SERP HTML (free, no key).""" try: url = "https://www.bing.com/search" params = {"q": query, "count": str(max_results)} html = requests.get(url, params=params, headers=HEADERS, timeout=15).text soup = BeautifulSoup(html, "html.parser") results = [] for item in soup.select("#b_results > .b_algo"): if len(results) >= max_results: break a = item.select_one("h2 a") if not a: continue title = a.get_text(" ", strip=True) link = a.get("href", "") snippet_tag = item.select_one(".b_caption p") snippet = snippet_tag.get_text(" ", strip=True) if snippet_tag else "" results.append({ "url": link, "title": title, "snippet": snippet, "date": "" }) # Politeness delay time.sleep(random.uniform(1.0, 2.0)) return results except Exception as e: print(f"⚠️ Error scraping Bing HTML: {e}") return [] def _search_google_html(self, query: str, max_results: int = 15, use_selenium: bool = False) -> List[Dict]: """Scrape Google SERP HTML using requests or Selenium.""" try: if use_selenium and SELENIUM_AVAILABLE: from webdriver_manager.chrome import ChromeDriverManager options = ChromeOptions() options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--no-sandbox') options.add_argument(f'user-agent={HEADERS["User-Agent"]}') driver = webdriver.Chrome(ChromeDriverManager().install(), options=options) driver.get(f"https://www.google.com/search?q={quote_plus(query)}&num={max_results}&hl=en") html = driver.page_source driver.quit() else: url = "https://www.google.com/search" params = {"q": query, "num": str(max_results), "hl": "en"} html = requests.get(url, params=params, headers=HEADERS, timeout=15).text return self._parse_google_html(html, max_results) except Exception as e: if not use_selenium and SELENIUM_AVAILABLE: # Retry once with selenium return self._search_google_html(query, max_results, use_selenium=True) print(f"⚠️ Error scraping Google HTML: {e}") return [] def _parse_google_html(self, html: str, max_results: int) -> List[Dict]: """Parse Google SERP HTML into result dicts using Scrapy Selector when possible.""" results = [] if SCRAPY_AVAILABLE: sel = ScrapySelector(text=html) for g in sel.css('div.g')[:max_results]: if len(results) >= max_results: break a = g.css('a::attr(href)').get() title = g.css('h3::text').get() or '' snippet = ' '.join(g.css('.VwiC3b *::text').getall()) or '' if a and not a.startswith('/url?'): results.append({ 'url': a, 'title': title.strip(), 'snippet': snippet.strip(), 'date': '' }) else: soup = BeautifulSoup(html, 'html.parser') for g in soup.select('div.g'): if len(results) >= max_results: break a_tag = g.select_one('a') if not a_tag or not a_tag.get('href'): continue link = a_tag['href'] if link.startswith('/url?'): continue title = a_tag.get_text(' ', strip=True) snippet_tag = g.select_one('.VwiC3b') or g.select_one('span.st') snippet = snippet_tag.get_text(' ', strip=True) if snippet_tag else '' results.append({ 'url': link, 'title': title, 'snippet': snippet, 'date': '' }) return results[:max_results] def _search_google_api(self, query: str, location: str, search_type: str, max_results: int) -> List[Dict]: base_url = "https://www.googleapis.com/customsearch/v1" params = { 'key': self.google_api_key, 'cx': self.google_cx, 'q': query, 'num': max_results, 'gl': location.lower(), 'dateRestrict': 'y1' } if search_type == "news": params['searchType'] = 'news' response = requests.get(base_url, params=params, timeout=20) response.raise_for_status() data = response.json() results = [] for item in data.get('items', []): results.append({ 'url': item.get('link', ''), 'title': item.get('title', ''), 'snippet': item.get('snippet', ''), 'date': item.get('pagemap', {}).get('metatags', [{}])[0].get('article:published_time', '') }) return results def web_search(query: str, location: str = "us", search_type: str = "organic") -> List[Dict]: """ Main web search function that returns real search results. Args: query: Search query location: Country code (e.g., 'us', 'uk', 'in') search_type: Type of search ('organic' or 'news') Returns: List of dictionaries containing search results """ try: searcher = WebSearcher() results = searcher.search(query, location, search_type) if not results: print("⚠️ No search results found") return [{ 'url': 'https://example.com', 'title': query.title(), 'snippet': f'No results found for {query}.' }] return results except Exception as e: print(f"❌ Error in web search: {str(e)}") return [{ 'url': 'https://example.com', 'title': query.title(), 'snippet': f'Error searching for {query}: {str(e)}' }] if __name__ == "__main__": results = web_search("renewable energy trends 2024", location="us", search_type="news") # Test the function print("\nSearch Results:") for result in results: print("\nResult:") print(f"URL: {result.get('url', '')}") print(f"Title: {result.get('title', '')}") print(f"Snippet: {result.get('snippet', '')}") if result.get('date'): print(f"Date: {result.get('date')}")