"""
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::text
                  AND rate_gross IS NOT NULL
            ),
            rates_scraped = (
                SELECT COUNT(*)
                FROM booking_com_rates
                WHERE scrape_batch_id = bsl.batch_id::text
                  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::text
                      AND rate_gross IS NOT NULL
                ),
                rates_scraped = (
                    SELECT COUNT(*)
                    FROM booking_com_rates
                    WHERE scrape_batch_id = bsl.batch_id::text
                      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
                    elif result['success']:
                        acc['rates'] += result['rates_count']
                        acc['completed'] += 1
                    else:
                        acc['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 _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 _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),
        }