forecasting/backend/jobs/resos_aggregation.py

"""
Resos Bookings Aggregation Job
Aggregates resos_bookings_data into:
- resos_bookings_stats: daily aggregated stats with period/source breakdowns
- resos_booking_pace: lead-time snapshots for pickup forecasting (3 types)
"""
import json
import logging
from datetime import date, datetime, timedelta
from typing import Set, Dict, Any, Optional, Tuple, List
from collections import defaultdict

from sqlalchemy import text
from database import SyncSessionLocal

logger = logging.getLogger(__name__)

# Valid booking statuses for aggregation
# Note: Resos uses 'approved' for confirmed future reservations, 'left' for completed meals
VALID_STATUSES = ('approved', 'arrived', 'seated', 'left')

# All tracked pace intervals (same as Newbook)
PACE_INTERVALS = [
    # Monthly (months 7-12)
    365, 330, 300, 270, 240, 210,
    # Weekly (weeks 5-25)
    177, 170, 163, 156, 149, 142, 135, 128, 121, 114,
    107, 100, 93, 86, 79, 72, 65, 58, 51, 44, 37,
    # Daily (days 0-30)
    30, 29, 28, 27, 26, 25, 24, 23, 22, 21,
    20, 19, 18, 17, 16, 15, 14, 13, 12, 11,
    10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0
]


def parse_group_exclude_field(group_exclude_field: Optional[str], primary_booking_number: Optional[str]) -> Tuple[List[str], List[str]]:
    """
    Parse group_exclude_field to extract linked bookings and exclude markers.
    Returns: (all_booking_numbers, exclude_numbers)
    """
    all_booking_numbers = []
    exclude_numbers = []

    # Always include primary booking number
    if primary_booking_number:
        all_booking_numbers.append(primary_booking_number)

    if not group_exclude_field:
        return all_booking_numbers, exclude_numbers

    # Parse comma-separated entries
    parts = group_exclude_field.split(',')
    for part in parts:
        part = part.strip()

        if part.upper().startswith('NOT-#'):
            # Exclude marker: NOT-#56748 → NB56748
            booking_num = part[5:]  # Remove "NOT-#"
            exclude_numbers.append(f"NB{booking_num}")

        elif part.startswith('#'):
            # Additional booking: #12346 → NB12346
            booking_num = part[1:]  # Remove "#"
            all_booking_numbers.append(f"NB{booking_num}")

    return all_booking_numbers, exclude_numbers


async def aggregate_resos_bookings(triggered_by: str = "manual"):
    """
    Aggregate Resos bookings into resos_bookings_stats.

    Flow:
    1. Find bookings changed since last_resos_aggregation_at
    2. Calculate affected dates
    3. Reaggregate each affected date
    4. Update booking pace table (3 types)
    5. Update last_resos_aggregation_at
    """
    logger.info(f"Starting Resos bookings aggregation (triggered_by={triggered_by})")

    db = SyncSessionLocal()

    try:
        # Get last aggregation timestamp
        result = db.execute(
            text("SELECT config_value FROM system_config WHERE config_key = 'last_resos_aggregation_at'")
        )
        row = result.fetchone()
        if row and row.config_value:
            try:
                last_ts = datetime.fromisoformat(row.config_value)
            except ValueError:
                last_ts = datetime.min
        else:
            last_ts = datetime.min

        logger.info(f"Last aggregation: {last_ts}")

        # Find bookings changed since last aggregation
        result = db.execute(
            text("""
                SELECT resos_id, booking_date
                FROM resos_bookings_data
                WHERE fetched_at > :last_ts
            """),
            {"last_ts": last_ts}
        )
        changed_bookings = result.fetchall()

        if not changed_bookings:
            logger.info("No changed bookings to aggregate")
            # Still update pace table
            await update_resos_booking_pace(db)
            db.commit()

            # Update timestamp
            db.execute(
                text("""
                    INSERT INTO system_config (config_key, config_value, updated_at)
                    VALUES ('last_resos_aggregation_at', :now, NOW())
                    ON CONFLICT (config_key) DO UPDATE SET
                        config_value = :now,
                        updated_at = NOW()
                """),
                {"now": datetime.now().isoformat()}
            )
            db.commit()
            return

        logger.info(f"Found {len(changed_bookings)} changed bookings")

        # Calculate affected dates
        affected_dates: Set[date] = set()
        for booking in changed_bookings:
            if booking.booking_date:
                affected_dates.add(booking.booking_date)

        logger.info(f"Reaggregating {len(affected_dates)} affected dates")

        # Aggregate each affected date
        for target_date in sorted(affected_dates):
            await aggregate_date(db, target_date)

        # Update booking pace table (3 types)
        await update_resos_booking_pace(db)

        # Update last aggregation timestamp
        db.execute(
            text("""
                INSERT INTO system_config (config_key, config_value, updated_at)
                VALUES ('last_resos_aggregation_at', :now, NOW())
                ON CONFLICT (config_key) DO UPDATE SET
                    config_value = :now,
                    updated_at = NOW()
            """),
            {"now": datetime.now().isoformat()}
        )

        db.commit()
        logger.info(f"Resos bookings aggregation completed: {len(affected_dates)} dates processed")

    except Exception as e:
        logger.error(f"Resos bookings aggregation failed: {e}", exc_info=True)
        db.rollback()
        raise
    finally:
        db.close()


async def aggregate_date(db, target_date: date):
    """
    Aggregate all Resos bookings for a specific date into resos_bookings_stats.
    """
    # Get all valid bookings for this date
    result = db.execute(
        text("""
            SELECT
                resos_id,
                period_type,
                covers,
                source,
                opening_hour_id,
                is_hotel_guest,
                is_dbb,
                is_package,
                total_guests,
                hotel_booking_number,
                group_exclude_field
            FROM resos_bookings_data
            WHERE booking_date = :target_date
            AND status IN :valid_statuses
        """),
        {"target_date": target_date, "valid_statuses": VALID_STATUSES}
    )
    bookings = result.fetchall()

    # Initialize counters
    breakfast_covers = 0
    lunch_covers = 0
    afternoon_covers = 0
    dinner_covers = 0
    other_covers = 0

    breakfast_bookings = 0
    lunch_bookings = 0
    afternoon_bookings = 0
    dinner_bookings = 0
    other_bookings = 0

    hotel_guest_covers = 0
    non_hotel_guest_covers = 0
    dbb_covers = 0
    package_covers = 0

    covers_by_source: Dict[str, int] = defaultdict(int)
    covers_by_period: Dict[str, Dict[str, Any]] = {}

    total_party_sizes = []
    party_sizes_by_period: Dict[str, list] = defaultdict(list)

    # Build hotel_booking_numbers mapping
    hotel_booking_numbers: Dict[str, str] = {}  # hotel_booking_number -> resos_id
    bookings_with_hotel_link = 0

    for booking in bookings:
        period = booking.period_type or 'other'
        covers = booking.covers or 0
        source = booking.source or 'unknown'
        opening_hour_id = booking.opening_hour_id
        resos_id = booking.resos_id

        # Count by period
        if period == 'breakfast':
            breakfast_covers += covers
            breakfast_bookings += 1
        elif period == 'lunch':
            lunch_covers += covers
            lunch_bookings += 1
        elif period == 'afternoon':
            afternoon_covers += covers
            afternoon_bookings += 1
        elif period == 'dinner':
            dinner_covers += covers
            dinner_bookings += 1
        else:
            other_covers += covers
            other_bookings += 1

        # Count by source
        covers_by_source[source] += covers

        # Count by period (detailed)
        if opening_hour_id:
            if opening_hour_id not in covers_by_period:
                covers_by_period[opening_hour_id] = {
                    "period_type": period,
                    "covers": 0,
                    "bookings": 0
                }
            covers_by_period[opening_hour_id]["covers"] += covers
            covers_by_period[opening_hour_id]["bookings"] += 1

        # Count business segments
        if booking.is_hotel_guest:
            hotel_guest_covers += covers
        else:
            non_hotel_guest_covers += covers

        if booking.is_dbb:
            dbb_covers += covers
        if booking.is_package:
            package_covers += covers

        # Track party sizes
        if covers > 0:
            total_party_sizes.append(covers)
            party_sizes_by_period[period].append(covers)

        # Build hotel booking numbers mapping
        all_booking_numbers, _ = parse_group_exclude_field(
            booking.group_exclude_field,
            booking.hotel_booking_number
        )

        if all_booking_numbers:
            bookings_with_hotel_link += 1
            for hotel_number in all_booking_numbers:
                hotel_booking_numbers[hotel_number] = resos_id

    # Calculate averages
    avg_party_size = sum(total_party_sizes) / len(total_party_sizes) if total_party_sizes else None

    avg_by_period = {}
    for period, sizes in party_sizes_by_period.items():
        avg_by_period[period] = sum(sizes) / len(sizes) if sizes else None

    total_covers = breakfast_covers + lunch_covers + afternoon_covers + dinner_covers + other_covers
    total_bookings = breakfast_bookings + lunch_bookings + afternoon_bookings + dinner_bookings + other_bookings
    distinct_hotel_bookings = len(hotel_booking_numbers)

    # Upsert stats
    db.execute(
        text("""
            INSERT INTO resos_bookings_stats (
                date,
                breakfast_covers, lunch_covers, afternoon_covers, dinner_covers, other_covers, total_covers,
                breakfast_bookings, lunch_bookings, afternoon_bookings, dinner_bookings, other_bookings, total_bookings,
                covers_by_source, covers_by_period,
                hotel_guest_covers, non_hotel_guest_covers, dbb_covers, package_covers,
                hotel_booking_numbers, distinct_hotel_bookings, bookings_with_hotel_link,
                avg_party_size, avg_party_size_by_period,
                aggregated_at
            ) VALUES (
                :date,
                :breakfast_covers, :lunch_covers, :afternoon_covers, :dinner_covers, :other_covers, :total_covers,
                :breakfast_bookings, :lunch_bookings, :afternoon_bookings, :dinner_bookings, :other_bookings, :total_bookings,
                :covers_by_source, :covers_by_period,
                :hotel_guest_covers, :non_hotel_guest_covers, :dbb_covers, :package_covers,
                :hotel_booking_numbers, :distinct_hotel_bookings, :bookings_with_hotel_link,
                :avg_party_size, :avg_party_size_by_period,
                NOW()
            )
            ON CONFLICT (date) DO UPDATE SET
                breakfast_covers = :breakfast_covers,
                lunch_covers = :lunch_covers,
                afternoon_covers = :afternoon_covers,
                dinner_covers = :dinner_covers,
                other_covers = :other_covers,
                total_covers = :total_covers,
                breakfast_bookings = :breakfast_bookings,
                lunch_bookings = :lunch_bookings,
                afternoon_bookings = :afternoon_bookings,
                dinner_bookings = :dinner_bookings,
                other_bookings = :other_bookings,
                total_bookings = :total_bookings,
                covers_by_source = :covers_by_source,
                covers_by_period = :covers_by_period,
                hotel_guest_covers = :hotel_guest_covers,
                non_hotel_guest_covers = :non_hotel_guest_covers,
                dbb_covers = :dbb_covers,
                package_covers = :package_covers,
                hotel_booking_numbers = :hotel_booking_numbers,
                distinct_hotel_bookings = :distinct_hotel_bookings,
                bookings_with_hotel_link = :bookings_with_hotel_link,
                avg_party_size = :avg_party_size,
                avg_party_size_by_period = :avg_party_size_by_period,
                aggregated_at = NOW()
        """),
        {
            "date": target_date,
            "breakfast_covers": breakfast_covers,
            "lunch_covers": lunch_covers,
            "afternoon_covers": afternoon_covers,
            "dinner_covers": dinner_covers,
            "other_covers": other_covers,
            "total_covers": total_covers,
            "breakfast_bookings": breakfast_bookings,
            "lunch_bookings": lunch_bookings,
            "afternoon_bookings": afternoon_bookings,
            "dinner_bookings": dinner_bookings,
            "other_bookings": other_bookings,
            "total_bookings": total_bookings,
            "covers_by_source": json.dumps(dict(covers_by_source)),
            "covers_by_period": json.dumps(covers_by_period),
            "hotel_guest_covers": hotel_guest_covers,
            "non_hotel_guest_covers": non_hotel_guest_covers,
            "dbb_covers": dbb_covers,
            "package_covers": package_covers,
            "hotel_booking_numbers": json.dumps(hotel_booking_numbers),
            "distinct_hotel_bookings": distinct_hotel_bookings,
            "bookings_with_hotel_link": bookings_with_hotel_link,
            "avg_party_size": avg_party_size,
            "avg_party_size_by_period": json.dumps(avg_by_period)
        }
    )
    db.commit()


async def update_resos_booking_pace(db):
    """
    Update resos_booking_pace table with lead-time snapshots.
    Creates 3 rows per date: total, resident, non_resident
    """
    logger.info("Updating Resos booking pace table...")

    today = date.today()

    # Process dates from -30 to +365 (historical + forecast window)
    from_date = today - timedelta(days=30)
    to_date = today + timedelta(days=365)

    current = from_date
    while current <= to_date:
        # Calculate pace for each type
        for pace_type in ['total', 'resident', 'non_resident']:
            pace_values = {}

            for days_out in PACE_INTERVALS:
                snapshot_date = current - timedelta(days=days_out)

                # Build query based on pace_type
                if pace_type == 'total':
                    # All valid bookings
                    result = db.execute(
                        text("""
                            SELECT COALESCE(SUM(covers), 0) as total_covers
                            FROM resos_bookings_data
                            WHERE booking_date = :target_date
                            AND status IN :valid_statuses
                            AND booking_placed <= :snapshot_date
                        """),
                        {
                            "target_date": current,
                            "valid_statuses": VALID_STATUSES,
                            "snapshot_date": snapshot_date
                        }
                    )
                elif pace_type == 'resident':
                    # Hotel guests only
                    result = db.execute(
                        text("""
                            SELECT COALESCE(SUM(covers), 0) as total_covers
                            FROM resos_bookings_data
                            WHERE booking_date = :target_date
                            AND status IN :valid_statuses
                            AND booking_placed <= :snapshot_date
                            AND is_hotel_guest = true
                        """),
                        {
                            "target_date": current,
                            "valid_statuses": VALID_STATUSES,
                            "snapshot_date": snapshot_date
                        }
                    )
                else:  # non_resident
                    # Non-hotel guests
                    result = db.execute(
                        text("""
                            SELECT COALESCE(SUM(covers), 0) as total_covers
                            FROM resos_bookings_data
                            WHERE booking_date = :target_date
                            AND status IN :valid_statuses
                            AND booking_placed <= :snapshot_date
                            AND (is_hotel_guest = false OR is_hotel_guest IS NULL)
                        """),
                        {
                            "target_date": current,
                            "valid_statuses": VALID_STATUSES,
                            "snapshot_date": snapshot_date
                        }
                    )

                row = result.fetchone()
                pace_values[f"d{days_out}"] = row.total_covers if row else 0

            # Upsert pace record
            columns = ", ".join(pace_values.keys())
            placeholders = ", ".join([f":{k}" for k in pace_values.keys()])
            updates = ", ".join([f"{k} = :{k}" for k in pace_values.keys()])

            db.execute(
                text(f"""
                    INSERT INTO resos_booking_pace (booking_date, pace_type, {columns}, updated_at)
                    VALUES (:booking_date, :pace_type, {placeholders}, NOW())
                    ON CONFLICT (booking_date, pace_type) DO UPDATE SET
                        {updates},
                        updated_at = NOW()
                """),
                {"booking_date": current, "pace_type": pace_type, **pace_values}
            )

        current += timedelta(days=1)

    db.commit()
    logger.info("Resos booking pace table updated (3 types: total, resident, non_resident)")