بناء لوحة تحكم KoboToolbox مدعومة بالذكاء الاصطناعي للمنظمات غير الحكومية: من Kobo API إلى Postgres وتقارير PDF للممولين بثلاث لغات

مشكلة تصدير KoboToolbox

إذا كنت مسؤول رصد وتقييم في منظمة غير حكومية تونسية أو مغربية أو لبنانية تدير برنامجاً ممولاً من جهة مانحة، فمن المحتمل أن يبدو أسبوعك هكذا:

• صباح الإثنين، تقوم بتنزيل تصدير CSV من KoboToolbox.
• يقوم الفريق الميداني في سيدي بوزيد بإرسال ثلاثين نموذجاً إضافياً بعد ظهر الإثنين. ملف CSV الخاص بك قديم بالفعل.
• الموعد النهائي لتقرير USAID هو الجمعة. يريد الممول PDF بالإنجليزية، وتريد لجنة التوجيه نفس الأرقام بالفرنسية، ويحتاجها الشريك المحلي بالعربية.
• تقضي ليلة الجمعة في Excel للتوفيق بين ثلاثة تصديرات، ثم في PowerPoint لنسخ ولصق الأرقام في قالب الممول.

هذه ليست مشكلة أدوات — KoboToolbox ممتاز في ما يفعله. إنها مشكلة خط معالجة. Kobo هو محرك نماذج، وليس منصة تقارير. الحل هو معاملة كل إرسال Kobo كحقيقة في مستودع بيانات حقيقي، ثم وضع التقارير والذكاء الاصطناعي فوقه.

هذا الدرس هو الرفيق العملي لمقالنا الركيزة حول لوحات تحكم الرصد والتقييم بالذكاء الاصطناعي للمنظمات غير الحكومية في منطقة MENA. اقرأه للحصول على التفكير المعماري؛ عُد إلى هنا عندما تريد فعلاً بناء الأداة.

ما الذي ستبنيه

بنهاية هذا الدرس، سيكون لديك:

1. مهمة ETL بلغة Python تسحب إرسالات KoboToolbox عبر REST API v2 وفق جدول ليلي
2. مستودع Postgres بثلاثة جداول: الإرسالات الخام، والمؤشرات المسطّحة، وسجل التدقيق
3. طبقة وكلاء ذكاء اصطناعي تُعلِّم الشذوذ وتلخص الملاحظات الصوتية العربية/الفرنسية إلى الإنجليزية لتقارير الممولين
4. محرك تقارير PDF يستخدم WeasyPrint ينتج نفس مجموعة البيانات بـ EN وFR وAR من قالب Jinja2 واحد
5. تكلفة تشغيل إجمالية أقل من 30 دولاراً شهرياً لبرنامج يصل إلى 50,000 إرسال

كل شيء مفتوح المصدر. المكون المدفوع الوحيد هو استدعاءات LLM API (بضعة دولارات شهرياً بأحجام المنظمات غير الحكومية).

المتطلبات المسبقة

قبل أن تبدأ:

• Python 3.11+ و pip
• حساب KoboToolbox مع نموذج واحد منشور على الأقل. الخادم المجتمعي (kf.kobotoolbox.org) يفي بالغرض.
• قاعدة بيانات Postgres 15+. للاختبار، يكفي حاوية Docker محلية. للإنتاج، تعمل نسخة VPS بقيمة 6 دولارات شهرياً من Hetzner أو DigitalOcean.
• مفتاح API من أي مزود LLM متوافق مع OpenAI (نستخدم Claude عبر Anthropic، لكن الكود لا يعتمد على المزود)
• إلمام بـ requests و pandas وأساسيات SQL

معمارية خط المعالجة

إليك خط المعالجة الكامل من البداية إلى النهاية الذي ستبنيه. كل صندوق قابل للاستبدال بشكل مستقل.

flowchart LR
  A[Field Teams<br/>Sidi Bouzid, Tataouine,<br/>Mafraq, Casablanca] -->|XLSForm offline| B[KoboToolbox<br/>Server]
  B -->|v2 REST API| C[Python ETL<br/>nightly cron]
  C -->|append-only| D[(Postgres<br/>warehouse)]
  D --> E[AI Agent Layer<br/>Anomaly + Summary]
  E -->|writes flags| D
  D --> F[WeasyPrint<br/>PDF Renderer]
  F --> G[Donor PDF<br/>EN/FR/AR]
  D --> H[Web Dashboard<br/>internal team]

قاعدتان من قواعد التصميم تهمان فعلاً:

• المستودع هو مصدر الحقيقة. كل مستهلك لاحق (تقارير PDF، لوحات تحكم، وكلاء) يقرأ من Postgres. لا شيء يقرأ مباشرة من Kobo سوى ETL.
• الذكاء الاصطناعي يكتب علامات، ولا يعيد كتابة البيانات أبداً. يمكن للوكلاء وضع علامة على الإرسال كمشبوه أو إرفاق ملخص به. لا يمكنهم تعديل قيم المؤشرات الأساسية.

الخطوة 1: أساسيات KoboToolbox API

يكشف KoboToolbox عن REST API على https://kf.kobotoolbox.org/api/v2/. تتم المصادقة باستخدام رمز طويل الأمد، يتم توليده من صفحة إعدادات حسابك.

ثلاث نقاط نهاية مهمة لخط المعالجة الخاص بنا:

• GET /api/v2/assets/{asset_uid}/ — بيانات وصفية للنموذج، والمخطط، وعنوان URL للنشر
• GET /api/v2/assets/{asset_uid}/data/ — بيانات الإرسال المقسمة إلى صفحات بصيغة JSON
• GET /api/v2/assets/{asset_uid}/data/{submission_id}/attachments/{attachment_id}/ — المرفقات (الصور، الصوت)

الـ asset_uid هو المعرف القصير في عنوان URL للنموذج الخاص بك — يبدو مثل aJk8Zq3wY7nF2cQv9pXt. حد المعدل حوالي 60 طلباً في الدقيقة. الإرسالات مقسمة إلى صفحات عبر معاملات الاستعلام ?start=N&limit=M.

احصل على الرمز المميز الخاص بك (افعل هذا مرة واحدة، وقم بتخزينه كمتغير بيئة):

curl -X POST https://kf.kobotoolbox.org/token/ \
  -d "username=YOUR_USER&password=YOUR_PASS"
# returns: {"token":"a1b2c3..."}
 
export KOBO_TOKEN="a1b2c3..."
export KOBO_ASSET="aJk8Zq3wY7nF2cQv9pXt"

الخطوة 2: ETL بلغة Python — سحب الإرسالات

إليك جوهر ETL. إنه ممل عمداً: اسحب JSON، وقم بـ upsert في Postgres، وسجّل ما فعلته. احفظه باسم etl_kobo.py.

import os
import json
import logging
from datetime import datetime, timezone
import requests
import pandas as pd
from sqlalchemy import create_engine, text
 
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
log = logging.getLogger(__name__)
 
KOBO_BASE = "https://kf.kobotoolbox.org/api/v2"
KOBO_TOKEN = os.environ["KOBO_TOKEN"]
KOBO_ASSET = os.environ["KOBO_ASSET"]
DB_URL = os.environ["DATABASE_URL"]  # postgresql://user:pass@host/db
 
HEADERS = {"Authorization": f"Token {KOBO_TOKEN}"}
engine = create_engine(DB_URL, pool_pre_ping=True)
 
 
def fetch_submissions(asset_uid: str, page_size: int = 30000) -> list[dict]:
    """Pull all submissions for a deployed form, paginated."""
    url = f"{KOBO_BASE}/assets/{asset_uid}/data/"
    out, start = [], 0
    while True:
        r = requests.get(
            url,
            headers=HEADERS,
            params={"start": start, "limit": page_size, "format": "json"},
            timeout=60,
        )
        r.raise_for_status()
        batch = r.json().get("results", [])
        if not batch:
            break
        out.extend(batch)
        log.info("fetched %d (total %d)", len(batch), len(out))
        if len(batch) < page_size:
            break
        start += page_size
    return out
 
 
def upsert_raw(submissions: list[dict]) -> int:
    """Append raw payloads to submissions table. Idempotent on _uuid."""
    if not submissions:
        return 0
    rows = [
        {
            "uuid": s["_uuid"],
            "submitted_at": s["_submission_time"],
            "asset_uid": KOBO_ASSET,
            "payload": json.dumps(s, ensure_ascii=False),
            "ingested_at": datetime.now(timezone.utc).isoformat(),
        }
        for s in submissions
    ]
    with engine.begin() as conn:
        conn.execute(
            text("""
                INSERT INTO submissions (uuid, submitted_at, asset_uid, payload, ingested_at)
                VALUES (:uuid, :submitted_at, :asset_uid, :payload, :ingested_at)
                ON CONFLICT (uuid) DO NOTHING
            """),
            rows,
        )
    return len(rows)
 
 
if __name__ == "__main__":
    subs = fetch_submissions(KOBO_ASSET)
    n = upsert_raw(subs)
    log.info("ingested %d submissions (duplicates skipped)", n)

ترحيل المخطط المقابل:

CREATE TABLE IF NOT EXISTS submissions (
    uuid          TEXT PRIMARY KEY,
    submitted_at  TIMESTAMPTZ NOT NULL,
    asset_uid     TEXT NOT NULL,
    payload       JSONB NOT NULL,
    ingested_at   TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS submissions_submitted_at_idx ON submissions(submitted_at);
CREATE INDEX IF NOT EXISTS submissions_asset_uid_idx ON submissions(asset_uid);

شغّله كل ليلة باستخدام cron:

30 2 * * * cd /opt/ngo-etl && /usr/bin/python3 etl_kobo.py >> /var/log/kobo-etl.log 2>&1

هذه هي كامل طبقة الاستيعاب. خمسون سطراً، بدون اشتراك Zapier، قابلة للتكرار بالكامل.

الخطوة 3: تسطيح البيانات إلى مؤشرات

payload JSONB الخام هو مسار التدقيق الخاص بك. للوحات التحكم وتقارير PDF للممولين، تريد جدول مؤشرات مسطح. عرّف المؤشرات في ملف YAML (واحد لكل قالب ممول) وقم بتشغيل عملية تسطيح بعد الاستيعاب.

# indicators.py — runs after etl_kobo.py
from sqlalchemy import text
import pandas as pd
 
INDICATOR_SQL = """
INSERT INTO indicators (uuid, region, beneficiary_count, indicator_code, value, captured_at)
SELECT
    s.uuid,
    s.payload->>'region' AS region,
    (s.payload->>'num_beneficiaries')::int AS beneficiary_count,
    'OUT-1.2-cash-distribution' AS indicator_code,
    (s.payload->>'amount_tnd')::numeric AS value,
    s.submitted_at
FROM submissions s
WHERE s.payload->>'form_type' = 'cash_distribution'
ON CONFLICT (uuid, indicator_code) DO UPDATE
   SET value = EXCLUDED.value, captured_at = EXCLUDED.captured_at;
"""
 
with engine.begin() as conn:
    conn.execute(text(INDICATOR_SQL))

لماذا تخزين المؤشرات بشكل مسطح؟ لأن الممولين يطلبون "أرني الحساب" في عمليات التدقيق، وعرض SQL واحد يمكن للجميع قراءته هو إجابة أفضل من دفتر Jupyter لا يمكن لأحد العثور عليه.

الخطوة 4: طبقة الذكاء الاصطناعي — الشذوذ وملخصات الملاحظات الصوتية

يكسب وكيلان قيمتهما على نطاق المنظمات غير الحكومية: مُعَلِّم الشذوذ الذي يعمل بعد كل ETL، ومُلخِّص النصوص الذي يحول الملاحظات الصوتية العربية/الفرنسية إلى نقاط إنجليزية لسرديات الممولين.

# agents.py
import os
import anthropic
from sqlalchemy import text
 
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
ANOMALY_PROMPT = """You are an M&E data quality assistant for a humanitarian NGO.
Below are this week's indicator values vs the rolling 8-week baseline.
Flag any value that is more than 2 standard deviations from the baseline.
Return a JSON list of {{indicator, region, value, baseline_mean, severity}}.
 
Data:
{rows}
"""
 
def flag_anomalies():
    with engine.begin() as conn:
        df = pd.read_sql(
            "SELECT region, indicator_code, value, captured_at "
            "FROM indicators WHERE captured_at > now() - interval '8 weeks'",
            conn,
        )
    msg = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": ANOMALY_PROMPT.format(rows=df.to_csv(index=False))}],
    )
    flags = msg.content[0].text
    with engine.begin() as conn:
        conn.execute(text("INSERT INTO agent_flags(kind, payload, created_at) "
                          "VALUES ('anomaly', :p, now())"), {"p": flags})
 
 
def summarise_voice_note(audio_path: str, lang_hint: str = "ar") -> str:
    """Whisper transcription + LLM summary in English."""
    with open(audio_path, "rb") as f:
        transcript = client.audio.transcribe(file=f, language=lang_hint).text  # pseudo
    summary = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"Summarise this field interview in 3 English bullets for a donor report:\n\n{transcript}",
        }],
    ).content[0].text
    return summary

لا يكتب الوكيل أبداً في indicators. يكتب في agent_flags، وهو جدول منفصل تعرضه لوحة التحكم في الشريط الجانبي. يقرر البشر ما يفعلونه بالعلامات.

بالنسبة لتلخيص النصوص متعدد اللغات، هنا حيث يحدث السحر للمنظمات غير الحكومية في المغرب العربي الناطق بالفرنسية. يسجل فريق ميداني في تطاوين ملاحظة صوتية مدتها 90 ثانية بالدارجة التونسية يشرح فيها سبب تأخر التوزيع. يقوم خط المعالجة بنسخها (يتعامل Whisper مع العربية والفرنسية بشكل جيد)، ثم يطلب من LLM إنتاج ثلاث نقاط إنجليزية يقوم مسؤول الرصد والتقييم بلصقها في سردية USAID. التكلفة الإجمالية: حوالي 0.02 دولار لكل ملاحظة صوتية.

لمعرفة المزيد حول بناء طبقات الوكلاء كهذه، راجع المقال المرافق لنا حول لوحات تحكم PM/QA بالذكاء الاصطناعي لـ GitLab — نفس النمط، مجال مختلف.

الخطوة 5: تقارير PDF للممولين بـ EN / FR / AR

لا يقوم الممول بتسجيل الدخول إلى لوحة التحكم الخاصة بك. يريد الممول PDF في قالبه الخاص، بلغته، في الموعد النهائي. يحول WeasyPrint HTML + CSS إلى تقارير PDF بجودة الطباعة، ويتعامل مع العربية RTL بشكل أصلي.

# render_pdf.py
from jinja2 import Environment, FileSystemLoader
from weasyprint import HTML
import pandas as pd
 
env = Environment(loader=FileSystemLoader("templates"))
 
def render_report(lang: str, period: str, output: str):
    tmpl = env.get_template(f"donor_report_{lang}.html.j2")
    with engine.begin() as conn:
        df = pd.read_sql(
            text("SELECT indicator_code, region, sum(value) AS total "
                 "FROM indicators WHERE captured_at >= :start "
                 "GROUP BY indicator_code, region"),
            conn, params={"start": period},
        )
    html = tmpl.render(rows=df.to_dict(orient="records"), period=period, lang=lang)
    HTML(string=html).write_pdf(output)
 
# One template, three renders:
for lang in ("en", "fr", "ar"):
    render_report(lang, "2026-04-01", f"reports/Q2-2026-{lang}.pdf")

يستخدم قالب Jinja2 dir="rtl" عندما يكون lang == "ar"، ومجموعة خطوط مختلفة (Noto Naskh Arabic) للنسخة العربية، وهيكل جدول الإطار المنطقي الفعلي للممول. ابنِ قالباً واحداً لكل ممول، وأعد استخدامه إلى الأبد.

مقارنة التكاليف

لبرنامج INGO نموذجي مع مسؤول رصد وتقييم واحد، وثلاثة موظفين ميدانيين، واثنين من موظفي المكتب، واثنين من ممثلي الممولين، إليك الفاتورة الشهرية:

يفوز خط المعالجة مفتوح المصدر أيضاً في قابلية النقل. عندما ينتهي برنامج الممول بعد ثمانية عشر شهراً ويريد ممول جديد نفس البيانات بجدول زمني مختلف، فإن قاعدة بيانات Postgres الخاصة بك تنتقل معك. أما نموذجك الدلالي في Power BI فلا.

قائمة التحقق للإنتاج

قبل توجيه هذا إلى برنامج ممول حقيقي:

• حساب خدمة لرمز Kobo، وليس حساباً شخصياً
• نسخ احتياطية لـ Postgres إلى تخزين متوافق مع S3 (Hetzner Storage Box بقيمة 4 دولارات شهرياً لـ 1 تيرابايت)
• نبضات قلب لمهام cron — نستخدم الطبقة المجانية من healthchecks.io
• أمان على مستوى الصف في Postgres حتى لا يستطيع دور لوحة التحكم قراءة أعمدة PII
• ملف pyproject.toml يثبّت كل تبعية (requests، pandas، sqlalchemy، weasyprint، anthropic)
• علامة dry-run على agents.py حتى تتمكن من مراجعة مخرجات LLM قبل أن تصل إلى قاعدة البيانات

إلى أين تذهب بعد ذلك

بمجرد تشغيل خط المعالجة، التوسعات الواضحة هي:

• خادم MCP فوق Postgres حتى يتمكن المدير القطري من سؤال "كم عدد المستفيدين الذين وصلنا إليهم في تطاوين الأسبوع الماضي؟" من Claude Desktop بالعربية
• لوحة تحكم ويب لفريق الرصد والتقييم (نشحن إعداداً بـ Next.js + Tremor؛ يعمل SvelteKit + ECharts بنفس الجودة)
• استيعاب برمجي لقوالب الممولين — ربط الإطار المنطقي لممول جديد برموز المؤشرات الخاصة بك عبر YAML

هل تريد منا أن نبنيه لك؟

إذا كنت تدير برنامجاً ممولاً وخط المعالجة أعلاه هو ما تتمنى لو كان لديك، أخبرنا عن برنامجك. نسلّم عادةً لوحة تحكم KoboToolbox عاملة مع تقارير PDF متعددة اللغات للممولين في غضون أسبوعين، من الإطلاق إلى البيانات الحية. اعتباراً من حوالي 4,500 دولار لبرنامج ممول واحد؛ أقل إذا كانت نماذج XLSForms الخاصة بك مستقرة بالفعل.

اقرأ السياق الاستراتيجي في مقالنا الركيزة: لوحات تحكم الرصد والتقييم بالذكاء الاصطناعي للمنظمات غير الحكومية في منطقة MENA.