Brandon LiandClaude Opus 5
3ff5728af9
Next's dev server intermittently answers a 500 while recompiling a route: it reads a build manifest mid-write and cannot parse it. A single one of those during the locate poll was fatal, so a blip in the pipeline killed a run partway through an auth flow on Windows. 5xx responses and dropped connections are now a distinct TransientError, retried until the step's own timeout. A 4xx still fails immediately — those are verdicts about the request, not blips. If the errors persist all the way to the timeout, the message says so rather than blaming a missing extension. Error bodies are also summarised. A dev-server 500 replies with a full HTML page, and printing it raw buried the one line that said what went wrong under kilobytes of script tags. This makes the client tolerant of the fault, which is not the same as fixing it: the real answer on an automation host is to run a production build rather than `next dev`, so those manifests are written once instead of continuously. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
315 lines
13 KiB
Python
315 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Desktop mouse control for the AutoFirmer autobuyer.
|
|
|
|
The browser extension can see the DOM but can't move the mouse; this script can
|
|
move the mouse but can't see inside Chrome. They meet at the dashboard API:
|
|
|
|
clicker --POST /api/autobuyer/locate--> "where is button.buy?"
|
|
extension raises the window, scrolls the
|
|
element into view, measures it
|
|
clicker --GET /api/autobuyer/locate--> desktop x,y
|
|
clicker moves the real mouse and clicks
|
|
|
|
Usage:
|
|
python clicker.py locate "button.buy" # measure only, no clicking
|
|
python clicker.py click "button.buy" # measure, then really click
|
|
python clicker.py click ".btn" --index 2 --url "https://tradeify.co/*"
|
|
python clicker.py click "button.buy" --dry-run # move the cursor, don't press
|
|
|
|
Requires the AutoBuyer page switch to be ON — that's the master arming switch.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import random
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import actions
|
|
import focus
|
|
|
|
DEFAULT_API = "http://localhost:3000"
|
|
POLL_INTERVAL = 0.25
|
|
|
|
|
|
class DashboardError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class TransientError(DashboardError):
|
|
"""A failure that is worth retrying: a 5xx, or the server briefly unreachable.
|
|
|
|
Next's dev server intermittently serves a 500 while it recompiles a route —
|
|
it reads a build manifest mid-write and fails to parse it. That is a blip in
|
|
the pipeline, not a verdict about the page, and it should not kill a run that
|
|
is halfway through spending money.
|
|
"""
|
|
|
|
|
|
class NotFoundError(DashboardError):
|
|
"""The extension reached the page and the element simply isn't there.
|
|
|
|
Kept distinct from every other DashboardError — a dead extension, an
|
|
unreachable dashboard, a missing tab — so that a `skipIfNotFound` step can skip a
|
|
genuinely absent element without also swallowing a broken pipeline. Those
|
|
look identical from a distance and must not be treated alike.
|
|
"""
|
|
|
|
|
|
# What the extension says when the page loaded fine but the selector matched
|
|
# nothing. See pageLocate in extension/background.js.
|
|
_ABSENT_MARKERS = ("No element matches", "no index")
|
|
|
|
|
|
def _summarise_error_body(body: str, limit: int = 200) -> str:
|
|
"""Keep an error readable.
|
|
|
|
A dev-server 500 answers with a full HTML page — several kilobytes of script
|
|
tags and a stack trace — and printing it raw buries the one line that says
|
|
what went wrong.
|
|
"""
|
|
try:
|
|
return str(json.loads(body).get("error", body))[:limit]
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if "<!DOCTYPE" in body or "<html" in body:
|
|
match = re.search(r'"message":"(.*?)"', body)
|
|
detail = match.group(1) if match else "no detail in the page"
|
|
return f"server returned an HTML error page ({detail[:limit]})"
|
|
|
|
return body[:limit]
|
|
|
|
|
|
def _is_absent(message: str) -> bool:
|
|
return any(marker in message for marker in _ABSENT_MARKERS)
|
|
|
|
|
|
class Dashboard:
|
|
"""Thin client for the autobuyer endpoints."""
|
|
|
|
def __init__(self, base: str, timeout: float = 10.0):
|
|
self.base = base.rstrip("/")
|
|
self.timeout = timeout
|
|
|
|
def _request(self, path: str, method: str = "GET", payload: dict | None = None) -> dict:
|
|
data = json.dumps(payload).encode() if payload is not None else None
|
|
headers = {"Content-Type": "application/json"} if data else {}
|
|
req = urllib.request.Request(self.base + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as exc:
|
|
message = _summarise_error_body(exc.read().decode(errors="replace"))
|
|
cls = TransientError if exc.code >= 500 else DashboardError
|
|
raise cls(f"{method} {path} -> HTTP {exc.code}: {message}") from None
|
|
except urllib.error.URLError as exc:
|
|
raise TransientError(f"Cannot reach {self.base} — {exc.reason}") from None
|
|
|
|
def status(self) -> dict:
|
|
return self._request("/api/autobuyer/status")
|
|
|
|
def locate(self, selector: str, index: int, url_pattern: str, timeout: float,
|
|
open_url: str = "", navigate_url: str = "") -> dict:
|
|
"""Queue a lookup and block until the extension answers it.
|
|
|
|
`open_url` is the page the extension should open if no tab matches
|
|
`url_pattern` — a match pattern isn't navigable, so it's passed separately.
|
|
"""
|
|
queued = self._request(
|
|
"/api/autobuyer/locate",
|
|
"POST",
|
|
{"selector": selector, "index": index, "urlPattern": url_pattern,
|
|
"openUrl": open_url, "navigateUrl": navigate_url},
|
|
)
|
|
request_id = queued["id"]
|
|
|
|
deadline = time.monotonic() + timeout
|
|
last_transient = None
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
row = self._request(f"/api/autobuyer/locate?id={request_id}")
|
|
except TransientError as exc:
|
|
# The extension may well answer while the server is having a
|
|
# moment; keep polling rather than failing the run over a blip.
|
|
last_transient = exc
|
|
time.sleep(POLL_INTERVAL)
|
|
continue
|
|
|
|
if row["status"] == "done":
|
|
return row["result"]
|
|
if row["status"] == "error":
|
|
detail = row["error"] or ""
|
|
cls = NotFoundError if _is_absent(detail) else DashboardError
|
|
raise cls(f"Extension could not locate it: {detail}")
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
if last_transient is not None:
|
|
raise DashboardError(
|
|
f"No answer within {timeout:g}s, and the dashboard kept erroring "
|
|
f"({last_transient})"
|
|
)
|
|
raise DashboardError(
|
|
f"No answer within {timeout:g}s. Is the extension installed, is Chrome "
|
|
f"running, and is the AutoBuyer switch ON?"
|
|
)
|
|
|
|
|
|
def describe(found: dict, factor: float) -> str:
|
|
x, y = found["screen"]["x"] * factor, found["screen"]["y"] * factor
|
|
lines = [
|
|
f" matched <{found['tag']}> {found['text']!r}"
|
|
+ (f" (1 of {found['matchCount']})" if found["matchCount"] > 1 else ""),
|
|
f" page {found['url']}",
|
|
f" viewport x={found['viewport']['x']:.0f} y={found['viewport']['y']:.0f}"
|
|
f" {found['viewport']['width']:.0f}x{found['viewport']['height']:.0f}",
|
|
f" desktop x={x:.0f} y={y:.0f}" + (f" (scale {factor:g})" if factor != 1 else ""),
|
|
]
|
|
if found.get("covered"):
|
|
lines.append(f" WARNING something is on top of it: {found['coveredBy']}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("action", choices=["locate", "click", "type"],
|
|
help="locate = measure only; click = measure then click; "
|
|
"type = click the field, then type into it")
|
|
parser.add_argument("selector", help="CSS selector of the target element")
|
|
parser.add_argument("--index", type=int, default=0, help="which match, if the selector hits several (default 0)")
|
|
parser.add_argument("--url", default="", help='Chrome match pattern for the tab, e.g. "https://tradeify.co/*"')
|
|
parser.add_argument("--open-url", default="", help="page to open if no tab matches --url")
|
|
parser.add_argument("--signed-out", default="", help="URL substring identifying the login page; abort if the tab lands there")
|
|
parser.add_argument("--api", default=DEFAULT_API, help=f"dashboard URL (default {DEFAULT_API})")
|
|
parser.add_argument("--timeout", type=float, default=45.0, help="seconds to wait for the extension (default 45; a cold page load takes time)")
|
|
parser.add_argument("--scale", type=float, default=None, help="CSS-to-desktop pixel ratio (default: auto-detect)")
|
|
parser.add_argument("--dry-run", action="store_true", help="move the cursor to the target but do not press")
|
|
parser.add_argument("--force", action="store_true", help="click even if something is covering the element")
|
|
parser.add_argument("--robotic", action="store_true",
|
|
help="straight-line move and instant click, skipping the human motion model")
|
|
parser.add_argument("--seed", type=int, default=None,
|
|
help="seed the motion RNG so a run is reproducible (for debugging)")
|
|
parser.add_argument("--no-activate", action="store_true",
|
|
help="do not raise the browser first (the click may be eaten by window activation)")
|
|
parser.add_argument("--text", default=None, help="text to type (action=type)")
|
|
parser.add_argument("--stdin", action="store_true",
|
|
help="read the text to type from stdin instead of --text, keeping it out of shell history")
|
|
parser.add_argument("--clear", action="store_true",
|
|
help="select-all and delete before typing, rather than appending at the caret")
|
|
parser.add_argument("--allow-enter", action="store_true",
|
|
help="permit newlines in the text (each one presses Enter, which may submit the form)")
|
|
parser.add_argument("--no-verify", action="store_true",
|
|
help="skip reading the field back after typing")
|
|
args = parser.parse_args()
|
|
|
|
# Anything that moves the mouse or presses a key needs pyautogui; check once,
|
|
# up front, so a missing dependency reports itself rather than surfacing as an
|
|
# ImportError from somewhere deeper.
|
|
if args.action in ("click", "type"):
|
|
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}")
|
|
|
|
# ── resolve and vet the text before touching anything ───────────────────
|
|
text = ""
|
|
if args.action == "type":
|
|
if args.stdin:
|
|
text = sys.stdin.read()
|
|
elif args.text is not None:
|
|
text = args.text
|
|
else:
|
|
print("error: type needs --text or --stdin", file=sys.stderr)
|
|
return 1
|
|
if not text:
|
|
print("error: nothing to type", file=sys.stderr)
|
|
return 1
|
|
|
|
import humanize as _h
|
|
bad = _h.untypeable(text)
|
|
if bad:
|
|
print(f"error: pyautogui cannot type {''.join(bad)!r} and would silently "
|
|
f"skip those characters, leaving a truncated value in the field.",
|
|
file=sys.stderr)
|
|
return 1
|
|
if "\n" in text and not args.allow_enter:
|
|
print("error: the text contains a newline, which presses Enter and may "
|
|
"submit the form. Pass --allow-enter if that is intended.", file=sys.stderr)
|
|
return 1
|
|
|
|
dash = Dashboard(args.api)
|
|
|
|
try:
|
|
if not dash.status().get("enabled"):
|
|
print("AutoBuyer capture is OFF — turn it on from the dashboard first.", file=sys.stderr)
|
|
return 2
|
|
except DashboardError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Locating {args.selector!r} …", flush=True)
|
|
|
|
# `locate` never touches the mouse, so it stays a plain lookup.
|
|
if args.action == "locate":
|
|
try:
|
|
found = dash.locate(args.selector, args.index, args.url, args.timeout, args.open_url)
|
|
except DashboardError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(describe(found, actions.scale_factor(found, args.scale)))
|
|
return 0
|
|
|
|
rng = random.Random(args.seed) if args.seed is not None else random.Random()
|
|
|
|
try:
|
|
actions.perform(
|
|
dash,
|
|
args.action,
|
|
args.selector,
|
|
index=args.index,
|
|
url=args.url,
|
|
open_url=args.open_url,
|
|
signed_out=args.signed_out,
|
|
text=text,
|
|
clear=args.clear,
|
|
scale=args.scale,
|
|
timeout=args.timeout,
|
|
rng=rng,
|
|
activate=not args.no_activate,
|
|
verify=not args.no_verify,
|
|
force=args.force,
|
|
dry_run=args.dry_run,
|
|
robotic=args.robotic,
|
|
on_located=lambda f: print(describe(f, actions.scale_factor(f, args.scale))),
|
|
report=lambda m: print(f" {m}"),
|
|
)
|
|
except DashboardError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
except actions.StepError as exc:
|
|
print(f"\nerror: {exc}", file=sys.stderr)
|
|
return exc.code
|
|
|
|
if args.dry_run:
|
|
print("\ndry run — cursor moved, nothing pressed or typed.")
|
|
elif args.action == "click":
|
|
print("\nclicked.")
|
|
else:
|
|
shown = text if len(text) <= 60 else text[:57] + "\u2026"
|
|
print(f"\ntyped {shown!r}" + (" (field cleared first)" if args.clear else ""))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|