Как решать капчи в Jupyter без ошибок asyncio.run

Чтобы решать капчи в Jupyter-ноутбуке, установите пакет capskip через %pip, чтобы он попал в собственный Python ядра. Затем вызывайте CapSkip из ячейки так же, как в скрипте, или используйте await с его клиентом AsyncCapSkip прямо на верхнем уровне ячейки. Чего делать нельзя, так это оборачивать асинхронный вызов в asyncio.run(). В ядре уже работает цикл событий, поэтому asyncio.run выбрасывает RuntimeError, и по той же причине отказывается запускаться синхронный API Playwright. У работы с капчей в Jupyter Notebook есть ещё две собственные ловушки: кнопка остановки не всегда может остановить решение, а в Windows старый ipykernel не даёт Playwright запустить браузер. В этом руководстве разобраны установка, оба клиента, браузеры и удалённые ядра; всё проверено на ipykernel 7.4 и 7.3.

Что понадобится

• CapSkip, запущенный на машине с Windows. Примеры решают reCAPTCHA v2, а все остальные типы, которые поддерживает SDK, работают в ячейке точно так же.
• Python 3.10 или новее и актуальная версия пакета capskip (на момент написания 1.3.0), а также requests и Beautiful Soup для работы со страницей. Playwright нужен, только если вам требуется браузер.
• Jupyter в любом виде: Notebook 7, JupyterLab или редактор ноутбуков в VS Code. Все три выполняют ваши ячейки на ipykernel, и именно от него зависит всё описанное ниже поведение.
• Адрес решателя. Режим Local отвечает на 127.0.0.1 и только для этого устройства; режим Server слушает ваш сетевой адрес или публичный IP, чтобы ядро на другой машине могло обращаться к нему через API. Оба режима задаются в разделе Настройки подключения, а когда нужен режим Server, объясняет последний раздел.

Шаг 1: устанавливаем пакет в Python ядра через %pip

Запустите установку из ячейки ноутбука, со знаком процента:

# pip install capskip requests beautifulsoup4, run inside the notebook:
%pip install capskip requests beautifulsoup4

Знак процента важен. Встроенная в IPython магическая команда %pip запускает pip с собственным интерпретатором ядра, sys.executable, поэтому пакет попадает туда, откуда импортируют ваши ячейки. Восклицательный знак, напротив, передаёт строку оболочке, а оболочка запускает тот pip, который первым найдётся в PATH. На машине с несколькими Python эти два pip относятся к разным установкам. В одном тесте %pip в ядре показал capskip 1.3.0, а pip оболочки из глобального Python 3.14 показал старую версию 1.0.1 без новых типов капчи. Прежде чем двигаться дальше, проверьте, какой Python и какая версия вам достались:

import sys
import capskip

print(sys.executable)        # the Python this kernel runs on
print(capskip.__version__)   # 1.3.0 or later

Если пакет уже был импортирован, когда вы его обновили, перезапустите ядро, чтобы загрузилась новая версия. IPython выводит такое же напоминание после %pip.

Шаг 2: решаем капчу в ячейке синхронным клиентом

Синхронному клиенту в ноутбуке не нужно ничего особенного. Загрузите страницу через requests.Session, найдите виджет и решите капчу:

import requests
from bs4 import BeautifulSoup
from capskip import CapSkip

PAGE_URL = "https://example.com/login"

# pollingInterval caps the wait between polls. At 1 instead of the
# default 5, the stop button takes effect within about a second.
solver = CapSkip(host="127.0.0.1", port=8080, pollingInterval=1)

session = requests.Session()
page = session.get(PAGE_URL, timeout=30)
soup = BeautifulSoup(page.text, "html.parser")
widget = soup.select_one(".g-recaptcha[data-sitekey]")
form = widget.find_parent("form")

token = solver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)["code"]
len(token)

Затем отправьте токен в следующей ячейке вместе с собственными полями формы и через ту же сессию:

from urllib.parse import urljoin

def form_fields(form):
    """The fields a browser would send: hidden, text and checked boxes."""
    fields = {}
    for tag in form.select("input[name]"):
        kind = (tag.get("type") or "text").lower()
        if kind in ("submit", "button", "image", "file", "reset"):
            continue
        if kind in ("checkbox", "radio") and not tag.has_attr("checked"):
            continue
        fields[tag["name"]] = tag.get("value", "on" if kind in ("checkbox", "radio") else "")
    return fields

data = form_fields(form)
data["username"] = "YOUR_USERNAME"
data["g-recaptcha-response"] = token
res = session.post(urljoin(PAGE_URL, form.get("action") or PAGE_URL), data=data,
                   headers={"Referer": PAGE_URL}, timeout=30)
res.status_code

Сессия переносит cookies страницы в отправку, и именно это привязывает скрытое поле CSRF к вашему визиту. Заголовок Referer заменяет тот, что отправляет браузер: requests его не отправляет, а некоторые фреймворки, в том числе Django, отклоняют отправку формы по HTTPS без Referer или Origin. Токен reCAPTCHA годится для одной отправки и живёт около двух минут, поэтому запускайте ячейку отправки сразу после ячейки решения, а не возвращайтесь к ней позже.

Именно с кнопкой остановки ноутбук начинает отличаться от скрипта. Прерывание выбрасывает KeyboardInterrupt в вашей ячейке, но в ядре на Windows оно срабатывает только тогда, когда текущий вызов вернёт управление. Большую часть решения SDK спит между опросами, поэтому ячейка останавливается, когда заканчивается этот сон: до пяти секунд при pollingInterval по умолчанию и около одной секунды с настройкой выше. Трассировка выглядит странно: сначала идёт NetworkException, а KeyboardInterrupt возникает во время его обработки. Такое NetworkException служит всего лишь сигналом SDK о том, что ответ ещё не готов.

Плохой случай: CapSkip, который принимает соединение и никогда не отвечает. Синхронный клиент не задаёт таймаут для своих HTTP-вызовов, поэтому в Windows ячейка застревает в чтении из сокета, до которого не доходит ни одно прерывание, и recaptchaTimeout не помогает, потому что SDK проверяет его только между опросами. Перезапуск ядра освобождает ячейку, но все переменные теряются. Закрытие CapSkip тоже освобождает её и сохраняет переменные, потому что разорванное соединение завершает чтение, и уже отправленное прерывание срабатывает сразу. Кроме того, прерывание останавливает только вашу ячейку, но никогда не CapSkip: в SDK нет вызова для отмены задания, поэтому брошенное решение продолжается, а его токен так никто и не прочитает.

Шаг 3: асинхронный код в ячейке: await на верхнем уровне и никогда asyncio.run

Обычный способ запустить асинхронный код из скрипта в ноутбуке не работает:

import asyncio
from capskip import AsyncCapSkip

asolver = AsyncCapSkip(host="127.0.0.1", port=8080, pollingInterval=1)

asyncio.run(asolver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL))
# RuntimeError: asyncio.run() cannot be called from a running event loop

ipykernel выполняет каждую ячейку внутри цикла событий, который он уже запустил, а asyncio.run настаивает на запуске нового. Старая идиома asyncio.get_event_loop().run_until_complete() тоже не работает и выдаёт RuntimeError: This event loop is already running. Ни один из этих вызовов не доходит до CapSkip, так что задание не отправляется. Переданная вами корутина так и не запускается, поэтому в одной из следующих ячеек может появиться RuntimeWarning о том, что для неё так и не был вызван await; это безобидно. Решение проще, чем проблема. IPython позволяет использовать await на верхнем уровне ячейки, вне какой-либо функции:

# No asyncio.run: the kernel's loop runs this await.
from capskip import AsyncCapSkip

asolver = AsyncCapSkip(host="127.0.0.1", port=8080, pollingInterval=1)
result = await asolver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)
token = result["code"]

Поэтому для ноутбука лучше подходит AsyncCapSkip, и дело не только в стиле: причин две. Прерывание сразу отменяет await, поэтому ячейка завершается немедленно, а не после следующего опроса. А его HTTP-вызовы идут через httpx, который по умолчанию сдаётся на молчащем соединении через пять секунд. Тогда если CapSkip перестанет отвечать, вызов завершится ошибкой SDK, а не зависнет: TimeoutException вскоре после истечения recaptchaTimeout, если завис опрос, или NetworkException через пять секунд, если зависла отправка задания. Кнопка остановки при этом всё время работает. Чтобы сократить это ожидание, задайте recaptchaTimeout в конструкторе; recaptcha() в Python SDK не принимает аргумент timeout, и его передача вызывает TypeError.

Чтобы запустить несколько решений одновременно, достаточно gather. По умолчанию CapSkip решает reCAPTCHA в 10 потоков (настройка Max. Threads), а задание, стоящее в очереди за ними, ждёт до 250 секунд, прежде чем CapSkip признает его неудачным, поэтому ограничьте размер пакета:

import asyncio

# Max. Threads for reCAPTCHA is 10 by default; more than that queues.
limit = asyncio.Semaphore(10)

async def solve(sitekey, url):
    async with limit:
        return (await asolver.recaptcha(sitekey=sitekey, url=url))["code"]

# jobs: (sitekey, url) pairs; here the Step 2 page three times
jobs = [(widget["data-sitekey"], PAGE_URL)] * 3
tokens = await asyncio.gather(*(solve(k, u) for k, u in jobs), return_exceptions=True)

gather возвращает все токены в конце, поэтому отправляйте их без промедления или перейдите на asyncio.as_completed и отправляйте каждый по мере поступления. С return_exceptions=True неудачное решение возвращается в списке в виде своего исключения, а не выбрасывается из ячейки, поэтому отправляйте только те элементы, которые являются строками. Решения в пакете делят одну машину с CapSkip, так что каждое будет идти медленнее одиночного решения. Кроме того, клиент создаёт новый HTTP-клиент для каждого опроса, а это около десятой доли секунды процессорного времени ядра Jupyter на каждый опрос в Windows, так что большой пакет держит занятым одно ядро процессора. Приёмы для больших пакетов описаны в руководстве по параллельному решению капчи на Python.

Старые ответы на ошибку asyncio.run советуют nest_asyncio. Не используйте его. Проект был заархивирован в марте 2024 года. В Python 3.14 его применение ломает ядро: начиная со следующей ячейки каждый вызов httpx, включая AsyncCapSkip, падает с TypeError о слабой ссылке на объект NoneType, пока вы не перезапустите ядро. Для await верхнего уровня никакие патчи вообще не нужны.

Шаг 4: браузер в ноутбуке

У Selenium нет собственного цикла событий, поэтому сессия webdriver работает в ячейке точно так же, как в скрипте: решите капчу синхронным клиентом из шага 2 и запишите токен в страницу. Синхронный API Playwright проверяет, не запущен ли цикл, и отказывается работать с ошибкой Error: It looks like you are using Playwright Sync API inside the asyncio loop. Please use the Async API instead. Поэтому в ноутбуке Playwright нужно управлять через его асинхронный API, с await на верхнем уровне:

# pip install playwright into the kernel, then download a browser once:
%pip install playwright
import sys
!"{sys.executable}" -m playwright install chromium

from playwright.async_api import async_playwright

# start() instead of "async with", so the browser outlives this cell.
pw = await async_playwright().start()
browser = await pw.chromium.launch()
page = await browser.new_page()
await page.goto(PAGE_URL)

sitekey = await page.locator(".g-recaptcha").get_attribute("data-sitekey")
token = (await asolver.recaptcha(sitekey=sitekey, url=page.url))["code"]

# The widget's hidden textarea is the field the form submits.
await page.locator("#g-recaptcha-response").evaluate("(el, t) => { el.value = t; }", token)
await page.locator("form button[type=submit]").click()

Если хранить pw и browser в переменных, со страницей можно экспериментировать из следующих ячеек, а ради этого ноутбук во многом и нужен. Когда закончите, закройте оба через await browser.close() и await pw.stop(). Вызов собственного sys.executable ядра подбирает браузер под тот Playwright, который импортируют ваши ячейки, по той же причине, по которой это делает %pip. Кавычки сохраняют путь с пробелами целым, а строка с восклицательным знаком показывает прогресс загрузки в ячейке. Подробнее об управлении самим браузером рассказано в руководстве по решению капчи в Playwright.

В Windows проверьте версию ipykernel, прежде чем отлаживать Playwright. До версии 7.3 включительно ipykernel выполнял там ячейки на цикле событий selector, который не умеет запускать подпроцессы asyncio, поэтому chromium.launch() падает с голым NotImplementedError. Версия 7.4.0, выпущенная в конце сентября 2026 года, перевела ячейки в Windows на цикл Proactor, и та же ячейка работает. Выведите ipykernel.__version__, чтобы узнать, какая версия у вас. Чтобы обновиться, выполните %pip install -U ipykernel в ноутбуке и перезапустите ядро. Если обновиться нельзя, фоновый поток не поможет, потому что политика selector действует на весь процесс ядра. Вместо этого управляйте браузером через Selenium, который запускает chromedriver без asyncio.

Когда ядро не на вашем ПК: режим Server

Всё сказанное выше предполагает, что ядро работает на той же машине, что и CapSkip, поэтому 127.0.0.1 до него достаёт. Это обычная схема решения капчи в Jupyter Notebook, но не единственная. Ноутбук, который обслуживает JupyterHub на сервере, удалённое ядро в VS Code или облачный ноутбук вроде Google Colab выполняют ваши ячейки где-то ещё, и там 127.0.0.1 означает ту машину. Тогда первый же вызов падает с NetworkException и отказом в соединении. Переключите CapSkip в режим Server в настройках подключения, и он будет слушать ваш сетевой адрес или публичный IP, так что удалённое ядро сможет обращаться к нему через тот же API, указав этот адрес как host.

Если маршрут идёт через интернет, используйте статический публичный IP. При запуске режима Server CapSkip открывает свой порт для всех адресов, поэтому, чтобы доступ был только у ожидаемых машин, добавьте для остальных блокирующее правило Windows Firewall. Такой подход работает, когда адрес ядра фиксирован, как на вашем собственном сервере JupyterHub. Исходящий адрес облачного ноутбука может меняться от сессии к сессии, поэтому в этом случае включите API key validation в настройках CapSkip и передавайте ключ как apiKey. Это по-прежнему ваше собственное железо, и по-прежнему без платы за каждое решение. Клиент сам не читает переменные окружения, поэтому читайте CAPSKIP_HOST через os.getenv в ячейке и передавайте его клиенту, как это делает полный пример.

Полный рабочий пример

Три ячейки, всё асинхронно: самый короткий полный рабочий процесс решения капчи в Jupyter Notebook, который переживает нажатие кнопки остановки.

# Cell 1: pip install capskip requests beautifulsoup4, then restart the kernel.
%pip install capskip requests beautifulsoup4

# Cell 2: the client and a helper. Run once per kernel.
import os
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from capskip import AsyncCapSkip, CapSkipError

PAGE_URL = "https://example.com/login"
solver = AsyncCapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
    pollingInterval=1,
)

def form_fields(form):
    fields = {}
    for tag in form.select("input[name]"):
        kind = (tag.get("type") or "text").lower()
        if kind in ("submit", "button", "image", "file", "reset"):
            continue
        if kind in ("checkbox", "radio") and not tag.has_attr("checked"):
            continue
        fields[tag["name"]] = tag.get("value", "on" if kind in ("checkbox", "radio") else "")
    return fields

# Cell 3: fetch, solve, post. Run it again for a fresh token.
session = requests.Session()
page = session.get(PAGE_URL, timeout=30)
soup = BeautifulSoup(page.text, "html.parser")
widget = soup.select_one(".g-recaptcha[data-sitekey]")
if widget is None:
    raise RuntimeError("No reCAPTCHA widget in the HTML; the page may build it with JavaScript.")
form = widget.find_parent("form")

try:
    result = await solver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)
except CapSkipError as err:
    raise RuntimeError(f"solve failed: {type(err).__name__}: {err}") from err

data = form_fields(form)
data["username"] = "YOUR_USERNAME"
data["g-recaptcha-response"] = result["code"]
res = session.post(urljoin(PAGE_URL, form.get("action") or PAGE_URL), data=data,
                   headers={"Referer": PAGE_URL}, timeout=30)
print(res.status_code)

Именно ячейку 3 вы запускаете повторно, и каждый запуск означает новое решение и новый токен, потому что токен reCAPTCHA срабатывает один раз. Если страница строит свой виджет из JavaScript, в загруженном HTML нет sitekey; в этом случае используйте ячейки с Playwright из шага 4 или прочитайте sitekey из скрипта, который рендерит виджет.

Частые ошибки и что они означают

FAQ

Коротко

Решение капчи в Jupyter Notebook сводится к пяти привычкам. Устанавливайте через %pip и никогда через оболочку. Синхронный клиент CapSkip работает в ячейке так же, как в скрипте; задайте pollingInterval равным 1, чтобы прерывания срабатывали быстро. Для асинхронного кода используйте await с AsyncCapSkip на верхнем уровне ячейки и никогда не оборачивайте его в asyncio.run и не прибегайте к nest_asyncio. Для браузера используйте Selenium как есть или асинхронный API Playwright, а в Windows работайте на ipykernel 7.4 или новее. Когда ядро находится на другой машине, переключите CapSkip в режим Server и передайте его адрес.

• Все типы капчи, которые решает пакет для Python, описаны на странице сервиса распознавания капч для Python.
• Как устроен сам виджет и что покрывает решатель, описано на странице сервиса распознавания reCAPTCHA v2.

Именно в ноутбуке парсер собирают по частям, раз за разом перезапуская ячейки. Если обход капчи выполняется на вашей собственной машине, каждый такой повторный запуск бесплатен.