autofirmer-expanded/clicker/captcha.py

"""Solving a page's captcha through the 2captcha API.

The old flow clicked the 2captcha browser extension's own "solve" button and
waited for it to turn green. This replaces it: read the widget's parameters out
of the page, ask 2captcha for a token, put the token back into the page.

Three round trips, all through the dashboard:

    runner  --locate captchaInfo-->   extension reads the widget's sitekey
    runner  --/api/autobuyer/captcha->  dashboard asks 2captcha, polls, returns a token
    runner  --locate captchaToken-->  extension fills the field and fires the callback

The API key lives on the dashboard, not here — see app/api/autobuyer/captcha.
"""

import time

import actions
from clicker import DashboardError

# 2captcha's own guidance: nothing is ready sooner than this, so the first poll
# is a wasted round trip if it goes out immediately.
FIRST_POLL_SECONDS = 10.0
POLL_SECONDS = 5.0

# A "task expired"/"unsolvable" verdict means this attempt is dead, not that the
# page is unsolvable — a fresh task usually goes through. Bounded so a site that
# genuinely cannot be solved fails the run instead of burning the balance.
MAX_ATTEMPTS = 3

# Timeouts on the two page ops. Reading the widget is a plain injection; posting
# the token runs the page's own callback, which can kick off a network request.
PAGE_TIMEOUT = 30.0


def _page_op(dash, step, opts, options: dict, timeout: float) -> dict:
    """Run one captcha op in the page. The selector is unused by these ops, but
    the locate endpoint requires one, so `body` stands in for it."""
    return dash.locate(
        "body", 0,
        step.get("urlPattern", "") or opts.url,
        timeout,
        step.get("openUrl", "") or "",
        options=options,
    )


def read_widget(dash, step, opts, provider: str, deadline: float, report) -> dict:
    """Wait for a captcha to exist on the page, then describe it.

    Polls rather than reading once: the step that opens the modal returns as soon
    as the click lands, and the widget's script still has to load and render.
    """
    announced = False
    last = "no captcha widget found"

    while True:
        try:
            return _page_op(dash, step, opts, {
                "op": "captchaInfo",
                "provider": provider,
                "scopeSelector": step.get("selector", "") or "",
            }, PAGE_TIMEOUT)
        except DashboardError as exc:
            # Includes "no captcha widget on the page yet", which is the normal
            # answer while the widget is still loading.
            last = str(exc)

        if time.time() >= deadline:
            raise actions.StepError(f"no captcha to solve: {last}")
        if not announced:
            report("waiting for the captcha widget to render")
            announced = True
        time.sleep(2.0)


def _await_token(dash, task_id, deadline: float, report, cancelled) -> str | None:
    """Poll one 2captcha task. Returns the token, None if the task failed and a
    retry is worth it, or raises when the run is out of time."""
    time.sleep(min(FIRST_POLL_SECONDS, max(0.0, deadline - time.time())))

    while True:
        if cancelled and cancelled():
            raise actions.StepError("cancelled while waiting for the captcha")

        try:
            row = dash._request(f"/api/autobuyer/captcha?taskId={task_id}")
        except DashboardError as exc:
            # A blip talking to our own dashboard should not throw away a task
            # that 2captcha is already working on.
            report(f"(poll failed, retrying: {exc})")
            row = {"status": "processing"}

        if row.get("status") == "ready":
            return row["token"]
        if row.get("status") == "failed":
            report(f"2captcha gave up on task {task_id}: {row.get('error')}")
            return None

        if time.time() >= deadline:
            raise actions.StepError(f"2captcha did not return a token in time (task {task_id})")
        time.sleep(POLL_SECONDS)


def solve(dash, step, opts, report, cancelled=None) -> None:
    """Run one `solveCaptcha` step. Raises StepError if the page is never solved."""
    deadline = time.time() + float(step.get("timeoutSeconds", 180))
    provider = step.get("provider") or "auto"

    info = read_widget(dash, step, opts, provider, deadline, report)
    sitekey = info.get("sitekey") or ""
    report(f"{info.get('provider')} widget, sitekey {sitekey[:16]}… (via {info.get('source')})")

    started = time.time()
    token = None
    for attempt in range(1, MAX_ATTEMPTS + 1):
        if time.time() >= deadline:
            break

        queued = dash._request("/api/autobuyer/captcha", "POST", {
            "provider": info.get("provider"),
            "sitekey": sitekey,
            # The page's own URL, not the firm's configured one: 2captcha scores
            # the token against the page it was asked for.
            "pageurl": info.get("pageurl"),
            "action": info.get("action") or "",
            "data": info.get("data") or "",
            "pagedata": info.get("pagedata") or "",
            "invisible": bool(info.get("invisible")),
            "enterprise": bool(info.get("enterprise")),
            # The browser that will submit the form, so a proxied solve can claim
            # the same one rather than 2captcha's default.
            "userAgent": info.get("userAgent") or "",
        })
        task_id = queued.get("taskId")
        report(f"2captcha task {task_id} queued ({queued.get('type')} via {queued.get('via')})"
               + (f", attempt {attempt}" if attempt > 1 else ""))

        token = _await_token(dash, task_id, deadline, report, cancelled)
        if token:
            break

    if not token:
        raise actions.StepError("2captcha could not solve the captcha")

    report(f"token in hand after {time.time() - started:.0f}s ({len(token)} chars)")

    delivered = _page_op(dash, step, opts, {
        "op": "captchaToken",
        "provider": info.get("provider"),
        "token": token,
    }, PAGE_TIMEOUT)

    called = delivered.get("called")
    fields = delivered.get("fieldIds") or []
    keys = delivered.get("clientKeys") or []
    report(f"filled {delivered.get('filled', 0)} response field(s)"
           + (f" {fields}" if fields else "")
           + (f", fired {called}" if called else ""))
    # Printed whenever the page is carrying more than one widget, because that is
    # the state in which handing the token to the wrong one is possible at all —
    # and it looks identical to success right up until the button stays dark.
    if len(keys) > 1 or (delivered.get("candidates") or 0) > 1:
        report(f"page had widgets {keys or '?'}, live one looked like "
               f"{delivered.get('liveWidgetId')}")
    if not called:
        # The token is in the form, so a plain submit still works — but a page
        # that gates its submit button on the callback will look unsolved, and
        # that is worth saying out loud rather than failing three steps later.
        report("no captcha callback found — if the next step fails, that is why")