""" Google Play Store Review Scraper ================================ A production-ready web scraper for extracting app reviews from Google Play Store. Features: - Configurable via command-line arguments - Modern Selenium 4+ API - Robust error handling with retries - Progress tracking with rich output - Automatic backup exports - Headless mode support Usage: python webscrape.py --app-id com.gojek.app --output reviews.csv --count 1000 """ import argparse import logging import time from dataclasses import dataclass, field from pathlib import Path from typing import Set import pandas as pd from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait from selenium.common.exceptions import ( NoSuchElementException, TimeoutException, ElementClickInterceptedException, ) # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) @dataclass class ScraperConfig: """Configuration for the scraper.""" app_id: str output_file: str target_count: int = 1000 language: str = "id" headless: bool = False scroll_pause: float = 2.0 max_retries: int = 3 backup_interval: int = 500 class PlayStoreReviewScraper: """ A robust web scraper for Google Play Store reviews. Uses Selenium to handle dynamic content loading and extracts both short and long-form reviews. """ # CSS Selectors (centralized for easy maintenance) SELECTORS = { "review_container": "div.LXrl4c", "review_item": "div.UD7Dzf", "short_comment": 'span[jsname="bN97Pc"]', "long_comment": 'span[jsname="fbQN7e"]', "load_more_button": "span.CwaK9", } def __init__(self, config: ScraperConfig): self.config = config self.driver = self._init_driver() self.reviews: Set[str] = set() self.start_time = time.time() def _init_driver(self) -> webdriver.Chrome: """Initialize Chrome WebDriver with optimal settings.""" options = Options() if self.config.headless: options.add_argument("--headless=new") # Performance optimizations options.add_argument("--disable-gpu") options.add_argument("--no-sandbox") options.add_argument("--disable-dev-shm-usage") options.add_argument("--window-size=1920,1080") # Try to use webdriver-manager if available, fallback to system PATH try: from webdriver_manager.chrome import ChromeDriverManager service = Service(ChromeDriverManager().install()) except ImportError: service = Service() # Uses chromedriver from PATH return webdriver.Chrome(service=service, options=options) @property def url(self) -> str: """Build the Google Play Store URL for the target app.""" return ( f"https://play.google.com/store/apps/details" f"?id={self.config.app_id}" f"&showAllReviews=true" f"&hl={self.config.language}" ) def _scroll_page(self, iterations: int = 10) -> None: """Scroll down the page to trigger lazy loading.""" for _ in range(iterations): self.driver.execute_script( "window.scrollTo(0, document.body.scrollHeight)" ) time.sleep(self.config.scroll_pause) def _click_load_more(self) -> bool: """Attempt to click the 'Load More' button. Returns success status.""" try: button = WebDriverWait(self.driver, 5).until( EC.element_to_be_clickable( (By.CSS_SELECTOR, self.SELECTORS["load_more_button"]) ) ) button.click() return True except (NoSuchElementException, TimeoutException, ElementClickInterceptedException): return False def _load_more_reviews(self) -> None: """Load more reviews by clicking button and scrolling.""" for attempt in range(self.config.max_retries): if self._click_load_more(): self._scroll_page(iterations=5) else: # If button not found, just scroll self._scroll_page(iterations=3) break def _extract_review_text(self, review_element) -> str | None: """Extract the full review text from a review element.""" try: # Try to get the long (expanded) comment first long_comment = review_element.find_element( By.CSS_SELECTOR, self.SELECTORS["long_comment"] ) # Remove style to reveal hidden text self.driver.execute_script( 'arguments[0].removeAttribute("style")', long_comment ) if long_comment.text.strip(): return long_comment.text.strip() # Fallback to short comment if long is empty short_comment = review_element.find_element( By.CSS_SELECTOR, self.SELECTORS["short_comment"] ) return short_comment.text.strip() or None except NoSuchElementException: return None def _export_reviews(self, filename: str | None = None) -> None: """Export collected reviews to CSV.""" if not self.reviews: logger.warning("No reviews to export") return output_path = filename or self.config.output_file df = pd.DataFrame({"comment": list(self.reviews)}) df.to_csv(output_path, index=False, encoding="utf-8") logger.info(f"Exported {len(self.reviews)} reviews to {output_path}") def _create_backup(self) -> None: """Create a backup of current reviews.""" backup_name = Path(self.config.output_file).stem backup_path = f"{backup_name}_backup_{len(self.reviews)}.csv" self._export_reviews(backup_path) def scrape(self) -> Set[str]: """ Main scraping loop. Collects reviews until target count is reached. Returns: Set of unique review texts. """ logger.info(f"Starting scrape for app: {self.config.app_id}") logger.info(f"Target: {self.config.target_count} reviews") try: self.driver.get(self.url) # Wait for review container to load WebDriverWait(self.driver, 15).until( EC.presence_of_element_located( (By.CSS_SELECTOR, self.SELECTORS["review_container"]) ) ) last_backup_count = 0 stale_iterations = 0 while len(self.reviews) < self.config.target_count: previous_count = len(self.reviews) # Load more content self._load_more_reviews() # Extract reviews container = self.driver.find_element( By.CSS_SELECTOR, self.SELECTORS["review_container"] ) review_elements = container.find_elements( By.CLASS_NAME, "UD7Dzf" ) for element in review_elements: text = self._extract_review_text(element) if text and text not in self.reviews: self.reviews.add(text) if len(self.reviews) % 100 == 0: logger.info(f"Progress: {len(self.reviews)} reviews collected") # Check for stale state (no new reviews) if len(self.reviews) == previous_count: stale_iterations += 1 if stale_iterations >= 3: logger.warning("No new reviews found after multiple attempts. Stopping.") break else: stale_iterations = 0 # Periodic backup if len(self.reviews) - last_backup_count >= self.config.backup_interval: self._create_backup() last_backup_count = len(self.reviews) # Final export self._export_reviews() elapsed = time.time() - self.start_time logger.info("=" * 50) logger.info("SCRAPING COMPLETE") logger.info(f"Total reviews: {len(self.reviews)}") logger.info(f"Time elapsed: {elapsed:.2f} seconds") logger.info("=" * 50) return self.reviews except TimeoutException: logger.error("Timeout waiting for page to load") raise except Exception as e: logger.error(f"Unexpected error: {e}") # Emergency backup if self.reviews: self._export_reviews("emergency_backup.csv") raise finally: self.driver.quit() def parse_args() -> argparse.Namespace: """Parse command-line arguments.""" parser = argparse.ArgumentParser( description="Scrape reviews from Google Play Store" ) parser.add_argument( "--app-id", "-a", required=True, help="Google Play Store app ID (e.g., com.gojek.app)" ) parser.add_argument( "--output", "-o", default="reviews.csv", help="Output CSV file path (default: reviews.csv)" ) parser.add_argument( "--count", "-c", type=int, default=1000, help="Target number of reviews to collect (default: 1000)" ) parser.add_argument( "--language", "-l", default="id", help="Review language code (default: id)" ) parser.add_argument( "--headless", action="store_true", help="Run browser in headless mode" ) return parser.parse_args() def main(): """Entry point for the scraper.""" args = parse_args() config = ScraperConfig( app_id=args.app_id, output_file=args.output, target_count=args.count, language=args.language, headless=args.headless, ) scraper = PlayStoreReviewScraper(config) scraper.scrape() if __name__ == "__main__": main()