Add hotel-page scraper backend with full room/rate plan extraction

Switches booking.com scraping from search-results page (Cloudflare-targeted)
to individual hotel property pages (not CF-protected). Each page load returns
all room types, all rate plan variants (room-only/B&B × refundable/non-ref ×
1-2 adults), and availability counts.

Key changes:
- New PlaywrightHotelPageBackend: proxy reuse until block, rotate on CF/WAF
- booking_scraper.py: _run_hotel_page_scrape(), scrape_hotel_date(),
  _scrape_hotels_concurrent() — sharded by hotel so one proxy session covers
  all dates for one hotel (looks human)
- schema.sql: ADD COLUMN rate_plan_id, max_persons on booking_com_rates
- competitors API: filter max_persons=2, order by rate_gross ASC as tiebreaker
  so DISTINCT ON returns cheapest 2-adult rate from latest batch

Enable via Settings → Scraper Backend → playwright_hotel_page

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/services/booking_scraper.py

@@ -25,7 +25,10 @@ from sqlalchemy.orm import Session
 
 from database import SyncSessionLocal
 from services import proxy as proxy_util
-from .scraper_backends import ScraperBackend, PlaywrightLocalBackend, HotelData, RateData, AvailabilityStatus
+from .scraper_backends import (
+    ScraperBackend, PlaywrightLocalBackend, PlaywrightHotelPageBackend,
+    HotelData, RateData, AvailabilityStatus,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -94,12 +97,17 @@ def get_scraper_backend(db: Session) -> ScraperBackend:
         # Future: Apify backend
         raise NotImplementedError("Apify backend not yet implemented")
 
+    proxy_cfg = proxy_util.load_config(db)
+
+    if backend_type == 'playwright_hotel_page':
+        return PlaywrightHotelPageBackend(proxy_config=proxy_cfg)
+
     if backend_type not in ('playwright_local', 'playwright_proxy'):
         logger.warning(f"Unknown backend type '{backend_type}', falling back to playwright_local")
 
     # Proxy config resolved by the shared module: system_config is authoritative
     # when booking_proxy_enabled is set, else BOOKING_PROXY_* env ({} = direct).
-    return PlaywrightLocalBackend(proxy_config=proxy_util.load_config(db))
+    return PlaywrightLocalBackend(proxy_config=proxy_cfg)
 
 
 def get_scrape_config(db: Session) -> Optional[Dict[str, Any]]:
@@ -189,9 +197,11 @@ def save_rate(db: Session, rate: RateData, hotel_id: int, batch_id: uuid.UUID):
         text("""
             INSERT INTO booking_com_rates
             (hotel_id, rate_date, availability_status, rate_gross, currency, room_type,
-             breakfast_included, free_cancellation, no_prepayment, rooms_left, scrape_batch_id)
+             breakfast_included, free_cancellation, no_prepayment, rooms_left,
+             rate_plan_id, max_persons, scrape_batch_id)
             VALUES (:hotel_id, :rate_date, :status, :rate, :currency, :room_type,
-                    :breakfast, :cancel, :prepay, :rooms_left, :batch_id)
+                    :breakfast, :cancel, :prepay, :rooms_left,
+                    :rate_plan_id, :max_persons, :batch_id)
         """),
         {
             'hotel_id': hotel_id,
@@ -204,6 +214,8 @@ def save_rate(db: Session, rate: RateData, hotel_id: int, batch_id: uuid.UUID):
             'cancel': rate.free_cancellation,
             'prepay': rate.no_prepayment,
             'rooms_left': rate.rooms_left,
+            'rate_plan_id': rate.rate_plan_id,
+            'max_persons': rate.max_persons,
             'batch_id': str(batch_id),
         }
     )
@@ -282,6 +294,69 @@ def cleanup_stale_batches(db: Session, max_age_minutes: int = 60):
     return len(cleaned)
 
 
+def get_active_hotels(db: Session) -> List[Dict[str, Any]]:
+    """Return all active hotels that have a booking_com_url (needed for hotel-page scraping)."""
+    rows = db.execute(
+        text("""
+            SELECT id, booking_com_id, name, booking_com_url
+            FROM booking_com_hotels
+            WHERE is_active = TRUE AND booking_com_url IS NOT NULL
+            ORDER BY display_order, id
+        """)
+    ).fetchall()
+    return [dict(r._mapping) for r in rows]
+
+
+async def scrape_hotel_date(
+    db: Session,
+    hotel: Dict[str, Any],
+    rate_date: date,
+    backend: 'PlaywrightHotelPageBackend',
+    batch_id: uuid.UUID,
+    adults: int = 2,
+    max_retries: int = 2,
+) -> Dict[str, Any]:
+    """
+    Scrape all rate plans for one hotel on one date via the hotel page backend.
+
+    On block detection, the backend rotates its proxy context automatically.
+    We retry up to max_retries times to handle the rotation.
+    """
+    check_out = rate_date + timedelta(days=1)
+    hotel_url = hotel['booking_com_url']
+
+    for attempt in range(max_retries + 1):
+        result = await backend.scrape_hotel_page(hotel_url, rate_date, check_out, adults)
+
+        if result.success:
+            rates_saved = 0
+            for rate in result.rates:
+                rate.rate_date = rate_date  # ensure date is set
+                try:
+                    save_rate(db, rate, hotel['id'], batch_id)
+                    rates_saved += 1
+                except Exception as e:
+                    logger.warning(f"Error saving rate plan for {hotel['name']} {rate_date}: {e}")
+                    db.rollback()
+            db.commit()
+            return {'success': True, 'blocked': False, 'rates_count': rates_saved}
+
+        if result.blocked and attempt < max_retries:
+            logger.info(
+                f"Hotel {hotel['name']} {rate_date} blocked (attempt {attempt + 1}), retrying…"
+            )
+            continue
+
+        return {
+            'success': False,
+            'blocked': result.blocked,
+            'error': result.error_message,
+            'rates_count': 0,
+        }
+
+    return {'success': False, 'blocked': True, 'rates_count': 0}
+
+
 async def scrape_date(
     db: Session,
     rate_date: date,
@@ -438,6 +513,13 @@ def get_scraper_concurrency(db: Session) -> int:
         return 6
 
 
+def _is_hotel_page_mode(db: Session) -> bool:
+    row = db.execute(
+        text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_backend'")
+    ).fetchone()
+    return (row and row.config_value) == 'playwright_hotel_page'
+
+
 def _effective_concurrency(db: Session, n_jobs: int) -> int:
     """Clamp configured concurrency to the workload, and force serial when the
     proxy is off — N workers would share one IP and hammer it, worse than 1."""
@@ -451,6 +533,82 @@ def _effective_concurrency(db: Session, n_jobs: int) -> int:
     return max(1, min(configured, n_jobs))
 
 
+async def _scrape_hotels_concurrent(
+    hotel_date_jobs: List[Tuple[Dict[str, Any], date]],
+    concurrency: int,
+    batch_id: uuid.UUID,
+    adults: int = 2,
+) -> Dict[str, int]:
+    """
+    Scrape (hotel, date) pairs using the hotel-page backend.
+
+    Jobs are sharded by HOTEL (not date) so each worker keeps its proxy session
+    alive across all dates for one hotel — looks like a single user checking
+    availability for a trip.
+    """
+    # Group by hotel to get stable shards
+    hotels_seen: List[Dict[str, Any]] = []
+    hotel_dates: Dict[int, List[date]] = {}
+    for hotel, rate_date in hotel_date_jobs:
+        hid = hotel['id']
+        if hid not in hotel_dates:
+            hotel_dates[hid] = []
+            hotels_seen.append(hotel)
+        hotel_dates[hid].append(rate_date)
+
+    # Shard hotels across workers
+    shards: List[List[Dict[str, Any]]] = [hotels_seen[i::concurrency] for i in range(concurrency)]
+    shards = [s for s in shards if s]
+
+    async def worker(hotel_shard: List[Dict[str, Any]], widx: int) -> Dict[str, int]:
+        acc = {'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0}
+        wdb = SyncSessionLocal()
+        backend = PlaywrightHotelPageBackend(proxy_config=proxy_util.load_config(wdb))
+        try:
+            for hotel in hotel_shard:
+                for rate_date in hotel_dates[hotel['id']]:
+                    try:
+                        result = await scrape_hotel_date(wdb, hotel, rate_date, backend, batch_id, adults)
+                    except Exception as e:
+                        logger.error(f"[worker {widx}] {hotel['name']} {rate_date} crashed: {e}")
+                        try:
+                            wdb.rollback()
+                        except Exception:
+                            pass
+                        acc['failed'] += 1
+                        continue
+
+                    if result.get('blocked'):
+                        acc['blocked'] += 1
+                        acc['failed'] += 1
+                    elif result['success']:
+                        acc['rates'] += result['rates_count']
+                        acc['completed'] += 1
+                    else:
+                        acc['failed'] += 1
+        finally:
+            try:
+                await backend.close()
+            except Exception:
+                pass
+            wdb.close()
+        return acc
+
+    results = await asyncio.gather(
+        *(worker(shard, i) for i, shard in enumerate(shards)),
+        return_exceptions=True,
+    )
+
+    agg = {'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0}
+    for r in results:
+        if isinstance(r, Exception):
+            logger.error(f"Hotel-page scrape worker crashed: {r}")
+            continue
+        for k in agg:
+            agg[k] += r[k]
+    return agg
+
+
 def _safe_mark_queue(db: Session, queue_id: Optional[int], status: str, error: str = None):
     """mark_queue_item that never raises — a marking failure shouldn't kill a worker."""
     if queue_id is None:
@@ -565,12 +723,105 @@ async def run_manual_scrape(
         _release_scrape_lock()
 
 
+async def _run_hotel_page_scrape(
+    db: Session,
+    from_date: date,
+    to_date: date,
+    scrape_type: str,
+) -> Dict[str, Any]:
+    """
+    Hotel-page scrape mode: load each known hotel's property page per date.
+
+    Unlike search-results mode, this scrapes known hotels (from booking_com_hotels
+    WHERE is_active AND booking_com_url IS NOT NULL) rather than discovering them
+    from search results. Workers are sharded by hotel so each worker's proxy session
+    covers all dates for one hotel before moving to the next.
+    """
+    hotels = get_active_hotels(db)
+    if not hotels:
+        return {
+            'success': False,
+            'error': 'No active hotels with booking_com_url in database. '
+                     'Run a search-results scrape first to populate hotels.',
+        }
+
+    dates: List[date] = []
+    current_date = from_date
+    while current_date <= to_date:
+        dates.append(current_date)
+        current_date += timedelta(days=1)
+
+    # Build (hotel, date) job pairs
+    hotel_date_jobs: List[Tuple[Dict[str, Any], date]] = [
+        (hotel, d) for hotel in hotels for d in dates
+    ]
+
+    batch_id = create_scrape_batch(db, scrape_type)
+    db.execute(
+        text("UPDATE booking_scrape_log SET dates_queued = :n WHERE batch_id = :bid"),
+        {'n': len(hotel_date_jobs), 'bid': str(batch_id)}
+    )
+    db.commit()
+
+    concurrency = _effective_concurrency(db, len(hotels))
+    logger.info(
+        f"Hotel-page scrape {from_date}..{to_date}: "
+        f"{len(hotels)} hotels × {len(dates)} dates = {len(hotel_date_jobs)} jobs, "
+        f"{concurrency} worker(s)"
+    )
+
+    try:
+        config = get_scrape_config(db)
+        adults = config['adults'] if config else 2
+        agg = await _scrape_hotels_concurrent(hotel_date_jobs, concurrency, batch_id, adults)
+
+        status = 'completed' if agg['completed'] else 'failed'
+        update_scrape_batch(
+            db, batch_id,
+            status=status,
+            hotels_found=len(hotels),
+            rates_scraped=agg['rates'],
+        )
+        db.execute(
+            text("""
+                UPDATE booking_scrape_log SET
+                    dates_completed = :completed, dates_failed = :failed
+                WHERE batch_id = :bid
+            """),
+            {'completed': agg['completed'], 'failed': agg['failed'], 'bid': str(batch_id)}
+        )
+        db.commit()
+
+        return {
+            'success': agg['completed'] > 0,
+            'blocked': agg['blocked'] > 0,
+            'dates_completed': agg['completed'],
+            'dates_failed': agg['failed'],
+            'hotels_found': len(hotels),
+            'rates_scraped': agg['rates'],
+        }
+
+    except Exception as e:
+        logger.error(f"Hotel-page scrape error: {e}")
+        update_scrape_batch(db, batch_id, status='failed', error_message=str(e))
+        return {
+            'success': False,
+            'error': str(e),
+            'dates_completed': 0,
+            'dates_failed': len(hotel_date_jobs),
+        }
+
+
 async def _run_manual_scrape_locked(
     db: Session,
     from_date: date,
     to_date: date
 ) -> Dict[str, Any]:
-    # Get config
+    # Hotel-page mode: iterate known hotels × dates
+    if _is_hotel_page_mode(db):
+        return await _run_hotel_page_scrape(db, from_date, to_date, 'manual')
+
+    # Search-results mode (original)
     config = get_scrape_config(db)
     if not config:
         return {
@@ -751,6 +1002,17 @@ async def process_queue(db: Session) -> Dict[str, Any]:
 
 
 async def _process_queue_locked(db: Session) -> Dict[str, Any]:
+    # Hotel-page mode ignores the queue and just scrapes the queued date range directly
+    if _is_hotel_page_mode(db):
+        items = get_pending_queue_items(db, limit=200)
+        if not items:
+            return {'success': True, 'dates_completed': 0, 'message': 'Queue empty'}
+        dates = [it['rate_date'] for it in items]
+        # Mark all as completed (hotel-page scrape manages its own tracking)
+        for it in items:
+            _safe_mark_queue(db, it['id'], 'completed')
+        return await _run_hotel_page_scrape(db, min(dates), max(dates), 'scheduled')
+
     # Get config
     config = get_scrape_config(db)
     if not config: