Notion API + CaptchaAI: إدخال البيانات تلقائيًا مع معالجة اختبار CAPTCHA

الفكرة الأساسية هنا مباشرة: حوّل قاعدة بيانات Notion إلى قائمة انتظار لمهام CAPTCHA، ودَع CaptchaAI يتولّى الحل بينما يكتفي فريقك بمتابعة الحالة في الجدول. سكربت واحد يقرأ المهام المعلّقة، يرسل كل مفتاح موقع وعنوان صفحة إلى الخدمة، ثم يعيد كتابة الرمز الناتج والحالة في السجل نفسه. النتيجة أن Notion يتحوّل من مجرد جدول إلى لوحة تشغيل حيّة لعملية استخراج بيانات كانت تتعثّر عند كل نموذج محمي.

هذا الدليل يبني العامل خطوة بخطوة: تجهيز قاعدة البيانات، ثم تنفيذ العامل بلغتَي Python وNode.js، ثم معالجة الأخطاء الشائعة التي تظهر عند الربط الفعلي.

لماذا تصلح Notion كطبقة تنسيق لمهام CAPTCHA

Notion ليست قاعدة بيانات تقليدية، لكنها توفّر ثلاثة عناصر تجعلها مناسبة لتنسيق أعمال الأتمتة. أولها واجهة برمجة تطبيقات تقرأ وتكتب في الجداول برمجيًا، فتسمح للسكربت بسحب المهام وإعادة كتابة النتائج. وثانيها حقول حالة من نوع Select يسهل تصفيتها، فتفصل المهام المعلّقة عن المحلولة عن الفاشلة بنقرة واحدة. وثالثها واجهة مرئية يفهمها غير المبرمجين، فيتابع مسؤول العمليات سير العمل دون فتح الطرفية.

الأهم أن هذا الأسلوب يفصل بوضوح بين طبقة العرض وطبقة الحل. Notion هو الواجهة التي يُدخل الفريق فيها المهام ويقرأ منها النتائج، وCaptchaAI هو المحرّك الذي يحلّ الاختبار في الخلفية. هذا الفصل يجعل الصيانة أبسط: يمكنك تعديل منطق الحل أو تبديل نوع CAPTCHA دون المساس بالطريقة التي يُدخل بها الفريق البيانات، والعكس صحيح. كما يقلّل الأخطاء البشرية، لأن لا أحد يحتاج إلى نسخ الرموز يدويًا بين النوافذ.

سيناريو من واقع فرق التشغيل

تخيّل فريق عمليات في متجر إلكتروني بالرياض يتابع أسعار الموردين على عشرات المواقع الإقليمية. بعض هذه المواقع يعرض reCAPTCHA v2 قبل السماح بالوصول إلى صفحات الأسعار. بدل أن يفتح أحد أفراد الفريق كل رابط يدويًا في وقت الذروة، يضيفون الروابط ومفاتيح المواقع إلى جدول Notion واحد. يعمل السكربت كل صباح: يقرأ الصفوف المعلّقة، يحلّ كل CAPTCHA عبر CaptchaAI، ويحدّث الحالة إلى Solved مع طابع زمني دقيق.

عند فتح Notion يرى الفريق بنظرة واحدة أي المواقع جاهز للمعالجة، وأيها ما زال معلّقًا، وأيها فشل ويحتاج تدخلاً. ولأن كل خطأ يُسجَّل في حقل مخصّص، تصبح المتابعة مسألة تصفية بسيطة لا تحقيقًا في سجلّات متفرّقة. الأسلوب نفسه يخدم فرق التسويق التي تجمع بيانات المنافسين، أو فرق البحث التي تراقب مصادر إقليمية بلغات متعددة.

ما تحتاجه قبل البدء

• تكامل داخلي في Notion يُنشأ من developers.notion.com
• قاعدة بيانات Notion تمّت مشاركتها مع هذا التكامل
• مفتاح CaptchaAI API من لوحة التحكم
• بيئة تشغيل: Python 3.8 أو أحدث، أو Node.js 18 أو أحدث

تجهيز قاعدة بيانات Notion

أنشئ قاعدة بيانات Notion بهذه الخصائص. احتفظ بأسماء الخصائص بالإنجليزية تمامًا كما هي أدناه، لأن السكربت يشير إليها حرفيًا:

بعد إنشاء الجدول، شاركه مع تكامل Notion الخاص بك من قائمة الاتصالات. تأكّد من أن أسماء الخصائص مطابقة حرفيًا لما في الجدول، فأسماء خصائص Notion حسّاسة لحالة الأحرف، وأي اختلاف بسيط في التسمية يؤدي إلى فشل السكربت في قراءة الحقل أو الكتابة فيه.

تنفيذ العامل بلغة Python

يجلب هذا العامل المهام المعلّقة، يحلّها عبر CaptchaAI، ثم يكتب الرمز والحالة في كل سجل. احفظ مفاتيح Notion وCaptchaAI في متغيّرات بيئة بدل كتابتها مباشرة في الكود:

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

يتبع السكربت أعلاه دورة من أربع خطوات لكل مهمة: يرسل مفتاح الموقع وعنوان الصفحة إلى نقطة النهاية in.php فيستقبل معرّف المهمة، ثم ينتظر 15 ثانية قبل بدء الاستطلاع الدوري على res.php كل 5 ثوانٍ حتى تجهز النتيجة. عند نجاح الحل يكتب الرمز في حقل Token ويضبط الحالة على Solved مع طابع زمني؛ وعند تجاوز المهلة أو ورود رمز خطأ يسجّل الرسالة في حقل Error ويضع الحالة Failed. هذا الفصل الصريح بين النجاح والفشل هو ما يجعل الجدول قابلاً للمتابعة بلا لبس.

تنفيذ العامل بلغة Node.js

النسخة نفسها بمنطق مطابق باستخدام مكتبة @notionhq/client الرسمية وAxios لطلبات CaptchaAI. تناسب هذه النسخة الفرق التي تعتمد بيئة 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);

معالجة الأخطاء الشائعة

معظم المشكلات عند أول تشغيل تعود إلى الربط بين التكامل وقاعدة البيانات، أو إلى تفاصيل صغيرة في التسمية والحدود. الجدول التالي يلخّص أكثرها تكرارًا:

الأسئلة الشائعة

هل يعالج السكربت عدة مهام في وقت واحد؟

في صورته الحالية يعالج السكربت المهام تسلسليًا، صفًا تلو الآخر، وهو مناسب للأحمال المتوسطة. لرفع الإنتاجية يمكنك تشغيل عدة نسخ أو اعتماد معالجة غير متزامنة، مع مراعاة عدد الـ Threads المتاحة في خطتك وحدّ الطلبات في Notion API. ابدأ بالنسخة التسلسلية ثم وسّع تدريجيًا بعد قياس السلوك الفعلي.

كيف أتحكّم في التكلفة عند معالجة آلاف الصفوف؟

يعتمد CaptchaAI نموذج تسعير قائمًا على الـ Threads وليس على عدد عمليات الحل: كل خطة تتيح عددًا من الـ Threads المتزامنة مع عمليات حل غير محدودة خلال الشهر. تبدأ خطة BASIC من 15 دولارًا شهريًا بـ 5 Threads، وترتفع حتى ADVANCE بسعر 90 دولارًا مع 50 Thread للأحمال الأكبر. عمليًا، معالجة آلاف الصفوف لا ترفع فاتورتك ما دمت ضمن حدود الـ Threads المتاحة.

ماذا أفعل إذا رفض الموقع المستهدف الرمز المحلول؟

إذا حُلّ الرمز لكن الموقع رفضه، فالسبب غالبًا اختلاف مفتاح الموقع أو عنوان الصفحة أو سياق الجلسة بين لحظة الحل ولحظة الإرسال. التقط مفتاح الموقع من الصفحة نفسها، واستخدم الرمز مباشرة داخل الجلسة التي أرسلت منها الطلب قبل انتهاء صلاحيته القصيرة نسبيًا.

هل يمكن استخدام الأسلوب نفسه مع أنواع CAPTCHA أخرى؟

نعم. يدعم CaptchaAI reCAPTCHA v2 وv3، وCloudflare Turnstile وChallenge، وGeeTest v3، إضافة إلى الصور وGrid وBLS. أضف خاصية "CAPTCHA Type" إلى الجدول وبدّل قيمة method في دالة الحل (مثل turnstile أو geetest) لتوجيه كل مهمة إلى النوع المناسب دون تغيير بقية المنطق.

الخطوات التالية

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• كيفية حلّ reCAPTCHA v2 عبر الـ API: دليل خطوة بخطوة
• كيفية حل Cloudflare Turnstile باستخدام واجهة API
• كيفية حل GeeTest v3 باستخدام API

أدلة ذات صلة

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• كيفية حلّ reCAPTCHA v2 عبر الـ API: دليل خطوة بخطوة