Construire un tableau de bord KoboToolbox propulsé par l'IA pour ONG : Kobo API vers Postgres, rapports bailleurs PDF en 3 langues

Le problème des exports KoboToolbox

Si vous êtes chargé S&E dans une ONG tunisienne, marocaine ou libanaise gérant un programme bailleur, votre semaine ressemble probablement à ceci :

• Lundi matin, vous téléchargez un export CSV depuis KoboToolbox.
• L'équipe terrain à Sidi Bouzid soumet trente formulaires supplémentaires lundi après-midi. Votre CSV est déjà périmé.
• L'échéance de reporting USAID est vendredi. Le bailleur veut un PDF en anglais, le comité de pilotage veut les mêmes chiffres en français, et le partenaire local les veut en arabe.
• Vous passez le vendredi soir dans Excel à réconcilier trois exports, puis dans PowerPoint à recopier les chiffres dans un template bailleur.

Ce n'est pas un problème d'outil — KoboToolbox fait très bien ce pour quoi il a été conçu. C'est un problème de pipeline. Kobo est un moteur de formulaires, pas une plateforme de reporting. La solution est de traiter chaque soumission Kobo comme un fait dans un véritable entrepôt de données, puis d'empiler reporting et IA par-dessus.

Ce tutoriel est le compagnon pratique de notre article pilier sur les tableaux de bord IA pour le S&E des ONG MENA. Lisez-le pour le raisonnement architectural ; revenez ici quand vous voulez réellement construire l'outil.

Ce que vous allez construire

À la fin de ce tutoriel, vous aurez :

1. Un job ETL Python qui récupère les soumissions KoboToolbox via l'API REST v2 sur un planning nocturne
2. Un entrepôt Postgres avec trois tables : soumissions brutes, indicateurs dénormalisés, et journal d'audit
3. Une couche d'agents IA qui flag les anomalies et résume les notes vocales arabes/françaises en anglais pour les rapports bailleurs
4. Un moteur de reporting PDF basé sur WeasyPrint qui produit le même jeu de données en EN, FR et AR depuis un seul template Jinja2
5. Un coût opérationnel total inférieur à 30 $/mois pour un programme jusqu'à 50 000 soumissions

Tout est open source. Le seul composant payant correspond aux appels API du LLM (quelques dollars par mois aux volumes ONG).

Prérequis

Avant de commencer :

• Python 3.11+ et pip
• Un compte KoboToolbox avec au moins un formulaire déployé. Le serveur communautaire (kf.kobotoolbox.org) suffit.
• Une base Postgres 15+. Pour tester, un conteneur Docker local convient. Pour la production, une instance VPS à 6 $/mois chez Hetzner ou DigitalOcean fait l'affaire.
• Une clé API d'un fournisseur LLM compatible OpenAI (nous utilisons Claude via Anthropic, mais le code est agnostique au fournisseur)
• Une familiarité avec requests, pandas et le SQL de base

Architecture du pipeline

Voici le pipeline complet que vous allez construire. Chaque case est remplaçable indépendamment.

flowchart LR
  A[Équipes terrain<br/>Sidi Bouzid, Tataouine,<br/>Mafraq, Casablanca] -->|XLSForm hors-ligne| B[Serveur<br/>KoboToolbox]
  B -->|API REST v2| C[ETL Python<br/>cron nocturne]
  C -->|append-only| D[(Postgres<br/>entrepôt)]
  D --> E[Couche IA<br/>Anomalies + Résumés]
  E -->|écrit des flags| D
  D --> F[WeasyPrint<br/>Moteur PDF]
  F --> G[PDF bailleur<br/>EN/FR/AR]
  D --> H[Tableau de bord web<br/>équipe interne]

Les deux règles de design qui comptent :

• L'entrepôt est la source de vérité. Chaque consommateur aval (PDF, dashboards, agents) lit depuis Postgres. Rien ne lit directement depuis Kobo, sauf l'ETL.
• L'IA écrit des flags, jamais ne réécrit les données. Les agents peuvent marquer une soumission comme suspecte ou y attacher un résumé. Ils ne peuvent pas muter les valeurs d'indicateur sous-jacentes.

Étape 1 : les bases de l'API KoboToolbox

KoboToolbox expose une API REST sur https://kf.kobotoolbox.org/api/v2/. L'authentification se fait via un token longue durée, généré depuis la page des paramètres de votre compte.

Trois endpoints comptent pour notre pipeline :

• GET /api/v2/assets/{asset_uid}/ — métadonnées du formulaire, schéma, URL de déploiement
• GET /api/v2/assets/{asset_uid}/data/ — données des soumissions paginées en JSON
• GET /api/v2/assets/{asset_uid}/data/{submission_id}/attachments/{attachment_id}/ — pièces jointes (images, audio)

L'asset_uid est l'ID court présent dans l'URL de votre formulaire — quelque chose comme aJk8Zq3wY7nF2cQv9pXt. La limite de débit est d'environ 60 requêtes/minute. Les soumissions sont paginées via les paramètres ?start=N&limit=M.

Récupérez votre token (à faire une fois, à stocker en variable d'environnement) :

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

Étape 2 : ETL Python — récupérer les soumissions

Voici le cœur de l'ETL. Volontairement ennuyeux : récupérer du JSON, upsert dans Postgres, journaliser ce qui a été fait. À sauvegarder sous 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]:
    """Récupère toutes les soumissions d'un formulaire déployé, paginées."""
    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("récupéré %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 les payloads bruts dans la table submissions. Idempotent sur _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("ingéré %d soumissions (doublons ignorés)", n)

La migration de schéma correspondante :

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

Lancez-le tous les soirs avec cron :

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

C'est toute la couche d'ingestion. Cinquante lignes, aucun abonnement Zapier, entièrement reproductible.

Étape 3 : dénormaliser en indicateurs

Le payload JSONB brut est votre piste d'audit. Pour les dashboards et les PDF bailleurs, vous voulez une table d'indicateurs à plat. Définissez les indicateurs dans un fichier YAML (un par template bailleur) et lancez une passe de dénormalisation après l'ingestion.

# indicators.py — s'exécute après 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))

Pourquoi stocker les indicateurs dénormalisés ? Parce que les bailleurs demandent « montrez-moi le calcul » en audit, et qu'une vue SQL unique que tout le monde peut lire est une meilleure réponse qu'un notebook Jupyter introuvable.

Étape 4 : la couche IA — anomalies et résumés de notes vocales

Deux agents gagnent leur place à l'échelle ONG : un agent d'anomalies qui tourne après chaque ETL et un agent de résumé de transcriptions qui transforme les notes vocales arabes/françaises en puces anglaises pour les narratifs bailleurs.

# agents.py
import os
import anthropic
from sqlalchemy import text
 
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
ANOMALY_PROMPT = """Tu es un assistant qualité de données S&E pour une ONG humanitaire.
Voici les valeurs d'indicateurs de la semaine vs le baseline glissant sur 8 semaines.
Flag toute valeur qui s'écarte de plus de 2 écarts-types du baseline.
Retourne une liste JSON de {{indicator, region, value, baseline_mean, severity}}.
 
Données :
{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:
    """Transcription Whisper + résumé LLM en anglais."""
    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"Résume cette interview terrain en 3 puces anglaises pour un rapport bailleur :\n\n{transcript}",
        }],
    ).content[0].text
    return summary

L'agent n'écrit jamais dans indicators. Il écrit dans agent_flags, une table séparée que le dashboard affiche en sidebar. Les humains décident quoi faire des flags.

Pour le résumé multilingue de transcriptions, c'est ici que la magie opère pour les ONG du Maghreb francophone. Une équipe terrain à Tataouine enregistre une note vocale de 90 secondes en darija tunisien expliquant pourquoi une distribution a été retardée. Le pipeline la transcrit (Whisper gère bien l'arabe et le français), puis demande au LLM de produire trois puces anglaises que le chargé S&E colle dans le narratif USAID. Coût de bout en bout : environ 0,02 $ par note vocale.

Pour aller plus loin sur la construction de couches d'agents comme celle-ci, voyez notre article compagnon sur les dashboards PM/QA IA GitLab — même schéma, autre domaine.

Étape 5 : PDF bailleurs en EN / FR / AR

Le bailleur ne se connecte pas à votre dashboard. Le bailleur veut un PDF dans son template, dans sa langue, à la deadline. WeasyPrint transforme HTML + CSS en PDF prêts pour l'impression et gère nativement l'arabe 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)
 
# Un template, trois rendus :
for lang in ("en", "fr", "ar"):
    render_report(lang, "2026-04-01", f"reports/T2-2026-{lang}.pdf")

Le template Jinja2 utilise dir="rtl" quand lang == "ar", une pile de polices différente (Noto Naskh Arabic) pour la version AR, et la structure de tableau du cadre logique réel du bailleur. Construisez un template par bailleur, réutilisez-le indéfiniment.

Comparaison de coûts

Pour un programme INGO typique avec un chargé S&E, trois agents terrain, deux chargés de bureau et deux représentants bailleur, voici la facture mensuelle :

Le pipeline open source gagne aussi en portabilité. Quand le programme bailleur se termine dans dix-huit mois et qu'un nouveau bailleur veut les mêmes données sur un autre calendrier, votre base Postgres vous suit. Pas votre modèle sémantique Power BI.

Checklist de production

Avant de pointer ça vers un vrai programme bailleur :

• Compte de service pour le token Kobo, pas un compte personnel
• Sauvegardes de Postgres vers un stockage S3-compatible (Hetzner Storage Box à 4 $/mois pour 1 To)
• Heartbeat de cron — nous utilisons le tier gratuit de healthchecks.io
• Sécurité row-level dans Postgres pour que le rôle dashboard ne puisse pas lire les colonnes PII
• Un pyproject.toml qui épingle chaque dépendance (requests, pandas, sqlalchemy, weasyprint, anthropic)
• Un flag dry-run sur agents.py pour que vous puissiez revoir les sorties LLM avant qu'elles n'atterrissent en base

Pour aller plus loin

Une fois le pipeline en place, les extensions évidentes sont :

• Un serveur MCP au-dessus de Postgres pour que le directeur pays puisse demander « combien de bénéficiaires avons-nous atteints à Tataouine la semaine dernière ? » depuis Claude Desktop en arabe
• Un tableau de bord web pour l'équipe S&E (nous livrons un setup Next.js + Tremor ; SvelteKit + ECharts marche aussi bien)
• L'ingestion programmatique de templates bailleurs — mapper le cadre logique d'un nouveau bailleur vers vos codes d'indicateur via YAML

Vous voulez qu'on le construise pour vous ?

Si vous pilotez un programme bailleur et que le pipeline ci-dessus est ce que vous aimeriez avoir, parlez-nous de votre programme. Nous livrons typiquement un tableau de bord KoboToolbox fonctionnel avec PDF bailleurs multilingues en deux semaines, du kickoff aux données en direct. À partir de 4 500 $ pour un programme mono-bailleur ; moins si vos XLSForms sont déjà stables.

Lisez le contexte stratégique dans notre article pilier : tableaux de bord IA pour le S&E des ONG MENA.