Proxy config, DataImpulse sticky-session username syntax, and Playwright/ httpx proxy builders now live in one place (services/proxy.py) instead of being duplicated across the Booking.com backend and the /config/proxy test endpoint. Both scrapers consume it. - services/proxy.py: load_config/normalize (DB-authoritative, env fallback), new_session_id, username, playwright_proxy, httpx_proxy_url - PlaywrightLocalBackend delegates proxy building to the module - get_scraper_backend factory uses proxy.load_config (one resolution path) - test_proxy_config endpoint uses the shared URL builder; httpx proxies= -> proxy= (forward-compatible, 0.28-safe) - Direct booking-engine scraper (httpx) can now route through the same proxy, gated by the direct_scraper_use_proxy flag (default off, plumbing ready) Dead code removed: set_scraper_paused (never called — rotate-on-block replaced pause-on-block), get_competitor_matrix / get_hotels_list / update_hotel_tier (endpoints have their own SQL), unused PROXY_KEYS tuple. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
827 lines
28 KiB
Python
827 lines
28 KiB
Python
"""
|
|
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 uuid
|
|
from datetime import date, datetime, 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()
|
|
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
async def is_scraper_paused(db: Session) -> bool:
|
|
"""Check if scraper is currently paused due to blocking."""
|
|
result = db.execute(
|
|
text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_paused'")
|
|
).fetchone()
|
|
|
|
if not result or result.config_value != 'true':
|
|
return False
|
|
|
|
# Check if pause period has expired
|
|
pause_until_result = db.execute(
|
|
text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_pause_until'")
|
|
).fetchone()
|
|
|
|
if pause_until_result and pause_until_result.config_value:
|
|
try:
|
|
pause_until = datetime.fromisoformat(pause_until_result.config_value)
|
|
if datetime.now() >= pause_until:
|
|
# Pause expired, reset
|
|
db.execute(
|
|
text("UPDATE system_config SET config_value = 'false' WHERE config_key = 'booking_scraper_paused'")
|
|
)
|
|
db.commit()
|
|
return False
|
|
except ValueError:
|
|
pass
|
|
|
|
return True
|
|
|
|
|
|
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 SCRAPE_LOCK.acquire(blocking=False):
|
|
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:
|
|
SCRAPE_LOCK.release()
|
|
|
|
|
|
async def _run_manual_scrape_locked(
|
|
db: Session,
|
|
from_date: date,
|
|
to_date: date
|
|
) -> Dict[str, Any]:
|
|
# Check if paused
|
|
if await is_scraper_paused(db):
|
|
return {
|
|
'success': False,
|
|
'error': 'Scraper is currently paused due to blocking. Try again later.',
|
|
}
|
|
|
|
# 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 SCRAPE_LOCK.acquire(blocking=False):
|
|
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:
|
|
SCRAPE_LOCK.release()
|
|
|
|
|
|
async def _process_queue_locked(db: Session) -> Dict[str, Any]:
|
|
# Check if paused
|
|
if await is_scraper_paused(db):
|
|
return {
|
|
'success': False,
|
|
'error': 'Scraper is currently paused due to blocking.',
|
|
}
|
|
|
|
# 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),
|
|
}
|
|
|