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>
334 lines
14 KiB
Python
334 lines
14 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
|
|
|
|
|
|
def force_utf8_output() -> None:
|
|
"""Make stdout/stderr accept the status glyphs this tool prints.
|
|
|
|
Windows picks the console code page for stdout, and under PM2 (where stdout
|
|
is a pipe, not a console) that is cp1252. It can encode em dashes but not
|
|
the run markers, so the first one raised UnicodeEncodeError mid-run and the
|
|
runner reported it as a step failure. errors="replace" is a backstop for any
|
|
stream that cannot be reconfigured at all.
|
|
"""
|
|
for stream in (sys.stdout, sys.stderr):
|
|
try:
|
|
stream.reconfigure(encoding="utf-8", errors="replace")
|
|
except (AttributeError, ValueError):
|
|
pass
|
|
|
|
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 = "",
|
|
options: dict | None = None) -> 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,
|
|
"options": options or {}},
|
|
)
|
|
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:
|
|
force_utf8_output()
|
|
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())
|