Notion API + CaptchaAI: Automatisierte Dateneingabe mit CAPTCHA-Verarbeitung

Eine Notion-Datenbank eignet sich erstaunlich gut als Steuerzentrale für wiederkehrende Automatisierungsaufgaben – auch dann, wenn einzelne Zielseiten hinter einem CAPTCHA liegen. Das Muster ist schlank: Notion hält die Aufgabenliste, CaptchaAI löst die reCAPTCHA-v2-Abfrage, und ein kleiner Worker schreibt Token, Zeitstempel und Status wieder in dieselbe Zeile zurück. So bleibt der Bearbeitungsstand jederzeit im Team sichtbar, ohne dass Sie eine separate Datenbank oder ein Dashboard aufbauen müssen.

Die Notion-API erlaubt lesenden und schreibenden Zugriff auf Datenbanken über HTTP. Genau das macht sie zu einer flexiblen Aufgabenebene: Jede Zeile ist ein Auftrag, jede Statusspalte ein Fortschrittsindikator. Dieser Leitfaden zeigt Schritt für Schritt, wie Sie eine Notion-Datenbank als CAPTCHA-Warteschlange betreiben – von der Datenmodellierung über den Python- und Node.js-Worker bis zur Fehlerbehebung.

Das Prinzip: Notion als Aufgaben-Warteschlange

Stellen Sie sich ein Team vor, das eine Notion-Datenbank mit Website-URLs pflegt, aus denen regelmäßig Daten extrahiert werden müssen. Ein Teil dieser Seiten ist durch reCAPTCHA v2 geschützt. Ein Worker-Skript übernimmt den Rest in drei klar getrennten Schritten:

1. Er liest alle Aufgaben mit dem Status Pending aus Notion.
2. Er lässt das CAPTCHA über CaptchaAI lösen und wartet auf das Token.
3. Er schreibt Token, Lösungszeitpunkt und den neuen Status in denselben Datensatz zurück.

Der Vorteil dieses Aufbaus: Notion ist gleichzeitig Eingabemaske, Warteschlange und Protokoll. Fachanwender pflegen neue URLs bequem in der Oberfläche, während der Worker im Hintergrund abarbeitet – kein zusätzliches Ticketsystem, keine parallele Datenhaltung.

Voraussetzungen

• Eine Notion-Integration (interne Integration über developers.notion.com)
• Eine Notion-Datenbank, die mit dieser Integration geteilt ist
• Einen CaptchaAI-API-Schlüssel
• Python 3.8+ oder Node.js 18+

Notion-Datenbank einrichten

Legen Sie eine Notion-Datenbank mit den folgenden Eigenschaften an. Wichtig: Die Eigenschaftsnamen und die Auswahlwerte sind in Notion case-sensitiv und müssen exakt mit den Zeichenketten im Code übereinstimmen – halten Sie deshalb die englischen Bezeichner bei.

Teilen Sie die Datenbank anschließend mit Ihrer Notion-Integration, sonst liefert die API einen 401-Fehler zurück.

Hinweis: Die Auswahlwerte der Status-Spalte (Pending, Solving, Solved, Failed) müssen exakt so angelegt sein wie im Code – Notion legt fehlende Optionen zwar automatisch an, filtert aber nur bei buchstabengenauer Übereinstimmung korrekt.

Python-Worker implementieren

Der Worker fragt offene Aufgaben ab, übermittelt Sitekey und Page-URL an CaptchaAI (method=userrecaptcha), fragt das Ergebnis im Polling ab und aktualisiert die Notion-Seite. CaptchaAI löst reCAPTCHA v2 typischerweise in unter 60 Sekunden mit hoher Erfolgsquote – das anfängliche time.sleep(15) vor der ersten Abfrage ist bewusst gesetzt, um leere Polling-Runden zu sparen.

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

Beachten Sie die abschließende time.sleep(1)-Pause: Sie hält die Notion-API unter ihrem Ratenlimit und verhindert 429-Fehler bei größeren Warteschlangen.

Node.js-Variante

Wer den Worker lieber in einem JavaScript-Stack betreibt, nutzt das offizielle @notionhq/client-SDK und axios für die CaptchaAI-Aufrufe. Ablauf und Statuslogik sind identisch – nur die Sprache wechselt.

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

Betrieb in der eigenen Infrastruktur

Der Worker ist zustandslos – die gesamte Warteschlange liegt in Notion. Damit lässt er sich problemlos als geplanter Job betreiben: als cron-Eintrag oder systemd-Timer auf einem kleinen VPS (etwa bei Hetzner, IONOS oder netcup), als Schritt in einer GitLab-CI-Pipeline oder als AWS-Lambda-Funktion mit EventBridge-Auslöser. Weil jeder Lauf nur Pending-Zeilen anfasst und den Status sofort auf Solving setzt, verarbeiten auch zwei parallele Instanzen keine Aufgabe doppelt.

Ein Hinweis zur Sorgfaltspflicht für Leser im DACH-Raum: Sobald Sie fremde Webseiten ansteuern und dabei personenbezogene Daten – IP-Adressen zählen bereits dazu – verarbeiten, sollten Sie Ihre Rechtsgrundlage nach DSGVO prüfen und die Nutzungsbedingungen der Zielseite beachten. Setzen Sie CaptchaAI für Aufgaben ein, für die Sie autorisiert sind.

Typische Fehler und ihre Behebung

Häufige Fragen

Wie viele CAPTCHAs kann ich parallel abarbeiten?

So viele, wie Ihr Plan Threads erlaubt. CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung – jeder Plan enthält unbegrenzte Lösungen pro Thread. BASIC (15 $/Monat) bietet 5 Threads, STANDARD (30 $/Monat) 15 Threads, ADVANCE (90 $/Monat) 50 Threads. Für einen Notion-Worker, der Aufgaben nacheinander abarbeitet, genügt oft schon der kleinste Plan; erst bei mehreren parallelen Instanzen lohnt ein größerer.

Was passiert, wenn das Token abläuft, bevor es verwendet wird?

reCAPTCHA-Token sind nur rund 120 Sekunden gültig. Speichern Sie das Token zwar in Notion, aber verwenden Sie es zeitnah in derselben Session – lösen Sie es also erst kurz vor der finalen Zielanfrage. Ein Token, das stunden- oder tagelang in der Datenbank liegt, wird beim Absenden abgelehnt.

Kann ich weitere CAPTCHA-Typen in derselben Warteschlange verarbeiten?

Ja. Ergänzen Sie eine Eigenschaft CAPTCHA Type in der Notion-Datenbank und wählen Sie im Worker die passende CaptchaAI-method – etwa turnstile für Cloudflare Turnstile oder post für Bild-CAPTCHAs. Beachten Sie, dass hCaptcha und FunCaptcha nicht unterstützt werden und GeeTest v4 erst bald verfügbar sein wird.

Lässt sich der Worker in eine bestehende CI-Pipeline einbinden?

Ja. Weil das Skript nur Umgebungsvariablen für die beiden Tokens benötigt, läuft es als Job in GitLab CI oder GitHub Actions genauso wie in einem systemd-Timer. Hinterlegen Sie NOTION_TOKEN und CAPTCHAAI_KEY als geschützte CI-Variablen und rufen Sie den Worker in einem geplanten Pipeline-Schritt auf.

Verwandte Leitfäden

• Airtable als CAPTCHA-Datenbank anbinden
• CaptchaAI in Retool-Tools integrieren
• Power Automate mit CaptchaAI verbinden