"""Reconstruct each campaign's budget state timeline from change events.

The export lists only *changes*, so the state between two rows has to be
inferred. Three things make that non-obvious:

1. Rows are written newest-first, so rows sharing a minute are also
   newest-first and must be reversed before walking the machine. Sorting on
   timestamp alone silently preserves the wrong intra-minute order -- on the
   reference file that costs 64 campaign-hours and manufactures 14 phantom
   inconsistencies.
2. Delivery state (Paused) overlays budget state. A paused campaign forgoes
   nothing to its budget, so paused minutes are excluded from the loss-eligible
   total and from the in-budget denominator that sets the spend rate.
3. De-duplication during export can drop an intermediate transition, leaving a
   row whose `From` disagrees with the running state. We trust the row over the
   inference and place the implied transition at the midpoint of the gap.
"""

from __future__ import annotations

from dataclasses import dataclass, field

from .ingest import BUDGET_STATES, Event, parse_money

IN, OOB, PAUSED, NA = 0, 1, 2, 3
STATE_NAMES = {IN: "in", OOB: "out_of_budget", PAUSED: "paused", NA: "not_eligible"}

MINUTES_PER_DAY = 1440
DEFAULT_MERGE_GAP_MIN = 5


@dataclass(slots=True)
class ChainBreak:
    at_min: int
    expected_from: str
    saw_from: str
    ambiguity_min: int


@dataclass(slots=True)
class Episode:
    index: int
    start_min: int
    end_min: int
    raw_min: int
    active_min: int  # raw minus any overlapping pause


@dataclass(slots=True)
class BudgetInfo:
    value: float | None = None
    source: str = "unknown"  # daily_budget_event | budget_rule | perf_report | unknown
    time_weighted: float | None = None
    changes: int = 0


@dataclass(slots=True)
class CampaignDay:
    campaign: str
    date_key: str
    t0: int
    t1: int

    in_min: int = 0
    oob_min: int = 0  # loss-eligible: pause and out-of-window already removed
    paused_min: int = 0
    na_min: int = 0
    oob_min_raw: int = 0  # what the budget machine alone says
    post_reset_oob_min: int = 0

    episodes: list[Episode] = field(default_factory=list)
    episodes_raw: int = 0
    episodes_merged: int = 0

    first_oob_min: int | None = None
    last_recovery_min: int | None = None
    opened_oob: bool = False
    closed_oob: bool = False

    hourly_oob: list[int] = field(default_factory=lambda: [0] * 24)
    hourly_paused: list[int] = field(default_factory=lambda: [0] * 24)
    hourly_na: list[int] = field(default_factory=lambda: [0] * 24)
    track: list[tuple[int, int, int]] = field(default_factory=list)  # (state, start, end)

    budget: BudgetInfo = field(default_factory=BudgetInfo)
    chain_breaks: list[ChainBreak] = field(default_factory=list)
    oob_uncertainty_min: float = 0.0
    confidence: str = "clean"  # clean | repaired | partial_day

    # filled in by metrics.py / perfjoin.py
    severity: float = 0.0
    diagnosis: str = ""
    lost: dict | None = None
    perf: object | None = None

    @property
    def eligible_min(self) -> int:
        return self.t1 - self.t0

    @property
    def active_min(self) -> int:
        """Eligible minutes where the campaign could actually have spent."""
        return self.eligible_min - self.paused_min

    @property
    def oob_share(self) -> float:
        return self.oob_min / self.active_min if self.active_min > 0 else 0.0

    @property
    def in_hours(self) -> float:
        return self.in_min / 60

    @property
    def oob_hours(self) -> float:
        return self.oob_min / 60

    @property
    def paused_hours(self) -> float:
        return self.paused_min / 60


def sort_key(e: Event) -> tuple[int, int]:
    """Chronological order. Descending source index reverses the newest-first file."""
    return (e.minute, -e.source_index)


def _walk(events: list[Event], t0: int, t1: int, initial: str):
    """Turn a two-state event stream into contiguous spans over [t0, t1)."""
    spans: list[tuple[str, int, int]] = []
    breaks: list[ChainBreak] = []
    state, prev = initial, t0

    for e in events:
        m = e.minute if e.minute > prev else prev
        if e.from_val != state:
            # A transition went missing. The row is observation, the running
            # state is inference, so trust the row and split the difference.
            mid = (prev + m) // 2
            if mid > prev:
                spans.append((state, prev, mid))
            if m > mid:
                spans.append((e.from_val, mid, m))
            breaks.append(ChainBreak(m, state, e.from_val, m - prev))
        elif m > prev:
            spans.append((state, prev, m))
        state, prev = e.to_val, m

    if t1 > prev:
        spans.append((state, prev, t1))
    return spans, breaks


def _budget_timeline(events: list[Event], rule_events: list[Event],
                     t0: int, t1: int) -> BudgetInfo:
    """Recover the daily budget, which can change during the day."""
    amounts = sorted(events, key=sort_key)
    if amounts:
        opening = amounts[0].from_num
        spans: list[tuple[float | None, int, int]] = []
        value, prev = opening, t0
        for e in amounts:
            m = max(e.minute, prev)
            if m > prev:
                spans.append((value, prev, m))
            value, prev = e.to_num, m
        if t1 > prev:
            spans.append((value, prev, t1))

        weighted = sum(v * (b - a) for v, a, b in spans if v is not None)
        covered = sum(b - a for v, a, b in spans if v is not None)
        return BudgetInfo(
            value=amounts[-1].to_num,
            source="daily_budget_event",
            time_weighted=(weighted / covered) if covered else None,
            changes=len(amounts),
        )

    for e in rule_events:
        amount = parse_money(e.from_val, e.to_val)
        if amount is not None:
            return BudgetInfo(value=amount, source="budget_rule", time_weighted=amount)

    return BudgetInfo()


def score_campaign_day(campaign: str, date_key: str, events: list[Event],
                       day_end_min: int = MINUTES_PER_DAY,
                       merge_gap_min: int = DEFAULT_MERGE_GAP_MIN) -> CampaignDay | None:
    """Score one campaign for one day. Returns None if budget state is unknowable."""
    budget_events = sorted((e for e in events if e.machine == "budget"), key=sort_key)
    if not budget_events:
        return None  # never impute a state we did not observe

    delivery_events = sorted((e for e in events if e.machine == "delivery"), key=sort_key)
    created = [e for e in events if e.machine == "created"]

    # Eligible window. A campaign created at 07:02 is scored over the remaining
    # 16.97h, not a full day -- but only if no budget event precedes creation.
    t1 = min(day_end_min, MINUTES_PER_DAY)
    t0 = 0
    if created:
        birth = min(e.minute for e in created)
        if birth <= budget_events[0].minute and birth < t1:
            t0 = birth
    budget_events = [e for e in budget_events if e.minute >= t0]
    if not budget_events:
        return None
    delivery_events = [e for e in delivery_events if e.minute >= t0]

    day = CampaignDay(campaign=campaign, date_key=date_key, t0=t0, t1=t1)

    budget_spans, breaks = _walk(budget_events, t0, t1, budget_events[0].from_val)
    day.chain_breaks = breaks
    day.oob_uncertainty_min = sum(b.ambiguity_min for b in breaks) / 2

    delivery_initial = delivery_events[0].from_val if delivery_events else "Delivering"
    delivery_spans, _ = _walk(delivery_events, t0, t1, delivery_initial)

    # Flatten: not-eligible > paused > out of budget > in budget.
    flat = bytearray([NA]) * MINUTES_PER_DAY
    for state, a, b in budget_spans:
        flat[a:b] = bytes([OOB if state == "Out of budget" else IN]) * (b - a)
    for state, a, b in delivery_spans:
        if state == "Paused":
            flat[a:b] = bytes([PAUSED]) * (b - a)

    day.in_min = flat.count(IN)
    day.oob_min = flat.count(OOB)
    day.paused_min = flat.count(PAUSED)
    day.na_min = flat.count(NA)
    day.post_reset_oob_min = flat[60:t1].count(OOB) if t1 > 60 else 0
    day.hourly_oob = [flat[h * 60:(h + 1) * 60].count(OOB) for h in range(24)]
    day.hourly_paused = [flat[h * 60:(h + 1) * 60].count(PAUSED) for h in range(24)]
    day.hourly_na = [flat[h * 60:(h + 1) * 60].count(NA) for h in range(24)]

    oob_spans = [(a, b) for s, a, b in budget_spans if s == "Out of budget"]
    day.oob_min_raw = sum(b - a for a, b in oob_spans)
    day.episodes_raw = len(oob_spans)

    # Amazon can release a sliver of budget that is consumed within the same
    # minute, producing zero-length recoveries. Those are pacing noise, not
    # genuine outages, so also report a merged count.
    merged: list[tuple[int, int]] = []
    for a, b in oob_spans:
        if merged and a - merged[-1][1] < merge_gap_min:
            merged[-1] = (merged[-1][0], b)
        else:
            merged.append((a, b))
    day.episodes_merged = len(merged)
    day.episodes = [
        Episode(i + 1, a, b, b - a, flat[a:b].count(OOB))
        for i, (a, b) in enumerate(merged)
    ]

    # State facts come from the budget machine; a concurrent pause must not
    # mask the fact that the budget itself was exhausted. Loss math, above,
    # uses the flattened track instead.
    day.opened_oob = budget_spans[0][0] == "Out of budget"
    day.closed_oob = budget_spans[-1][0] == "Out of budget"
    day.first_oob_min = oob_spans[0][0] if oob_spans else None
    day.last_recovery_min = None if day.closed_oob or not oob_spans else oob_spans[-1][1]

    day.budget = _budget_timeline(
        [e for e in events if e.machine == "budget_amount"],
        [e for e in events if e.machine == "budget_rule"],
        t0, t1,
    )

    # Run-length encode for the report; spans are contiguous and tile the day.
    track: list[tuple[int, int, int]] = []
    run_state, run_start = flat[0], 0
    for m in range(1, MINUTES_PER_DAY):
        if flat[m] != run_state:
            track.append((run_state, run_start, m))
            run_state, run_start = flat[m], m
    track.append((run_state, run_start, MINUTES_PER_DAY))
    day.track = track

    if breaks:
        day.confidence = "repaired"
    if day.na_min > 0:
        day.confidence = "partial_day"
    return day


def score_all(events: list[Event], merge_gap_min: int = DEFAULT_MERGE_GAP_MIN) -> list[CampaignDay]:
    """Score every campaign in every day present in the event stream."""
    by_day: dict[str, dict[str, list[Event]]] = {}
    for e in events:
        by_day.setdefault(e.date_key, {}).setdefault(e.campaign, []).append(e)

    results: list[CampaignDay] = []
    for date_key in sorted(by_day):
        campaigns = by_day[date_key]
        # If the export was cut short (a "Today" pull), do not score the
        # unobserved remainder of the day as if it were in budget.
        last_seen = max(e.minute for evs in campaigns.values() for e in evs)
        day_end = MINUTES_PER_DAY if last_seen >= MINUTES_PER_DAY - 1 else last_seen + 1
        for campaign, evs in campaigns.items():
            day = score_campaign_day(campaign, date_key, evs, day_end, merge_gap_min)
            if day is not None:
                results.append(day)
    return results


def check_invariants(days: list[CampaignDay]) -> list[str]:
    """Structural checks that make chart/table disagreement impossible."""
    problems: list[str] = []
    for d in days:
        total = d.in_min + d.oob_min + d.paused_min + d.na_min
        if total != MINUTES_PER_DAY:
            problems.append(f"{d.campaign} [{d.date_key}]: minutes sum to {total}, not 1440")
        if sum(e.active_min for e in d.episodes) != d.oob_min:
            problems.append(f"{d.campaign} [{d.date_key}]: episode minutes != out-of-budget minutes")
        if sum(d.hourly_oob) != d.oob_min:
            problems.append(f"{d.campaign} [{d.date_key}]: hourly buckets != out-of-budget minutes")
        for i in range(1, len(d.track)):
            if d.track[i][1] != d.track[i - 1][2]:
                problems.append(f"{d.campaign} [{d.date_key}]: timeline has a gap or overlap")
                break
    return problems