开发者文档

快速开始:5 分钟完成第一次采集

AntsData 的 scraper 采集接口默认异步:一次采集是一个 run(任务),提交后立即拿到任务 id,后台执行,你按 id 轮询状态,成功后再按 id 分页取记录或下载。也可加 deliveryMode:"sync" 阻塞等终态;若只要搜索结果、想一次调用直接拿数据,见同步搜索 · Google SERP


前置条件

  • AntsData 账号(注册
  • API Key(在控制台生成,格式 ants_xxxxxxxxxxxx
  • Python 3.8+ / Node.js 16+ 或 cURL

Step 1:提交采集任务(创建 run)

以采集两个 TikTok 账号资料为例。默认走"纯拉取":提交后轮询、再取记录,无需配置任何回调:

curl -X POST "https://api.antsdata.com/v1/scraper/tiktok/profile" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "deliveryMode": "async",
    "recordLimit": 1000,
    "targets": [
      { "username": "khaby.lame" },
      { "username": "nike" }
    ]
  }'
字段 必填 说明
deliveryMode async(提交+轮询/取数,返回 201)或 sync(阻塞等终态,见下方「同步模式」)
recordLimit 每个目标的记录上限(默认 1000;0=不限,按可用积分反算)
targets 采集目标数组(≥1);每个元素的字段见对应平台文档
deliveries 可选:配置 webhook 等渠道自动推送结果;不配则用轮询+取记录(纯拉取)。sync 模式不支持 deliveries

可选推送:若想任务完成后自动把结果推到你的服务,加 deliveries"deliveries": [{ "channelType": "webhook", "config": { "url": "https://your.app/hook" }, "outputFormat": "json" }]

同步模式(deliveryMode: "sync":适合小任务。提交后服务端阻塞等待采集完成,按结果返回:

  • 完成(服务端超时窗口内到终态)→ 200 + run 对象(statussucceeded/failed,含 tasks[]);随后照常 GET .../records 取数(省掉自行轮询)。
  • 超时未完成202 + run 对象(带 idstatus 仍为 queued/running);任务继续在后台跑,转为轮询 GET /v1/scraper/runs/{id} 即可。

sync尽力而为的快路径,不保证一定在阻塞内完成(大任务/目标多会走 202)。⚠️ 非幂等:请求若在客户端侧超时后重试,会重复创建 run 并重复冻结积分——重试前先按已拿到的 id 查询确认。

响应(HTTP 201)——记下 id,这是后续所有查询的任务号:

{
  "code": 0,
  "message": "ok",
  "data": {
    "id": 1287345,
    "actorName": "TikTok",
    "endpointName": "Profile",
    "status": "queued",
    "targetCount": 2,
    "maxCost": "2.0000",
    "chargedCredits": "0",
    "createdAt": "2026-07-06T09:00:00Z"
  }
}

提交即冻结 maxCost 的积分(不是扣费);成功后按实际产出结算,全部失败则释放不计费。


Step 2:轮询任务状态

id 查询,直到状态进入终态 succeededfailed(建议退避 1s→2s→5s):

curl "https://api.antsdata.com/v1/scraper/runs/1287345" \
  -H "Authorization: Bearer YOUR_API_KEY"

{
  "code": 0,
  "message": "ok",
  "data": {
    "id": 1287345,
    "status": "succeeded",
    "targetCount": 2,
    "targetsOk": 2,
    "targetsFailed": 0,
    "recordCount": 2,
    "chargedCredits": "0.0040",
    "tasks": [
      { "targetIndex": 0, "taskId": "1279...41", "submitState": "submitted", "status": "succeeded", "recordCount": 1 },
      { "targetIndex": 1, "taskId": "1279...42", "submitState": "submitted", "status": "succeeded", "recordCount": 1 }
    ]
  }
}
  • 状态只有 4 态queuedrunningsucceeded | failed(后两个是终态)。
  • 进度看计数targetCount / targetsOk / targetsFailed(无百分比)。
  • 单目标成败tasks[]:某个目标失败时该元素 status:"failed" 且带 errorMessage,其余目标仍可成功(部分成功照常出数并计费)。

Step 3:取数据

任务 succeeded 后,按 id 分页取记录:

curl "https://api.antsdata.com/v1/scraper/runs/1287345/records?page=1&pageSize=20" \
  -H "Authorization: Bearer YOUR_API_KEY"

{
  "code": 0,
  "message": "ok",
  "data": {
    "total": 2,
    "page": 1,
    "pageSize": 20,
    "items": [
      { "username": "khaby.lame", "nickname": "Khaby Lame", "followers": 162000000, "isVerified": true },
      { "username": "nike", "nickname": "Nike", "followers": 302000000, "isVerified": true }
    ]
  }
}

下载整份产物(formatjson / csv / excel):

curl "https://api.antsdata.com/v1/scraper/runs/1287345/download?format=csv" \
  -H "Authorization: Bearer YOUR_API_KEY" -o result.csv

records 里每条的字段(items[])随平台/端点而定,见各平台文档的"输出字段"。run 未成功时取记录返回空集。


完整示例(Python:提交 → 轮询 → 取数)

import requests, time

BASE = "https://api.antsdata.com/v1"
H = {"Authorization": "Bearer YOUR_API_KEY", "Content-Type": "application/json"}

# 1. 提交
run = requests.post(f"{BASE}/scraper/tiktok/profile", headers=H, json={
    "deliveryMode": "async",
    "recordLimit": 1000,
    "targets": [{"username": "khaby.lame"}, {"username": "nike"}]
}).json()["data"]
run_id = run["id"]

# 2. 轮询
for delay in [1, 2, 5, 5, 5, 5, 5]:
    r = requests.get(f"{BASE}/scraper/runs/{run_id}", headers=H).json()["data"]
    if r["status"] in ("succeeded", "failed"):
        break
    time.sleep(delay)

# 3. 取数
if r["status"] == "succeeded":
    page = requests.get(f"{BASE}/scraper/runs/{run_id}/records",
                        headers=H, params={"page": 1, "pageSize": 100}).json()["data"]
    print(f"共 {page['total']} 条 | 实扣 {r['chargedCredits']} 积分")
    for item in page["items"]:
        print(item)
else:
    print("任务失败:", [t.get("errorMessage") for t in r["tasks"]])

计费

按成功产出的数据条数计费(每 1000 条记录一个单价):

  1. 提交时冻结 maxCost = 目标数 × ⌈recordLimit/1000⌉ × 单价;余额不足则提交被拒(HTTP 402)。
  2. 成功结算 chargedCredits = ⌈成功记录数/1000 × 单价⌉(不超过 maxCost)。
  3. 全部失败释放:run failed 时释放冻结、不计费

各端点单价与套餐见定价页面


错误处理

统一信封错误:真实 HTTP 状态码 + {code, message}

r = requests.post(f"{BASE}/scraper/tiktok/profile", headers=H, json=payload)
if r.status_code == 201:
    run_id = r.json()["data"]["id"]     # 提交成功
elif r.status_code == 401:
    print("认证失败,请检查 API Key")
elif r.status_code == 402:
    print("积分不足,请充值")
elif r.status_code == 429:
    print("请求频率超限,稍后重试")
else:
    print("提交失败:", r.json().get("message"))
状态码 含义 处理方式
400 参数/目标不符要求 检查 targets 字段
401 API Key 无效 检查 Key
402 积分不足 充值
404 任务不存在 检查 run id
429 频率超限 降频,见 Retry-After
500 服务端错误 稍后重试

完整信封、状态机与计费细节见 认证与错误处理


下一步


FAQ

Q: 为什么不是调用后直接返回数据?

采集是异步的:提交拿任务 id,后台执行,你轮询 id 取结果。这样才能支撑大批量、多目标、可下载。

Q: 一次可以采集多个目标吗?

可以。targets 是数组,一个 run 可含多个目标,内部并行执行,逐目标状态见 tasks[]

Q: 支持哪些编程语言?

任何支持 HTTP 的语言。官方给 Python/Node.js/cURL 示例。

Q: 采集失败会扣费吗?

不会。全部目标失败释放冻结、不计费;部分成功只按成功产出的条数结算。

Q: 结果能保存多久 / 什么格式?

结果可分页查看或下载为 json/csv/excel,留存约 90 天,请及时取用。