Notion API + CaptchaAI: Entri Data Otomatis dengan Penanganan CAPTCHA

Database Notion Anda bisa jadi antrean tugas CAPTCHA tanpa infrastruktur tambahan: satu baris per URL, worker yang membaca baris berstatus Pending, mengirim sitekey ke CaptchaAI, lalu menulis token dan status baru kembali ke baris yang sama. Panduan ini menunjukkan pola itu dari nol sampai jalan — persiapan database, kode Python dan Node.js siap pakai, sampai perkiraan kebutuhan thread untuk tim yang mau menjalankannya rutin.

Yang perlu disiapkan sebelum mulai

Sebelum menulis satu baris kode pun, pastikan empat hal ini sudah beres:

Menyiapkan database Notion

Buat database dengan tujuh properti berikut. Nama properti harus persis sama seperti di tabel ini, sebab kode di bawah mengaksesnya langsung berdasarkan nama — bukan berdasarkan posisi kolom:

Perhatikan spasi pada Solved At dan huruf besar-kecil di keempat opsi Status — keduanya harus sama persis dengan yang dibaca dan ditulis oleh kode di bawah.

Langkah terakhir sebelum coding: bagikan database dengan integrasi Notion Anda. Tanpa langkah ini, setiap panggilan API akan ditolak dengan 401 Unauthorized.

Kode Python: baca, solve, tulis kembali

Worker ini membaca task Pending, submit tiap sitekey ke in.php, polling res.php sampai token siap, lalu menulis hasilnya ke record Notion yang sama:

# 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()

Versi Node.js

Logika yang identik, pakai SDK resmi @notionhq/client dan axios:

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

Kapan pola ini benar-benar layak dipakai

Pola "Notion sebagai antrean" paling sering muncul di tim data kecil: agensi price-monitoring di Jakarta yang memantau puluhan URL e-commerce kompetitor setiap hari, atau freelancer scraping bergaya kontrak Upwork/Fastwork yang mengelola beberapa proyek klien sekaligus. Alih-alih membangun database dan scheduler terpisah, tim cukup menambah baris baru di Notion tiap kali ada URL yang perlu diproses. Dari sisi worker, alurnya persis tiga langkah yang barusan Anda lihat di kode:

1. Membaca baris berstatus Pending dari Notion
2. Mengirim sitekey tiap baris ke CaptchaAI untuk diselesaikan
3. Menulis token dan status baru kembali ke baris yang sama

Begitu volume URL naik ke ratusan per hari, pola ini masih jalan — tinggal tambah thread, bukan tambah komponen infrastruktur baru.

Perkiraan kebutuhan thread

Satu thread CaptchaAI menangani satu solve yang sedang berjalan, jadi kebutuhan thread kira-kira mengikuti jumlah task paralel yang diproses worker Anda, bukan jumlah total baris di database:

Mulai dari BASIC dulu — endpoint dan payload yang dipakai sama persis di semua tier, jadi naik plan nanti tidak butuh perubahan kode.

Error yang sering muncul dan cara mengatasinya

Pertanyaan seputar automasi Notion + CaptchaAI

Bisakah proses ini dijadwalkan otomatis, tanpa saya jalankan manual?

Bisa. Jalankan skrip lewat cron (Linux), Task Scheduler (Windows), atau scheduler cloud seperti AWS EventBridge maupun Google Cloud Scheduler — panggil notion_captcha_worker.py atau versi Node.js sesuai interval yang Anda mau, misalnya tiap 15 menit.

Bisa satu worker menangani beberapa database Notion sekaligus?

Bisa. Jalankan instance worker terpisah dengan NOTION_DB_ID berbeda untuk tiap database, atau modifikasi main() agar melakukan looping atas beberapa ID database secara berurutan sebelum lanjut ke database berikutnya.

Apakah scraping data untuk antrean Notion ini perlu memperhatikan UU PDP?

Ya, kalau data yang Anda ekstrak memuat data pribadi. UU Pelindungan Data Pribadi (UU 27/2022) dan UU ITE tetap berlaku untuk data yang Anda kumpulkan lewat automasi ini — CaptchaAI hanya menangani penyelesaian CAPTCHA-nya, jadi kepatuhan atas data yang disimpan tetap tanggung jawab tim Anda.

Bisa dipakai untuk CAPTCHA selain reCAPTCHA v2?

Bisa. Tambahkan properti CAPTCHA Type ke database Notion, lalu sesuaikan solve_captcha() untuk memakai method CaptchaAI yang sesuai (turnstile, geetest, base64, dan lainnya) berdasarkan nilai properti tersebut.

Artikel terkait

• cara solve callback reCAPTCHA v2 lewat API
• menangani reCAPTCHA v2 dan Turnstile di situs yang sama
• cara kerja mekanisme callback reCAPTCHA v2

Langkah selanjutnya

Ubah database Notion Anda jadi antrean penyelesaian CAPTCHA otomatis — dapatkan API key CaptchaAI Anda.

Panduan terkait:

• Airtable + CaptchaAI
• menyimpan hasil CAPTCHA di PostgreSQL
• Zapier + CaptchaAI