Brandon LiandClaude Opus 5
fbe2c47ed4
A run died with "'charmap' codec can't encode character '▶'" at step 1. Python picks the console code page for stdout, and under PM2 - where stdout is a pipe rather than a console - that is cp1252 on Windows. cp1252 handles the em dashes in these files but not the run markers, so the first one raised UnicodeEncodeError from inside run_steps and the runner reported it as a step failure rather than an output problem. Reconfigure stdout and stderr to UTF-8 at the top of each entry point, with errors="replace" as a backstop for streams that cannot be reconfigured. Fixing the encoding once beats stripping the glyphs from ~30 call sites, and covers manual runs in a plain console too. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
429 lines
18 KiB
Python
429 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Automation runner — the process that makes dashboard buttons do something.
|
|
|
|
Leave this running. It polls the dashboard for queued runs, and when one appears
|
|
it works through that automation's steps, driving the real mouse and keyboard,
|
|
reporting each step back so the page can show progress.
|
|
|
|
python runner.py # against http://localhost:3000
|
|
python runner.py --api http://vps:3000
|
|
|
|
Steps come from the server, so adding a new button means editing
|
|
lib/automations.ts — nothing here needs to change.
|
|
|
|
Stop with Ctrl-C. A run in progress can be halted from the dashboard's Stop
|
|
button; the runner notices between steps.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import random
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import actions
|
|
import focus
|
|
from clicker import Dashboard, DashboardError, NotFoundError, force_utf8_output
|
|
|
|
POLL_SECONDS = 1.0
|
|
HEARTBEAT_SECONDS = 2.0
|
|
|
|
# Bumped whenever the step vocabulary or the locate protocol changes. Reported in
|
|
# the heartbeat so the dashboard can say "restart your runner" instead of letting
|
|
# a stale process fail on a step type it has never heard of.
|
|
VERSION = "0.16.0"
|
|
|
|
# Shared with the heartbeat thread: whether a run is currently executing.
|
|
_busy = threading.Event()
|
|
|
|
|
|
def heartbeat_loop(dash: Dashboard, opts, stop: threading.Event) -> None:
|
|
"""Check in on our own thread.
|
|
|
|
Deliberately not folded into the main poll loop: a single step can block for
|
|
twenty seconds waiting on the extension, and a runner that goes quiet that
|
|
long would show as offline in the middle of the run it is executing.
|
|
"""
|
|
payload = {"host": socket.gethostname(), "pid": os.getpid(),
|
|
"dryRun": opts.dry_run, "version": VERSION}
|
|
while not stop.is_set():
|
|
try:
|
|
dash._request("/api/autobuyer/runner", "POST", {**payload, "busy": _busy.is_set()})
|
|
except DashboardError:
|
|
pass # the main loop reports connectivity; don't double up
|
|
stop.wait(HEARTBEAT_SECONDS)
|
|
|
|
|
|
def run_one_step(dash: Dashboard, step: dict, opts, rng, report, run_id: int | None = None) -> None:
|
|
"""Dispatch a single step. Raises StepError / DashboardError on failure."""
|
|
if step["action"] == "wait":
|
|
time.sleep(float(step.get("seconds", 1)))
|
|
report(f"waited {step.get('seconds', 1)}s")
|
|
|
|
elif step["action"] == "navigate":
|
|
# No mouse involved: the extension points the tab at the page and waits
|
|
# for it to load. Locating <body> confirms it is really there.
|
|
target = step.get("url", "")
|
|
landed = dash.locate("body", 0, step.get("urlPattern", "") or opts.url,
|
|
opts.timeout, step.get("openUrl", "") or "",
|
|
navigate_url=target)
|
|
# Navigating to a protected page is where an expired session shows up:
|
|
# the site answers with a redirect to its login form.
|
|
actions.check_signed_in(landed, step.get("signedOut", "") or "")
|
|
report(f"tab is on {landed.get('url', target)}")
|
|
|
|
elif step["action"] == "waitFor":
|
|
# A gate, not an action: block until the element exists — or, with
|
|
# `absent`, until it is gone. Nothing is clicked or typed. Whatever has to
|
|
# change the page — a person solving a challenge, a modal closing itself,
|
|
# a slow server — happens outside this run.
|
|
selector = step["selector"]
|
|
absent = bool(step.get("absent"))
|
|
goal = "disappear" if absent else "appear"
|
|
timeout_s = float(step.get("timeoutSeconds", 120))
|
|
deadline = time.time() + timeout_s
|
|
announced = False
|
|
|
|
while True:
|
|
# Long gates must still honour the dashboard's Stop button.
|
|
if run_id is not None and run_status(dash, run_id) == "cancelled":
|
|
raise actions.StepError("cancelled while waiting")
|
|
|
|
try:
|
|
dash.locate(selector, int(step.get("index", 0)),
|
|
step.get("urlPattern", "") or opts.url,
|
|
opts.timeout, step.get("openUrl", "") or "")
|
|
if not absent:
|
|
report(f"{selector} appeared")
|
|
break
|
|
# Still there. Keep waiting for it to go.
|
|
except NotFoundError:
|
|
if absent:
|
|
report(f"{selector} is gone")
|
|
break
|
|
# Not there yet; keep waiting for it to arrive.
|
|
except DashboardError:
|
|
# A dead extension or an unreachable dashboard must not be read
|
|
# as "the element is gone" — that would satisfy an absent gate
|
|
# for entirely the wrong reason.
|
|
pass
|
|
|
|
if time.time() >= deadline:
|
|
raise actions.StepError(
|
|
f"{selector} did not {goal} within {timeout_s:.0f}s"
|
|
)
|
|
if not announced:
|
|
report(f"waiting for {selector} to {goal} (up to {timeout_s:.0f}s)")
|
|
announced = True
|
|
time.sleep(2.0)
|
|
|
|
elif step["action"] == "scrollToLoad":
|
|
# Lists that load progressively need walking to the bottom before the
|
|
# steps that act on their items can see everything.
|
|
result = dash.locate(
|
|
step["selector"], 0,
|
|
step.get("urlPattern", "") or opts.url,
|
|
max(opts.timeout, 120.0), # scrolling a long list outlasts a normal step
|
|
step.get("openUrl", "") or "",
|
|
options={
|
|
"op": "scrollToLoad",
|
|
"containerSelector": step.get("containerSelector", ""),
|
|
"maxScrolls": int(step.get("maxScrolls", 25)),
|
|
"settleMs": int(step.get("settleMs", 800)),
|
|
},
|
|
)
|
|
found_n = result.get("after", 0)
|
|
report(f"{found_n} match(es) after {result.get('scrolls', 0)} scroll(s) "
|
|
f"of {result.get('container', '?')} (was {result.get('before', 0)})")
|
|
if not result.get("exhausted"):
|
|
# Stopping on the scroll cap is not a failure, but it does mean the
|
|
# list may still have more below — worth saying so rather than
|
|
# letting a later step quietly work on a partial list.
|
|
report("hit the scroll limit — there may be more not loaded")
|
|
|
|
elif step["action"] not in ("click", "type"):
|
|
# Almost always a stale runner: the server defines the step vocabulary,
|
|
# so a step type this process has never heard of means automations.ts has
|
|
# moved on and this file has not been restarted.
|
|
raise actions.StepError(
|
|
f"unknown step action {step['action']!r} — this runner is "
|
|
f"v{VERSION}; restart it to pick up newer step types"
|
|
)
|
|
|
|
else:
|
|
try:
|
|
actions.perform(
|
|
dash,
|
|
step["action"],
|
|
step["selector"],
|
|
index=int(step.get("index", 0)),
|
|
url=step.get("urlPattern", "") or opts.url,
|
|
open_url=step.get("openUrl", "") or "",
|
|
signed_out=step.get("signedOut", "") or "",
|
|
text=step.get("text"),
|
|
clear=bool(step.get("clear")),
|
|
scale=opts.scale,
|
|
timeout=opts.timeout,
|
|
rng=rng,
|
|
activate=not opts.no_activate,
|
|
dry_run=opts.dry_run,
|
|
report=report,
|
|
)
|
|
except NotFoundError:
|
|
# Only absence is skippable. A covered element, an off-screen target
|
|
# or an unreachable extension still fails the run.
|
|
if not step.get("skipIfNotFound"):
|
|
raise
|
|
report(f"{step['selector']} not present — skipped")
|
|
|
|
|
|
def run_status(dash: Dashboard, run_id: int) -> str | None:
|
|
"""Current status, so a wait can notice the dashboard's Stop button."""
|
|
try:
|
|
return dash._request(f"/api/autobuyer/runs?id={run_id}")["run"]["status"]
|
|
except (DashboardError, KeyError):
|
|
return None
|
|
|
|
|
|
def wait_for_sign_in(dash: Dashboard, step: dict, run_id: int, opts, deadline: float) -> bool:
|
|
"""Block until the tab leaves the login page.
|
|
|
|
The runner does not log in — entering credentials is the human's job, and a
|
|
keystroke aimed at the wrong field is not something to risk automating. All
|
|
this does is watch the tab and pick the run back up once you are through.
|
|
"""
|
|
signed_out = step.get("signedOut", "") or ""
|
|
pattern = step.get("urlPattern", "") or opts.url
|
|
open_url = step.get("openUrl", "") or ""
|
|
|
|
while time.time() < deadline:
|
|
if run_status(dash, run_id) == "cancelled":
|
|
return False
|
|
time.sleep(3.0)
|
|
try:
|
|
landed = dash.locate("body", 0, pattern, opts.timeout, open_url)
|
|
except DashboardError:
|
|
continue # tab busy mid-redirect; look again shortly
|
|
if signed_out not in (landed.get("url") or ""):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def signed_in_now(dash: Dashboard, step: dict, opts, settle_seconds: float = 20.0) -> bool:
|
|
"""Did we actually get back in?
|
|
|
|
Auth steps can 'succeed' — every click landing — while the site still rejects
|
|
the sign-in, so this confirms rather than assumes before the failed step is
|
|
retried.
|
|
|
|
It polls rather than checking once: submitting a login form starts a network
|
|
round trip and then a redirect, so the tab is still sitting on the login URL
|
|
for a second or two afterwards. Checking immediately reports a failure for a
|
|
sign-in that is merely still in flight.
|
|
"""
|
|
signed_out = step.get("signedOut", "") or ""
|
|
if not signed_out:
|
|
return True
|
|
|
|
deadline = time.time() + settle_seconds
|
|
while True:
|
|
try:
|
|
landed = dash.locate("body", 0, step.get("urlPattern", "") or opts.url,
|
|
opts.timeout, step.get("openUrl", "") or "")
|
|
if signed_out not in (landed.get("url") or ""):
|
|
return True
|
|
except DashboardError:
|
|
pass # mid-redirect the tab can be briefly un-injectable
|
|
|
|
if time.time() >= deadline:
|
|
return False
|
|
time.sleep(1.0)
|
|
|
|
|
|
def run_steps(dash: Dashboard, run: dict, opts) -> None:
|
|
"""Work through one automation. Reports every step; stops on the first
|
|
failure, because a half-completed purchase flow should not barrel on."""
|
|
run_id = run["id"]
|
|
steps = run["steps"]
|
|
auth_steps = run.get("authSteps") or []
|
|
rng = random.Random(opts.seed) if opts.seed is not None else random.Random()
|
|
|
|
print(f"\n▶ run #{run_id} — {run['label']} ({len(steps)} steps)")
|
|
|
|
for i, step in enumerate(steps):
|
|
label = step.get("label") or f"{step['action']} {step.get('selector', '')}".strip()
|
|
print(f" [{i + 1}/{len(steps)}] {label}")
|
|
|
|
detail_parts: list[str] = []
|
|
|
|
def report(message: str) -> None:
|
|
detail_parts.append(message)
|
|
print(f" {message}")
|
|
|
|
# One retry after a sign-in: the session can only be expired once per step.
|
|
for attempt in range(2):
|
|
try:
|
|
run_one_step(dash, step, opts, rng, report, run_id)
|
|
break
|
|
|
|
except actions.SignedOutError as exc:
|
|
wait_seconds = float(step.get("signedOutWait", 0) or 0)
|
|
|
|
# Scripted auth first, if the firm defines any; otherwise fall
|
|
# back to pausing for a human. Only ever attempted once per step —
|
|
# auth that "succeeds" without signing in would otherwise loop.
|
|
if auth_steps and attempt == 0:
|
|
post_progress(dash, run_id, i, f"signed out — running {len(auth_steps)} auth step(s)",
|
|
False, exc.args[0])
|
|
try:
|
|
for astep in auth_steps:
|
|
alabel = astep.get("label") or f"auth: {astep['action']}"
|
|
print(f" [auth] {alabel}")
|
|
run_one_step(dash, astep, opts, rng, report, run_id)
|
|
except (actions.StepError, DashboardError) as aexc:
|
|
msg = f"auth steps failed: {aexc}"
|
|
post_progress(dash, run_id, i, label, False, msg)
|
|
finish(dash, run_id, msg)
|
|
print(f"✗ run #{run_id} stopped: {msg}", file=sys.stderr)
|
|
return
|
|
|
|
if not signed_in_now(dash, step, opts):
|
|
msg = "auth steps ran but the session is still signed out"
|
|
post_progress(dash, run_id, i, label, False, msg)
|
|
finish(dash, run_id, msg)
|
|
print(f"✗ run #{run_id} stopped: {msg}", file=sys.stderr)
|
|
return
|
|
|
|
report("signed in via auth steps — retrying")
|
|
continue
|
|
|
|
if wait_seconds <= 0 or attempt > 0:
|
|
print(f" FAILED: {exc}", file=sys.stderr)
|
|
post_progress(dash, run_id, i, label, False, str(exc))
|
|
finish(dash, run_id, str(exc))
|
|
print(f"✗ run #{run_id} stopped at step {i + 1}")
|
|
return
|
|
|
|
mins = wait_seconds / 60
|
|
note = f"signed out — waiting up to {mins:.0f} min for you to log in"
|
|
print(f" {note}", file=sys.stderr)
|
|
post_progress(dash, run_id, i, note, False, exc.args[0])
|
|
|
|
if not wait_for_sign_in(dash, step, run_id, opts, time.time() + wait_seconds):
|
|
if run_status(dash, run_id) == "cancelled":
|
|
print(f"■ run #{run_id} cancelled while waiting for sign-in")
|
|
return
|
|
msg = f"still signed out after {mins:.0f} min"
|
|
post_progress(dash, run_id, i, label, False, msg)
|
|
finish(dash, run_id, msg)
|
|
print(f"✗ run #{run_id} gave up waiting for sign-in")
|
|
return
|
|
|
|
report("signed in — retrying the step")
|
|
|
|
except (actions.StepError, DashboardError) as exc:
|
|
print(f" FAILED: {exc}", file=sys.stderr)
|
|
post_progress(dash, run_id, i, label, False, str(exc))
|
|
finish(dash, run_id, str(exc))
|
|
print(f"✗ run #{run_id} stopped at step {i + 1}")
|
|
return
|
|
|
|
status = post_progress(dash, run_id, i + 1, label, True, "; ".join(detail_parts) or None)
|
|
|
|
# The dashboard's Stop button shows up here.
|
|
if status == "cancelled":
|
|
print(f"■ run #{run_id} cancelled from the dashboard")
|
|
return
|
|
|
|
finish(dash, run_id, None)
|
|
print(f"✓ run #{run_id} complete")
|
|
|
|
|
|
def post_progress(dash, run_id, step_index, step, ok, detail):
|
|
try:
|
|
res = dash._request("/api/autobuyer/runs/progress", "POST", {
|
|
"id": run_id, "stepIndex": step_index, "step": step, "ok": ok, "detail": detail,
|
|
})
|
|
return res.get("status")
|
|
except DashboardError as exc:
|
|
print(f" (could not report progress: {exc})", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def finish(dash, run_id, error):
|
|
try:
|
|
dash._request("/api/autobuyer/runs/progress", "POST",
|
|
{"id": run_id, "finish": True, "error": error})
|
|
except DashboardError as exc:
|
|
print(f" (could not report completion: {exc})", file=sys.stderr)
|
|
|
|
|
|
def main() -> int:
|
|
force_utf8_output()
|
|
parser = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("--api", default="http://localhost:3000", help="dashboard URL")
|
|
parser.add_argument("--url", default="", help="fallback Chrome match pattern for steps that omit one")
|
|
parser.add_argument("--timeout", type=float, default=45.0, help="seconds to wait for the extension per step (a cold page load takes time)")
|
|
parser.add_argument("--scale", type=float, default=None, help="CSS-to-desktop pixel ratio")
|
|
parser.add_argument("--seed", type=int, default=None, help="seed the motion RNG (debugging)")
|
|
parser.add_argument("--dry-run", action="store_true", help="move the cursor through the steps but never press or type")
|
|
parser.add_argument("--no-activate", action="store_true", help="do not raise the browser before each step")
|
|
opts = parser.parse_args()
|
|
|
|
try:
|
|
import pyautogui # noqa: F401
|
|
except ImportError:
|
|
print("error: pyautogui is not installed — run: pip install -r requirements.txt", file=sys.stderr)
|
|
return 1
|
|
|
|
# Before anything asks the OS how big the screen is. No-op off Windows.
|
|
dpi = focus.enable_dpi_awareness()
|
|
if dpi:
|
|
print(f" {dpi}")
|
|
|
|
dash = Dashboard(opts.api)
|
|
print(f"Runner watching {opts.api}" + (" [DRY RUN — nothing will be pressed]" if opts.dry_run else ""))
|
|
print("Waiting for a run. Press a button on the AutoBuyer page. Ctrl-C to stop.")
|
|
|
|
stop = threading.Event()
|
|
beat = threading.Thread(target=heartbeat_loop, args=(dash, opts, stop), daemon=True)
|
|
beat.start()
|
|
|
|
idle_warned = False
|
|
while True:
|
|
try:
|
|
claim = dash._request("/api/autobuyer/runs/claim", "POST", {})
|
|
except DashboardError as exc:
|
|
if not idle_warned:
|
|
print(f" ({exc})", file=sys.stderr)
|
|
idle_warned = True
|
|
time.sleep(POLL_SECONDS * 3)
|
|
continue
|
|
|
|
idle_warned = False
|
|
run = claim.get("run")
|
|
if not run:
|
|
time.sleep(POLL_SECONDS)
|
|
continue
|
|
|
|
_busy.set()
|
|
try:
|
|
run_steps(dash, run, opts)
|
|
except Exception as exc: # keep the daemon alive
|
|
print(f"✗ run failed unexpectedly: {exc}", file=sys.stderr)
|
|
finish(dash, run["id"], f"runner error: {exc}")
|
|
finally:
|
|
_busy.clear()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyboardInterrupt:
|
|
print("\nstopped.")
|
|
sys.exit(0)
|