Notion API + CaptchaAI：使用验证码处理自动数据输入 (实用指南)

Notion 的 API 允许您以编程方式读取和写入数据库，使其成为自动化工作流程的灵活任务管理层。当这些工作流程遇到验证码保护的表单时，CaptchaAI处理求解。本指南展示了如何使用 Notion 数据库作为 CAPTCHA 任务队列 - 存储 URL 和站点密钥、触发解决以及写回结果。

真实场景

您的团队维护着一个需要定期提取数据的网站 URL 的概念数据库。其中一些 URL 受验证码保护。一个脚本：

从 Notion 读取挂起的任务
通过 CaptchaAI 解决验证码
使用令牌和状态更新每个 Notion 记录

先决条件

概念集成（内部集成通过开发者.notion.com）
与集成共享的概念数据库
CaptchaAI API 密钥
Python 3.8+ 或 Node.js 18+

概念数据库设置

创建具有以下属性的概念数据库：

财产	类型	目的
姓名	标题	任务标识符
网址	网址	带有验证码的目标页面
站点密钥	富文本	reCAPTCHA 站点密钥
地位	选择	待处理、正在解决、已解决、失败
代币	富文本	已解决验证码令牌
解决于	日期	解决时间戳
错误	富文本	如果失败则出现错误消息

与您的 Notion 集成共享数据库。

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);

故障排除

问题	原因	处理方式
来自 Notion 的 `401 Unauthorized`	集成未连接到数据库	与您在 Notion 中的集成共享数据库
属性名称不匹配	区分大小写	概念属性名称区分大小写——完全匹配
令牌被截断	概念 rich_text 限制为 2000 个字符	CAPTCHA 令牌通常小于 1000 个字符；这不应该是问题
概念速率限制 (429)	API 调用过多	在 Notion 更新之间添加 1 秒延迟

常问问题

我可以将其作为计划作业运行吗？

是的。使用 cron (Linux)、任务计划程序 (Windows) 或云计划程序（AWS EventBridge、Google Cloud Scheduler）按计划运行脚本。

如何设置 Notion 集成？

前往概念.so/my-integrations，创建新的内部集成，复制秘密令牌，并与集成共享您的数据库。

我可以处理其他验证码类型吗？

是的。将“CAPTCHA Type”属性添加到 Notion 数据库并修改求解函数以使用适当的 CaptchaAI 方法（turnstile、geetest、base64 等）。

如何使用Api解决Recaptcha V2回调
Recaptcha V2 Turnstile同一站点处理
Recaptcha V2回调机制

下一步

将您的概念数据库转变为自动验证码解决队列 -”获取您的 CaptchaAI API 密钥。

相关指南：

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

真实场景

先决条件

概念数据库设置

Python实现

JavaScript 实现

故障排除

常问问题

我可以将其作为计划作业运行吗？

如何设置 Notion 集成？

我可以处理其他验证码类型吗？

相关文章

下一步

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

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

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

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

使用 Espresso 和 CaptchaAI 进行 Android CAPTCHA 测试

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