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