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 DuckDuckGoSearchService: def __init__(self): self.search_engine = "duckduckgo" self.search_url = "https://duckduckgo.com/?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_duckduckgo_results(self, max_results: int = 10) -> List[SearchResult]: """Extract search results from DuckDuckGo""" results = [] result_elements = self.driver.find_elements(By.CSS_SELECTOR, '.result') 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, '.result__title a') result.title = title_link.text result.url = title_link.get_attribute('href') except: continue # Extract snippet try: snippet = element.find_element(By.CSS_SELECTOR, '.result__snippet') result.snippet = snippet.text except: pass # Extract displayed URL try: cite_element = element.find_element(By.CSS_SELECTOR, '.result__url') result.displayed_url = cite_element.text except: pass if result.url: results.append(result) except Exception as e: logger.debug(f"Error extracting DuckDuckGo result {i}: {e}") continue return results def perform_search(self, request: SearchEngineRequest) -> SearchResponse: """Perform DuckDuckGo search""" start_time = time.time() try: # Build search URL search_url = self.search_url.format(request.query.replace(' ', '+')) logger.info(f"Searching DuckDuckGo for: {request.query}") # Navigate to search page self.driver.get(search_url) time.sleep(request.delay) # Extract results results = self.extract_duckduckgo_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"DuckDuckGo search completed for '{request.query}': {len(results)} results in {processing_time:.2f}s") return response except Exception as e: logger.error(f"Error performing DuckDuckGo 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 DuckDuckGo 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"DuckDuckGo search response sent for request: {request.request_id}") except Exception as e: logger.error(f"Error handling DuckDuckGo search request: {e}") def start_consuming(self): """Start consuming messages from Kafka""" logger.info("Starting DuckDuckGo Search Service...") try: self.kafka_client.consume_messages( topics=[TOPICS['DUCKDUCKGO_SEARCH_REQUESTS']], group_id='duckduckgo-search-service', message_handler=self.handle_search_request ) except KeyboardInterrupt: logger.info("DuckDuckGo Search Service stopped by user") except Exception as e: logger.error(f"Error in DuckDuckGo 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 = DuckDuckGoSearchService() service.start_consuming()