import os import sys import time import logging import platform import subprocess import re from typing import List, Dict from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.service import Service as ChromeService from selenium.webdriver.chrome.options import Options as ChromeOptions from selenium.webdriver.firefox.service import Service as FirefoxService from selenium.webdriver.firefox.options import Options as FirefoxOptions from webdriver_manager.chrome import ChromeDriverManager from webdriver_manager.firefox import GeckoDriverManager # Add shared modules to path sys.path.append('/app') sys.path.append('/app/shared') from shared.models.schemas import SearchEngineRequest, SearchResponse, SearchResult from shared.utils.kafka_utils import KafkaClient, TOPICS from shared.utils.redis_utils import RedisClient # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class GoogleSearchService: def __init__(self): self.search_engine = "google" self.search_url = "https://www.google.com/search?q={}" self.kafka_client = KafkaClient() self.redis_client = RedisClient() self.driver = None self.setup_driver() def detect_browsers(self) -> Dict[str, str]: """Detect installed browsers""" browsers = {} system = platform.system() # Try Chrome try: if system == "Darwin": # macOS result = subprocess.run([ "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "--version" ], capture_output=True, text=True, timeout=10) elif system == "Linux": result = subprocess.run(["google-chrome", "--version"], capture_output=True, text=True, timeout=10) else: # Windows result = subprocess.run(["chrome", "--version"], capture_output=True, text=True, timeout=10) if result.returncode == 0: version_match = re.search(r'(\d+\.\d+\.\d+)', result.stdout) if version_match: browsers['chrome'] = version_match.group(1) except: pass # Try Firefox try: result = subprocess.run(["firefox", "--version"], capture_output=True, text=True, timeout=10) if result.returncode == 0: version_match = re.search(r'(\d+\.\d+)', result.stdout) if version_match: browsers['firefox'] = version_match.group(1) except: pass return browsers def setup_driver(self): """Setup WebDriver with cross-platform compatibility""" available_browsers = self.detect_browsers() if 'chrome' in available_browsers: self.setup_chrome_driver() elif 'firefox' in available_browsers: self.setup_firefox_driver() else: raise Exception("No supported browsers detected") def setup_chrome_driver(self): """Setup Chrome WebDriver""" options = ChromeOptions() options.add_argument("--headless=new") options.add_argument("--no-sandbox") options.add_argument("--disable-dev-shm-usage") options.add_argument("--disable-gpu") options.add_argument("--window-size=1920,1080") options.add_argument("--disable-blink-features=AutomationControlled") options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36") # Clear cache and get driver path driver_path = ChromeDriverManager().install() # Fix for WebDriver Manager cache issue if driver_path.endswith('THIRD_PARTY_NOTICES.chromedriver'): import os driver_dir = os.path.dirname(driver_path) actual_driver = os.path.join(driver_dir, 'chromedriver') if os.path.exists(actual_driver): driver_path = actual_driver # Fix permissions import stat os.chmod(driver_path, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH) service = ChromeService(driver_path) self.driver = webdriver.Chrome(service=service, options=options) self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") logger.info("Chrome WebDriver initialized") def setup_firefox_driver(self): """Setup Firefox WebDriver""" options = FirefoxOptions() options.add_argument("--headless") options.add_argument("--width=1920") options.add_argument("--height=1080") options.set_preference("general.useragent.override", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/109.0") service = FirefoxService(GeckoDriverManager().install()) self.driver = webdriver.Firefox(service=service, options=options) logger.info("Firefox WebDriver initialized") def extract_google_results(self, max_results: int = 10) -> List[SearchResult]: """Extract search results from Google""" results = [] # Google search result selectors result_selectors = [ 'div.g', # Main result container 'div[data-sokoban-container] div.g', # Alternative container 'div.tF2Cxc' # Another result container ] result_elements = [] for selector in result_selectors: elements = self.driver.find_elements(By.CSS_SELECTOR, selector) if elements: result_elements = elements break for i, element in enumerate(result_elements[:max_results]): try: result = SearchResult(rank=i + 1) # Extract title and URL try: title_link = element.find_element(By.CSS_SELECTOR, 'h3') if title_link: result.title = title_link.text link_element = title_link.find_element(By.XPATH, './..') result.url = link_element.get_attribute('href') except: continue # Extract snippet snippet_selectors = ['.VwiC3b', '.s3v9rd', '.st'] for selector in snippet_selectors: try: snippet = element.find_element(By.CSS_SELECTOR, selector) result.snippet = snippet.text break except: continue # Extract displayed URL try: cite_element = element.find_element(By.CSS_SELECTOR, 'cite') result.displayed_url = cite_element.text except: pass if result.url: results.append(result) except Exception as e: logger.debug(f"Error extracting Google result {i}: {e}") continue return results def perform_search(self, request: SearchEngineRequest) -> SearchResponse: """Perform Google search""" start_time = time.time() try: # Build search URL search_url = self.search_url.format(request.query.replace(' ', '+')) logger.info(f"Searching Google for: {request.query}") # Navigate to search page self.driver.get(search_url) time.sleep(request.delay) # Extract results results = self.extract_google_results(request.max_results) processing_time = time.time() - start_time response = SearchResponse( request_id=request.request_id, search_engine=self.search_engine, query=request.query, results=results, total_results=len(results), processing_time=processing_time ) logger.info(f"Google search completed for '{request.query}': {len(results)} results in {processing_time:.2f}s") return response except Exception as e: logger.error(f"Error performing Google search: {e}") return SearchResponse( request_id=request.request_id, search_engine=self.search_engine, query=request.query, results=[], total_results=0, error=str(e), processing_time=time.time() - start_time ) def handle_search_request(self, topic: str, message: dict, key: str): """Handle incoming search requests""" try: request = SearchEngineRequest(**message) logger.info(f"Processing Google search request: {request.request_id}") # Perform search response = self.perform_search(request) # Send response back via Kafka self.kafka_client.send_message( topic=TOPICS['SEARCH_RESULTS'], message=response.dict(), key=request.request_id ) logger.info(f"Google search response sent for request: {request.request_id}") except Exception as e: logger.error(f"Error handling Google search request: {e}") def start_consuming(self): """Start consuming messages from Kafka""" logger.info("Starting Google Search Service...") try: self.kafka_client.consume_messages( topics=[TOPICS['GOOGLE_SEARCH_REQUESTS']], group_id='google-search-service', message_handler=self.handle_search_request ) except KeyboardInterrupt: logger.info("Google Search Service stopped by user") except Exception as e: logger.error(f"Error in Google Search Service: {e}") finally: self.cleanup() def cleanup(self): """Clean up resources""" if self.driver: self.driver.quit() logger.info("WebDriver closed") if self.kafka_client: self.kafka_client.close() if self.redis_client: self.redis_client.close() if __name__ == "__main__": service = GoogleSearchService() service.start_consuming()