"""Core Google Maps scraper engine using Playwright."""
from __future__ import annotations
import asyncio
import logging
import random
from types import TracebackType
from playwright.async_api import Browser, BrowserContext, Page, Route, async_playwright
try:
from playwright_stealth import stealth_async
except ImportError:
stealth_async = None
from gmaps_scraper.models import PlaceDetails, ScrapeConfig, ScrapeResult
from gmaps_scraper.parser import parse_place_details
from gmaps_scraper.utils import (
get_random_user_agent,
normalize_google_maps_url,
parse_google_maps_url,
retry_async,
)
logger = logging.getLogger("gmaps_scraper")
# Resource types to block for faster loading and lower memory
# NOTE: Do NOT block 'stylesheet' โ Google Maps SPA needs CSS for JS initialization
# NOTE: Unblocked 'image' so parser can scrape the hero image `image_url`
_BLOCKED_RESOURCE_TYPES = {"font", "media"}
# Known Google Maps search page titles (not actual place names)
# When the scraper lands on a search results page instead of a place,
# the h1 element contains one of these localized strings.
_SEARCH_PAGE_TITLES = {
"็ตๆ", "results", "Results", "ๆค็ดข็ตๆ",
"ๆๅฐ็ตๆ", "ๆ็ดข็ปๆ",
}
class GoogleMapsScraper:
"""Async Google Maps scraper using Playwright.
Usage:
async with GoogleMapsScraper() as scraper:
result = await scraper.scrape("https://www.google.com/maps/search/...")
print(result.place.name, result.place.rating)
"""
def __init__(self, config: ScrapeConfig | None = None) -> None:
self.config = config or ScrapeConfig()
self._playwright = None
self._browser: Browser | None = None
self._context: BrowserContext | None = None
async def __aenter__(self) -> GoogleMapsScraper:
"""Start the browser."""
await self.start()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
"""Close the browser."""
await self.close()
async def start(self) -> None:
"""Launch the Playwright browser."""
self._playwright = await async_playwright().start()
launch_args = {
"headless": self.config.headless,
}
if self.config.proxy:
launch_args["proxy"] = {"server": self.config.proxy}
# Firefox: avoids Google's bot detection that strips review data in Chromium headless
self._browser = await self._playwright.firefox.launch(**launch_args)
# Create a shared context for all pages (reused across scrapes)
self._context = await self._create_context()
logger.info(
"Browser launched (headless=%s, engine=firefox)", self.config.headless
)
async def close(self) -> None:
"""Close the browser and cleanup."""
if self._context:
await self._context.close()
self._context = None
if self._browser:
await self._browser.close()
self._browser = None
if self._playwright:
await self._playwright.stop()
self._playwright = None
logger.info("Browser closed")
async def _create_context(self) -> BrowserContext:
"""Create a new browser context with stealth settings."""
assert self._browser is not None, "Browser not started. Call start() first."
context_options = {
"viewport": {"width": 1920, "height": 1080},
"timezone_id": "Asia/Tokyo",
"java_script_enabled": True,
}
if self.config.language:
context_options["locale"] = self.config.language
context = await self._browser.new_context(**context_options)
# Stealth: override navigator.webdriver
await context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
// Override permissions
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications'
? Promise.resolve({ state: Notification.permission })
: originalQuery(parameters)
);
""")
# Bypass consent dialog which blocks review rendering
await context.add_cookies([{
"name": "CONSENT",
"value": "YES+cb.20240101-01-p0.en+FX+430",
"domain": ".google.com",
"path": "/"
}, {
"name": "CONSENT",
"value": "YES+cb.20240101-01-p0.en+FX+430",
"domain": ".google.com.tw",
"path": "/"
}])
return context
async def scrape(self, url: str) -> ScrapeResult:
"""Scrape a single Google Maps URL.
Args:
url: Google Maps search or place URL.
Returns:
ScrapeResult with place details.
"""
normalized_url = normalize_google_maps_url(url, self.config.language)
assert self._context is not None, "Context not initialized. Call start() first."
try:
result = await self._scrape_with_retry(normalized_url, self._context)
return result
except Exception as e:
logger.error("All retries failed for %s: %s", url, str(e))
return ScrapeResult(
input_url=url,
success=False,
error=str(e),
)
@retry_async(max_retries=3)
async def _scrape_with_retry(
self, url: str, context: BrowserContext
) -> ScrapeResult:
"""Internal scrape with retry logic."""
page = await context.new_page()
# Block heavy resources (images, fonts, stylesheets, media) to save memory + speed
async def _block_resources(route: Route) -> None:
if route.request.resource_type in _BLOCKED_RESOURCE_TYPES:
await route.abort()
else:
await route.continue_()
await page.route("**/*", _block_resources)
# Apply stealth if installed
if stealth_async:
await stealth_async(page)
try:
# Navigate to the URL
logger.info("Navigating to: %s", url)
await page.goto(url, wait_until="domcontentloaded", timeout=self.config.timeout)
# Handle cookie consent dialog
await self._handle_consent(page)
# Wait for the place panel to load
await self._wait_for_place_panel(page)
# Parse place details
place = await parse_place_details(page)
# Guard: reject if the parsed name is a search page title
if place.name and place.name.strip() in _SEARCH_PAGE_TITLES:
logger.warning(
"Parsed name '%s' is a search page title, not a real place. "
"Marking as failed for URL: %s", place.name, url
)
return ScrapeResult(
input_url=url,
success=False,
error=f"Landed on search results page (name='{place.name}')",
)
return ScrapeResult(
input_url=url,
place=place,
success=True,
)
except Exception as e:
logger.error("Failed to scrape %s: %s", url, str(e))
raise
finally:
await page.close()
async def _handle_consent(self, page: Page) -> None:
"""Handle Google cookie consent dialog if it appears."""
consent_selectors = [
# English
'button:has-text("Reject all")',
'button:has-text("Accept all")',
# Japanese
'button:has-text("ใในใฆๆๅฆ")',
'button:has-text("ใในใฆๅๆ")',
# Chinese Traditional
'button:has-text("ๅ
จ้จๆ็ต")',
'button:has-text("ๅ
จ้จๆฅๅ")',
# Generic form consent
'form[action*="consent"] button',
]
for selector in consent_selectors:
try:
btn = await page.query_selector(selector)
if btn and await btn.is_visible():
await btn.click()
logger.debug("Clicked consent button: %s", selector)
await page.wait_for_timeout(1000)
return
except Exception:
continue
async def _wait_for_place_panel(self, page: Page) -> None:
"""Wait for the Google Maps place panel or search feed to load.
When the page lands on a search results list (multiple results),
automatically click the first result. Includes a reload-retry loop
for resilience against non-deterministic rendering.
"""
place_selectors = [
'h1.DUwDvf',
'h1.lfPIob',
'div[role="main"] h1',
'h1.fontHeadlineLarge',
'h1[class*="header"]',
]
# Semantic selectors first (stable), class-based as fallback.
# div[role="feed"] is the container Google Maps uses for search
# result lists โ much more stable than individual class names.
search_result_selectors = [
'div[role="feed"] a[href*="/maps/place/"]',
'div[role="feed"] a',
'a.hfpxzc',
'a.hfpxV',
'div.Nv2PK',
]
search_result_combined = ", ".join(search_result_selectors)
all_selectors = place_selectors + search_result_selectors
combined_selector = ", ".join(all_selectors)
max_search_retries = 2
for attempt in range(max_search_retries + 1):
try:
# Wait for either a place panel or a search result list to appear
await page.wait_for_selector(
combined_selector, timeout=self.config.timeout, state="attached",
)
await page.wait_for_timeout(500) # Brief settle
except Exception:
logger.warning(
"Timeout waiting for place panel or search feed, waiting 3s as fallback..."
)
await page.wait_for_timeout(3000)
# โโ Check if a place panel loaded โโโโโโโโโโโโโโโโโโโโโโโโโโโ
found_place = False
for selector in place_selectors:
try:
el = await page.query_selector(selector)
if el and await el.is_visible():
text = (await el.inner_text()).strip()
if text in _SEARCH_PAGE_TITLES:
logger.info(
"h1 text '%s' indicates search results page (attempt %d/%d).",
text,
attempt + 1,
max_search_retries + 1,
)
break # Fall through to search result click logic
# It's a real place panel!
logger.info("Place panel found. Polling for review count hydration...")
await self._poll_for_review_count(page)
return
except Exception:
continue
# โโ Try to click the first search result โโโโโโโโโโโโโโโโโโโโ
clicked = await self._click_first_search_result(
page, search_result_combined, place_selectors,
)
if clicked:
return
# โโ Retry: reload the page and try again โโโโโโโโโโโโโโโโโโโโ
if attempt < max_search_retries:
logger.info(
"Search result click failed. Reloading page (retry %d/%d)...",
attempt + 1,
max_search_retries,
)
try:
await page.reload(
wait_until="domcontentloaded",
timeout=self.config.timeout,
)
await self._handle_consent(page)
except Exception as e:
logger.debug("Page reload failed: %s", e)
await page.wait_for_timeout(2000)
logger.warning("Could not navigate to a place panel after retries. Proceeding anyway.")
async def _click_first_search_result(
self,
page: Page,
search_result_selector: str,
place_selectors: list[str],
) -> bool:
"""Try to find and click the first search result.
Returns True if a place panel was successfully loaded after clicking.
"""
try:
# Wait for search result items to render
try:
await page.wait_for_selector(
search_result_selector,
timeout=8000,
state="attached",
)
await page.wait_for_timeout(300) # Brief settle
except Exception:
logger.debug("No search result items appeared within 8s.")
return False
first_result = await page.query_selector(search_result_selector)
if not first_result:
logger.debug("No clickable search results found on the page.")
return False
logger.info("Found search results list. Clicking first result...")
await first_result.click()
# Wait for navigation (URL changes from /search/ to /place/)
try:
await page.wait_for_url("**/place/**", timeout=self.config.timeout)
except Exception:
logger.debug("URL did not change to /place/ pattern, settling...")
await page.wait_for_timeout(2000)
# Wait for the place panel h1 to appear
place_combined = ", ".join(place_selectors)
await page.wait_for_selector(
place_combined, timeout=self.config.timeout, state="visible",
)
await page.wait_for_timeout(500) # Brief settle
# Verify we actually reached a place page (not still on search)
for selector in place_selectors:
try:
el = await page.query_selector(selector)
if el and await el.is_visible():
text = (await el.inner_text()).strip()
if text in _SEARCH_PAGE_TITLES:
logger.warning("Still on search results page after click.")
return False
except Exception:
continue
# Smart poll for review count
await self._poll_for_review_count(page)
return True
except Exception as e:
logger.debug("Failed to click search result: %s", e)
return False
async def _poll_for_review_count(self, page: Page, max_polls: int = 6, poll_interval: int = 500) -> None:
"""Wait for review count data to appear, with hotel layout fallback.
Strategy:
1. Poll every 500ms (up to 3s) for review data signals
2. If not found (hotel/lodging layout), reload the page โ Google often
serves the normal layout on a second request to the resolved place URL
"""
_JS_CHECK = r"""() => {
// Signal 1: Reviews/ใฏใใณใ tab exists in tablist
let tabs = document.querySelectorAll('div[role="tablist"] button');
for (let tab of tabs) {
let text = tab.innerText || '';
if (text.includes('ใฏใใณใ') || text.includes('Review') || text.includes('่ฉ่ซ')) {
return true;
}
}
// Signal 2: F7nice has review count displayed (e.g. "4.2\n(1,296)")
let f7 = document.querySelector('div.F7nice');
if (f7 && f7.innerText) {
let text = f7.innerText.trim();
if (/[(]/.test(text) && /[0-9]/.test(text)) return true;
let nums = text.match(/[0-9]+(?:[.,][0-9]+)?/g);
if (nums && nums.length >= 2 && nums.some(n => parseFloat(n.replace(',','')) > 9)) return true;
}
return false;
}"""
# Step 1: Poll for review data (most pages hydrate in 1-2s)
for i in range(max_polls):
if await page.evaluate(_JS_CHECK):
logger.debug("Review count data detected after %dms", (i + 1) * poll_interval)
return
await page.wait_for_timeout(poll_interval)
# Step 2: Hotel/lodging layout detected โ reload the page
logger.info("Review count not in DOM (hotel layout). Reloading page...")
try:
await page.reload(wait_until="domcontentloaded", timeout=self.config.timeout)
await self._handle_consent(page)
# Poll again after reload (up to 2s)
for i in range(4):
if await page.evaluate(_JS_CHECK):
logger.debug("Review data found after reload + %dms", (i + 1) * poll_interval)
return
await page.wait_for_timeout(poll_interval)
except Exception as e:
logger.debug("Page reload failed: %s", e)
await page.wait_for_timeout(2000)
async def scrape_sync_wrapper(self, url: str) -> ScrapeResult:
"""Synchronous wrapper for scrape(). Used by CLI."""
return await self.scrape(url)
def scrape_place(url: str, config: ScrapeConfig | None = None) -> ScrapeResult:
"""Synchronous convenience function to scrape a single URL.
Args:
url: Google Maps URL to scrape.
config: Optional scraper configuration.
Returns:
ScrapeResult with place details.
Example:
>>> from gmaps_scraper import scrape_place
>>> result = scrape_place("https://www.google.com/maps/search/?api=1&query=...")
>>> print(result.place.name, result.place.rating)
"""
config = config or ScrapeConfig()
async def _run() -> ScrapeResult:
async with GoogleMapsScraper(config) as scraper:
return await scraper.scrape(url)
return asyncio.run(_run())