К содержимому
Akmelo

Документация

Страница Cloudflare: проверить полученное содержимое (Python)

Проверить доступность контрольного содержимого демонстрационной страницы ScrapingCourse. Результатом будет одна запись о проверке, а не каталог товаров.

Выберите язык Python, вставьте полный исходник из этой статьи и задайте параметры запуска ниже. Источник для всей подборки: ScrapingCourse. Разбор относится к указанной демонстрационной странице.

Настройки примера

{
  "path": "/cloudflare-challenge",
  "mode": "challenge",
  "expected": 1
}

path выбирает страницу, mode выбирает ветку алгоритма, expected задаёт контрольное число записей для этого примера. Значение expected не является лимитом тарифа. Если набор данных источника меняется, сверяйте ожидаемое количество с ним.

Как работает алгоритм

  1. Выбираем /cloudflare-challenge и режим challenge. Этот режим выделен отдельно от сбора карточек.
  2. Открываем страницу браузерным SDK с текущим значением auto_captcha. Исходник занимается проверкой полученного содержимого.
  3. Читаем текст body через DOM-инструменты. Повторяющиеся пробелы и переводы строк сводим к одному пробелу.
  4. Ищем контрольную фразу демонстрационной страницы регулярным выражением: варианты с bypassed или Congratulations без учёта регистра.
  5. Только при совпадении сохраняем name с путём страницы и passed: true. Ключом записи становится тот же путь.
  6. Закрываем браузер в finally. Обычный сбор карточек и сравнение числа товаров для режима challenge не выполняются.

На что обратить внимание

Успех определяется контрольным текстом именно этой страницы. expected: 1 описывает одну запись результата; число товаров здесь не подсчитывается.

Особенности Python

В Python SDK работает последовательно: async и await в этом примере не нужны. Имена методов записаны через подчёркивание: without_cache, wait_for, wait_for_response. Вложенная collect использует общий set с ключами и возвращает число карточек в текущем HTML. В finally вызывается page.close().

Что получится

Результат этой ветки: одна запись с полями name и passed. name равен /cloudflare-challenge, passed равен true только после совпадения контрольного текста. Отдельная проверка seen в конце для режима challenge пропускается.

Полный код сценария

Исходник общий для всей подборки. Для этого примера в нём выполняется ветка challenge, выбранная параметрами выше. Блок можно целиком скопировать в редактор Python; реквизиты доступа и ключи сервисов в него добавлять не требуется.

import re


def run(ctx):
    origin = "https://www.scrapingcourse.com"
    path = str(ctx.param("path", "/pagination"))
    mode = str(ctx.param("mode", "http"))
    auto = ctx.param("auto_captcha", True) is not False
    seen = set()

    def collect(html):
        dom = ctx.dom(html)
        for card in dom.qa(".product-item, li.product"):
            title = card.q(".product-name, .woocommerce-loop-product__title")
            name = title.text().strip() if title else ""
            link = card.q("a")
            key = (link.attr("href") if link else "") or name
            if not name or key in seen:
                continue
            price = card.q(".product-price, .price")
            image = card.q("img")
            seen.add(key)
            ctx.emit({"name": name, "price": ctx.text().money(price.text() if price else ""),
                      "image": image.attr("src") if image else ""}, key)
        return len(dom.qa(".product-item, li.product"))

    if mode == "http":
        url = origin + path
        visited = set()
        while url and len(visited) < 50:
            if url in visited:
                raise ValueError("Пагинация вернула уже обработанную страницу")
            visited.add(url)
            response = ctx.http().without_cache().get(url, auto)
            if not response.ok():
                raise ValueError("Источник не открылся: " + str(response.status))
            html = response.body
            collect(html)
            next_url = ctx.dom(html).attr('a[rel="next"], a.next.page-numbers', "href")
            url = ctx.url().absolutize(url, next_url) if next_url else None
        if url:
            raise ValueError("Превышен защитный предел пагинации")
    else:
        page = ctx.browser().open(origin + path, ["*/ajax/products*"], {}, auto)
        try:
            if mode == "login":
                dom = ctx.dom(page.content())
                hint = next((p.text() for p in dom.qa("p") if "Demo Credentials:" in p.text()), "")
                password = re.search(r"Password:\s*([^\s|]+)", hint, re.I)
                email_match = re.search(r"Email:\s*([^\s|]+@[^\s|]+)", hint, re.I)
                email = email_match.group(1) if email_match else ""
                encoded = dom.attr("[data-cfemail]", "data-cfemail")
                if not email and encoded and len(encoded) > 2:
                    key = int(encoded[:2], 16)
                    email = "".join(chr(int(encoded[i:i + 2], 16) ^ key) for i in range(2, len(encoded), 2))
                if not email or not password:
                    raise ValueError("Страница не содержит учебных реквизитов")
                page.type("#email", email)
                page.type("#password", password.group(1))
                page.click('button[type="submit"]')
            if mode == "challenge":
                text = re.sub(r"\s+", " ", ctx.dom(page.content()).text("body") or "")
                if not re.search(r"You (?:successfully )?bypassed|successfully bypassed|Congratulations", text, re.I):
                    raise ValueError("Защищённое содержимое не получено")
                ctx.emit({"name": path, "passed": True}, path)
            else:
                page.wait_for(".product-name:not(:empty)", 15000)
                if mode == "button":
                    complete = False
                    for step in range(30):
                        html = page.content()
                        before = ctx.dom(html)
                        collect(html)
                        button = before.q("#load-more-btn")
                        if not button or re.search(r"display:\s*none", button.attr("style") or ""):
                            complete = True
                            break
                        offset = button.attr("data-offset")
                        count = len(before.qa(".product-item"))
                        page.click("#load-more-btn")
                        response = page.wait_for_response("*offset=" + offset, 15000)
                        if not response or not 200 <= response["status"] < 300 or response["truncated"] or response["error"]:
                            raise ValueError("Не удалось загрузить следующую порцию")
                        if not (response.get("body") or "").strip():
                            complete = True
                            break
                        page.wait_for(".product-item:nth-child(" + str(count + 1) + ")")
                    if not complete:
                        raise ValueError("Превышен защитный предел загрузки")
                if mode == "scroll":
                    stable, previous = 0, 0
                    for step in range(40):
                        page.scroll(3, 400)
                        count = collect(page.content())
                        stable = stable + 1 if count == previous else 0
                        previous = count
                        if stable >= 3:
                            break
                    if stable < 3:
                        raise ValueError("Лента продолжает загружаться: увеличьте предел прокрутки")
                collect(page.content())
        finally:
            page.close()
    expected = int(ctx.param("expected", 1))
    if mode != "challenge" and len(seen) != expected:
        raise ValueError("Неполный результат: " + str(len(seen)) + " вместо " + str(expected))

Куда посмотреть дальше

Общий алгоритм и параметры · HTTP-запросы · Действия браузера · Поиск элементов HTML