Notion API + CaptchaAI：验证码自动识别实战

跨境团队的老问题：Notion 数据库里挂着几十个待抓取网址，脚本跑到一半撞上验证码，流水线就停在那儿。

思路很简单：把 Notion 数据库当队列用，一列存 URL 和 sitekey，一列存状态和结果，CaptchaAI 负责识别，脚本负责把 token 写回去——不需要额外的消息队列，Notion 本身就够用。

整体流程

从 Notion 读取状态为 Pending 的任务
调用 CaptchaAI 识别对应的 reCAPTCHA v2
把 token 和最新状态写回对应的 Notion 记录

适用场景

适合跨境电商、比价工具这类需要定期抓取多个站点、团队本来就用 Notion 管任务的场景——不用另外搭后台，运营也能直接在 Notion 里看进度。

准备工作

一个 Notion 集成（在 developers.notion.com 创建 Internal Integration）
一个已与该集成共享的 Notion 数据库
CaptchaAI API Key
Python 3.8+ 或 Node.js 18+

国内环境安装依赖建议用镜像源加速，例如 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple。

Notion 数据库结构设计

新建一个 Notion 数据库，包含以下属性。属性名要和代码里的字符串完全一致（区分大小写），下面直接用英文属性名：

Name（标题）：任务标识
URL（网址）：带验证码的目标页面
Sitekey（富文本）：reCAPTCHA sitekey
Status（单选）：Pending、Solving、Solved、Failed
Token（富文本）：已识别的 CAPTCHA token
Solved At（日期）：求解完成时间
Error（富文本）：失败时的错误信息

建好之后一定要把数据库分享给你的 Notion 集成，否则脚本会拿到 401。

Python 实现

# notion_captcha_worker.py
import os
import time
import requests

NOTION_TOKEN = os.environ.get("NOTION_TOKEN")
NOTION_DB_ID = os.environ.get("NOTION_DB_ID")
CAPTCHAAI_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")

NOTION_HEADERS = {
    "Authorization": f"Bearer {NOTION_TOKEN}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28",
}

def get_pending_tasks():
    """Fetch tasks with Status = Pending from Notion."""
    url = f"https://api.notion.com/v1/databases/{NOTION_DB_ID}/query"
    payload = {
        "filter": {
            "property": "Status",
            "select": {"equals": "Pending"},
        }
    }
    resp = requests.post(url, headers=NOTION_HEADERS, json=payload)
    resp.raise_for_status()
    return resp.json()["results"]

def update_task(page_id, properties):
    """Update a Notion page with new property values."""
    url = f"https://api.notion.com/v1/pages/{page_id}"
    payload = {"properties": properties}
    resp = requests.patch(url, headers=NOTION_HEADERS, json=payload)
    resp.raise_for_status()

def set_status(page_id, status, token=None, error=None):
    """Update task status in Notion."""
    props = {"Status": {"select": {"name": status}}}

    if token:
        props["Token"] = {"rich_text": [{"text": {"content": token[:2000]}}]}
        props["Solved At"] = {"date": {"start": time.strftime("%Y-%m-%dT%H:%M:%S")}}

    if error:
        props["Error"] = {"rich_text": [{"text": {"content": error[:200]}}]}

    update_task(page_id, props)

def solve_captcha(sitekey, pageurl):
    """Submit to CaptchaAI and poll for result."""
    # Submit
    resp = requests.get("https://ocr.captchaai.com/in.php", params={
        "key": CAPTCHAAI_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": "1",
    })
    result = resp.json()

    if result.get("status") != 1:
        raise Exception(f"Submit failed: {result.get('request')}")

    task_id = result["request"]

    # Poll
    time.sleep(15)
    for _ in range(25):
        poll = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": CAPTCHAAI_KEY,
            "action": "get",
            "id": task_id,
            "json": "1",
        })
        poll_result = poll.json()

        if poll_result.get("status") == 1:
            return poll_result["request"]
        if poll_result.get("request") != "CAPCHA_NOT_READY":
            raise Exception(f"Solve failed: {poll_result.get('request')}")

        time.sleep(5)

    raise Exception("Polling timeout")

def extract_property(page, prop_name, prop_type="rich_text"):
    """Extract a property value from a Notion page."""
    prop = page["properties"].get(prop_name, {})
    if prop_type == "rich_text":
        texts = prop.get("rich_text", [])
        return texts[0]["plain_text"] if texts else ""
    elif prop_type == "url":
        return prop.get("url", "")
    return ""

def main():
    tasks = get_pending_tasks()
    print(f"Found {len(tasks)} pending tasks")

    for task in tasks:
        page_id = task["id"]
        sitekey = extract_property(task, "Sitekey")
        pageurl = extract_property(task, "URL", "url")

        if not sitekey or not pageurl:
            set_status(page_id, "Failed", error="Missing sitekey or URL")
            continue

        print(f"Solving: {pageurl}")
        set_status(page_id, "Solving")

        try:
            token = solve_captcha(sitekey, pageurl)
            set_status(page_id, "Solved", token=token)
            print(f"  Solved successfully")
        except Exception as e:
            set_status(page_id, "Failed", error=str(e))
            print(f"  Failed: {e}")

        time.sleep(1)  # Rate limit for Notion API

    print("All tasks processed")

if __name__ == "__main__":
    main()

JavaScript 实现

// notion_captcha_worker.js
const { Client } = require('@notionhq/client');
const axios = require('axios');

const notion = new Client({ auth: process.env.NOTION_TOKEN });
const DB_ID = process.env.NOTION_DB_ID;
const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';

async function getPendingTasks() {
  const response = await notion.databases.query({
    database_id: DB_ID,
    filter: { property: 'Status', select: { equals: 'Pending' } },
  });
  return response.results;
}

async function updateTask(pageId, status, token, error) {
  const properties = {
    Status: { select: { name: status } },
  };
  if (token) {
    properties.Token = { rich_text: [{ text: { content: token.slice(0, 2000) } }] };
    properties['Solved At'] = { date: { start: new Date().toISOString() } };
  }
  if (error) {
    properties.Error = { rich_text: [{ text: { content: error.slice(0, 200) } }] };
  }
  await notion.pages.update({ page_id: pageId, properties });
}

async function solveCaptcha(sitekey, pageurl) {
  const submit = await axios.get('https://ocr.captchaai.com/in.php', {
    params: {
      key: API_KEY, method: 'userrecaptcha',
      googlekey: sitekey, pageurl, json: '1',
    },
  });
  if (submit.data.status !== 1) throw new Error(submit.data.request);

  await new Promise(r => setTimeout(r, 15000));

  for (let i = 0; i < 25; i++) {
    const poll = await axios.get('https://ocr.captchaai.com/res.php', {
      params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
    });
    if (poll.data.status === 1) return poll.data.request;
    if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
    await new Promise(r => setTimeout(r, 5000));
  }
  throw new Error('Timeout');
}

async function main() {
  const tasks = await getPendingTasks();
  console.log(`Found ${tasks.length} pending tasks`);

  for (const task of tasks) {
    const sitekey = task.properties.Sitekey?.rich_text?.[0]?.plain_text;
    const pageurl = task.properties.URL?.url;

    if (!sitekey || !pageurl) {
      await updateTask(task.id, 'Failed', null, 'Missing sitekey or URL');
      continue;
    }

    console.log(`Solving: ${pageurl}`);
    await updateTask(task.id, 'Solving');

    try {
      const token = await solveCaptcha(sitekey, pageurl);
      await updateTask(task.id, 'Solved', token);
      console.log('  Solved');
    } catch (e) {
      await updateTask(task.id, 'Failed', null, e.message);
      console.log(`  Failed: ${e.message}`);
    }

    await new Promise(r => setTimeout(r, 1000));
  }
}

main().catch(console.error);

常见故障排查

401 Unauthorized：数据库没有与集成共享 → 把数据库分享给你创建的集成
属性名对不上：区分大小写 → 必须和代码中的字符串逐字一致
Token 被截断：rich_text 单字段上限 2000 字符 → token 通常不到 1000 字符，一般不会触发
429（限流）：调用太频繁 → 每次更新之间加 1 秒延迟，参考代码里的 time.sleep(1)

常见问题

Notion 数据库要放国内还是国际版？会不会有网络问题？

reCAPTCHA 要加载 Google 脚本，国内网络不一定稳定，但影响的是目标站点，跟 Notion 读写无关。若目标是国内站点，大概率遇到 GeeTest（极验）而非 reCAPTCHA，把 solve_captcha 里的 method 换成 geetest 即可。

除了 reCAPTCHA v2，这套流程能改成识别 Turnstile 或 GeeTest v3 吗？

可以。加一个"CAPTCHA Type"属性，按类型分支调用对应的 method（turnstile、geetest 等），其余逻辑不用改。

能不能用 cron 或云函数定时跑这个脚本？

可以，而且推荐——脚本不需要常驻运行
Linux 用 cron，Windows 用任务计划程序
云端用 AWS EventBridge、Google Cloud Scheduler 或阿里云函数计算

轮询频率设多少合适，会不会触发 Notion 的限流？

真正容易触发 429 的是批量更新状态间隔太短，保留代码里的 1 秒延迟即可，跟轮询 CaptchaAI 结果无关。

如何通过 API 处理 reCAPTCHA v2 回调
reCAPTCHA v2 与 Turnstile 同站点处理
reCAPTCHA v2 回调机制详解

下一步

把你的 Notion 数据库变成一条自动验证码识别流水线——获取 CaptchaAI API Key。相关场景还可以看看：

Notion API + CaptchaAI：使用验证码处理自动数据输入

整体流程

适用场景

准备工作

Notion 数据库结构设计

Python 实现

JavaScript 实现

常见故障排查

常见问题

Notion 数据库要放国内还是国际版？会不会有网络问题？

除了 reCAPTCHA v2，这套流程能改成识别 Turnstile 或 GeeTest v3 吗？

能不能用 cron 或云函数定时跑这个脚本？

轮询频率设多少合适，会不会触发 Notion 的限流？

相关文章

下一步

使用验证码处理自动提交表单

注册流程测试中的验证码处理

持续集成测试中的验证码处理

使用 CaptchaAI 构建客户端验证码管道

弹出模式中的验证码：检测和token 提交

使用 Espresso 和 CaptchaAI 进行 Android CAPTCHA 测试