jtricerolph
8a80c66a3b
scrape_batch_id in booking_com_rates is a uuid column, not text. Casting bsl.batch_id::text made the comparison uuid = text which Postgres rejects. Drop the cast — both sides are uuid. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1216 lines
42 KiB
Python
1216 lines
42 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 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, PlaywrightHotelPageBackend,
|
||
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, back-filling actual counts
|
||
db.execute(text("""
|
||
UPDATE booking_scrape_log bsl
|
||
SET status = 'interrupted', completed_at = NOW(),
|
||
error_message = 'Force-reset by admin',
|
||
hotels_found = (
|
||
SELECT COUNT(DISTINCT hotel_id)
|
||
FROM booking_com_rates
|
||
WHERE scrape_batch_id = bsl.batch_id
|
||
AND rate_gross IS NOT NULL
|
||
),
|
||
rates_scraped = (
|
||
SELECT COUNT(*)
|
||
FROM booking_com_rates
|
||
WHERE scrape_batch_id = bsl.batch_id
|
||
AND rate_gross IS NOT NULL
|
||
)
|
||
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")
|
||
|
||
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_cfg)
|
||
|
||
|
||
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,
|
||
rate_plan_id, max_persons, scrape_batch_id,
|
||
breakfast_text, cancel_text, payment_text)
|
||
VALUES (:hotel_id, :rate_date, :status, :rate, :currency, :room_type,
|
||
:breakfast, :cancel, :prepay, :rooms_left,
|
||
:rate_plan_id, :max_persons, :batch_id,
|
||
:breakfast_text, :cancel_text, :payment_text)
|
||
"""),
|
||
{
|
||
'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,
|
||
'rate_plan_id': rate.rate_plan_id,
|
||
'max_persons': rate.max_persons,
|
||
'batch_id': str(batch_id),
|
||
'breakfast_text': rate.breakfast_text,
|
||
'cancel_text': rate.cancel_text,
|
||
'payment_text': rate.payment_text,
|
||
}
|
||
)
|
||
|
||
|
||
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.
|
||
Counts are back-filled from booking_com_rates so the history shows what
|
||
was actually scraped before the interruption, not zeroes.
|
||
"""
|
||
result = db.execute(
|
||
text("""
|
||
UPDATE booking_scrape_log bsl SET
|
||
status = 'failed',
|
||
completed_at = NOW(),
|
||
error_message = 'Interrupted (container restart or timeout)',
|
||
hotels_found = (
|
||
SELECT COUNT(DISTINCT hotel_id)
|
||
FROM booking_com_rates
|
||
WHERE scrape_batch_id = bsl.batch_id
|
||
AND rate_gross IS NOT NULL
|
||
),
|
||
rates_scraped = (
|
||
SELECT COUNT(*)
|
||
FROM booking_com_rates
|
||
WHERE scrape_batch_id = bsl.batch_id
|
||
AND rate_gross IS NOT NULL
|
||
)
|
||
WHERE bsl.status = 'running'
|
||
AND bsl.started_at < NOW() - INTERVAL ':mins minutes'
|
||
RETURNING bsl.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)
|
||
|
||
|
||
def get_active_hotels(db: Session) -> List[Dict[str, Any]]:
|
||
"""Return own + competitor hotels with a booking_com_url (for hotel-page scraping).
|
||
Market hotels are excluded — they are auto-discovered from search results and not
|
||
individually monitored. Own hotel IS included so its Booking.com listing is tracked."""
|
||
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
|
||
AND tier IN ('own', 'competitor')
|
||
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 = 4,
|
||
) -> 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,
|
||
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 6). Each worker
|
||
runs its own browser on its own residential proxy IP (~0.4 GB each), so
|
||
keep this below RAM / 0.4 GB. Requires proxy — serial fallback if off."""
|
||
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 6
|
||
except (ValueError, TypeError):
|
||
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, require_proxy: bool = True) -> int:
|
||
"""Clamp configured concurrency to the workload.
|
||
|
||
require_proxy=True (search-results mode): force serial without a proxy because
|
||
all workers would hit the same search URL from the same IP.
|
||
require_proxy=False (hotel-page mode): allow parallelism even without proxy —
|
||
each worker hits a different hotel's URL so concurrent requests look like normal
|
||
multi-tab browsing rather than a hammered aggregation endpoint.
|
||
"""
|
||
configured = get_scraper_concurrency(db)
|
||
if configured <= 1 or n_jobs <= 1:
|
||
return 1
|
||
if require_proxy and not get_scraper_backend(db)._proxy_enabled():
|
||
logger.info("Proxy disabled — running search-results scrape serially (parallelism needs per-worker IPs)")
|
||
return 1
|
||
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 DATE so each worker owns a slice of dates and scrapes
|
||
all hotels for each date before moving to the next. This means any
|
||
interruption or block leaves whole dates complete rather than partial
|
||
(all hotels on date N done, none on date N+1) rather than some hotels
|
||
fully done and others not started.
|
||
"""
|
||
# Collect unique hotels (ordered) and unique dates (sorted)
|
||
hotels_seen: List[Dict[str, Any]] = []
|
||
hotels_index: Dict[int, Dict[str, Any]] = {}
|
||
unique_dates: List[date] = []
|
||
dates_seen: set = set()
|
||
for hotel, rate_date in hotel_date_jobs:
|
||
hid = hotel['id']
|
||
if hid not in hotels_index:
|
||
hotels_index[hid] = hotel
|
||
hotels_seen.append(hotel)
|
||
if rate_date not in dates_seen:
|
||
dates_seen.add(rate_date)
|
||
unique_dates.append(rate_date)
|
||
unique_dates.sort()
|
||
|
||
# Shard dates across workers
|
||
date_shards: List[List[date]] = [unique_dates[i::concurrency] for i in range(concurrency)]
|
||
date_shards = [s for s in date_shards if s]
|
||
|
||
async def worker(date_shard: List[date], 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 rate_date in date_shard:
|
||
for hotel in hotels_seen:
|
||
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
|
||
_increment_batch_progress(wdb, batch_id, delta_failed=1)
|
||
elif result['success']:
|
||
acc['rates'] += result['rates_count']
|
||
acc['completed'] += 1
|
||
_increment_batch_progress(wdb, batch_id,
|
||
delta_rates=result['rates_count'],
|
||
delta_completed=1)
|
||
else:
|
||
acc['failed'] += 1
|
||
_increment_batch_progress(wdb, batch_id, delta_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(date_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 _increment_batch_progress(
|
||
db: Session, batch_id: uuid.UUID,
|
||
delta_hotels: int = 0, delta_rates: int = 0,
|
||
delta_completed: int = 0, delta_failed: int = 0,
|
||
):
|
||
"""Atomically increment running-batch counters. Safe for concurrent workers
|
||
because it uses += rather than read-modify-write."""
|
||
try:
|
||
db.execute(text("""
|
||
UPDATE booking_scrape_log
|
||
SET hotels_found = COALESCE(hotels_found, 0) + :dh,
|
||
rates_scraped = COALESCE(rates_scraped, 0) + :dr,
|
||
dates_completed = COALESCE(dates_completed, 0) + :dc,
|
||
dates_failed = COALESCE(dates_failed, 0) + :df
|
||
WHERE batch_id = :bid
|
||
"""), {'bid': str(batch_id), 'dh': delta_hotels, 'dr': delta_rates,
|
||
'dc': delta_completed, 'df': delta_failed})
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning(f"Progress update failed for batch {batch_id}: {e}")
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
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))
|
||
_increment_batch_progress(wdb, batch_id, delta_failed=1)
|
||
continue
|
||
|
||
if result.get('blocked'):
|
||
acc['blocked'] += 1
|
||
acc['failed'] += 1
|
||
_safe_mark_queue(wdb, queue_id, 'failed', f"Blocked: {result.get('block_reason')}")
|
||
_increment_batch_progress(wdb, batch_id, delta_failed=1)
|
||
elif result['success']:
|
||
acc['hotels'] += result['hotels_count']
|
||
acc['rates'] += result['rates_count']
|
||
acc['completed'] += 1
|
||
_safe_mark_queue(wdb, queue_id, 'completed')
|
||
_increment_batch_progress(wdb, batch_id,
|
||
delta_hotels=result['hotels_count'],
|
||
delta_rates=result['rates_count'],
|
||
delta_completed=1)
|
||
else:
|
||
acc['failed'] += 1
|
||
_safe_mark_queue(wdb, queue_id, 'failed', result.get('error'))
|
||
_increment_batch_progress(wdb, batch_id, delta_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 = {'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_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), require_proxy=False)
|
||
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]:
|
||
# 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 {
|
||
'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,
|
||
}
|
||
|
||
|
||
async def run_discovery_scrape(db: Session) -> Dict[str, Any]:
|
||
"""
|
||
Run a search-results scrape for a single date to discover market hotels.
|
||
Always uses playwright_local (search-results) regardless of the configured
|
||
backend — intended for hotel discovery, not rate tracking.
|
||
Returns new_hotels: count of hotels added to the DB this run.
|
||
"""
|
||
config = get_scrape_config(db)
|
||
if not config:
|
||
return {'success': False, 'error': 'No scrape location configured.'}
|
||
|
||
hotels_before = db.execute(
|
||
text("SELECT COUNT(*) FROM booking_com_hotels WHERE is_active = TRUE")
|
||
).scalar() or 0
|
||
|
||
batch_id = create_scrape_batch(db, 'discovery')
|
||
check_in = date.today() + timedelta(days=14)
|
||
jobs: List[Tuple[date, Optional[int]]] = [(check_in, None)]
|
||
|
||
try:
|
||
agg = await _scrape_dates_concurrent(jobs, config, concurrency=1, batch_id=batch_id)
|
||
update_scrape_batch(
|
||
db, batch_id,
|
||
status='completed' if agg['completed'] else 'failed',
|
||
hotels_found=agg['hotels'],
|
||
rates_scraped=agg['rates'],
|
||
)
|
||
|
||
hotels_after = db.execute(
|
||
text("SELECT COUNT(*) FROM booking_com_hotels WHERE is_active = TRUE")
|
||
).scalar() or 0
|
||
|
||
return {
|
||
'success': agg['completed'] > 0,
|
||
'hotels_found': agg['hotels'],
|
||
'new_hotels': max(0, hotels_after - hotels_before),
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Discovery scrape error: {e}")
|
||
update_scrape_batch(db, batch_id, status='failed', error_message=str(e))
|
||
return {'success': False, 'error': str(e)}
|
||
|
||
|
||
# ============================================
|
||
# 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]:
|
||
# 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:
|
||
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),
|
||
}
|
||
|