rates/backend/scheduler.py

"""
APScheduler configuration for rate monitor jobs
"""
import logging
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy import text

from database import SyncSessionLocal

logger = logging.getLogger(__name__)

scheduler = AsyncIOScheduler(
    job_defaults={
        'misfire_grace_time': 3600,
        'coalesce': True,
    }
)


def get_config_value(key: str, default: str = None) -> str:
    db = SyncSessionLocal()
    try:
        result = db.execute(
            text("SELECT config_value FROM system_config WHERE config_key = :key"),
            {"key": key}
        )
        row = result.fetchone()
        if row and row.config_value:
            return row.config_value
        return default
    except Exception as e:
        logger.error(f"Error getting config {key}: {e}")
        return default
    finally:
        db.close()


def is_sync_enabled(source: str) -> bool:
    value = get_config_value(f"sync_{source}_enabled")
    if value:
        return value.lower() in ('true', '1', 'yes', 'enabled')
    return False


def get_sync_time(source: str, default_hour: int = 5, default_minute: int = 0) -> tuple:
    time_str = get_config_value(f"sync_{source}_time")
    if time_str:
        try:
            parts = time_str.split(':')
            return (int(parts[0]), int(parts[1]))
        except (ValueError, IndexError):
            pass
    return (default_hour, default_minute)


async def run_scheduled_direct_scrape():
    from jobs.scrape_direct_rates import run_scrape_all_direct
    import asyncio
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, run_scrape_all_direct)


async def run_scheduled_parity_check():
    from jobs.check_rate_parity import run_parity_check
    import asyncio
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, run_parity_check)


async def run_scrape_watchdog():
    """Auto-release the scrape lock if held for >3 hours (hung Playwright browser)."""
    from services.booking_scraper import get_lock_status, force_reset_scraper
    status = get_lock_status()
    held = status.get("held_seconds")
    if held and held > 3 * 3600:
        logger.warning(f"Scrape watchdog: lock held for {held}s — force releasing")
        db = SyncSessionLocal()
        try:
            force_reset_scraper(db)
        except Exception as e:
            logger.error(f"Scrape watchdog reset failed: {e}")
        finally:
            db.close()


async def run_scheduled_booking_scrape_async():
    from jobs.scrape_booking_rates import run_scheduled_booking_scrape
    import asyncio
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, run_scheduled_booking_scrape)


async def run_scheduled_fetch_current_rates():
    if is_sync_enabled("newbook_current_rates"):
        # run_fetch_current_rates is a coroutine — await it directly;
        # run_in_executor would return the coroutine object unawaited
        from jobs.fetch_current_rates import run_fetch_current_rates
        from jobs.sync_occupancy import run_sync_occupancy
        try:
            await run_sync_occupancy()
        except Exception as e:
            logger.warning(f"Occupancy sync failed (continuing with rates): {e}")
        await run_fetch_current_rates()
    else:
        logger.debug("Newbook current rates sync skipped (disabled)")


def start_scheduler():
    # Booking.com scrape — daily at configurable time (default 05:30)
    scrape_time = get_config_value('booking_scraper_daily_time', '05:30')
    try:
        h, m = scrape_time.split(':')
        scrape_hour, scrape_minute = int(h), int(m)
    except Exception:
        scrape_hour, scrape_minute = 5, 30

    scheduler.add_job(
        run_scheduled_booking_scrape_async,
        CronTrigger(hour=scrape_hour, minute=scrape_minute),
        id='booking_scrape',
        replace_existing=True,
    )

    # Newbook current rates fetch — daily at 05:20
    rates_hour, rates_minute = get_sync_time('newbook_current_rates', 5, 20)
    scheduler.add_job(
        run_scheduled_fetch_current_rates,
        CronTrigger(hour=rates_hour, minute=rates_minute),
        id='fetch_current_rates',
        replace_existing=True,
    )

    # Direct booking engine scrape — daily at 06:00
    scheduler.add_job(
        run_scheduled_direct_scrape,
        CronTrigger(hour=6, minute=0),
        id='scrape_direct_rates',
        replace_existing=True,
    )

    # Rate parity check — daily at 06:45, after Newbook fetch + Booking scrape
    scheduler.add_job(
        run_scheduled_parity_check,
        CronTrigger(hour=6, minute=45),
        id='parity_check',
        replace_existing=True,
    )

    # Scrape lock watchdog — every 30 min; force-releases if held >3 hours
    from apscheduler.triggers.interval import IntervalTrigger
    scheduler.add_job(
        run_scrape_watchdog,
        IntervalTrigger(minutes=30),
        id='scrape_watchdog',
        replace_existing=True,
    )

    scheduler.start()
    logger.info(f"Scheduler started: booking scrape at {scrape_hour:02d}:{scrape_minute:02d}, rates fetch at {rates_hour:02d}:{rates_minute:02d}, direct scrape at 06:00, parity check at 06:45, watchdog every 30m")


def shutdown_scheduler():
    if scheduler.running:
        scheduler.shutdown()