DataDome任务

通过使用同步接口创建任务 https://sync.ez-captcha.com/createSyncTask, 并直接获取结果

如果您得到了一个无效的Payload，请联系我们，我们将会为您解决和优化

任务类型

任务类型	描述	价格(美元)

任务类型

描述

价格(美元)

DataDomeTaskProxyless

DataDome 解决方案

step1: $0.1/k

step2: $2.5/k

支持能力

类型	支持情况

类型	支持情况
Slider 验证	✔ 支持
Interstitial 验证	✔ 支持

请求参数

参数	类型	是否必须	描述

参数	类型	是否必须	描述
clientKey	string	yes	你的clientkey
type	string	yes	DataDomeTaskProxyless
step	string	yes	1 = 取挑战地址; 2 = 取放行结果
referer	string	yes	step1 传页面 URL; step2 传 step1 返回的地址
html_b64	string	yes	HTML 文本的 base64
user_agent	string	no	step1 可不传，step2 必传。目前仅支持以下三个版本： `Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36` `Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36` `Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36`

注意事项：

目前仅支持的语言为：en-US;q=0.9

使用流程

EZ DataDome 处理流程分为两个步骤：

Step 1：提交初始页面（获取挑战）

接口

POST https://sync.ez-captcha.com/createSyncTask

请求参数

{
  "clientKey": "your_client_key",
  "task": {
    "type": "DataDomeTaskProxyless",
    "step": "1",                      # 1 = 取挑战地址; 2 = 取放行结果
    "html_b64": "<HTML 文本的 base64>",
    "referer": "https://target.com/page"
  }
}

参数说明

参数	说明

参数	说明
step=1	获取 DataDome challenge 地址
html_b64	页面 HTML（必须 base64 编码）
referer	当前访问页面 URL

返回结果

{
  "errorId": 0,
  "solution": "https://geo.captcha-delivery.com/captcha/xxx"
}

Step 2：提交挑战页面（获取解）

请求参数

{
  "clientKey": "your_client_key",
  "task": {
    "type": "DataDomeTaskProxyless",
    "step": "2",
    "html_b64": "<base64挑战页HTML>",
    "referer": "https://geo.captcha-delivery.com/captcha/xxx",
    "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
  }
}

返回结果

{
  "errorId": 0,
  "solution": {
    "kind": "slider", // 说明触发的是 Slider 还是 Interstitial 
    "url": "https://geo.captcha-delivery.com/validate/xxx",
    "body": null
  }
}

Step 3：提交验证结果

根据返回 kind 执行不同请求：

1. slider 类型

GET {solution.url}

2. interstitial 类型

POST {solution.url}
Content-Type: application/x-www-form-urlencoded

{solution.body}

返回结果

{
  "errorId": 0,
  "view": "redirect",
  "cookie": "datadome=xxxxx"
}

Cookie 使用说明

成功后会返回：

datadome=xxxxxxxxxxxx

使用方式：

写入当前 session
用于后续请求鉴权

Demo

# -*- coding: utf-8 -*-
"""
DataDome 求解服务 —— 滑块(slider)接入示例

依赖:
    pip install requests

使用:
    1) 填好下方 API_URL / CLIENT_KEY / USER_AGENT
    2) 在 PROXY 处配置住宅代理(滑块站点必须用干净住宅 IP, 且整个流程同一出口 IP)

    POST  {API_URL}
    Body(JSON):
    {
        "clientkey": "<你的密钥>",
        "task": {
            "type": "DataDomeTaskProxyless",
            "step": "1" 或 "2",          # 1 = 取挑战地址; 2 = 取放行结果
            "html_b64": "<HTML 文本的 base64>",
            "referer": "<step1 传页面 URL; step2 传 step1 返回的地址>",
            "user_agent": "<必须用ez指定的的固定 UA>"
        }
    }
    返回(JSON):成功 errorId=0(solution 为结果);失败 errorId!=0(见 errorCode / errorDescription)。
    滑块的 step2 结果 solution = {"url": 校验地址(GET), "body": null}。UA 一律用本地写死的 USER_AGENT。

重要约定:
    - user_agent 只能使用ez提供的【固定 UA】, 不要自行修改或更换, 否则无法通过。
    - 所有 HTML 必须先 base64 再放进 html_b64 字段。
    - 整个流程必须使用同一个会话(保留站点 cookie)与同一出口 IP。
"""

import base64
import json
import re
import time
import random

import requests

def generate_random_number(min_length=8, max_length=13):
    length = random.randint(min_length, max_length)

    first_digit = random.randint(1, 9)

    remaining_digits = [random.randint(0, 9) for _ in range(length - 1)]

    number = int(''.join(map(str, [first_digit] + remaining_digits)))

    return number


class Banned(Exception):
    """当前出口 IP/会话已被 DataDome 封禁(被 ban), 需更换代理/IP 后重试。"""


# ============================ 需要你填的配置 ============================

# 求解服务地址
API_URL = "https://sync.ez-captcha.com/createSyncTask"

# 你的密钥
CLIENT_KEY = "11b6106e35e0476890a95291a543657658"

# 固定 UA。
# 仅支持以下几类
# Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36
# Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36
# Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36
USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
)

# 代理(滑块站点必须用住宅代理)。例: "http://user:pass@host:port"
PROXY = "" + str(
        generate_random_number(8)) + ""

# xxx 触发 DataDome 的页面
TARGET_URL = "https://www.xxx.com/xxx/xxx"

# 发 check 前模拟"真人解题"停留秒数(太快会被判机器)
SOLVE_DELAY = 3

# ======================================================================


def _b64(text: str) -> str:
    """把 HTML 文本转成 base64。"""
    return base64.b64encode(text.encode("utf-8")).decode("ascii")


def _is_blocked(html: str) -> bool:
    """是否为 DataDome 封禁页(被 ban)。"""
    return "hardblock" in html


def _sec_ch_ua(ua: str) -> str:
    """根据固定 UA 生成匹配的 sec-ch-ua 头(版本必须与 UA 一致)。"""
    m = re.search(r"Chrome/(\d+)", ua)
    v = m.group(1) if m else "148"
    return f'"Chromium";v="{v}", "Google Chrome";v="{v}", "Not/A)Brand";v="99"'


def call_solver(step: int, html: str, referer: str) -> object:
    """
    调用求解服务。
        step    : 1 = 取挑战地址; 2 = 取放行结果
        html    : 当前页面/挑战页的 HTML 原文(函数内部会做 base64)
        referer : step1 传页面 URL; step2 传 step1 返回的地址
    返回:
        step1 -> 挑战地址字符串
        step2 -> dict: {"url", "body", "kind"};kind="slider"→GET url / "interstitial"→POST body
    """
    payload = {
        "clientKey": CLIENT_KEY,
        "task": {
            "type": "DataDomeTaskProxyless",
            "step": str(step),
            "html_b64": _b64(html),
            "referer": referer,
            "user_agent": USER_AGENT,
        },
    }
    resp = requests.post(API_URL, json=payload, timeout=60)
    data = resp.json()
    print(f"step{step} 求解服务返回: {data}")
    resp.raise_for_status()

    # 失败处理:errorId != 0 即失败, 原因在 errorCode / errorDescription。
    if data.get("errorId", 0) != 0:
        code = data.get("errorCode", "")
        desc = data.get("errorDescription", "")
        # 收到 blocked 表示该 IP/会话被 DataDome 封禁, 需更换代理/IP 重来
        if "blocked" in f"{code} {desc}".lower():
            raise Banned(desc or code or "blocked")
        raise RuntimeError(f"求解失败: {code} {desc}".strip())

    solution = data.get("solution")
    # 部分网关把 solution 以 JSON 字符串返回, 这里统一解开
    if isinstance(solution, str):
        try:
            solution = json.loads(solution)
        except json.JSONDecodeError:
            pass
    return solution


def _nav_headers(fetch_site: str) -> dict:
    """导航请求头(抓页面 / 复验用)。fetch_site: none=首访, same-origin=复验。"""
    return {
        "User-Agent": USER_AGENT,
        "Accept": ("text/html,application/xhtml+xml,application/xml;q=0.9,"
                   "image/avif,image/webp,image/apng,*/*;q=0.8"),
        "Accept-Language": "en-US;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "sec-ch-ua": _sec_ch_ua(USER_AGENT),
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "Upgrade-Insecure-Requests": "1",
        "Sec-Fetch-Site": fetch_site,
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-User": "?1",
        "Sec-Fetch-Dest": "document",
    }


def _challenge_headers(referer: str) -> dict:
    """抓滑块挑战页请求头(从页面里以 iframe 加载, 跨站)。"""
    return {
        "User-Agent": USER_AGENT,
        "Accept": ("text/html,application/xhtml+xml,application/xml;q=0.9,"
                   "image/avif,image/webp,image/apng,*/*;q=0.8"),
        "Accept-Language": "en-US;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "sec-ch-ua": _sec_ch_ua(USER_AGENT),
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "Sec-Fetch-Site": "cross-site",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Dest": "iframe",
        "Referer": referer,
    }


def _check_headers(referer: str, ua: str) -> dict:
    """校验请求头(在挑战页里 fetch /captcha/check, 同源 cors)。"""
    return {
        "User-Agent": ua,
        "Accept": "*/*",
        "Accept-Language": "en-US;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "sec-ch-ua": _sec_ch_ua(ua),
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "Sec-Fetch-Site": "same-origin",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Dest": "empty",
        "Referer": referer,
    }


def _post_headers(referer: str, ua: str) -> dict:
    """发包请求头(interstitial:把 payload POST 给 DataDome)。"""
    return {
        "User-Agent": ua,
        "Accept": "*/*",
        "Accept-Language": "en-US;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://geo.captcha-delivery.com",
        "Referer": referer,
        "sec-ch-ua": _sec_ch_ua(ua),
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "Sec-Fetch-Site": "same-origin",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Dest": "empty",
    }


def _submit(session, sol: dict, challenge_url: str, used_ua: str):
    """按 kind 把求解结果发给 DataDome, 返回 datadome cookie 串(未通过返回 None)。
       kind="slider" → GET 校验地址;"interstitial" → POST 表单 body。
       kind 缺失时按 body 是否为空兜底(服务端约定:body 为空即 slider GET)。"""
    kind = (sol.get("kind") or "").strip().lower()
    body = sol.get("body")
    submit_url = sol["url"]
    is_slider = (kind == "slider") if kind else (not body)
    if is_slider:
        res = session.get(
            submit_url, headers=_check_headers(challenge_url, used_ua), timeout=30
        ).json()
    else:
        res = session.post(
            submit_url, headers=_post_headers(challenge_url, used_ua),
            data=(body or "").encode("utf-8"), timeout=30,
        ).json()
    print(f"结果返回: {res}")
    # interstitial 放行 = view==redirect;slider 放行 = 回 cookie。两者放行都带 cookie。
    if not is_slider and res.get("view") != "redirect":
        print(f"未通过, 返回 view={res.get('view')}(可更换代理/IP 后重试)")
        return None
    cookie_field = res.get("cookie", "")
    if not cookie_field:
        print(f"未通过(未返回 cookie): {res}(可更换代理/IP 后重试)")
        return None
    return cookie_field


def main() -> None:
    session = requests.Session()
    if PROXY:
        session.proxies = {"http": PROXY, "https": PROXY}

    try:
        # 1) 抓页面(xxx), 期望 403 + 挑战脚本
        print("[1/5] 抓取页面 ...")
        html = session.get(TARGET_URL, headers=_nav_headers("none"), timeout=30).text
        if _is_blocked(html):                       # 初始页即封禁页 = 被 ban
            raise Banned("初始页面即为封禁页")
        if "var dd" not in html:
            print("页面未触发 DataDome(可能已放行, 或当前 IP 未被挑战), 无需处理。")
            return

        # 2) step1: 取滑块挑战地址
        print("[2/5] 取挑战地址 ...")
        challenge_url = call_solver(1, html, TARGET_URL)
        if not isinstance(challenge_url, str) or not challenge_url:
            raise RuntimeError(f"step1 返回异常: {challenge_url!r}")

        # 3) 抓滑块挑战页
        print("[3/5] 抓取挑战页 ...")
        challenge_html = session.get(
            challenge_url, headers=_challenge_headers(TARGET_URL), timeout=30
        ).text
        if _is_blocked(challenge_html):             # 挑战页是封禁页 = 被 ban
            raise Banned("挑战页为封禁页")

        # 4) step2: 求解, 拿放行结果(含 kind:slider=GET 校验地址 / interstitial=POST 表单 body)
        print("[4/5] 求解 ...")
        sol = call_solver(2, challenge_html, challenge_url)
        used_ua = USER_AGENT                 # UA 写死, 不依赖服务端返回

        # 模拟真人解题耗时, 再发送结果
        time.sleep(SOLVE_DELAY)

        # 5) 按 kind 发送结果(slider→GET 校验 / interstitial→POST 表单), 拿 datadome cookie
        print("[5/5] 发送结果 ...")
        cookie_field = _submit(session, sol, challenge_url, used_ua)
        if not cookie_field:
            return

        print("通过, 已获得 datadome cookie")

        # 6) 用新 datadome 复验 xxx。
        #    先清掉会话里旧的 datadome(首访 403 下发的旧挑战 cookie), 否则与新 cookie 冲突 -> 仍 403。
        #    站点自身的会话 cookie(由 Session 自动保存)保留, 复验时一并带回。
        new_dd = cookie_field.split(";")[0].split("=", 1)[-1]
        for c in list(session.cookies):
            if c.name == "datadome":
                session.cookies.clear(c.domain, c.path, c.name)
        session.cookies.set("datadome", new_dd, domain=".xxx.com", path="/")

        v = session.get(TARGET_URL, headers=_nav_headers("same-origin"), timeout=30)
        ok = v.status_code in (200, 302) and "var dd" not in v.text and not _is_blocked(v.text)
        print(f"复验 xxx: HTTP {v.status_code}, {'已放行' if ok else '仍被拦截'}")

    except Banned as e:
        # 被 ban:当前出口 IP/会话已被 DataDome 封禁, 换干净的住宅代理重试即可。
        print(f" 被封禁(banned): {e}。请更换代理/IP(住宅)后重试。")


if __name__ == "__main__":
    main()