Notion API + CaptchaAI: Nhập dữ liệu tự động với xử lý CAPTCHA

API của Notion cho phép bạn đọc và ghi vào cơ sở dữ liệu theo chương trình, biến nó thành lớp quản lý tác vụ linh hoạt cho quy trình làm việc tự động hóa. Khi những quy trình công việc đó gặp phải các biểu mẫu được bảo vệ bằng CAPTCHA,CaptchaAIxử lý việc giải quyết. Hướng dẫn này cho biết cách sử dụng cơ sở dữ liệu Notion làm hàng đợi tác vụ CAPTCHA - lưu trữ URL và khóa trang, kích hoạt các giải pháp và ghi lại kết quả.

Kịch bản thế giới thực

Nhóm của bạn duy trì cơ sở dữ liệu Notion về các URL trang web cần trích xuất dữ liệu định kỳ. Một số URL này được bảo vệ bằng CAPTCHA. Một kịch bản:

1. Đọc các tác vụ đang chờ xử lý từ Notion
2. Giải CAPTCHA qua CaptchaAI
3. Cập nhật từng bản ghi Notion bằng mã thông báo và trạng thái

Điều kiện tiên quyết

• Tích hợp khái niệm (Tích hợp nội bộ thông quanhà phát triển.notion.com)
• Cơ sở dữ liệu Notion được chia sẻ với sự tích hợp
• Khóa API CaptchaAI
• Python 3.8+ hoặc Node.js 18+

Thiết lập cơ sở dữ liệu khái niệm

Tạo cơ sở dữ liệu Notion với các thuộc tính sau:

Chia sẻ cơ sở dữ liệu với sự tích hợp Notion của bạn.

Triển khai 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()

Triển khai 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);

Khắc phục sự cố

Câu hỏi thường gặp

Tôi có thể chạy công việc này như một công việc theo lịch trình không?

Vâng. Sử dụng cron (Linux), Trình lập lịch tác vụ (Windows) hoặc trình lập lịch trên đám mây (AWS EventBridge, Google Cloud Scheduler) để chạy tập lệnh theo lịch.

Làm cách nào để thiết lập tích hợp Notion?

đi tớikhái niệm.so/my-integrations, tạo tiện ích tích hợp nội bộ mới, sao chép mã thông báo bí mật và chia sẻ cơ sở dữ liệu của bạn với tiện ích tích hợp.

Tôi có thể xử lý các loại CAPTCHA khác không?

Vâng. Thêm thuộc tính "Loại CAPTCHA" vào cơ sở dữ liệu Notion và sửa đổi hàm giải để sử dụng phương pháp CaptchaAI thích hợp (turnstile, geetest, base64, v.v.).

bài viết liên quan

• Cách giải quyết cuộc gọi lại Recaptcha V2 bằng Api
• Xử lý cửa quay Recaptcha V2 trên cùng một trang web
• Cơ chế gọi lại Recaptcha V2

Các bước tiếp theo

Biến cơ sở dữ liệu Notion của bạn thành hàng đợi giải CAPTCHA tự động —lấy khóa API CaptchaAI của bạn.

Hướng dẫn liên quan:

• Máy bay + CaptchaAI
• Lưu trữ kết quả CAPTCHA trong PostgreSQL
• Zapier + CaptchaAI