用 WeaveBit API 批量检测 DNS 污染与中国大陆域名可达性

批量检查域名时，先明确你要回答的是“是否报告 DNS 污染”，还是“DNS 污染、HTTP Host、TLS SNI 三个维度是否出现异常”。WeaveBit API v2 支持这两种异步批量工作流：整理目标，预估费用，创建批次，再按批次 ID 查询和保存结果。它适合域名清单检查、发布前检查，以及接入自己的运维系统。

以下示例展示最小批量工作流。运行 /task/create 会创建可能产生费用的检测批次；先在自己的受控服务器配置环境变量，替换示例域名，并审核预估。完整字段以API 文档为准。

选择 DNS 批量或 GFW 全量批量

GFW 全量包含 DNS 污染、HTTP Host、TLS SNI 三个维度与 overall_result，不是完整浏览器加载，也不是全部七种检测方法。需要帮助选择时，阅读GFW 域名检测操作指南。

每次预估或创建接收 1–1000 个目标，target 是对象数组。域名应为不带协议、路径、查询参数、端口或空格的纯域名；提交前去重，避免同一目标被重复创建为独立任务。国际化域名会规范化为小写 ASCII Punycode，返回的 check_id 可用于关联自己的业务记录。

DNS 工作流请求体如下：

{
  "check_type": "dns_pollution_check",
  "target": [{"domain": "example.com"}, {"domain": "example.net"}]
}

GFW 全量工作流使用相同的域名数组，并增加节点：

{
  "check_type": "gfw_full_domain_check",
  "target": [{"domain": "example.com"}, {"domain": "example.net"}],
  "nodes": ["NODE_A_FROM_GET_NODES", "NODE_B_FROM_GET_NODES"]
}

节点占位符必须替换为 /task/get_nodes 当前返回的 node_id。该接口使用 POST 和空对象 {}，父节点还会返回 children[]。nodes 可为最多 100 个节点代码的数组或字符串 "all"；父节点会展开为当前可用子节点。不要把显示名称当代码，也不要在不知道费用影响时直接选择所有节点。

先预估，再创建

v2 Base URL 为 https://api.weavebit.com/v2，请求使用 POST、JSON 和 Authorization: Bearer。先将最终请求体发到 /task/estimate，审核返回的 data，尤其是目标数、节点数、计划数量、预估费用与可用余额，再把相同的检测参数发到 /task/create。增加目标、节点或更换方法后，应重新预估。

金额字段是十进制字符串。保存原值，进行账务计算时使用十进制类型，不用二进制浮点数。价格以当前价格页和实际预估为准，不把文章中的示例配置当成固定费用。

创建返回 batch_id、datas[].check_id、规范化目标、execution_deadline_at 和 expires_at 等信息。创建成功只是受理成功，不是检测已经完成。平台按有效返回结果结算；超时、节点错误及余额不足而未接收的结果不计费，重复查询不重复扣费。

可复制的 Python 双工作流示例

下面代码仅使用 Python 3 标准库。把它保存为自己的 batch_check.py，在每个逻辑批次的独立工作目录运行。通过环境变量提供 WEAVEBIT_API_KEY；GFW 模式还需要 WEAVEBIT_NODE_IDS，其值是当前有效节点代码的 JSON 数组。创建前设置并持久保存本次 WEAVEBIT_REQUEST_ID，审核后才将 WEAVEBIT_CONFIRM_CREATE 设置为 yes。不要把密钥写进代码、浏览器脚本、版本库或调试日志。

import json
import os
import sys
from pathlib import Path
from urllib.request import Request, urlopen

BASE = "https://api.weavebit.com/v2/"
KEY = os.environ["WEAVEBIT_API_KEY"]
STATE = Path("weavebit-batch.json")

def api(path, body, request_id=None):
    headers = {"Authorization": "Bearer " + KEY,
               "Content-Type": "application/json"}
    if request_id is not None:
        headers["Idempotency-Key"] = request_id
    request = Request(BASE + path, method="POST", headers=headers,
                      data=json.dumps(body).encode("utf-8"))
    with urlopen(request, timeout=30) as response:
        result = json.load(response)
    if result.get("code") != 0:
        raise RuntimeError("API code: " + str(result.get("code")))
    return result

def payload(mode):
    body = {"target": [{"domain": "example.com"},
                       {"domain": "example.net"}]}
    if mode == "dns":
        body["check_type"] = "dns_pollution_check"
    elif mode == "full":
        body["check_type"] = "gfw_full_domain_check"
        body["nodes"] = json.loads(os.environ["WEAVEBIT_NODE_IDS"])
    else:
        raise ValueError("Choose dns or full")
    return body

action = sys.argv[1]
if action == "estimate":
    print(json.dumps(api("task/estimate", payload(sys.argv[2])), indent=2))
elif action == "create":
    if os.environ.get("WEAVEBIT_CONFIRM_CREATE") != "yes":
        raise RuntimeError("Review the estimate before enabling creation")
    body = payload(sys.argv[2])
    request_id = os.environ["WEAVEBIT_REQUEST_ID"]
    created = api("task/create", body, request_id)
    STATE.write_text(json.dumps({"request_id": request_id,
        "request": body, "created": created}, indent=2), encoding="utf-8")
    print("Saved batch_id:", created["batch_id"])
elif action == "collect":
    saved = json.loads(STATE.read_text(encoding="utf-8"))
    batch_id = saved["created"]["batch_id"]
    first = api("task/result", {"batch_id": batch_id,
                               "page": 1, "page_size": 200})
    pages = int(first["pagination"]["pages"])
    if not 1 <= pages <= 5:
        raise RuntimeError("Unexpected page count")
    for page in range(1, pages + 1):
        result = first if page == 1 else api("task/result",
            {"batch_id": batch_id, "page": page, "page_size": 200})
        Path("weavebit-results-page-" + str(page) + ".json").write_text(
            json.dumps(result, indent=2), encoding="utf-8")
    print("Saved all pages; batch_status:", first["batch_status"])
else:
    raise ValueError("Choose estimate, create or collect")

先运行 python batch_check.py estimate dns，审核后运行 python batch_check.py create dns，再运行 python batch_check.py collect。GFW 工作流将 dns 换成 full，并先获取节点。两个独立批次使用不同的幂等 ID 与工作目录。

示例读取一轮所有页并保存原始 JSON，不会无限轮询，也不会自动重试创建。生产集成还应在发请求前持久保存业务请求、参数和幂等 ID，保护本地结果文件，并对错误设置有限重试。示例输出仅供自己审核，不应进入包含真实目标或账户信息的公共日志。

创建超时，先保护幂等关系

Idempotency-Key 建议用于每次创建：长度 1–128，可用字母、数字及 . _ : -。同一账户、同一 Key、相同参数会返回原批次。如果创建请求超时，使用原 Key 和原请求体重试，不要生成新 Key；新的逻辑批次才使用新 Key。

40901 表示相同 Key 的参数冲突，应检查自己保存的请求；40902 表示仍在处理中，稍后用原 Key 重试。42900 时遵守 Retry-After，并采用有上限的退避。幂等记录随批次保留 24 小时，到期后复用 Key 会作为新请求处理，不能把它当成永久去重机制。

查询全部页，区分完成与正常

使用 /task/result 和创建返回的 batch_id 查询；check_id 不能替代批次 ID。page 默认 1，page_size 默认及最大值为 200 个目标。按响应的 pagination 读取所有页，不能只取第一页就认为已经覆盖 1000 个目标。

pending、running、completed 表示执行进度，completed 不等于检测正常。查询到尚未完成的批次时，保留原批次 ID，稍后进行下一轮读取，不要重新创建。多页读取不应当作同一瞬间的快照；需要最终归档时，在批次完成后再读取并保存所有页及最终计费信息。

DNS 的 datas[].result 是对象，判定码位于 datas[].result.result_code，同时保留其中的 status。GFW 全量的每个节点结果也为对象：汇总位于 datas[].results[].result.overall_result，还应保留三个分项对象及其状态。不要把整个 result 对象当成判定码。对判定字段保留 0、1、null 的区别，不把 null 自动转成正常。判读方法见结果文档。

保存截止时间，不把临时批次当数据库

批次最长执行六小时；到 execution_deadline_at 尚未返回的项目进入最终超时状态。目标与结果在创建后二十四小时到期清理，不是完成后再保留二十四小时。按 UTC ISO 8601 处理返回的截止时间，在 expires_at 前保存业务所需的结果。

/task/get_batches 可以查询最近 24 小时的批次元数据，不能替代逐批读取目标结果。自己的数据库至少应保存业务请求、方法、规范化目标、节点、batch_id、check_id、结果状态和相关时间。Console 监控历史与 API 临时批次生命周期不同，不要互相推断。

API v1 的 /v1/task/pollute 仅提供同步单域名 DNS 污染检测，不是本文的批量接口。需要平台安排长期频率与通知时，改用Console 监控任务；需要自行调度和保存数据时，继续使用上述 v2 工作流。