hotel-manage-stack/rates
1
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

Switch scrape sharding from hotel-first to date-first

Browse source 
Workers now own a slice of dates and scrape all hotels per date before
advancing. A block or interruption leaves complete dates rather than
some hotels fully done and others not started.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jtricerolph 2026-07-10 01:34:46 +00:00
parent 8ccc0e272d
commit effa982a8e
1 changed files with 21 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

35
backend/services/booking_scraper.py
View file

@ -551,31 +551,38 @@ async def _scrape_hotels_concurrent(
"""
Scrape (hotel, date) pairs using the hotel-page backend.
Jobs are sharded by HOTEL (not date) so each worker keeps its proxy session
alive across all dates for one hotel — looks like a single user checking
availability for a trip.
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.
"""
# Group by hotel to get stable shards
# Collect unique hotels (ordered) and unique dates (sorted)
hotels_seen: List[Dict[str, Any]] = []
hotel_dates: Dict[int, List[date]] = {}
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 hotel_dates:
hotel_dates[hid] = []
if hid not in hotels_index:
hotels_index[hid] = hotel
hotels_seen.append(hotel)
hotel_dates[hid].append(rate_date)
if rate_date not in dates_seen:
dates_seen.add(rate_date)
unique_dates.append(rate_date)
unique_dates.sort()
# Shard hotels across workers
shards: List[List[Dict[str, Any]]] = [hotels_seen[i::concurrency] for i in range(concurrency)]
shards = [s for s in shards if s]
# 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(hotel_shard: List[Dict[str, Any]], widx: int) -> Dict[str, int]:
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 hotel in hotel_shard:
for rate_date in hotel_dates[hotel['id']]:
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: