diff --git a/backend/services/booking_scraper.py b/backend/services/booking_scraper.py index c3d400a..b1c2797 100644 --- a/backend/services/booking_scraper.py +++ b/backend/services/booking_scraper.py @@ -11,16 +11,18 @@ Features: - Anti-scrape detection and pause/resume """ +import asyncio import logging import threading import uuid from datetime import date, datetime, timedelta from decimal import Decimal -from typing import List, Optional, Dict, Any +from typing import List, Optional, Dict, Any, Tuple from sqlalchemy import text from sqlalchemy.orm import Session +from database import SyncSessionLocal from .scraper_backends import ScraperBackend, PlaywrightLocalBackend, HotelData, RateData, AvailabilityStatus logger = logging.getLogger(__name__) @@ -446,6 +448,114 @@ async def scrape_date( } +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, @@ -501,67 +611,33 @@ async def _run_manual_scrape_locked( # Create batch batch_id = create_scrape_batch(db, 'manual') - # Get backend - backend = get_scraper_backend(db) + # 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) - total_hotels = 0 - total_rates = 0 - dates_completed = 0 - dates_failed = 0 + concurrency = _effective_concurrency(db, len(jobs)) + logger.info(f"Manual scrape {from_date}..{to_date}: {len(jobs)} date(s) across {concurrency} worker(s)") try: - current_date = from_date - while current_date <= to_date: - logger.info(f"Scraping date: {current_date}") + agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id) - result = await scrape_date(db, current_date, backend, config, batch_id) - - if result['blocked']: - # Blocking detected - pause and exit - await set_scraper_paused(db, True, hours=2) - update_scrape_batch( - db, batch_id, - status='blocked', - hotels_found=total_hotels, - rates_scraped=total_rates, - error_message=f"Blocked: {result.get('block_reason', 'unknown')}", - blocked=True - ) - return { - 'success': False, - 'blocked': True, - 'block_reason': result.get('block_reason'), - 'dates_completed': dates_completed, - 'dates_failed': dates_failed, - 'hotels_found': total_hotels, - 'rates_scraped': total_rates, - } - - if result['success']: - total_hotels += result['hotels_count'] - total_rates += result['rates_count'] - dates_completed += 1 - else: - dates_failed += 1 - logger.warning(f"Failed to scrape {current_date}: {result.get('error')}") - - current_date += timedelta(days=1) - - # Update batch as completed update_scrape_batch( db, batch_id, - status='completed', - hotels_found=total_hotels, - rates_scraped=total_rates + status='completed' if agg['completed'] or not jobs else 'failed', + hotels_found=agg['hotels'], + rates_scraped=agg['rates'], ) return { - 'success': True, - 'blocked': False, - 'dates_completed': dates_completed, - 'dates_failed': dates_failed, - 'hotels_found': total_hotels, - 'rates_scraped': total_rates, + '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: @@ -569,20 +645,18 @@ async def _run_manual_scrape_locked( update_scrape_batch( db, batch_id, status='failed', - hotels_found=total_hotels, - rates_scraped=total_rates, + hotels_found=0, + rates_scraped=0, error_message=str(e) ) return { 'success': False, 'error': str(e), - 'dates_completed': dates_completed, - 'dates_failed': dates_failed, - 'hotels_found': total_hotels, - 'rates_scraped': total_rates, + 'dates_completed': 0, + 'dates_failed': len(jobs), + 'hotels_found': 0, + 'rates_scraped': 0, } - finally: - await backend.close() # ============================================ @@ -737,72 +811,19 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]: ) db.commit() - # Get backend - backend = get_scraper_backend(db) - - total_hotels = 0 - total_rates = 0 - dates_completed = 0 - dates_failed = 0 + # 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: - for item in items: - rate_date = item['rate_date'] - queue_id = item['id'] + agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id) - logger.info(f"Queue processing: {rate_date} (priority={item['priority']}, attempt={item['attempts']+1})") - - result = await scrape_date(db, rate_date, backend, config, batch_id) - - if result['blocked']: - # Mark this item as failed, pause, and stop - mark_queue_item(db, queue_id, 'failed', f"Blocked: {result.get('block_reason')}") - await set_scraper_paused(db, True, hours=2) - update_scrape_batch( - db, batch_id, - status='blocked', - hotels_found=total_hotels, - rates_scraped=total_rates, - error_message=f"Blocked: {result.get('block_reason', 'unknown')}", - blocked=True - ) - # Update dates counters - db.execute( - text(""" - UPDATE booking_scrape_log SET - dates_completed = :completed, - dates_failed = :failed - WHERE batch_id = :bid - """), - {'completed': dates_completed, 'failed': dates_failed + 1, 'bid': str(batch_id)} - ) - db.commit() - return { - 'success': False, - 'blocked': True, - 'block_reason': result.get('block_reason'), - 'dates_completed': dates_completed, - 'dates_failed': dates_failed + 1, - 'hotels_found': total_hotels, - 'rates_scraped': total_rates, - } - - if result['success']: - mark_queue_item(db, queue_id, 'completed') - total_hotels += result['hotels_count'] - total_rates += result['rates_count'] - dates_completed += 1 - else: - mark_queue_item(db, queue_id, 'failed', result.get('error')) - dates_failed += 1 - logger.warning(f"Queue: failed to scrape {rate_date}: {result.get('error')}") - - # Update batch as completed update_scrape_batch( db, batch_id, - status='completed', - hotels_found=total_hotels, - rates_scraped=total_rates + status='completed' if agg['completed'] else 'failed', + hotels_found=agg['hotels'], + rates_scraped=agg['rates'], ) db.execute( text(""" @@ -811,17 +832,17 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]: dates_failed = :failed WHERE batch_id = :bid """), - {'completed': dates_completed, 'failed': dates_failed, 'bid': str(batch_id)} + {'completed': agg['completed'], 'failed': agg['failed'], 'bid': str(batch_id)} ) db.commit() return { - 'success': True, - 'blocked': False, - 'dates_completed': dates_completed, - 'dates_failed': dates_failed, - 'hotels_found': total_hotels, - 'rates_scraped': total_rates, + '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: @@ -829,18 +850,16 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]: update_scrape_batch( db, batch_id, status='failed', - hotels_found=total_hotels, - rates_scraped=total_rates, + hotels_found=0, + rates_scraped=0, error_message=str(e) ) return { 'success': False, 'error': str(e), - 'dates_completed': dates_completed, - 'dates_failed': dates_failed, + 'dates_completed': 0, + 'dates_failed': len(jobs), } - finally: - await backend.close() def get_competitor_matrix( diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 0f2fd9b..1c74c6f 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -123,6 +123,12 @@ function ProxyTab() { const [password, setPassword] = useState('') const [testResult, setTestResult] = useState(null) const [testError, setTestError] = useState(null) + const [workers, setWorkers] = useState('3') + + const { data: sysCfg } = useQuery({ + queryKey: ['system-config'], + queryFn: () => api.get('/competitors/config/system').then(r => r.data), + }) // Seed form once the stored config arrives (password stays blank by design) useEffect(() => { @@ -135,6 +141,17 @@ function ProxyTab() { } }, [cfg]) + useEffect(() => { + if (sysCfg?.booking_scraper_concurrency) setWorkers(String(sysCfg.booking_scraper_concurrency)) + }, [sysCfg]) + + const saveWorkers = useMutation({ + mutationFn: () => api.post('/competitors/config/system', { + key: 'booking_scraper_concurrency', value: String(Math.max(1, parseInt(workers) || 1)), + }), + onSuccess: () => qc.invalidateQueries({ queryKey: ['system-config'] }), + }) + const save = useMutation({ mutationFn: () => api.post('/competitors/config/proxy', { @@ -248,6 +265,38 @@ function ProxyTab() { )} + +
+
+ + + Parallel Workers + +
+
+

+ How many dates to scrape at once. Each worker runs its own browser on its own + residential IP, so a run finishes roughly this many times faster. Only takes effect + when the proxy is enabled (without it, all workers would share one IP). Budget about + 0.4 GB of RAM per worker — 3 is safe on the current 4 GB; raise it after + adding RAM. +

+
+ setWorkers(e.target.value)} style={{ width: 90 }} /> + + {!enabled && ( + + Enable the proxy above for this to apply + + )} +
+
+
) } @@ -493,6 +542,7 @@ function SystemTab({ config, isLoading }: { config: SystemConfig | undefined; is 'booking_proxy_enabled', 'booking_proxy_host', 'booking_proxy_country', + 'booking_scraper_concurrency', 'sync_newbook_current_rates_enabled', 'sync_newbook_current_rates_time', ]