""" Scheduled Booking.com Rate Scraping Job Tiered scheduling for 365-day coverage: - High (priority 10): next 30 days — all 31 dates scraped every day - Medium (priority 5): days 31-180 — up to 60 oldest-scraped dates per day - Low (priority 2): days 181-365 — up to 30 oldest-scraped dates per day Per-run budget: ~121 dates (30 high + 60 medium + 30 low). Medium and low tiers select whichever dates in their window have the oldest (or missing) scrape data first, naturally distributing coverage across the full year without hammering all 365 dates every day. Approximate refresh cadence: - High: daily - Medium (150 dates / 60 per day): every ~2-3 days - Low (185 dates / 30 per day): every ~6 days Uses a queue-based approach: 1. Populate the queue with the selected dates and priorities 2. Process the queue in priority order 3. Failed dates are retried (up to 3 attempts) 4. On blocking, the queue pauses and resumes after cooldown Schedule: Daily at configurable time (default 05:30) """ import asyncio import logging from datetime import date, timedelta from sqlalchemy import text from sqlalchemy.orm import Session from database import SyncSessionLocal from services.booking_scraper import ( populate_queue, process_queue, clear_old_queue_items, cleanup_stale_batches, get_scrape_config, ) logger = logging.getLogger(__name__) # Priority levels (higher = processed first) PRIORITY_HIGH = 10 # 0-30 days PRIORITY_MEDIUM = 5 # 31-180 days PRIORITY_LOW = 2 # 181-365 days # Per-run date budget for medium and low tiers MEDIUM_DATES_PER_RUN = 60 # out of ~150 in window LOW_DATES_PER_RUN = 30 # out of ~185 in window def get_high_priority_dates() -> list[date]: """High priority: all of today through +30 days, scraped every run.""" today = date.today() return [today + timedelta(days=i) for i in range(31)] def _get_oldest_scraped_dates(db: Session, start: date, end: date, limit: int) -> list[date]: """ From the date window [start, end], return up to `limit` dates ordered by oldest last-scrape first (never-scraped dates come first via NULLS FIRST). Joining against booking_com_rates means we naturally pick whichever dates in the window are most stale or have no data yet. """ rows = db.execute( text(""" SELECT s.rate_date::date AS rate_date, MAX(r.scraped_at) AS last_scraped FROM generate_series(CAST(:start AS date), CAST(:end AS date), INTERVAL '1 day') AS s(rate_date) LEFT JOIN booking_com_rates r ON r.rate_date = s.rate_date::date GROUP BY s.rate_date ORDER BY last_scraped ASC NULLS FIRST LIMIT :limit """), {'start': start, 'end': end, 'limit': limit} ).fetchall() return [row.rate_date for row in rows] def get_medium_priority_dates(db: Session, limit: int = MEDIUM_DATES_PER_RUN) -> list[date]: """Medium priority: up to `limit` oldest-scraped dates in the days 31-180 window.""" today = date.today() return _get_oldest_scraped_dates( db, start=today + timedelta(days=31), end=today + timedelta(days=180), limit=limit, ) def get_low_priority_dates(db: Session, limit: int = LOW_DATES_PER_RUN) -> list[date]: """Low priority: up to `limit` oldest-scraped dates in the days 181-365 window.""" today = date.today() return _get_oldest_scraped_dates( db, start=today + timedelta(days=181), end=today + timedelta(days=365), limit=limit, ) def compute_next_scrape_for_date(target_date: date) -> tuple[str, date | None]: """ For a target date, determine its priority tier and approximate next scrape date. Returns (tier, next_scrape_date) where tier is 'high'/'medium'/'low'/'none'. High dates are scraped every run; medium/low estimates reflect typical cadence based on the per-run budget vs window size. """ today = date.today() offset = (target_date - today).days if offset < 0: return ('none', None) if offset > 365: return ('none', None) if offset <= 30: # Scraped every day return ('high', today) elif offset <= 180: # ~150 dates, 60/day → ~2-3 day cadence return ('medium', today + timedelta(days=3)) else: # ~185 dates, 30/day → ~6-7 day cadence return ('low', today + timedelta(days=7)) def run_scheduled_booking_scrape(): """ Main scheduled job: populate queue with today's dates, then process. """ db = SyncSessionLocal() try: # Check if scraper is enabled result = db.execute( text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_enabled'") ).fetchone() if not result or result.config_value != 'true': logger.debug("Scheduled booking scrape skipped (disabled)") return if not get_scrape_config(db): logger.warning("Scheduled booking scrape skipped (no location configured)") return # Clean up stale running batches and old queue items cleanup_stale_batches(db, max_age_minutes=120) clear_old_queue_items(db, days=3) # Gather dates with priorities — medium/low select oldest-scraped first high = get_high_priority_dates() medium = get_medium_priority_dates(db) low = get_low_priority_dates(db) priorities = {} for d in high: priorities[d] = PRIORITY_HIGH for d in medium: priorities[d] = max(priorities.get(d, 0), PRIORITY_MEDIUM) for d in low: priorities[d] = max(priorities.get(d, 0), PRIORITY_LOW) all_dates = sorted(priorities.keys()) if not all_dates: logger.info("Scheduled booking scrape: no dates to scrape today") return logger.info( f"Scheduled booking scrape: queuing {len(all_dates)} dates " f"(high={len(high)}, medium={len(medium)}, low={len(low)})" ) # Populate queue populate_queue(db, all_dates, priorities) # Process queue loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(process_queue(db)) if result.get('success'): logger.info( f"Scheduled booking scrape completed: " f"{result.get('dates_completed', 0)} dates, " f"{result.get('rates_scraped', 0)} rates" ) elif result.get('blocked'): logger.warning( f"Scheduled booking scrape blocked: {result.get('block_reason')}. " f"Completed {result.get('dates_completed', 0)} dates. " f"Remaining dates stay queued for retry." ) else: logger.error(f"Scheduled booking scrape failed: {result.get('error')}") finally: loop.close() except Exception as e: logger.error(f"Scheduled booking scrape error: {e}", exc_info=True) finally: db.close() async def run_scheduled_booking_scrape_async(): """Async wrapper for APScheduler.""" loop = asyncio.get_event_loop() await loop.run_in_executor(None, run_scheduled_booking_scrape)