""" Booking.com Rate Scraper Service Main service for scraping competitor rates from booking.com. Uses pluggable backends (Playwright local, proxy, Apify) via factory pattern. Features: - Location-based search (1 query = 40+ hotels) - Hotel discovery and tier management - Rate extraction with availability status - Anti-scrape detection and pause/resume """ import asyncio import logging import threading import time import uuid from datetime import date, timedelta from decimal import Decimal from typing import List, Optional, Dict, Any, Tuple from sqlalchemy import text 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 logger = logging.getLogger(__name__) # One scrape at a time, process-wide: concurrent Chromium runs starve each # other of memory (page timeouts → partial results) and double the request # rate at Booking.com. Guards both manual scrapes and queue processing. SCRAPE_LOCK = threading.Lock() _lock_acquired_at: Optional[float] = None # monotonic timestamp when lock was acquired def _acquire_scrape_lock() -> bool: global _lock_acquired_at if SCRAPE_LOCK.acquire(blocking=False): _lock_acquired_at = time.monotonic() return True return False def _release_scrape_lock(): global _lock_acquired_at _lock_acquired_at = None try: SCRAPE_LOCK.release() except RuntimeError: pass # already released — race between watchdog and normal release def get_lock_status() -> dict: """Return lock state and how long it has been held (seconds).""" locked = SCRAPE_LOCK.locked() held = int(time.monotonic() - _lock_acquired_at) if (locked and _lock_acquired_at) else None return {"locked": locked, "held_seconds": held} def force_reset_scraper(db: Session) -> dict: """Force-release the scrape lock and tidy up any stuck DB state. Safe to call even when the lock is not held.""" was_locked = SCRAPE_LOCK.locked() _release_scrape_lock() # Mark any batch left in 'running' state as interrupted db.execute(text(""" UPDATE booking_scrape_log SET status = 'interrupted', completed_at = NOW(), error_message = 'Force-reset by admin' WHERE status = 'running' """)) db.commit() logger.warning(f"Scrape lock force-reset (was_locked={was_locked})") return {"was_locked": was_locked, "reset": True} def get_scraper_backend(db: Session) -> ScraperBackend: """ Factory to get configured scraper backend. Reads backend type from system_config and returns appropriate instance. """ # Get backend configuration result = db.execute( text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_backend'") ).fetchone() backend_type = result.config_value if result and result.config_value else 'playwright_local' if backend_type == 'apify': # Future: Apify backend raise NotImplementedError("Apify backend not yet implemented") 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)) def get_scrape_config(db: Session) -> Optional[Dict[str, Any]]: """Get the active scrape location configuration.""" result = db.execute( text(""" SELECT id, location_name, location_search_url, pages_to_scrape, adults, dest_id FROM booking_scrape_config WHERE is_active = TRUE ORDER BY id LIMIT 1 """) ).fetchone() if not result: return None return { 'id': result.id, 'location_name': result.location_name, 'location_search_url': result.location_search_url, 'pages_to_scrape': result.pages_to_scrape or 2, 'adults': result.adults or 2, 'dest_id': result.dest_id, } def save_hotel(db: Session, hotel: HotelData) -> int: """ Save or update a hotel in the database. Returns the hotel's database ID. """ # Check if hotel exists existing = db.execute( text("SELECT id FROM booking_com_hotels WHERE booking_com_id = :bid"), {'bid': hotel.booking_com_id} ).fetchone() if existing: # Update last_seen_at and any changed fields db.execute( text(""" UPDATE booking_com_hotels SET name = COALESCE(:name, name), booking_com_url = COALESCE(:url, booking_com_url), star_rating = COALESCE(:stars, star_rating), review_score = COALESCE(:score, review_score), review_count = COALESCE(:count, review_count), last_seen_at = NOW() WHERE booking_com_id = :bid """), { 'bid': hotel.booking_com_id, 'name': hotel.name, 'url': hotel.booking_com_url, 'stars': float(hotel.star_rating) if hotel.star_rating else None, 'score': float(hotel.review_score) if hotel.review_score else None, 'count': hotel.review_count, } ) return existing.id else: # Insert new hotel (default tier is 'market') result = db.execute( text(""" INSERT INTO booking_com_hotels (booking_com_id, name, booking_com_url, star_rating, review_score, review_count, tier) VALUES (:bid, :name, :url, :stars, :score, :count, 'market') RETURNING id """), { 'bid': hotel.booking_com_id, 'name': hotel.name, 'url': hotel.booking_com_url, 'stars': float(hotel.star_rating) if hotel.star_rating else None, 'score': float(hotel.review_score) if hotel.review_score else None, 'count': hotel.review_count, } ) return result.fetchone().id def save_rate(db: Session, rate: RateData, hotel_id: int, batch_id: uuid.UUID): """Save a rate to the database.""" db.execute( 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) VALUES (:hotel_id, :rate_date, :status, :rate, :currency, :room_type, :breakfast, :cancel, :prepay, :rooms_left, :batch_id) """), { 'hotel_id': hotel_id, 'rate_date': rate.rate_date, 'status': rate.availability_status.value, 'rate': float(rate.rate_gross) if rate.rate_gross else None, 'currency': rate.currency, 'room_type': rate.room_type, 'breakfast': rate.breakfast_included, 'cancel': rate.free_cancellation, 'prepay': rate.no_prepayment, 'rooms_left': rate.rooms_left, 'batch_id': str(batch_id), } ) def create_scrape_batch(db: Session, scrape_type: str) -> uuid.UUID: """Create a new scrape batch log entry.""" batch_id = uuid.uuid4() db.execute( text(""" INSERT INTO booking_scrape_log (batch_id, scrape_type, started_at, status) VALUES (:batch_id, :scrape_type, NOW(), 'running') """), {'batch_id': str(batch_id), 'scrape_type': scrape_type} ) db.commit() return batch_id def update_scrape_batch( db: Session, batch_id: uuid.UUID, status: str, hotels_found: int = 0, rates_scraped: int = 0, error_message: str = None, blocked: bool = False ): """Update scrape batch log with results.""" db.execute( text(""" UPDATE booking_scrape_log SET completed_at = CASE WHEN :status IN ('completed', 'failed', 'blocked') THEN NOW() ELSE NULL END, status = :status, hotels_found = :hotels, rates_scraped = :rates, error_message = :error, blocked_at = CASE WHEN :blocked THEN NOW() ELSE NULL END, resume_after = CASE WHEN :blocked THEN NOW() + INTERVAL '2 hours' ELSE NULL END WHERE batch_id = :batch_id """), { 'batch_id': str(batch_id), 'status': status, 'hotels': hotels_found, 'rates': rates_scraped, 'error': error_message, 'blocked': blocked, } ) db.commit() def cleanup_stale_batches(db: Session, max_age_minutes: int = 60): """ Mark any 'running' scrape batches as 'failed' if they've been running longer than max_age_minutes. This handles orphaned batches from container restarts or crashes. """ result = db.execute( text(""" UPDATE booking_scrape_log SET status = 'failed', completed_at = NOW(), error_message = 'Interrupted (container restart or timeout)' WHERE status = 'running' AND started_at < NOW() - INTERVAL ':mins minutes' RETURNING batch_id """.replace(':mins', str(int(max_age_minutes)))) ) cleaned = result.fetchall() db.commit() if cleaned: logger.info(f"Cleaned up {len(cleaned)} stale running scrape batch(es)") return len(cleaned) async def scrape_date( db: Session, rate_date: date, backend: ScraperBackend, config: Dict[str, Any], batch_id: uuid.UUID ) -> Dict[str, Any]: """ Scrape rates for a single date. Args: db: Database session rate_date: Date to scrape rates for backend: Scraper backend instance config: Scrape configuration batch_id: Current batch ID Returns: Dict with 'success', 'blocked', 'hotels_count', 'rates_count' """ check_in = rate_date check_out = rate_date + timedelta(days=1) # Single night result = await backend.scrape_location_search( location=config['location_name'], check_in=check_in, check_out=check_out, adults=config['adults'], pages=config['pages_to_scrape'], dest_id=config.get('dest_id'), search_url=config.get('location_search_url'), ) if result.blocked: return { 'success': False, 'blocked': True, 'block_reason': result.block_reason, 'hotels_count': 0, 'rates_count': 0, } if not result.success: return { 'success': False, 'blocked': False, 'error': result.error_message, 'hotels_count': 0, 'rates_count': 0, } # Save hotels and rates hotels_saved = 0 rates_saved = 0 for hotel, rate in zip(result.hotels, result.rates): if not hotel.booking_com_id: continue try: hotel_id = save_hotel(db, hotel) save_rate(db, rate, hotel_id, batch_id) hotels_saved += 1 rates_saved += 1 except Exception as e: logger.warning(f"Error saving hotel/rate: {e}") # A failed statement aborts the transaction — roll back so the # remaining hotels in this batch can still be saved db.rollback() continue # Flag known hotels absent from this successful scrape as 'not_listed' # (sold out or pushed off the search results). Without this their last # 'available' rate stays the latest row for the date and reads as a # current price, skewing market averages. # # Absence is only trustworthy when the scrape saw the WHOLE market: # a partial scrape (a page timed out / never rendered) or one that saw # far fewer hotels than this date normally lists would mark still-listed # hotels not_listed and wrongly suppress their last known rates. In # those cases keep the rates we did save but skip the flagging, so # unseen hotels retain their last known rate and scrape time. seen_ids = [h.booking_com_id for h in result.hotels if h.booking_com_id] flag_absent = bool(seen_ids) if flag_absent and result.pages_ok < result.pages_requested: logger.warning( f"Partial scrape for {rate_date} ({result.pages_ok}/{result.pages_requested} " f"pages ok, {len(seen_ids)} hotels) — skipping not_listed flagging" ) flag_absent = False if flag_absent: try: baseline = db.execute( text(""" SELECT COUNT(DISTINCT hotel_id) FROM booking_com_rates WHERE rate_date = :rate_date AND availability_status IN ('available', 'sold_out') AND scraped_at > NOW() - INTERVAL '7 days' AND scrape_batch_id != :batch_id """), {'rate_date': rate_date, 'batch_id': str(batch_id)} ).scalar() or 0 if baseline and len(seen_ids) < baseline * 0.6: logger.warning( f"Scrape for {rate_date} saw {len(seen_ids)} hotels vs recent " f"baseline {baseline} — skipping not_listed flagging" ) flag_absent = False except Exception as e: logger.warning(f"Baseline check failed for {rate_date}: {e}") db.rollback() flag_absent = False if flag_absent: try: db.execute( text(""" INSERT INTO booking_com_rates (hotel_id, rate_date, availability_status, rate_gross, scrape_batch_id) SELECT h.id, :rate_date, 'not_listed', NULL, :batch_id FROM booking_com_hotels h WHERE h.is_active = TRUE AND h.booking_com_id != ALL(:seen_ids) """), {'rate_date': rate_date, 'batch_id': str(batch_id), 'seen_ids': seen_ids} ) except Exception as e: logger.warning(f"Error flagging unlisted hotels for {rate_date}: {e}") db.rollback() db.commit() return { 'success': True, 'blocked': False, 'hotels_count': hotels_saved, 'rates_count': rates_saved, } def get_scraper_concurrency(db: Session) -> int: """Number of parallel scrape workers (config key, default 3). Each worker runs its own browser on its own residential proxy IP, so raise this only with proxy IPs and RAM to spare (~0.4 GB per worker).""" row = db.execute( text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_concurrency'") ).fetchone() try: return max(1, int(row.config_value)) if row and row.config_value else 3 except (ValueError, TypeError): return 3 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.""" configured = get_scraper_concurrency(db) if configured <= 1 or n_jobs <= 1: return 1 # get_scraper_backend does no I/O beyond the config read; safe to probe. if not get_scraper_backend(db)._proxy_enabled(): logger.info("Proxy disabled — running scrape serially (parallelism needs per-worker IPs)") return 1 return max(1, min(configured, n_jobs)) 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: return try: mark_queue_item(db, queue_id, status, error) except Exception as e: logger.warning(f"Failed to mark queue item {queue_id} as {status}: {e}") try: db.rollback() except Exception: pass async def _scrape_dates_concurrent( jobs: List[Tuple[date, Optional[int]]], config: Dict[str, Any], concurrency: int, batch_id: uuid.UUID, ) -> Dict[str, int]: """Scrape a list of (date, queue_id) jobs across `concurrency` workers. Each worker owns a DB session and a scraper backend — and because every backend picks its own random sticky-session id, each worker scrapes from a distinct residential IP. Dates are interleaved across workers so each covers a spread of the range. Returns aggregate counts.""" shards = [jobs[i::concurrency] for i in range(concurrency)] shards = [s for s in shards if s] async def worker(shard: List[Tuple[date, Optional[int]]], widx: int) -> Dict[str, int]: acc = {'hotels': 0, 'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0} wdb = SyncSessionLocal() backend = get_scraper_backend(wdb) try: for rate_date, queue_id in shard: try: result = await scrape_date(wdb, rate_date, backend, config, batch_id) except Exception as e: logger.error(f"[worker {widx}] {rate_date} crashed: {e}") try: wdb.rollback() except Exception: pass acc['failed'] += 1 _safe_mark_queue(wdb, queue_id, 'failed', str(e)) continue if result.get('blocked'): acc['blocked'] += 1 acc['failed'] += 1 _safe_mark_queue(wdb, queue_id, 'failed', f"Blocked: {result.get('block_reason')}") elif result['success']: acc['hotels'] += result['hotels_count'] acc['rates'] += result['rates_count'] acc['completed'] += 1 _safe_mark_queue(wdb, queue_id, 'completed') else: acc['failed'] += 1 _safe_mark_queue(wdb, queue_id, 'failed', result.get('error')) 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 = {'hotels': 0, 'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0} for r in results: if isinstance(r, Exception): logger.error(f"Scrape worker crashed: {r}") continue for k in agg: agg[k] += r[k] return agg async def run_manual_scrape( db: Session, from_date: date, to_date: date = None ) -> Dict[str, Any]: """ Run a manual scrape for testing/on-demand use. Args: db: Database session from_date: Start date to_date: End date (defaults to from_date for single day) Returns: Dict with scrape results summary """ if to_date is None: to_date = from_date if not _acquire_scrape_lock(): logger.warning(f"Manual scrape {from_date}..{to_date} refused — another scrape is running") return { 'success': False, 'already_running': True, 'error': 'Another scrape is already running. Try again when it finishes.', } try: return await _run_manual_scrape_locked(db, from_date, to_date) finally: _release_scrape_lock() async def _run_manual_scrape_locked( db: Session, from_date: date, to_date: date ) -> Dict[str, Any]: # Get config config = get_scrape_config(db) if not config: return { 'success': False, 'error': 'No scrape location configured. Add a location in settings.', } # Create batch batch_id = create_scrape_batch(db, 'manual') # Build the date list and fan it out across the worker pool jobs: List[Tuple[date, Optional[int]]] = [] current_date = from_date while current_date <= to_date: jobs.append((current_date, None)) current_date += timedelta(days=1) concurrency = _effective_concurrency(db, len(jobs)) logger.info(f"Manual scrape {from_date}..{to_date}: {len(jobs)} date(s) across {concurrency} worker(s)") try: agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id) update_scrape_batch( db, batch_id, status='completed' if agg['completed'] or not jobs else 'failed', hotels_found=agg['hotels'], rates_scraped=agg['rates'], ) return { 'success': agg['completed'] > 0 or not jobs, 'blocked': agg['blocked'] > 0, 'dates_completed': agg['completed'], 'dates_failed': agg['failed'], 'hotels_found': agg['hotels'], 'rates_scraped': agg['rates'], } except Exception as e: logger.error(f"Scrape error: {e}") update_scrape_batch( db, batch_id, status='failed', hotels_found=0, rates_scraped=0, error_message=str(e) ) return { 'success': False, 'error': str(e), 'dates_completed': 0, 'dates_failed': len(jobs), 'hotels_found': 0, 'rates_scraped': 0, } # ============================================ # QUEUE MANAGEMENT # ============================================ def populate_queue(db: Session, dates: List[date], priorities: Dict[date, int] = None): """ Add dates to the scrape queue, skipping any already pending/processing. Args: db: Database session dates: Dates to add to the queue priorities: Optional priority map (higher = scraped first). Default: 0 """ if not dates: return 0 added = 0 for rate_date in dates: priority = (priorities or {}).get(rate_date, 0) try: db.execute( text(""" INSERT INTO booking_scrape_queue (rate_date, status, priority) VALUES (:rate_date, 'pending', :priority) ON CONFLICT (rate_date, status) DO UPDATE SET priority = GREATEST(booking_scrape_queue.priority, :priority) """), {'rate_date': rate_date, 'priority': priority} ) added += 1 except Exception: # Ignore duplicates or constraint issues pass db.commit() logger.info(f"Queue: added/updated {added} dates") return added def get_pending_queue_items(db: Session, limit: int = 50) -> List[Dict[str, Any]]: """Get pending queue items ordered by priority (highest first), then date.""" result = db.execute( text(""" SELECT id, rate_date, priority, attempts, max_attempts FROM booking_scrape_queue WHERE status = 'pending' AND attempts < max_attempts ORDER BY priority DESC, rate_date ASC LIMIT :limit """), {'limit': limit} ) return [dict(row._mapping) for row in result.fetchall()] def mark_queue_item(db: Session, queue_id: int, status: str, error_message: str = None): """Update a queue item's status.""" if status == 'completed': db.execute( text(""" UPDATE booking_scrape_queue SET status = 'completed', completed_at = NOW(), last_attempt_at = NOW(), attempts = attempts + 1 WHERE id = :id """), {'id': queue_id} ) elif status == 'failed': db.execute( text(""" UPDATE booking_scrape_queue SET status = CASE WHEN attempts + 1 >= max_attempts THEN 'failed' ELSE 'pending' END, last_attempt_at = NOW(), attempts = attempts + 1, error_message = :error WHERE id = :id """), {'id': queue_id, 'error': error_message} ) db.commit() def clear_old_queue_items(db: Session, days: int = 7): """Remove completed/failed queue items older than N days.""" db.execute( text(""" DELETE FROM booking_scrape_queue WHERE status IN ('completed', 'failed') AND created_at < NOW() - INTERVAL ':days days' """.replace(':days', str(int(days)))) ) db.commit() async def process_queue(db: Session) -> Dict[str, Any]: """ Process pending items from the scrape queue. Picks up pending items in priority order, scrapes each date, and handles blocking/retries. Returns: Dict with processing results """ if not _acquire_scrape_lock(): logger.warning("Queue processing skipped — another scrape is running") return { 'success': False, 'already_running': True, 'error': 'Another scrape is already running.', } try: return await _process_queue_locked(db) finally: _release_scrape_lock() async def _process_queue_locked(db: Session) -> Dict[str, Any]: # Get config config = get_scrape_config(db) if not config: return { 'success': False, 'error': 'No scrape location configured.', } # Get pending items items = get_pending_queue_items(db, limit=200) if not items: return {'success': True, 'dates_completed': 0, 'message': 'Queue empty'} # Create batch batch_id = create_scrape_batch(db, 'scheduled') # Update batch with queue count db.execute( text("UPDATE booking_scrape_log SET dates_queued = :count WHERE batch_id = :bid"), {'count': len(items), 'bid': str(batch_id)} ) db.commit() # Fan the queue out across the worker pool (each worker marks its own items) jobs: List[Tuple[date, Optional[int]]] = [(it['rate_date'], it['id']) for it in items] concurrency = _effective_concurrency(db, len(jobs)) logger.info(f"Queue processing: {len(jobs)} date(s) across {concurrency} worker(s)") try: agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id) update_scrape_batch( db, batch_id, status='completed' if agg['completed'] else 'failed', hotels_found=agg['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': agg['hotels'], 'rates_scraped': agg['rates'], } except Exception as e: logger.error(f"Queue processing error: {e}") update_scrape_batch( db, batch_id, status='failed', hotels_found=0, rates_scraped=0, error_message=str(e) ) return { 'success': False, 'error': str(e), 'dates_completed': 0, 'dates_failed': len(jobs), }