ba-auto-daily/ba_auto/tasks/event_sweep.py

"""Event AP sweep. Reference: baas-reference/module/sweep_activity.py and
module/activities/activity_utils.py's activity_sweep/start_sweep.

Ported per explicit user direction (2026-07-10): the currently-running event
("鉄道爆走事件") exposes 12 stages via a 任務情報 (task info) modal that is
structurally identical to story_sweep.py's own stage modal (MIN/-/+/MAX
count stepper, the same AP-usage confirm dialog, the same 掃討完了 SKIP/OK
result screen) -- see config.py's "Event sweep" section for the full live
calibration writeup. Rather than porting the reference's config-string
sweep-list parsing (arbitrary stage lists with per-stage float/fraction
counts), this sweeps exactly one stage per run, chosen from a fixed 9-12
sub-range via the same date-ordinal-modulo rotation story_sweep.py already
uses for its own daily-rotating target.

Key differences from story_sweep.py, all confirmed live during calibration:
- No region concept/navigation -- one flat stage list, reached via the home
  screen's event badge -> Quest tab, not story_sweep's Work-hub card ->
  region browser.
- The stage list only ever needs its bottom scroll extreme, which always
  reveals stages 9-12 (this event's last 5 stages) regardless of starting
  scroll position -- no separate top/bottom row-position sets needed.
- Row height is constant regardless of 1-line vs 2-line title wrapping,
  unlike story_sweep's stage list.
- The stage modal has ONE fixed layout (confirmed against both stage 09 and
  12), no tabbed-vs-plain variant to account for.
- The stage modal DOES close on Escape, unlike story_sweep's (X-button-only).
- Both stages tested were already 3-starred; the reference's SSS-availability
  gate for never-cleared stages was never actually exercised (see config.py).

Live-testing after initial calibration found the home screen's event badge
(config.EVENT_BADGE_ICON) is itself a rotating carousel, not a stable
single-event slot as calibration happened to suggest: it cycles between the
current event's countdown AND other notices (e.g. an already-finished
event's remaining reward-claim-period reminder), so a single click can land
on a stale/wrong event page instead of the current one. _find_stage_row now
distinguishes "wrong page entirely" (no stage-row numbers OCR at all) from
"right page, this specific stage just isn't there" (some numbers found, just
not the target) -- the former retries via navigation.return_to_home + a
re-click of the badge, hoping the carousel has moved on by the next attempt;
the latter is reported and left alone, since retrying won't fix a genuinely
different stage list.

A second live run (after the above fix) correctly recovered from a wrong
first attempt and correctly OCR'd the target stage's row number, but then
failed to open that stage's info modal: the row-enter click was a single
click-then-check with no retry, unlike every other click-then-confirm step
in this module. _open_stage_modal now retries it the same way, matching
CLAUDE.md's own documented history of the mailbox/cafe icons missing their
first click and working on retry.

A third live run surfaced two more instances of the same underlying
problem class:

1. _find_stage_row's "wrong page" detection (all 5 rows OCR empty) fired
   as a FALSE POSITIVE twice in a row on a run that was actually on the
   right page the whole time -- the Quest tab's stage list evidently
   hadn't finished rendering yet on the first OCR pass right after
   navigating there fresh, and 1 second wasn't always enough settle time.
   Burning a full return-home-and-re-navigate cycle (the "wrong page"
   recovery path) on a pure rendering race wastes retry budget that should
   be reserved for an actually-wrong page. _find_stage_row now does a
   cheap in-place rescan (re-read the same rows, no re-scroll/re-navigate)
   a couple of times before concluding "no valid rows at all."
2. Once past that, the MAX-count-raise succeeded (on its own internal
   retry) but the following 掃討開始 (start sweep) click was, again, a bare
   single click with no retry -- the click missed, no confirm dialog ever
   appeared, and the whole run aborted right at the last step before
   actually spending AP. _click_sweep_start_and_verify now wraps it the
   same way as every other click in this file.
"""
import datetime

from ba_auto import detector, navigation


OPEN_RETRIES = 3
POST_SWEEP_DISMISS_ROUNDS = 6
MAX_BUTTON_RETRIES = 3
MODAL_CLOSE_RETRIES = 3
WRONG_PAGE_RETRIES = 3
WRONG_PAGE_RETRY_WAIT = 3
STAGE_ENTER_RETRIES = 3
STAGE_ROW_SCAN_ATTEMPTS = 3
STAGE_ROW_SCAN_RETRY_WAIT = 1.5
SWEEP_START_RETRIES = 3


def _is_stage_modal_open(driver, config):
    r, g, b = driver.color_at(*config.EVENT_STAGE_MODAL_PROBE)
    return r < config.EVENT_STAGE_MODAL_DIM_MAX_CHANNEL and g < config.EVENT_STAGE_MODAL_DIM_MAX_CHANNEL and b < config.EVENT_STAGE_MODAL_DIM_MAX_CHANNEL


def _color_in_range(rgb, rgb_range):
    lo, hi = rgb_range
    r, g, b = rgb
    return lo[0] <= r <= hi[0] and lo[1] <= g <= hi[1] and lo[2] <= b <= hi[2]


def _is_sweep_usage_confirm(driver, config):
    return _color_in_range(driver.color_at(*config.SWEEP_CONFIRM_BUTTON), config.SWEEP_CONFIRM_CYAN)


def _is_ap_purchase_prompt(driver, config):
    return _color_in_range(driver.color_at(*config.SWEEP_CONFIRM_BUTTON), config.SWEEP_CONFIRM_GOLD)


def _find_result_button(driver, config):
    return detector.find_color_centroid(config.SWEEP_RESULT_BUTTON_REGION, *config.SWEEP_CONFIRM_CYAN)


def _count_raised_above_one(driver, config):
    # This modal's "-" stepper turns a saturated coral/orange once raised,
    # a subtler shift than story_sweep's vivid-orange indicator -- told
    # apart from its flat-grey default by color *spread* (max-min channel)
    # rather than a fixed g/b ceiling. See config.py's EVENT_SWEEP_MINUS_
    # BUTTON_PROBE comment.
    r, g, b = driver.color_at(*config.EVENT_SWEEP_MINUS_BUTTON_PROBE)
    return (max(r, g, b) - min(r, g, b)) > 40


def _open_event_screen(driver, config):
    for attempt in range(1, OPEN_RETRIES + 1):
        driver.click(*config.EVENT_BADGE_ICON)
        driver.wait(2)
        if navigation.is_on_subscreen(driver):
            break
        print(f"[event_sweep] event screen not detected after click (attempt {attempt}/{OPEN_RETRIES})")
    else:
        return False

    driver.click(*config.EVENT_QUEST_TAB)
    driver.wait(1)
    return True


def _row_number_rect(config, row_y):
    x1, x2 = config.EVENT_STAGE_NUMBER_OCR_X
    top_pad, bottom_pad = config.EVENT_STAGE_NUMBER_OCR_Y_PAD
    return (x1, row_y - top_pad, x2, row_y + bottom_pad)


def _scan_stage_rows_once(driver, config, stage):
    saw_any_valid_row = False
    for row_y in config.EVENT_STAGE_ROW_Y:
        label = detector.read_int(_row_number_rect(config, row_y))
        print(f"[event_sweep] row @ {row_y}: read '{label}'")
        if label is not None:
            saw_any_valid_row = True
        if label == stage:
            return row_y, saw_any_valid_row
    return None, saw_any_valid_row


def _find_stage_row(driver, config, stage):
    # This event's target range (9-12) always sits within the last 5 rows,
    # confirmed live regardless of starting scroll position -- so unlike
    # story_sweep._find_stage_row, only the bottom extreme is ever checked.
    #
    # Also tracks whether ANY row OCR'd a real number at all -- a finished/
    # stale event's Quest tab shows plain "period ended" text instead of
    # stage-row cards, so all 5 reads coming back empty is a strong signal
    # we're on the wrong page entirely (see module docstring), distinct from
    # "right page, this stage just isn't among the visible rows."
    #
    # A fresh navigation's list can still be settling/rendering when the
    # first OCR pass runs -- confirmed live, this produced two consecutive
    # false-positive "wrong page" reads on a run that was actually on the
    # right page the whole time. Rescan in place (no re-scroll/re-navigate)
    # a couple of times before concluding "no valid rows at all" -- cheap
    # compared to the caller's much more expensive return-home-and-retry
    # recovery, which should be reserved for an actually-wrong page.
    x, y = config.EVENT_STAGE_LIST_SCROLL_POINT
    driver.scroll(x, y, "down", config.EVENT_STAGE_LIST_SCROLL_CLICKS)
    driver.wait(0.5)

    for attempt in range(1, STAGE_ROW_SCAN_ATTEMPTS + 1):
        row_y, saw_any_valid_row = _scan_stage_rows_once(driver, config, stage)
        if row_y is not None or saw_any_valid_row:
            return row_y, saw_any_valid_row
        print(f"[event_sweep] no stage-row numbers recognized on scan {attempt}/{STAGE_ROW_SCAN_ATTEMPTS} -- screen may still be settling")
        if attempt < STAGE_ROW_SCAN_ATTEMPTS:
            driver.wait(STAGE_ROW_SCAN_RETRY_WAIT)
    return None, False


def _click_max_and_verify(driver, config):
    for attempt in range(1, MAX_BUTTON_RETRIES + 1):
        driver.click(*config.EVENT_SWEEP_MAX_BUTTON)
        driver.wait(0.8)
        if _count_raised_above_one(driver, config):
            return True
        print(f"[event_sweep] MAX click not detected (attempt {attempt}/{MAX_BUTTON_RETRIES})")
    return False


def _click_plus_and_verify(driver, config, count):
    for attempt in range(1, MAX_BUTTON_RETRIES + 1):
        for _ in range(count - 1):
            driver.click(*config.EVENT_SWEEP_PLUS_BUTTON)
        driver.wait(0.8)
        if _count_raised_above_one(driver, config):
            return True
        print(f"[event_sweep] count-raise via '+' not detected (attempt {attempt}/{MAX_BUTTON_RETRIES})")
    return False


def _set_sweep_count(driver, config, count):
    if count == "max":
        return _click_max_and_verify(driver, config)
    return _click_plus_and_verify(driver, config, count)


def _close_stage_modal(driver, config):
    # Confirmed live this modal DOES close on Escape (unlike story_sweep's,
    # which needs its own X button) -- try Escape first, fall back to the X
    # button if it somehow doesn't clear.
    for _ in range(MODAL_CLOSE_RETRIES):
        if not _is_stage_modal_open(driver, config):
            return True
        driver.keypress("Escape")
        driver.wait(1)
        if not _is_stage_modal_open(driver, config):
            return True
        driver.click(*config.EVENT_STAGE_MODAL_CLOSE_BUTTON)
        driver.wait(1)
    return not _is_stage_modal_open(driver, config)


def _watch_sweep_result(driver, config):
    def click_result_button(d):
        pos = _find_result_button(d, config)
        if pos:
            d.click(*pos)
        d.wait(1.5)

    ends = {
        (lambda d, c: _is_stage_modal_open(d, c) and _find_result_button(d, c) is None): "swept",
    }
    reactions = {
        (lambda d, c: _find_result_button(d, c) is not None): click_result_button,
    }
    outcome = navigation.wait_for_state(
        driver, config, reactions, ends,
        max_iterations=POST_SWEEP_DISMISS_ROUNDS, poll_interval=1.5,
    )
    return outcome or "unrecognized_state"


def _open_stage_modal(driver, config, row_y):
    # A single click-then-check here was found live to intermittently miss
    # (CLAUDE.md already documents this exact failure mode for the
    # mailbox/cafe icons: "missed the first click and worked on retry") --
    # retry-with-verify like every other click-then-confirm step in this
    # module, rather than aborting on the first miss.
    for attempt in range(1, STAGE_ENTER_RETRIES + 1):
        driver.click(config.EVENT_STAGE_ENTER_X, row_y)
        driver.wait(2)
        if _is_stage_modal_open(driver, config):
            return True
        print(f"[event_sweep] stage info panel not detected after click (attempt {attempt}/{STAGE_ENTER_RETRIES})")
    return False


def _click_sweep_start_and_verify(driver, config):
    # Same missed-click failure class as _open_stage_modal -- a bare single
    # click here was found live to leave neither the AP-usage-confirm nor
    # the AP-purchase dialog detected, aborting the run at the very last
    # step before it would have actually spent AP. Retry-with-verify like
    # every other click in this module.
    for attempt in range(1, SWEEP_START_RETRIES + 1):
        driver.click(*config.EVENT_SWEEP_START_BUTTON)
        driver.wait(1.5)
        if _is_sweep_usage_confirm(driver, config) or _is_ap_purchase_prompt(driver, config):
            return True
        print(f"[event_sweep] sweep confirm/AP-purchase dialog not detected after 掃討開始 click (attempt {attempt}/{SWEEP_START_RETRIES})")
    return False


def _sweep_target(driver, config, stage, count):
    print(f"[event_sweep] --- target stage {stage} x {count} ---")

    row_y, saw_any_valid_row = _find_stage_row(driver, config, stage)
    if row_y is None:
        if not saw_any_valid_row:
            print("[event_sweep] no stage-row numbers recognized at all -- likely on the wrong/stale event page")
            return "wrong_page"
        print(f"[event_sweep] stage {stage} not found in the visible stage list")
        return "stage_not_found"

    if not _open_stage_modal(driver, config, row_y):
        print("[event_sweep] stage info panel not detected, aborting")
        return "unrecognized_state"

    if not _set_sweep_count(driver, config, count):
        print("[event_sweep] could not confirm sweep count was raised (stage may not be SSS-cleared/sweepable yet) -- aborting without spending AP")
        _close_stage_modal(driver, config)
        return "not_sweepable"

    if not _click_sweep_start_and_verify(driver, config):
        print("[event_sweep] sweep-usage confirmation not detected, aborting without further input")
        _close_stage_modal(driver, config)
        return "unrecognized_state"

    if _is_ap_purchase_prompt(driver, config):
        print("[event_sweep] insufficient AP for this sweep -- cancelling without purchasing")
        driver.click(*config.SWEEP_CONFIRM_CANCEL_BUTTON)
        driver.wait(1)
        _close_stage_modal(driver, config)
        return "inadequate_ap"

    driver.click(*config.SWEEP_CONFIRM_BUTTON)
    driver.wait(1.5)
    print("[event_sweep] sweep confirmed, waiting for results")
    outcome = _watch_sweep_result(driver, config)
    print(f"[event_sweep] result: {outcome}")

    if not _close_stage_modal(driver, config):
        print("[event_sweep] warning: could not confirm stage info modal closed -- leaving it open rather than pressing further keys blindly")
    return outcome


def _rotation_target(config):
    stage_min = getattr(config, "EVENT_SWEEP_ROTATION_STAGE_MIN", None)
    if not stage_min:
        return None
    stage_max = config.EVENT_SWEEP_ROTATION_STAGE_MAX
    span = stage_max - stage_min + 1
    stage = stage_min + (datetime.date.today().toordinal() % span)
    return (stage, config.EVENT_SWEEP_ROTATION_COUNT)


def run(driver, config):
    driver.focus_game()

    rotation = _rotation_target(config)
    if not rotation:
        print("[event_sweep] no rotation target configured (config.EVENT_SWEEP_ROTATION_STAGE_MIN is unset), nothing to do")
        return
    stage, count = rotation
    print(f"[event_sweep] today's rotation target: stage {stage}")

    outcome = None
    for attempt in range(1, WRONG_PAGE_RETRIES + 1):
        if not _open_event_screen(driver, config):
            print(f"[event_sweep] could not confirm event screen is open (attempt {attempt}/{WRONG_PAGE_RETRIES})")
            navigation.return_to_home(driver)
            driver.wait(WRONG_PAGE_RETRY_WAIT)
            continue

        outcome = _sweep_target(driver, config, stage, count)
        if outcome == "wrong_page":
            print(f"[event_sweep] landed on the wrong event page (attempt {attempt}/{WRONG_PAGE_RETRIES}) -- returning home to retry")
            navigation.return_to_home(driver)
            driver.wait(WRONG_PAGE_RETRY_WAIT)
            continue
        break
    else:
        print(f"[event_sweep] could not reach the current event's stage list after {WRONG_PAGE_RETRIES} attempts, giving up")

    if outcome is not None and outcome not in ("swept", "wrong_page"):
        print(f"[event_sweep] target stage {stage} ended in '{outcome}'")

    if not navigation.return_to_home(driver):
        print("[event_sweep] warning: could not confirm return to home screen")

    print("[event_sweep] Done.")