Add automation framework, typing, and runner for the autobuyer
Turns the autobuyer from a page scraper into something that acts. A dashboard button queues a run; a desktop process executes it against the real browser. lib/automations.ts — automations are declarative step lists nested inside the firm whose site they drive. Steps are click / type / wait / navigate, and they inherit the firm's tab pattern and URL, so one firm's automation can't act on another's tab. Adding a button means adding an entry here; the page renders buttons from the API and the runner receives steps from the server, so neither needs editing. Runs key on firm:automation — every firm will plausibly have its own "buy-accounts", and a bare id would resolve to the wrong one. clicker/runner.py — the daemon behind the buttons. Claims a queued run, works through the steps, reports each one back for the page's live log. Only one run executes at a time: two processes driving one physical mouse would interleave clicks. Heartbeats on its own thread, because a step can block for tens of seconds and folding the beat into the main loop would show the runner as offline in the middle of the run it was executing. clicker/actions.py — one implementation of the safety checks, shared by the CLI and the runner. Refuses to act when the element is covered by an overlay, when coordinates fall off-screen, when the browser can't be confirmed frontmost, or (for type) when the target isn't an editable field. Typing: uneven human cadence, and the field is read back afterwards and compared against what was typed — a field that never took focus fails silently and looks identical to success otherwise. Non-ASCII is rejected because pyautogui skips those characters without complaint, and newlines because Enter may submit the form. Typos are deliberately not simulated: a mistyped digit in a trading form is a real loss, and the correction is the part that can go wrong. Extension: opens the firm's page when no tab matches, navigates to a specific page for a navigate step (skipped when already there, so page state survives), and retries the locate while a freshly loaded React app mounts — `complete` only means the document loaded. Staleness reporting, after it cost three debugging rounds: Chrome doesn't reload an unpacked extension and Python doesn't reload a running process, so both now report their version. A stale runner gets a red banner naming both versions and the automation buttons are disabled, rather than failing mid-run on a step type it predates. Scale detection is now conservative: a raw OS/browser width ratio is only trusted when it lands on a real scaling factor. On this multi-monitor desktop the previous logic would have silently halved every coordinate. Verified end to end against the live browser: navigate, locate, and a real click (run #12, all three steps). API round-trips, claim-once semantics, run cancellation, the heartbeat online/offline lifecycle, motion geometry and timing, focus activation, and typing verification all pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
54221bbc0c
commit
b748f95372
+101
-80
@@ -27,6 +27,7 @@ import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
import actions
|
||||
import focus
|
||||
|
||||
DEFAULT_API = "http://localhost:3000"
|
||||
@@ -64,12 +65,18 @@ class Dashboard:
|
||||
def status(self) -> dict:
|
||||
return self._request("/api/autobuyer/status")
|
||||
|
||||
def locate(self, selector: str, index: int, url_pattern: str, timeout: float) -> dict:
|
||||
"""Queue a lookup and block until the extension answers it."""
|
||||
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},
|
||||
{"selector": selector, "index": index, "urlPattern": url_pattern,
|
||||
"openUrl": open_url, "navigateUrl": navigate_url},
|
||||
)
|
||||
request_id = queued["id"]
|
||||
|
||||
@@ -88,28 +95,6 @@ class Dashboard:
|
||||
)
|
||||
|
||||
|
||||
def scale_factor(found: dict, override: float | None) -> float:
|
||||
"""CSS pixels and desktop pixels are the same on macOS and on unscaled Windows,
|
||||
but Windows display scaling and some Linux setups break that. Compare the
|
||||
screen size the browser reports against the one the OS reports."""
|
||||
if override is not None:
|
||||
return override
|
||||
|
||||
try:
|
||||
import pyautogui
|
||||
except ImportError:
|
||||
return 1.0 # `locate` is useful without the mouse library installed
|
||||
|
||||
os_width = pyautogui.size().width
|
||||
css_width = (found.get("screenSize") or {}).get("width")
|
||||
if not css_width:
|
||||
return 1.0
|
||||
ratio = os_width / css_width
|
||||
# Only trust a clean-ish ratio; anything odd means a multi-monitor layout we
|
||||
# shouldn't guess at, so fall back to 1:1 and let --scale override.
|
||||
return ratio if 0.4 < ratio < 4.0 else 1.0
|
||||
|
||||
|
||||
def describe(found: dict, factor: float) -> str:
|
||||
x, y = found["screen"]["x"] * factor, found["screen"]["y"] * factor
|
||||
lines = [
|
||||
@@ -127,13 +112,15 @@ def describe(found: dict, factor: float) -> str:
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument("action", choices=["locate", "click"],
|
||||
help="locate = measure only; click = measure then click")
|
||||
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("--api", default=DEFAULT_API, help=f"dashboard URL (default {DEFAULT_API})")
|
||||
parser.add_argument("--timeout", type=float, default=20.0, help="seconds to wait for the extension (default 20)")
|
||||
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")
|
||||
@@ -143,79 +130,113 @@ def main() -> int:
|
||||
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
|
||||
|
||||
# ── 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
|
||||
|
||||
print(f"Locating {args.selector!r} …", flush=True)
|
||||
found = dash.locate(args.selector, args.index, args.url, args.timeout)
|
||||
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":
|
||||
print(describe(found, scale_factor(found, args.scale)))
|
||||
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
|
||||
|
||||
if found.get("covered") and not args.force:
|
||||
print(describe(found, 1.0))
|
||||
print("\nRefusing to click: the element is covered — the click would hit "
|
||||
f"{found['coveredBy']} instead. Pass --force to click anyway.", file=sys.stderr)
|
||||
return 3
|
||||
rng = random.Random(args.seed) if args.seed is not None else random.Random()
|
||||
|
||||
try:
|
||||
import pyautogui
|
||||
except ImportError:
|
||||
print("error: pyautogui is not installed — run: pip install -r requirements.txt", file=sys.stderr)
|
||||
actions.perform(
|
||||
dash,
|
||||
args.action,
|
||||
args.selector,
|
||||
index=args.index,
|
||||
url=args.url,
|
||||
open_url=args.open_url,
|
||||
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
|
||||
|
||||
# Slamming the cursor into a screen corner aborts the script.
|
||||
pyautogui.FAILSAFE = True
|
||||
|
||||
factor = scale_factor(found, args.scale)
|
||||
x = found["screen"]["x"] * factor
|
||||
y = found["screen"]["y"] * factor
|
||||
|
||||
screen_w, screen_h = pyautogui.size()
|
||||
if not (0 <= x < screen_w and 0 <= y < screen_h):
|
||||
print(describe(found, factor))
|
||||
print(f"\nerror: target ({x:.0f}, {y:.0f}) is off-screen ({screen_w}x{screen_h}). "
|
||||
f"Is the Chrome window partly off the display, or on a second monitor?", file=sys.stderr)
|
||||
return 3
|
||||
|
||||
print(describe(found, factor))
|
||||
|
||||
# A click on a background window is eaten by the window manager activating it,
|
||||
# so the first attempt silently does nothing. Raise the browser first, and do it
|
||||
# here rather than before locating: the measurement takes seconds, and focus is
|
||||
# only required at the moment of the press.
|
||||
if not args.no_activate:
|
||||
focused = focus.ensure_frontmost()
|
||||
print(f" focus {focused.detail}")
|
||||
if not focused.ok:
|
||||
print("\nRefusing to click: the browser is not frontmost, so the click "
|
||||
"would be consumed activating its window instead of pressing the "
|
||||
"element. Pass --no-activate to override.", file=sys.stderr)
|
||||
return 3
|
||||
|
||||
if args.robotic:
|
||||
pyautogui.moveTo(x, y, duration=0.25)
|
||||
if not args.dry_run:
|
||||
pyautogui.click()
|
||||
else:
|
||||
import humanize
|
||||
rng = random.Random(args.seed) if args.seed is not None else random.Random()
|
||||
humanize.click(x, y, rng=rng, press=not args.dry_run)
|
||||
except actions.StepError as exc:
|
||||
print(f"\nerror: {exc}", file=sys.stderr)
|
||||
return exc.code
|
||||
|
||||
if args.dry_run:
|
||||
print("\ndry run — cursor moved, no click sent.")
|
||||
else:
|
||||
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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user