wyndham-ARR/monthly_reports/core.py

"""Pure deterministic rules for monthly channel report payloads."""

from __future__ import annotations

import calendar
import re
from collections import defaultdict
from datetime import date
from decimal import Decimal
from typing import Dict, Iterable, List, Sequence, Tuple

from monthly_reports.contracts import (
    KB_SHEET,
    STANDARD_SHEETS,
    ChannelObservation,
    ChannelSheet,
    DailyVersionPin,
    ErrorCode,
    MonthlyFact,
    MonthlyReport,
    MonthlyReportError,
    MonthlySnapshot,
)


INVALID_SHEET_CHARS = re.compile(r"[:\\/?*\[\]]")


def month_bounds(year: int, month: int) -> Tuple[date, date]:
    if year < 1900 or year > 9999 or month < 1 or month > 12:
        raise MonthlyReportError(ErrorCode.REQUEST_INVALID, "report month is invalid")
    return date(year, month, 1), date(year, month, calendar.monthrange(year, month)[1])


def validate_as_of(year: int, month: int, as_of_date: date) -> None:
    start, end = month_bounds(year, month)
    if as_of_date < start or as_of_date > end:
        raise MonthlyReportError(
            ErrorCode.REQUEST_INVALID,
            "as-of date must fall inside the report month",
        )


def _valid_sheet_name(value: str) -> bool:
    return (
        bool(value)
        and value == value.strip()
        and len(value) <= 31
        and INVALID_SHEET_CHARS.search(value) is None
    )


def _append_unique(target: List[str], seen: set[str], values: Iterable[str]) -> None:
    for value in values:
        if not _valid_sheet_name(value):
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "source channel worksheet name is invalid",
            )
        if value not in seen:
            target.append(value)
            seen.add(value)


def channel_order(snapshot: MonthlySnapshot) -> Tuple[str, ...]:
    ordered: List[str] = []
    seen: set[str] = set()
    _append_unique(ordered, seen, STANDARD_SHEETS)
    _append_unique(
        ordered,
        seen,
        (name for name in snapshot.preferred_channel_order if name not in STANDARD_SHEETS),
    )
    observations: Sequence[ChannelObservation] = sorted(
        snapshot.channel_observations,
        key=lambda item: (
            item.business_date,
            item.worksheet_order is None,
            item.worksheet_order if item.worksheet_order is not None else 2**31,
            item.worksheet,
        ),
    )
    _append_unique(ordered, seen, (item.worksheet for item in observations))
    _append_unique(
        ordered,
        seen,
        sorted({fact.channel_key for fact in snapshot.facts if fact.channel_key not in seen}),
    )
    return tuple(ordered)


def _require_nonnegative(value: int, label: str) -> None:
    if not isinstance(value, int) or isinstance(value, bool) or value < 0:
        raise MonthlyReportError(ErrorCode.SOURCE_INVALID, f"{label} is invalid")


def _validate_fact(fact: MonthlyFact, start: date, as_of_date: date) -> None:
    if (
        fact.daily_record_id <= 0
        or fact.daily_version_id <= 0
        or fact.business_date < start
        or fact.business_date > as_of_date
        or fact.arrival != fact.business_date
        or fact.departure < fact.arrival
        or fact.nights != (fact.departure - fact.arrival).days
        or not _valid_sheet_name(fact.channel_key)
        or fact.no_of_rooms <= 0
    ):
        raise MonthlyReportError(
            ErrorCode.SOURCE_INVALID,
            "monthly source fact shape is invalid",
        )
    _require_nonnegative(fact.adults, "adults")
    _require_nonnegative(fact.children, "children")
    for text in (
        fact.company_name,
        fact.confirmation_no,
        fact.disp_room_no,
        fact.full_name,
        fact.rate_code,
    ):
        if not str(text).strip():
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly source text is missing",
            )
    for amount in (
        fact.effective_rate_amount,
        fact.real_price,
        fact.total_price,
    ):
        if not isinstance(amount, Decimal) or not amount.is_finite() or amount < 0:
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly source amount is invalid",
            )
    if fact.total_price != fact.real_price * fact.no_of_rooms * fact.nights:
        raise MonthlyReportError(
            ErrorCode.SOURCE_INVALID,
            "monthly source total does not match static fact inputs",
        )
    if fact.channel_key == KB_SHEET:
        if fact.kb_amount != Decimal(fact.no_of_rooms * 100):
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly KB source amount is invalid",
            )
    elif fact.kb_amount is not None:
        raise MonthlyReportError(
            ErrorCode.SOURCE_INVALID,
            "non-KB channel unexpectedly contains a KB amount",
        )


def _validate_pins(
    facts: Sequence[MonthlyFact],
    pins: Sequence[DailyVersionPin],
    start: date,
    as_of_date: date,
) -> Tuple[DailyVersionPin, ...]:
    by_date: Dict[date, int] = {}
    for pin in pins:
        if (
            pin.business_date < start
            or pin.business_date > as_of_date
            or pin.daily_version_id <= 0
            or (
                pin.business_date in by_date
                and by_date[pin.business_date] != pin.daily_version_id
            )
        ):
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly daily-version manifest is invalid",
            )
        by_date[pin.business_date] = pin.daily_version_id
    for fact in facts:
        if by_date.get(fact.business_date) != fact.daily_version_id:
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly fact is not covered by the current daily manifest",
            )
    return tuple(
        DailyVersionPin(business_date=current, daily_version_id=version_id)
        for current, version_id in sorted(by_date.items())
    )


def _validate_observations(
    facts: Sequence[MonthlyFact],
    observations: Sequence[ChannelObservation],
    pins: Sequence[DailyVersionPin],
) -> None:
    pin_by_date = {pin.business_date: pin.daily_version_id for pin in pins}
    observed: Dict[Tuple[int, str], int] = {}
    orders: set[Tuple[int, int]] = set()
    for item in observations:
        if (
            pin_by_date.get(item.business_date) != item.daily_version_id
            or not _valid_sheet_name(item.worksheet)
            or item.row_count < 0
            or (item.worksheet_order is not None and item.worksheet_order <= 0)
        ):
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly channel observation is invalid",
            )
        key = (item.daily_version_id, item.worksheet)
        if key in observed:
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly channel observation is duplicated",
            )
        observed[key] = item.row_count
        if item.worksheet_order is not None:
            order_key = (item.daily_version_id, item.worksheet_order)
            if order_key in orders:
                raise MonthlyReportError(
                    ErrorCode.SOURCE_INVALID,
                    "monthly channel observation order is duplicated",
                )
            orders.add(order_key)

    actual: Dict[Tuple[int, str], int] = defaultdict(int)
    for fact in facts:
        actual[(fact.daily_version_id, fact.channel_key)] += 1
    for key, row_count in actual.items():
        if observed.get(key) != row_count:
            raise MonthlyReportError(
                ErrorCode.SOURCE_INVALID,
                "monthly channel observation does not match retained facts",
            )


def build_monthly_report(
    year: int,
    month: int,
    as_of_date: date,
    snapshot: MonthlySnapshot,
) -> MonthlyReport:
    validate_as_of(year, month, as_of_date)
    start, _end = month_bounds(year, month)
    facts = tuple(sorted(snapshot.facts, key=lambda item: item.daily_record_id))
    if len({fact.daily_record_id for fact in facts}) != len(facts):
        raise MonthlyReportError(ErrorCode.SOURCE_INVALID, "monthly source rows are duplicated")
    for fact in facts:
        _validate_fact(fact, start, as_of_date)
    if facts and max(fact.arrival for fact in facts) != as_of_date:
        raise MonthlyReportError(
            ErrorCode.SOURCE_INVALID,
            "monthly as-of date must equal the greatest included ARRIVAL",
        )
    pins = _validate_pins(facts, snapshot.daily_versions, start, as_of_date)
    _validate_observations(facts, snapshot.channel_observations, pins)
    ordered_channels = channel_order(snapshot)
    grouped: Dict[str, List[MonthlyFact]] = defaultdict(list)
    for fact in facts:
        grouped[fact.channel_key].append(fact)
    channels = tuple(
        ChannelSheet(
            worksheet=name,
            worksheet_order=index,
            rows=tuple(
                sorted(grouped.get(name, []), key=lambda item: (item.arrival, item.daily_record_id))
            ),
        )
        for index, name in enumerate(ordered_channels, 1)
    )
    if sum(len(channel.rows) for channel in channels) != len(facts):
        raise MonthlyReportError(
            ErrorCode.SOURCE_INVALID,
            "monthly source channel coverage is incomplete",
        )
    filename = (
        f"各渠道情况-{year:04d}年{month:02d}月-"
        f"更新至{as_of_date.month}.{as_of_date.day}.xlsx"
    )
    return MonthlyReport(
        report_year=year,
        report_month=month,
        as_of_date=as_of_date,
        filename=filename,
        channels=channels,
        daily_versions=pins,
    )