rates/backend/services/scraper_backends/playwright_hotel_page.py

"""
Booking.com Hotel Page Scraper Backend

Scrapes individual hotel property pages instead of search results.
Returns all room types and rate plan variants per hotel per date.

Advantages over search-results approach:
- Individual hotel pages are not Cloudflare-protected (search results are)
- Captures every room type, rate plan, meal plan, and availability count
- Proxy reuse: one residential IP stays valid across many hotel pages
  (looks like a human browsing properties), rotates only when blocked

Rate plan variants per room type (typical):
  2-adult + room only + non-refundable
  2-adult + room only + free cancellation
  2-adult + breakfast  + non-refundable
  2-adult + breakfast  + free cancellation
  1-adult + room only / breakfast (filtered out by caller if not needed)
"""

import asyncio
import logging
import re
import uuid
from datetime import date
from decimal import Decimal
from typing import List, Optional, Tuple
from urllib.parse import urlparse

from playwright.async_api import async_playwright, Browser, BrowserContext, Page

from .base import ScraperBackend, ScraperResult, HotelData, RateData, AvailabilityStatus
from services import proxy as proxy_util

logger = logging.getLogger(__name__)

# JS that extracts all rate plan rows from the room availability table.
# Runs inside the page after the room table has loaded.
_EXTRACT_RATES_JS = """
() => {
    const results = [];

    document.querySelectorAll('[id^="room_type_id_"]').forEach(roomEl => {
        const roomId  = roomEl.getAttribute('data-room-id') || roomEl.id.replace('room_type_id_', '');
        const roomName = (
            roomEl.querySelector('.hprt-roomtype-icon-link')?.innerText ||
            roomEl.querySelector('span')?.innerText || ''
        ).trim();

        // Availability count from scarcity indicator ("We have 2 left")
        const availText = roomEl.closest('tr')
            ?.querySelector('.only_x_left, .thisRoomAvailabilityNew span')
            ?.innerText?.trim() || '';
        const availMatch = availText.match(/\\d+/);
        const availCount = availMatch ? parseInt(availMatch[0]) : null;

        // Walk sibling <tr> rows that belong to this room type
        let tr = roomEl.closest('tr');
        while (tr) {
            if (tr.classList.contains('js-rt-block-row')) {
                const blockId   = tr.getAttribute('data-block-id') || '';
                const priceRaw  = tr.getAttribute('data-hotel-rounded-price') || '';

                let fltrs = {};
                try { fltrs = JSON.parse(tr.getAttribute('data-fltrs') || '{}'); } catch(e) {}

                // Conditions cell (3rd <td>) holds meal plan + cancel info
                const cells = tr.querySelectorAll('td');
                const condCell = cells.length >= 3 ? cells[2].innerText || '' : '';

                const breakfastIncluded = condCell.toLowerCase().includes('breakfast');
                const nonRefundable     = (fltrs.non_refundable === 1);

                // Free cancellation date: "Free cancellation before DD Month YYYY"
                const cancelMatch = condCell.match(/free cancellation before ([\\w\\s]+)/i);
                const freeCancelText = cancelMatch ? cancelMatch[1].trim() : null;

                // Persons: extract from block_id format {room_id}_{rate_plan_id}_{persons}_{meal}_0
                // More reliable than cell text parsing which can pick up the price instead.
                const blockParts = blockId.split('_');
                const maxPersons = blockParts.length >= 3 ? parseInt(blockParts[2]) : null;

                // Quantity dropdown: last <td> has a <select> with options 0..N
                // where N = actual available qty (capped at 10 by Booking.com).
                // This is more reliable than the "X left" scarcity text which only
                // appears when availability is low (typically ≤5).
                let qtyAvailable = availCount; // fallback to scarcity text
                const qtyCell = cells.length >= 4 ? cells[cells.length - 1] : null;
                if (qtyCell) {
                    const sel = qtyCell.querySelector('select');
                    if (sel) {
                        const vals = Array.from(sel.options)
                            .map(o => parseInt(o.value))
                            .filter(v => !isNaN(v) && v > 0);
                        if (vals.length > 0) qtyAvailable = Math.max(...vals);
                    } else {
                        // Fallback: parse max number from cell text
                        // "Select rooms\\n0\\n1  (£208)\\n2  (£416)" → [1,2] → max 2
                        const nums = (qtyCell.innerText || '').match(/^(\\d+)/gm);
                        if (nums && nums.length > 0) {
                            const parsed = nums.map(n => parseInt(n)).filter(v => v > 0);
                            if (parsed.length > 0) qtyAvailable = Math.max(...parsed);
                        }
                    }
                }

                results.push({
                    room_id:            roomId,
                    room_name:          roomName,
                    avail_count:        qtyAvailable,
                    block_id:           blockId,
                    price:              priceRaw ? parseInt(priceRaw) : null,
                    breakfast_included: breakfastIncluded,
                    non_refundable:     nonRefundable,
                    free_cancel_text:   freeCancelText,
                    max_persons:        maxPersons,
                });
            }

            tr = tr.nextElementSibling;
            if (!tr) break;
            // Stop at the next room type's header row
            if (tr.querySelector('[id^="room_type_id_"]')) break;
        }
    });

    return results;
}
"""


def _slug_from_url(url: str) -> Optional[str]:
    """Extract hotel slug from Booking.com hotel page URL."""
    if not url:
        return None
    m = re.search(r'/hotel/\w+/([^.]+)\.', url)
    return m.group(1) if m else None


def _build_hotel_url(hotel_url: str, checkin: date, checkout: date, adults: int) -> str:
    base = hotel_url.split('?')[0]
    return (
        f"{base}?checkin={checkin}&checkout={checkout}"
        f"&group_adults={adults}&no_rooms=1&selected_currency=GBP"
    )


class PlaywrightHotelPageBackend(ScraperBackend):
    """
    Scrapes individual Booking.com hotel property pages.

    Keeps one browser context alive across multiple hotel+date requests.
    Rotates proxy only when a CF/WAF block is detected.
    """

    def __init__(self, proxy_config: dict = None):
        self._proxy_config = proxy_config or {}
        self._pw = None
        self._browser: Optional[Browser] = None
        self._context: Optional[BrowserContext] = None
        self._requests_on_context = 0
        # Rotate after this many requests even without a block (keeps session fresh)
        self._max_requests_per_context = 40

    def _proxy_enabled(self) -> bool:
        return proxy_util.is_enabled(self._proxy_config)

    def _proxy_kwargs(self, session_id: Optional[str] = None) -> dict:
        # Credentials embedded in URL — separate username/password fields cause
        # a 407 round-trip that takes ~14s on DataImpulse (see proxy.py comment).
        proxy = proxy_util.playwright_proxy(self._proxy_config, session_id)
        return {'proxy': proxy} if proxy else {}

    async def _start(self):
        if not self._pw:
            self._pw = await async_playwright().start()
        if not self._browser:
            self._browser = await self._pw.chromium.launch(
                headless=True,
                args=['--no-sandbox', '--disable-dev-shm-usage'],
            )

    async def _get_context(self) -> BrowserContext:
        await self._start()
        if self._context is None or self._requests_on_context >= self._max_requests_per_context:
            if self._context:
                try:
                    await self._context.close()
                except Exception:
                    pass
            session_id = uuid.uuid4().hex[:12] if self._proxy_enabled() else None
            self._context = await self._browser.new_context(
                user_agent=(
                    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                    'AppleWebKit/537.36 (KHTML, like Gecko) '
                    'Chrome/125.0.0.0 Safari/537.36'
                ),
                **self._proxy_kwargs(session_id),
            )
            self._requests_on_context = 0
            logger.debug("Opened new browser context" + (f" (proxy session {session_id})" if session_id else ""))
        return self._context

    async def _rotate_context(self):
        """Force a new browser context (new proxy session)."""
        if self._context:
            try:
                await self._context.close()
            except Exception:
                pass
            self._context = None
        self._requests_on_context = 0
        logger.info("Rotated browser context (new proxy session)")

    async def scrape_hotel_page(
        self,
        hotel_url: str,
        check_in: date,
        check_out: date,
        adults: int = 2,
    ) -> ScraperResult:
        """
        Load a hotel page and extract all room types + rate plan variants.

        On block detection, rotates proxy and returns blocked=True so the
        caller can retry with the fresh context.
        """
        page_url = _build_hotel_url(hotel_url, check_in, check_out, adults)
        context = await self._get_context()
        self._requests_on_context += 1
        page: Optional[Page] = None

        try:
            page = await context.new_page()

            # Load the page
            page_loaded = False
            try:
                await page.goto(page_url, wait_until='domcontentloaded', timeout=30000)
                page_loaded = True
            except Exception as e:
                logger.warning(f"goto timeout for {hotel_url} {check_in}: {e}")

            if not page_loaded:
                return ScraperResult(success=False, blocked=False, error_message='page load timeout')

            # Wait for the room table (prices render via JS after DOM)
            room_table_appeared = False
            try:
                await page.wait_for_selector('[id^="room_type_id_"]', timeout=15000)
                room_table_appeared = True
            except Exception:
                pass

            if not room_table_appeared:
                # Could be sold-out, no-availability, or a block page
                content = await page.content()
                is_blocked, reason = self.detect_blocking(content)
                if is_blocked or len(content) < 20000:
                    logger.warning(f"Block detected for {hotel_url} {check_in}: {reason or 'small page'}")
                    await self._rotate_context()
                    return ScraperResult(success=False, blocked=True, block_reason=reason or 'small page')

                # No availability for this date
                return ScraperResult(
                    success=True,
                    rates=[RateData(
                        rate_date=check_in,
                        availability_status=AvailabilityStatus.SOLD_OUT,
                        currency='GBP',
                    )],
                )

            # Extra wait for JS prices to render
            await asyncio.sleep(10)

            # Check for explicit "no availability" message
            no_avail_el = await page.query_selector('#no_availability_msg')
            if no_avail_el and await no_avail_el.is_visible():
                return ScraperResult(
                    success=True,
                    rates=[RateData(
                        rate_date=check_in,
                        availability_status=AvailabilityStatus.SOLD_OUT,
                        currency='GBP',
                    )],
                )

            # Extract all rate plan rows
            raw_plans = await page.evaluate(_EXTRACT_RATES_JS)

            if not raw_plans:
                return ScraperResult(success=True, rates=[])

            rates: List[RateData] = []
            for plan in raw_plans:
                if plan['price'] is None:
                    continue
                rates.append(RateData(
                    rate_date=check_in,
                    availability_status=AvailabilityStatus.AVAILABLE,
                    rate_gross=Decimal(plan['price']),
                    currency='GBP',
                    room_type=plan['room_name'] or None,
                    breakfast_included=plan['breakfast_included'],
                    free_cancellation=not plan['non_refundable'],
                    rooms_left=plan['avail_count'],   # qty from dropdown (max 10) or scarcity text
                    available_qty=plan['avail_count'],
                    rate_plan_id=plan['block_id'] or None,
                    max_persons=plan['max_persons'],
                ))

            logger.info(
                f"Hotel page {_slug_from_url(hotel_url)} {check_in}: "
                f"{len(raw_plans)} rate plans extracted"
            )
            return ScraperResult(success=True, rates=rates)

        except Exception as e:
            logger.error(f"scrape_hotel_page error for {hotel_url} {check_in}: {e}")
            return ScraperResult(success=False, error_message=str(e))
        finally:
            if page:
                try:
                    await page.close()
                except Exception:
                    pass

    async def scrape_location_search(self, *args, **kwargs) -> ScraperResult:
        """Not used by this backend — hotel pages replace location search."""
        raise NotImplementedError("PlaywrightHotelPageBackend does not support location search")

    async def close(self):
        if self._context:
            try:
                await self._context.close()
            except Exception:
                pass
            self._context = None
        if self._browser:
            try:
                await self._browser.close()
            except Exception:
                pass
            self._browser = None
        if self._pw:
            try:
                await self._pw.stop()
            except Exception:
                pass
            self._pw = None