autofirmer-expanded/clicker/clicker.py

#!/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 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 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 _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:
            body = exc.read().decode(errors="replace")
            try:
                message = json.loads(body).get("error", body)
            except json.JSONDecodeError:
                message = body
            raise DashboardError(f"{method} {path} -> HTTP {exc.code}: {message}") from None
        except urllib.error.URLError as exc:
            raise DashboardError(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
        while time.monotonic() < deadline:
            row = self._request(f"/api/autobuyer/locate?id={request_id}")
            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)

        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())