bacancydataprophets/AI-Generated_FAQs
0
1#!/usr/bin/env python32"""3Enhanced Google Maps Reviews Scraper for David's Bridal4Scrapes reviews from Google Maps with parallel processing and improved element detection5"""6 7import csv8import time9import random10import asyncio11from concurrent.futures import ThreadPoolExecutor, as_completed12from selenium import webdriver13from selenium.webdriver.common.by import By14from selenium.webdriver.support.ui import WebDriverWait15from selenium.webdriver.support import expected_conditions as EC16from selenium.webdriver.chrome.options import Options17from selenium.webdriver.chrome.service import Service18from selenium.common.exceptions import TimeoutException, NoSuchElementException, WebDriverException, ElementClickInterceptedException19from webdriver_manager.chrome import ChromeDriverManager20import pandas as pd21from datetime import datetime22import logging23import sys24import threading25from queue import Queue26 27# Set up logging28logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')29logger = logging.getLogger(__name__)30 31class EnhancedGoogleMapsReviewsScraper:32 def __init__(self, headless=True, wait_time=10, max_workers=3):33 """Initialize the scraper with Chrome driver options"""34 self.wait_time = wait_time35 self.max_workers = max_workers36 self.reviews_queue = Queue()37 self.processed_reviews = []38 self.lock = threading.Lock()39 self.setup_driver(headless)40 41 def setup_driver(self, headless):42 """Set up Chrome driver with appropriate options"""43 try:44 chrome_options = Options()45 if headless:46 chrome_options.add_argument("--headless")47 chrome_options.add_argument("--no-sandbox")48 chrome_options.add_argument("--disable-dev-shm-usage")49 chrome_options.add_argument("--disable-blink-features=AutomationControlled")50 chrome_options.add_argument("--disable-extensions")51 chrome_options.add_argument("--disable-gpu")52 chrome_options.add_argument("--remote-debugging-port=9222")53 chrome_options.add_argument("--window-size=1920,1080")54 chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])55 chrome_options.add_experimental_option('useAutomationExtension', False)56 chrome_options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")57 58 logger.info("Setting up ChromeDriver...")59 service = Service(ChromeDriverManager().install())60 61 self.driver = webdriver.Chrome(service=service, options=chrome_options)62 self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")63 self.wait = WebDriverWait(self.driver, self.wait_time)64 logger.info("ChromeDriver setup successful")65 66 except WebDriverException as e:67 logger.error(f"Failed to setup ChromeDriver: {e}")68 sys.exit(1)69 70 def search_location(self, query):71 """Search for David's Bridal location on Google Maps"""72 try:73 search_url = f"https://www.google.com/maps/search/{query.replace(' ', '+')}"74 logger.info(f"Navigating to: {search_url}")75 self.driver.get(search_url)76 77 # Wait for page to load78 time.sleep(5)79 80 # Look for search results81 result_selectors = [82 "button.hh2c6.G7m0Af", # Button with class for location83 ]84 85 result_found = False86 for selector in result_selectors:87 try:88 first_result = self.wait.until(89 EC.element_to_be_clickable((By.CSS_SELECTOR, selector))90 )91 self.driver.execute_script("arguments[0].click();", first_result)92 time.sleep(3)93 result_found = True94 break95 except TimeoutException:96 continue97 98 return result_found99 100 except Exception as e:101 logger.error(f"Error in search_location: {e}")102 return False103 104 def click_reviews_tab(self):105 """Click on the reviews tab using the specific element structure"""106 try:107 # Wait for the reviews tab to be clickable108 reviews_button = self.wait.until(109 EC.element_to_be_clickable((By.CSS_SELECTOR, "button[data-tab-index='1'][aria-label*='Reviews']"))110 )111 112 # Scroll the button into view113 self.driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", reviews_button)114 time.sleep(1)115 116 # Click the reviews button117 self.driver.execute_script("arguments[0].click();", reviews_button)118 logger.info("Successfully clicked reviews tab")119 120 # Wait for reviews to load121 time.sleep(3)122 return True123 124 except Exception as e:125 logger.error(f"Could not click reviews tab: {e}")126 return False127 128 def expand_review_text(self, review_element):129 """Expand review text by clicking 'More' button if present"""130 try:131 # Look for the 'More' button within this review132 more_button = review_element.find_element(133 By.CSS_SELECTOR, 134 "button.w8nwRe.kyuRq[aria-label='See more']"135 )136 137 # Scroll button into view and click138 self.driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", more_button)139 time.sleep(0.5)140 self.driver.execute_script("arguments[0].click();", more_button)141 time.sleep(1) # Wait for text to expand142 return True143 144 except NoSuchElementException:145 # No 'More' button found - review is already fully visible146 return False147 except Exception as e:148 logger.warning(f"Error expanding review text: {e}")149 return False150 151 def scroll_and_load_reviews(self, target_count=5000):152 """Scroll through reviews to load all available reviews"""153 try:154 scrollable_container = self.driver.find_element(By.CSS_SELECTOR, "div.m6QErb.DxyBCb.kA9KIf.dS8AEf.XiKgde")155 last_review_count = 0156 stagnant_rounds = 0157 max_stagnant_rounds = 5158 scroll_attempts = 0159 max_scroll_attempts = 1000 # increased max160 161 while scroll_attempts < max_scroll_attempts:162 # Scroll down163 self.driver.execute_script(164 "arguments[0].scrollTo(0, arguments[0].scrollHeight);", 165 scrollable_container166 )167 168 # Wait for content to load169 time.sleep(random.uniform(2, 4))170 171 # Count reviews172 current_reviews = len(self.driver.find_elements(By.CSS_SELECTOR, "div[data-review-id]"))173 logger.info(f"Attempt {scroll_attempts + 1}: Loaded {current_reviews} reviews (target: {target_count})")174 175 # Check if we’ve hit the target176 if current_reviews >= target_count:177 logger.info("Reached target review count.")178 break179 180 # Check if no new reviews are loading181 if current_reviews == last_review_count:182 stagnant_rounds += 1183 logger.info(f"No new reviews this round. Stagnant rounds: {stagnant_rounds}/{max_stagnant_rounds}")184 if stagnant_rounds >= max_stagnant_rounds:185 logger.info("No new reviews after several attempts. Stopping scroll.")186 break187 else:188 stagnant_rounds = 0 # reset if progress made189 190 last_review_count = current_reviews191 scroll_attempts += 1192 193 # Occasionally wait longer to mimic human behavior194 if scroll_attempts % 10 == 0:195 logger.info("Taking a longer pause to mimic human browsing...")196 time.sleep(random.uniform(5, 8))197 198 logger.info(f"Finished scrolling. Total reviews found: {current_reviews}")199 return current_reviews200 201 except Exception as e:202 logger.error(f"Error scrolling reviews: {e}")203 return 0204 205 def extract_single_review_data(self, review_element):206 """Extract data from a single review element"""207 try:208 review_data = {}209 210 # First, try to expand the review text if there's a 'More' button211 self.expand_review_text(review_element)212 213 # Extract reviewer name214 try:215 name_element = review_element.find_element(By.CSS_SELECTOR, "div[class*='d4r55']")216 review_data['reviewer_name'] = name_element.text.strip()217 except NoSuchElementException:218 review_data['reviewer_name'] = "Anonymous"219 220 # Extract rating221 try:222 rating_element = review_element.find_element(By.CSS_SELECTOR, "span[role='img'][aria-label*='star']")223 rating_text = rating_element.get_attribute('aria-label')224 review_data['rating'] = self.extract_rating_from_text(rating_text)225 except NoSuchElementException:226 review_data['rating'] = None227 228 # Extract review text using the specific selector you provided229 try:230 text_element = review_element.find_element(By.CSS_SELECTOR, "span.wiI7pd")231 review_data['review_text'] = text_element.text.strip()232 except NoSuchElementException:233 review_data['review_text'] = ""234 235 # Extract date236 try:237 date_element = review_element.find_element(By.CSS_SELECTOR, "span.rsqaWe")238 review_data['date'] = date_element.text.strip()239 except NoSuchElementException:240 review_data['date'] = ""241 242 # Extract owner response if any243 try:244 response_element = review_element.find_element(By.CSS_SELECTOR, "div[class*='wiI7pd']")245 review_data['owner_response'] = response_element.text.strip()246 except NoSuchElementException:247 review_data['owner_response'] = ""248 249 # Add metadata250 review_data['scraped_at'] = datetime.now().isoformat()251 review_data['review_id'] = review_element.get_attribute('data-review-id') or f"review_{int(time.time() * 1000)}"252 253 return review_data254 255 except Exception as e:256 logger.error(f"Error extracting single review: {e}")257 return None258 259 def extract_rating_from_text(self, text):260 """Extract numeric rating from aria-label text"""261 if not text:262 return None263 264 import re265 # Look for patterns like "5 stars", "Rated 4 out of 5 stars"266 match = re.search(r'(\d+)\s*(?:out of \d+\s*)?stars?', text.lower())267 if match:268 return int(match.group(1))269 270 # Fallback: count star characters271 star_count = text.count('★') or text.count('⭐')272 if star_count > 0:273 return star_count274 275 return None276 277 def process_reviews_batch(self, review_elements, start_idx, end_idx):278 """Process a batch of reviews in parallel"""279 batch_results = []280 281 for i in range(start_idx, min(end_idx, len(review_elements))):282 try:283 review_data = self.extract_single_review_data(review_elements[i])284 if review_data:285 batch_results.append(review_data)286 logger.info(f"Processed review {i+1}/{len(review_elements)}")287 except Exception as e:288 logger.warning(f"Error processing review {i+1}: {e}")289 continue290 291 return batch_results292 293 def extract_all_reviews_parallel(self):294 """Extract all reviews using parallel processing with duplicate removal"""295 try:296 # Get all review elements using a single, specific selector297 review_elements = self.driver.find_elements(By.CSS_SELECTOR, "div[data-review-id]")298 total_reviews = len(review_elements)299 logger.info(f"Found {total_reviews} review elements to process")300 301 if total_reviews == 0:302 return []303 304 # Use a set to track processed review IDs and avoid duplicates305 processed_review_ids = set()306 all_reviews = []307 308 # Process reviews sequentially to better control duplicates309 for i, review_element in enumerate(review_elements):310 try:311 # Get review ID first to check for duplicates312 review_id = review_element.get_attribute('data-review-id')313 314 if review_id and review_id in processed_review_ids:315 logger.debug(f"Skipping duplicate review ID: {review_id}")316 continue317 318 # Extract review data319 review_data = self.extract_single_review_data(review_element)320 321 if review_data and review_data.get('review_id'):322 # Add to processed set to prevent duplicates323 processed_review_ids.add(review_data['review_id'])324 all_reviews.append(review_data)325 logger.info(f"Processed review {len(all_reviews)}/{total_reviews}")326 327 except Exception as e:328 logger.warning(f"Error processing review {i+1}: {e}")329 continue330 331 logger.info(f"Successfully extracted {len(all_reviews)} unique reviews")332 return all_reviews333 334 except Exception as e:335 logger.error(f"Error in review extraction: {e}")336 return []337 338 def save_to_csv(self, reviews_data, filename="davids_bridal_reviews.csv"):339 """Save reviews data to CSV file with duplicate removal and better formatting"""340 if not reviews_data:341 logger.warning("No reviews data to save")342 return343 344 try:345 df = pd.DataFrame(reviews_data)346 347 # Remove duplicates based on review_id and review_text348 initial_count = len(df)349 df = df.drop_duplicates(subset=['review_id'], keep='first')350 351 # If review_id duplicates removed, also check for text duplicates as backup352 df = df.drop_duplicates(subset=['reviewer_name', 'review_text', 'date'], keep='first')353 354 final_count = len(df)355 if initial_count > final_count:356 logger.info(f"Removed {initial_count - final_count} duplicate reviews")357 358 # Reorder columns for better readability359 column_order = ['reviewer_name', 'rating', 'date', 'review_text', 'owner_response', 'review_id', 'scraped_at']360 df = df.reindex(columns=column_order)361 362 # Save to CSV with proper encoding363 df.to_csv(filename, index=False, encoding='utf-8')364 logger.info(f"Successfully saved {len(df)} unique reviews to {filename}")365 366 # Print summary statistics367 if 'rating' in df.columns and len(df) > 0:368 avg_rating = df['rating'].mean()369 logger.info(f"Average rating: {avg_rating:.2f}")370 logger.info(f"Rating distribution:\n{df['rating'].value_counts().sort_index()}")371 372 except Exception as e:373 logger.error(f"Error saving to CSV: {e}")374 375 def scrape_reviews(self, location_query, output_file="davids_bridal_reviews.csv"):376 """Main method to scrape all reviews"""377 try:378 logger.info("Starting enhanced review scraping...")379 380 # Search for the location381 if not self.search_location(location_query):382 logger.error("Failed to find location")383 return None384 385 # Click reviews tab386 if not self.click_reviews_tab():387 logger.error("Failed to access reviews tab")388 return None389 390 # Scroll to load all reviews391 total_loaded = self.scroll_and_load_reviews(target_count=2394)392 393 if total_loaded == 0:394 logger.error("No reviews found after scrolling")395 return None396 397 # Extract all reviews using parallel processing398 reviews_data = self.extract_all_reviews_parallel()399 400 # Save to CSV401 if reviews_data:402 self.save_to_csv(reviews_data, output_file)403 logger.info(f"Successfully scraped {len(reviews_data)} reviews")404 return reviews_data405 else:406 logger.warning("No reviews extracted")407 return None408 409 except Exception as e:410 logger.error(f"Error during scraping: {e}")411 return None412 finally:413 self.close()414 415 def close(self):416 """Close the browser driver"""417 if hasattr(self, 'driver'):418 self.driver.quit()419 420def main():421 """Enhanced main function with better error handling"""422 try:423 # Initialize scraper424 scraper = EnhancedGoogleMapsReviewsScraper(425 headless=False, # Set to True for background operation426 max_workers=3 # Adjust based on your system427 )428 429 # Define search query430 search_query = "David's Bridal Middletown NY"431 432 logger.info(f"Starting scrape for: {search_query}")433 434 # Scrape reviews435 reviews = scraper.scrape_reviews(436 location_query=search_query,437 output_file="davids_bridal_middletown_reviews.csv"438 )439 440 if reviews:441 logger.info(f"Scraping completed successfully! Total reviews: {len(reviews)}")442 else:443 logger.error("Scraping failed - no reviews collected")444 445 except KeyboardInterrupt:446 logger.info("Scraping interrupted by user")447 except Exception as e:448 logger.error(f"Unexpected error in main: {e}")449 450if __name__ == "__main__":451 main()