""" Local Playwright backend for booking.com scraping. Uses Playwright with Chromium to scrape search results. No proxy - direct connection. Suitable for low-volume scraping. """ import asyncio import logging import random import re from datetime import date from decimal import Decimal, InvalidOperation from typing import List, Optional from urllib.parse import urlencode, urlparse, parse_qs from services import proxy as proxy_util def location_params_from_url(url: str) -> dict: """Pull the stable destination params (ss, dest_id, dest_type) out of a pasted Booking.com search URL, ignoring volatile session/click params and the date/occupancy params we set ourselves per scrape.""" if not url: return {} try: q = parse_qs(urlparse(url).query) except Exception: return {} out = {} for key in ('ss', 'dest_id', 'dest_type'): if q.get(key): out[key] = q[key][0] return out from playwright.async_api import async_playwright, Browser, BrowserContext, Page try: from playwright_stealth import stealth_async except ImportError: # pragma: no cover - stealth is optional at runtime stealth_async = None from .base import ( ScraperBackend, ScraperResult, HotelData, RateData, AvailabilityStatus ) logger = logging.getLogger(__name__) class PlaywrightLocalBackend(ScraperBackend): """ Local Playwright backend using Chromium. Stealth measures: - Runs headful (via Xvfb in the container) — headless leaks SwiftShader WebGL, empty plugins, missing chrome.runtime; Xvfb needs shm_size 256m - playwright-stealth patches navigator.webdriver, plugins, WebGL vendor - ONE consistent modern-Chrome identity: UA + matching sec-ch-ua client hints + platform (a mismatched UA is worse than none) - Warms up via the homepage so search requests carry real session cookies - Paginates by clicking "next" rather than deep-linking ?offset=25 - Random delays, viewport jitter, mouse movement, human-like scroll Proxy (optional, via BOOKING_PROXY_* env vars): - Routes through a residential proxy when configured, keeping ONE sticky IP per session so cache stays warm and behaviour looks like one user browsing many dates. rotate_session() burns the IP for a fresh one. - The context is persistent across dates within a session — the first page pays the full ~5 MB cold-cache cost, every date after it is ~0.3 MB on the wire. Rotating the IP resets the cache. - Blocks images/media/fonts and third-party ad/consent/analytics domains to cut bandwidth, but NEVER touches the anti-bot challenge (awswaf). """ # A single coherent identity. The UA, the sec-ch-ua hints and the # platform must all agree or the mismatch itself is a bot signal. CHROME_VERSION = "121" USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36" ) CLIENT_HINT_HEADERS = { "sec-ch-ua": '"Not A(Brand";v="99", "Google Chrome";v="121", "Chromium";v="121"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "accept-language": "en-GB,en;q=0.9", } HOMEPAGE = "https://www.booking.com/index.en-gb.html" # Bandwidth trimming. Heavy asset types the parser never needs, plus # third-party ad/consent/analytics hosts. The anti-bot challenge host # (awswaf) is ALWAYS allowed — a browser that skips it fails silently. BLOCK_RESOURCE_TYPES = {"image", "media", "font"} BLOCK_DOMAINS = ( "googlesyndication.com", "pagead2", "google-analytics.com", "googletagmanager.com", "doubleclick.net", "cookielaw.org", "onetrust.com", "accounts.google.com", "connect.facebook.net", "facebook.com", "hotjar.com", "bat.bing.com", "clarity.ms", "criteo.com", "adnxs.com", ) def __init__(self, proxy_config: dict = None): """ Initialize the backend. Args: proxy_config: Optional proxy configuration. Falls back to the BOOKING_PROXY_* environment variables when not supplied. See services/proxy.py for the shared config/build helpers. """ self.proxy_config = proxy_config if proxy_config is not None else proxy_util.config_from_env() self._playwright = None self._browser: Optional[Browser] = None self._context: Optional[BrowserContext] = None self._session_id: Optional[str] = None if self._proxy_enabled(): self._new_session_id() logger.info( f"Proxy enabled via {self.proxy_config['host']}:{self.proxy_config['port']} " f"(country={self.proxy_config.get('country', 'gb')}, session={self._session_id})" ) def _proxy_enabled(self) -> bool: return proxy_util.is_enabled(self.proxy_config) def _new_session_id(self): """Pick a fresh sticky-session id — DataImpulse pins one IP per id.""" self._session_id = proxy_util.new_session_id() async def rotate_session(self): """Burn the current proxy IP and warm cache; the next scrape gets a fresh IP + fresh context. No-op when the proxy is disabled.""" if not self._proxy_enabled(): return old = self._session_id self._new_session_id() logger.info(f"Rotating proxy session {old} -> {self._session_id}") # Session lives in the launch username, so we must drop both. if self._context is not None: try: await self._context.close() except Exception: pass self._context = None if self._browser is not None: try: await self._browser.close() except Exception: pass self._browser = None async def _ensure_browser(self) -> Browser: """Ensure browser is running, start if needed.""" if self._browser is None or not self._browser.is_connected(): if self._playwright is None: self._playwright = await async_playwright().start() launch_kwargs = dict( headless=False, args=[ '--disable-blink-features=AutomationControlled', '--no-sandbox', '--disable-dev-shm-usage', '--disable-features=IsolateOrigins,site-per-process', '--start-maximized', ], ) # Proxy is set at launch (Chromium binds the sticky session, which # lives in the username, at the network layer). Rotating the IP # therefore relaunches the browser — see rotate_session(). proxy = proxy_util.playwright_proxy(self.proxy_config, self._session_id) if proxy: launch_kwargs['proxy'] = proxy self._browser = await self._playwright.chromium.launch(**launch_kwargs) return self._browser async def _route_handler(self, route): """Drop heavy assets and third-party ad/consent/analytics requests to save bandwidth. The anti-bot challenge host is always allowed through.""" try: req = route.request host = urlparse(req.url).netloc if "awswaf" in host: await route.continue_() return if req.resource_type in self.BLOCK_RESOURCE_TYPES: await route.abort() return if any(d in host for d in self.BLOCK_DOMAINS): await route.abort() return await route.continue_() except Exception: # Never let a routing error kill the page load try: await route.continue_() except Exception: pass async def _ensure_context(self) -> BrowserContext: """Return a persistent context with a single coherent Chrome identity. The context is reused across every date in a session so the browser cache stays warm (first page ~5 MB, each date after ~0.3 MB). It is warmed up on the homepage exactly once, here at creation time.""" if self._context is not None: return self._context browser = await self._ensure_browser() # Small viewport jitter so every session isn't pixel-identical width = 1920 + random.randint(-40, 40) height = 1080 + random.randint(-30, 30) context = await browser.new_context( user_agent=self.USER_AGENT, viewport={'width': width, 'height': height}, locale='en-GB', timezone_id='Europe/London', extra_http_headers=self.CLIENT_HINT_HEADERS, ) await context.route("**/*", self._route_handler) # Warm up once per context (per IP) — subsequent date searches reuse # these cookies and cache rather than re-fetching the homepage. page = await context.new_page() await self._prepare_page(page) await self._warm_up(page) await page.close() self._context = context return context async def _prepare_page(self, page: Page): """Apply stealth patches to a freshly created page.""" if stealth_async is not None: try: await stealth_async(page) except Exception as e: logger.warning(f"stealth_async failed, continuing without: {e}") async def _warm_up(self, page: Page): """Visit the homepage first so search requests carry session cookies.""" try: await page.goto(self.HOMEPAGE, wait_until='domcontentloaded', timeout=90000) await asyncio.sleep(random.uniform(1.5, 3.0)) # Dismiss the cookie-consent dialog if present for sel in ['#onetrust-accept-btn-handler', '[aria-label="Accept"]', '[data-testid="cookie-banner-accept"]']: try: btn = await page.query_selector(sel) if btn: await btn.click() await asyncio.sleep(random.uniform(0.4, 0.9)) break except Exception: continue except Exception as e: logger.warning(f"Homepage warm-up failed, continuing: {e}") def _build_search_url( self, location: str, check_in: date, check_out: date, adults: int, offset: int = 0, dest_id: Optional[str] = None, location_params: Optional[dict] = None ) -> str: """Build booking.com search URL. We always own the date/occupancy/paging params; the destination comes from (in priority order) a pasted search URL's params, an explicit dest_id, or the free-text location name.""" params = { 'checkin': check_in.isoformat(), 'checkout': check_out.isoformat(), 'group_adults': adults, 'no_rooms': 1, 'group_children': 0, } if location_params: # ss + dest_id + dest_type lifted from the user's pasted URL — # the most reliable pin (it's exactly what their browser resolved). params.update(location_params) else: params['ss'] = location if dest_id: # Pin the destination — without this, free-text ss= intermittently # resolves to the wrong place entirely. params['dest_id'] = dest_id params['dest_type'] = 'city' if offset > 0: params['offset'] = offset return f"https://www.booking.com/searchresults.en-gb.html?{urlencode(params)}" def _parse_price(self, price_text: str) -> Optional[Decimal]: """Parse price from text like '£150' or 'GBP 150'.""" if not price_text: return None # Remove currency symbols and extract number cleaned = re.sub(r'[£$€,\s]', '', price_text) # Find first number (including decimals) match = re.search(r'[\d,]+(?:\.\d{2})?', cleaned) if match: try: return Decimal(match.group().replace(',', '')) except InvalidOperation: return None return None def _extract_hotel_id(self, url: str) -> Optional[str]: """Extract hotel ID from booking.com URL.""" if not url: return None # URL format: /hotel/gb/hotel-name.en-gb.html or ?dest_id=123 # Try to extract from URL path match = re.search(r'/hotel/[a-z]{2}/([^/]+)\.', url) if match: return match.group(1) # Try dest_id parameter match = re.search(r'dest_id=(-?\d+)', url) if match: return match.group(1) return None async def _human_like_scroll(self, page: Page): """Simulate human-like scrolling behavior.""" # Scroll down in increments for _ in range(random.randint(3, 6)): await page.mouse.wheel(0, random.randint(300, 600)) await asyncio.sleep(random.uniform(0.3, 0.8)) async def _human_like_mouse(self, page: Page): """A few random mouse moves — real users generate pointer events.""" try: for _ in range(random.randint(2, 4)): await page.mouse.move(random.randint(100, 1400), random.randint(150, 800)) await asyncio.sleep(random.uniform(0.1, 0.4)) except Exception: pass async def _go_to_next_page(self, page: Page) -> bool: """Click the pagination 'next' control. Returns True if navigation happened.""" selectors = [ '[data-testid="pagination-next-btn"]', 'button[aria-label="Next page"]', 'a[aria-label="Next page"]', ] for sel in selectors: try: btn = await page.query_selector(sel) if btn and await btn.is_enabled(): await btn.scroll_into_view_if_needed() await asyncio.sleep(random.uniform(0.3, 0.7)) await btn.click() # Results re-render in place; wait for network to settle try: await page.wait_for_load_state('networkidle', timeout=15000) except Exception: await asyncio.sleep(2) return True except Exception as e: logger.debug(f"Next-page selector {sel} failed: {e}") continue return False async def _extract_search_results(self, page: Page, rate_date: date) -> tuple[List[HotelData], List[RateData]]: """Extract hotel and rate data from search results page.""" hotels = [] rates = [] # Wait for property cards - booking.com uses data-testid. Retry once # with a scroll nudge before giving up: cards can lazy-load late. try: await page.wait_for_selector('[data-testid="property-card"]', timeout=30000) except Exception as e: logger.warning(f"No property cards on first wait, retrying: {e}") try: await self._human_like_scroll(page) await page.wait_for_selector('[data-testid="property-card"]', timeout=30000) except Exception as e2: logger.warning(f"No property cards found after retry: {e2}") return hotels, rates # Get all property cards cards = await page.query_selector_all('[data-testid="property-card"]') logger.info(f"Found {len(cards)} property cards") for card in cards: try: hotel = HotelData(booking_com_id='', name='') rate = RateData(rate_date=rate_date) # Hotel name name_el = await card.query_selector('[data-testid="title"]') if name_el: hotel.name = (await name_el.inner_text()).strip() if not hotel.name: continue # Skip if no name found # Hotel URL and ID link_el = await card.query_selector('[data-testid="title-link"]') if link_el: hotel.booking_com_url = await link_el.get_attribute('href') hotel.booking_com_id = self._extract_hotel_id(hotel.booking_com_url) or '' rate.booking_com_id = hotel.booking_com_id # Star rating - look for star icons or rating text stars_el = await card.query_selector('[data-testid="rating-stars"]') if stars_el: stars_text = await stars_el.get_attribute('aria-label') or '' match = re.search(r'(\d+)', stars_text) if match: hotel.star_rating = Decimal(match.group(1)) # Review score score_el = await card.query_selector('[data-testid="review-score"]') if score_el: score_text = await score_el.inner_text() match = re.search(r'([\d.]+)', score_text) if match: try: hotel.review_score = Decimal(match.group(1)) except InvalidOperation: pass # Check for no availability message FIRST no_avail_el = await card.query_selector('[data-testid="availability-message"]') if no_avail_el: avail_text = (await no_avail_el.inner_text()).lower() if 'no availability' in avail_text or 'sold out' in avail_text: rate.availability_status = AvailabilityStatus.SOLD_OUT hotels.append(hotel) rates.append(rate) continue # Price price_el = await card.query_selector('[data-testid="price-and-discounted-price"]') if not price_el: # Try alternative selector price_el = await card.query_selector('[data-testid="price"]') if price_el: price_text = await price_el.inner_text() rate.rate_gross = self._parse_price(price_text) if rate.rate_gross: rate.availability_status = AvailabilityStatus.AVAILABLE # Room type room_el = await card.query_selector('[data-testid="recommended-units"]') if room_el: rate.room_type = (await room_el.inner_text()).strip() # Rate option badges - try multiple selectors # Breakfast included breakfast_el = await card.query_selector('[data-testid="breakfast-included"]') if not breakfast_el: # Check text content for breakfast mentions card_text = (await card.inner_text()).lower() rate.breakfast_included = 'breakfast included' in card_text else: rate.breakfast_included = True # Free cancellation cancel_el = await card.query_selector('[data-testid="cancellation-policy"]') if cancel_el: cancel_text = (await cancel_el.inner_text()).lower() rate.free_cancellation = 'free cancellation' in cancel_text else: card_text = (await card.inner_text()).lower() rate.free_cancellation = 'free cancellation' in card_text # No prepayment prepay_el = await card.query_selector('[data-testid="no-prepayment"]') if prepay_el: rate.no_prepayment = True else: card_text = (await card.inner_text()).lower() rate.no_prepayment = 'no prepayment' in card_text # Rooms left / scarcity indicator scarcity_el = await card.query_selector('[data-testid="availability-rate"]') if scarcity_el: scarcity_text = await scarcity_el.inner_text() match = re.search(r'(\d+)\s*room', scarcity_text.lower()) if match: rate.rooms_left = int(match.group(1)) hotels.append(hotel) rates.append(rate) except Exception as e: logger.warning(f"Error extracting hotel data: {e}") continue return hotels, rates async def scrape_location_search( self, location: str, check_in: date, check_out: date, adults: int = 2, pages: int = 2, dest_id: Optional[str] = None, search_url: Optional[str] = None ) -> ScraperResult: """ Scrape booking.com location search results, rotating the proxy IP if a session looks soft-blocked (challenge page, or page 1 rendered nothing). With no proxy configured this is a single attempt. Args: location: Location name check_in: Check-in date check_out: Check-out date (check_in + 1 for single night rate) adults: Number of adults pages: Number of result pages to scrape dest_id: Booking.com numeric destination id (pins the search) search_url: A pasted Booking.com search URL — its ss/dest_id/ dest_type win over dest_id/location when present Returns: ScraperResult with hotels and rates found """ location_params = location_params_from_url(search_url) if search_url else None max_attempts = 3 if self._proxy_enabled() else 1 result = None for attempt in range(max_attempts): result = await self._scrape_once( location, check_in, check_out, adults, pages, dest_id, location_params) # Soft-block signals: an explicit challenge, or a "successful" load # that yielded zero hotels (page 1 never rendered results). soft_blocked = result.blocked or (result.success and not result.hotels) if not soft_blocked: return result if attempt < max_attempts - 1: logger.warning( f"Attempt {attempt + 1}/{max_attempts} for {check_in} looked blocked " f"(blocked={result.blocked}, hotels={len(result.hotels)}) — rotating IP" ) await self.rotate_session() await asyncio.sleep(random.uniform(1.0, 2.0)) return result async def _scrape_once( self, location: str, check_in: date, check_out: date, adults: int, pages: int, dest_id: Optional[str] = None, location_params: Optional[dict] = None ) -> ScraperResult: """A single scrape attempt for one date on the current IP/session.""" all_hotels = [] all_rates = [] seen_hotel_ids = set() pages_ok = 0 page = None try: # Persistent, already-warmed context (cache stays hot across dates) context = await self._ensure_context() page = await context.new_page() await self._prepare_page(page) for page_num in range(pages): # Short jittered delay between pages. Kept modest because with # rotate-on-block a burned IP costs nothing — throughput wins. if page_num > 0: delay = random.uniform(1.5, 3.5) logger.info(f"Waiting {delay:.1f}s before page {page_num + 1}") await asyncio.sleep(delay) page_loaded = True if page_num == 0: # Navigate to page 1 by URL url = self._build_search_url( location, check_in, check_out, adults, dest_id=dest_id, location_params=location_params) logger.info(f"Scraping page {page_num + 1}: {url}") try: await page.goto(url, wait_until='domcontentloaded', timeout=90000) except Exception as e: logger.warning(f"Page load timeout, continuing: {e}") page_loaded = False else: # Paginate by clicking "next" like a human — deep-linking # ?offset=25 is a stronger bot signal and gets throttled clicked = await self._go_to_next_page(page) if not clicked: # Fall back to offset URL if the control isn't found url = self._build_search_url( location, check_in, check_out, adults, offset=page_num * 25, dest_id=dest_id, location_params=location_params ) logger.info(f"Next-button not found, offset fallback: {url}") try: await page.goto(url, wait_until='domcontentloaded', timeout=30000) except Exception as e: logger.warning(f"Page load timeout, continuing: {e}") page_loaded = False # Check for blocking content = await page.content() is_blocked, reason = self.detect_blocking(content) if is_blocked: logger.warning(f"Blocking detected: {reason}") return ScraperResult( success=False, blocked=True, block_reason=reason, hotels=all_hotels, rates=all_rates, page_content_sample=content[:1000], pages_requested=pages, pages_ok=pages_ok, ) # Human-like mouse movement + scrolling await self._human_like_mouse(page) await self._human_like_scroll(page) # Extract data (retries the selector once on timeout) hotels, rates = await self._extract_search_results(page, check_in) # A page counts as clean if it loaded fully and parsed. # An empty page 1 means the results never rendered; an empty # later page can legitimately be the end of the results. if page_loaded and (hotels or page_num > 0): pages_ok += 1 # Deduplicate by booking_com_id for hotel, rate in zip(hotels, rates): if hotel.booking_com_id and hotel.booking_com_id not in seen_hotel_ids: seen_hotel_ids.add(hotel.booking_com_id) all_hotels.append(hotel) all_rates.append(rate) logger.info(f"Page {page_num + 1}: found {len(hotels)} hotels, {len(all_hotels)} total unique") return ScraperResult( success=True, blocked=False, hotels=all_hotels, rates=all_rates, pages_requested=pages, pages_ok=pages_ok, ) except Exception as e: logger.error(f"Scrape error: {e}") return ScraperResult( success=False, blocked=False, error_message=str(e), hotels=all_hotels, rates=all_rates, pages_requested=pages, pages_ok=pages_ok, ) finally: # Close only the page — the context persists across dates so its # cache stays warm. rotate_session()/close() dispose of the context. if page: await page.close() async def scrape_hotel_page( self, hotel_url: str, check_in: date, check_out: date, adults: int = 2 ) -> ScraperResult: """ Scrape individual hotel page for detailed rates. Future expansion - placeholder for now. Will extract available_qty from room dropdowns. """ # Not implemented in Phase 2a logger.warning("scrape_hotel_page not yet implemented") return ScraperResult( success=False, error_message="Hotel page scraping not yet implemented" ) async def close(self): """Clean up browser resources.""" if self._context: try: await self._context.close() except Exception: pass self._context = None if self._browser: await self._browser.close() self._browser = None if self._playwright: await self._playwright.stop() self._playwright = None