nik/ba-auto-daily
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
ba-auto-daily/ba_daily.py

335 lines
16 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Python CLI entry point: ba_dailies.sh -> ba_daily.py -> ba_auto/tasks/*.py"""
import os
import sys
from datetime import datetime
from ba_auto import config, driver, navigation
from ba_auto.tasks import arena, battle_pass, bounty, cafe, circle, event_sweep, exit_game, gem_shop, lesson, login, mailbox, scrimmage, shop_common, shop_tactical, stamina, story_sweep, story_sweep_hard
class _TimestampedStream:
"""Wraps a stream so every printed line gets a '[YYYY-MM-DD HH:MM:SS]'
prefix. Installed once on sys.stdout/sys.stderr at process start (see
the __main__ guard at the bottom of this file) so every task module's
existing print() call -- ~300 call sites across ba_daily.py and
ba_auto/ -- gains a timestamp for free, without touching any of them
individually. This is what lets the per-preset cron logs in
~/ba_logs/ show how long each task/step within a run actually took,
not just the run's own start/finish (which ba_cron_run.sh already
timestamps on its own bracketing lines).
Buffers partial writes until a newline so a single print() call (which
itself may issue more than one underlying write()) still produces one
timestamped line rather than one timestamp per fragment.
"""
def __init__(self, stream):
self._stream = stream
self._buffer = ""
def write(self, data):
self._buffer += data
while "\n" in self._buffer:
line, self._buffer = self._buffer.split("\n", 1)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self._stream.write(f"[{timestamp}] {line}\n")
self._stream.flush()
return len(data)
def flush(self):
self._stream.flush()
def isatty(self):
return self._stream.isatty()
def _install_timestamped_output():
sys.stdout = _TimestampedStream(sys.stdout)
sys.stderr = _TimestampedStream(sys.stderr)
TASKS = {
"login": login.run,
"mailbox": mailbox.run,
"cafe": cafe.run,
"stamina": stamina.run,
"gem_shop": gem_shop.run,
"circle": circle.run,
"battle_pass": battle_pass.run,
"story_sweep": story_sweep.run,
# Bypasses story_sweep's campaign-active guard -- see that module's
# docstring and config.TASK_CAMPAIGN_BADGE_RECT for why the guard exists
# (no reference equivalent; per explicit user direction, added 2026-07-21
# to match story_sweep_hard's own guard).
"story_sweep_force": lambda d, c: story_sweep.run(d, c, force=True),
"story_sweep_hard": story_sweep_hard.run,
# Bypasses story_sweep_hard's campaign-active guard -- see that module's
# docstring and config.TASK_CAMPAIGN_BADGE_RECT for why the guard exists
# (no reference equivalent; per explicit user direction, Hard sweep
# normally refuses to spend AP without an active reward campaign).
"story_sweep_hard_force": lambda d, c: story_sweep_hard.run(d, c, force=True),
"event_sweep": event_sweep.run,
"shop_common": shop_common.run,
"shop_tactical": shop_tactical.run,
"lesson": lesson.run,
"arena": arena.run,
"bounty": bounty.run,
"scrimmage": scrimmage.run,
"exit_game": exit_game.run,
}
# story_sweep (and its story_sweep_force variant), story_sweep_hard (and its
# story_sweep_hard_force variant), event_sweep, both shop tasks, lesson,
# arena, bounty, and scrimmage are opt-in only (not in the default flow):
# they spend AP/credits/tactical coin/lesson tickets/an arena ticket/a
# bounty ticket/a scrimmage ticket on an automated choice rather than
# reclaiming something free, which is a real resource decision the default
# unattended run shouldn't make blindly. scrimmage additionally refuses to
# confirm any sweep whose own confirm dialog doesn't read an OCR'd AP cost
# of exactly 0 (the account's monthly pass is expected to make every
# scrimmage sweep ticket-only) -- see ba_auto/tasks/scrimmage.py's module
# docstring. Both
# story_sweep and story_sweep_hard additionally have their own campaign-active
# guard on top of being opt-in (see each module's own docstring) --
# story_sweep_force/story_sweep_hard_force are separate opt-in commands for
# bypassing those guards, not something DEFAULT_ORDER/PRESETS should ever
# invoke unattended. Arena specifically
# fights a real ranked PvP battle each run -- see ba_auto/tasks/arena.py's
# module docstring. gem_shop and circle are the opposite case -- like
# mailbox/cafe/stamina, they only ever reclaim a genuinely free (0 yen),
# once-per-day resource with no choice to make (claim it or don't, nothing
# to select), so they belong in the default flow rather than opt-in. login
# is first, ahead of everything else: every other task's own navigation
# assumes the home screen is already reachable via the ordinary
# Escape/BACK_BUTTON press loop in navigation.return_to_home, which has no
# way to get there from the title/loading/attendance-card states login.py
# handles -- see that module's docstring. login.run() checks true-home
# first on every internal poll, so this is a fast no-op on a session that's
# already logged in, not a risky blind click every single day. battle_pass
# is the same free-reclaim category as gem_shop/circle -- see its own
# module docstring.
DEFAULT_ORDER = ["login", "mailbox", "cafe", "stamina", "gem_shop", "circle", "battle_pass"]
# Named multi-task sequences, run via `./ba_dailies.sh <preset-name>` --
# distinct from DEFAULT_ORDER (the plain no-args flow, deliberately kept to
# free/no-decision tasks only, see the comment above). A preset is an
# explicit, named choice the user opts into, so unlike DEFAULT_ORDER it's
# free to include resource-spending tasks and repeat tasks (e.g. event_sweep
# multiple times, to re-check a rotating target). Ported directly from the
# user's own `~/.bashrc` `ba_daily()` wrapper (which looped
# `./ba_dailies.sh "$task"` once per task, a fresh Python process every
# time, stopping on the first failure) -- Python owns the loop here
# instead, matching CLAUDE.md's "Python owns automation logic" rule, and
# `_run_task`'s existing try/finally means an uncaught exception from one
# task still halts the whole sequence exactly like the bash `|| return` did.
# exit_game is appended to the end of both presets below -- it deliberately
# ends the session (Escape then Enter on the confirmed true home screen
# raises and confirms Blue Archive's own "exit the game?" dialog), so like
# every other resource-spending/session-affecting task it's opt-in only, not
# in DEFAULT_ORDER. Confirmed live standalone (2026-07-18); not yet
# exercised specifically as the tail end of a full daily/q4h preset run --
# see ba_auto/reference_notes/mapping.md's "Exit game" row.
PRESETS = {
"daily": [
"login", "event_sweep", "cafe", "event_sweep", "circle", "lesson",
"arena", "shop_common", "shop_tactical", "event_sweep", "bounty",
"scrimmage", "gem_shop", "mailbox", "stamina", "battle_pass", "event_sweep", "exit_game",
],
# "q4h": [
# "login", "cafe", "mailbox", "stamina", "event_sweep",
# "story_sweep_hard", "event_sweep", "story_sweep", "exit_game"
# ],
"q4h": [
"login", "cafe", "event_sweep", "mailbox", "stamina",
"event_sweep", "exit_game"
]
}
# How many times _ensure_home retries navigation.return_to_home as a whole
# (not to be confused with that function's own internal
# RETURN_HOME_MAX_ROUNDS press-loop) before giving up, and how long it waits
# between attempts. A single return_to_home call already presses through
# ordinary stuck subscreens/modals -- this outer retry exists for the
# genuinely transient case a single pass can't help with (a screen still
# mid-transition/loading at the moment it was checked, a slow network
# hiccup dialog that needs a few seconds to settle), giving that condition
# real time to resolve on its own between attempts rather than hammering
# the same check back-to-back.
PRE_TASK_HOME_RETRIES = 3
PRE_TASK_HOME_RETRY_WAIT = 5
def _ensure_home(name):
"""Self-healing pre-task recovery: retry navigation.return_to_home
across multiple bounded attempts (with a wait between, for a genuinely
transient condition to clear) rather than giving up after a single
pass. Returns True once home is confirmed, False if still not home
after PRE_TASK_HOME_RETRIES attempts -- callers should NOT abort on
False (see _run_task's own docstring for why), just proceed with
whatever confidence they have.
"""
for attempt in range(1, PRE_TASK_HOME_RETRIES + 1):
if navigation.return_to_home(driver):
return True
print(f"[{name}] still not confirmed home after recovery attempt {attempt}/{PRE_TASK_HOME_RETRIES}")
if attempt < PRE_TASK_HOME_RETRIES:
driver.wait(PRE_TASK_HOME_RETRY_WAIT)
return False
def _run_task(name):
"""Run one task, guaranteeing the game is at the home screen both before
it starts and after it ends -- success, a task's own early "abort
without pressing further keys" return, or an uncaught exception.
Centralized here rather than duplicated inside every task's own run()
so it can't be missed by an early-return path a task's own author
didn't think to guard against, or skipped entirely by a new task that
forgets to add it. An audit (2026-07-11, per explicit user request)
found this was a real, widespread gap: mailbox/cafe/stamina/shop_*/
lesson only pressed a single conditional Escape on some paths (nothing
at all on several early-failure paths), story_sweep.py had no home-
return cleanup anywhere including its own success path, and arena.py
only called navigation.return_to_home at the START of a run (recovering
from the *previous* run's leftover state) rather than at the end of its
own. Only event_sweep.py and bounty.py already called it on every path
-- see plan.md's "Return-to-home audit" entry for the full writeup.
The pre-task call turned out to matter just as much as the post-task
one, confirmed live the same session: every task's own navigation
(MAILBOX_ICON, WORK_ICON, etc.) is a fixed, home-screen-relative
coordinate. mailbox.py was run once with the game left mid-navigation
on an unrelated Event Quest screen (a leftover from an unrelated
network-disconnect popup, nothing to do with this project) -- its
MAILBOX_ICON click landed on that screen instead, and a chain of
false-positive state checks (see navigation.return_to_home's own
_not_home fix from the same incident) let it click blindly into a
completely unrelated event stage-info modal rather than the mailbox.
No resource was actually spent (confirmed by unchanged AP/credits), but
calling return_to_home before every task closes the gap outright rather
than relying on chance.
Per explicit user direction (2026-07-12): this is self-healing, not a
hard gate. _ensure_home retries across several bounded attempts rather
than giving up after one, and even if it still can't confirm home, the
task is attempted anyway -- never hard-aborted here. A task's own
individual navigation steps already verify their own state before
acting (mailbox's MAILBOX_ICON click, story_sweep's task-screen check,
etc.), so a task started from an unconfirmed state still fails safely
at its own first click-verify step rather than cascading blindly, the
same protection every task already has for a missed click mid-run.
A bare try/finally (no except) is used deliberately for the post-task
call: navigation.return_to_home always runs before control leaves this
function, but any exception a task raises still propagates normally
afterward rather than being swallowed -- a crash should still surface
as a crash, just with the game safely back at home first instead of
stuck wherever it failed.
The post-task check is skipped entirely (no call, no warning) if the
game window no longer exists at all -- added alongside exit_game, the
first task able to legitimately close it. Without this, exit_game
succeeding would still make navigation.return_to_home return False (it
already bails out gracefully on a missing window, see its own
docstring) and print a "could not confirm return to home" warning on
every single daily/q4h run that reaches it, even though nothing is
actually wrong -- exactly the kind of noise the cron log convention
(`grep -E 'FAILED|SKIPPED' ~/ba_logs/*.log`) is meant to avoid.
"""
if not _ensure_home(name):
print(f"[{name}] warning: could not confirm starting from the home screen after {PRE_TASK_HOME_RETRIES} attempts -- proceeding anyway")
try:
TASKS[name](driver, config)
finally:
if driver.window_exists() and not navigation.return_to_home(driver):
print(f"[{name}] warning: could not confirm return to home screen after task finished")
def _run_sequence(names):
for name in names:
_run_task(name)
print("All done.")
# Distinct from the generic error exit code (1) so ba_cron_run.sh can log a
# paused run as "SKIPPED", matching the existing lock-skip convention
# instead of a misleading "FAILED (exit N)" -- see that file's own handling
# of this exact value.
PAUSE_EXIT_CODE = 75
def _pause():
open(config.PAUSE_FLAG_PATH, "w").close()
print(f"Paused -- automated and manual runs will be skipped until 'resume'. Flag: {config.PAUSE_FLAG_PATH}")
def _resume():
if os.path.exists(config.PAUSE_FLAG_PATH):
os.remove(config.PAUSE_FLAG_PATH)
print("Resumed.")
else:
print("Not paused.")
def _pause_status():
print("PAUSED" if os.path.exists(config.PAUSE_FLAG_PATH) else "ACTIVE")
# Meta-commands: control the pause flag itself rather than the game, so
# they must stay reachable even while paused -- handled in main() before
# the pause check, not added to TASKS (no driver/game interaction at all).
META_COMMANDS = {
"pause": _pause,
"resume": _resume,
"pause_status": _pause_status,
}
def main(argv):
args = argv[1:]
if args and args[0] == "--list-commands":
# Machine-readable command list for shell completion (see
# completions/ba_dailies.bash) -- kept as a thin read of TASKS/
# PRESETS/META_COMMANDS themselves so the completion list can never
# drift from what's actually dispatchable.
print(" ".join(list(TASKS.keys()) + list(PRESETS.keys()) + list(META_COMMANDS.keys())))
return 0
if args and args[0] in META_COMMANDS:
META_COMMANDS[args[0]]()
return 0
if os.path.exists(config.PAUSE_FLAG_PATH):
print(f"Paused (flag present at {config.PAUSE_FLAG_PATH}) -- skipping this run. Run './ba_dailies.sh resume' to re-enable.", file=sys.stderr)
return PAUSE_EXIT_CODE
if not args:
_run_sequence(DEFAULT_ORDER)
return 0
command = args[0]
if command in PRESETS:
_run_sequence(PRESETS[command])
return 0
if command not in TASKS:
print(f"Unknown phase: {command}", file=sys.stderr)
print(f"Valid phases: {' '.join(TASKS.keys())}", file=sys.stderr)
print(f"Valid presets: {' '.join(PRESETS.keys())}", file=sys.stderr)
print(f"Valid meta-commands: {' '.join(META_COMMANDS.keys())}", file=sys.stderr)
return 1
_run_task(command)
return 0
if __name__ == "__main__":
# Skipped for --list-commands: that output is consumed verbatim by
# completions/ba_dailies.bash as a plain space-separated list, and a
# timestamp prefix would break its parsing.
if not (len(sys.argv) > 1 and sys.argv[1] == "--list-commands"):
_install_timestamped_output()
sys.exit(main(sys.argv))
Reference in New Issue View Git Blame Copy Permalink