Récupération des tâches par Polling

Présentation

Cette documentation démontre l'intégration de l'API Koncile en utilisant la bibliothèque requests de Python via une méthode de récupération de tâches basée sur le polling.

Étapes clés

Le script illustre trois processus principaux :

1. Téléchargement de fichier — Les fichiers sont envoyés à un endpoint API dédié avec les paramètres de Template d'extraction et d'ID de Folder.

2. Récupération de l'ID de tâche — Une fois traité, l'API génère un ID de tâche pour suivre le statut.

3. Surveillance du traitement de la tâche — Une fonction de polling vérifie périodiquement la complétion de la tâche avant d'extraire les résultats.

Détails d'implémentation

L'exemple Python fourni sert de base pour l'intégration automatisée de Koncile. Les utilisateurs doivent remplacer les valeurs placeholder incluant your-api-key, folder-id, template-id et path-to-file par leurs propres identifiants et références de fichiers.

Workflow principal

Le script effectue les actions suivantes :

• Initialise la configuration API avec les en-têtes d'authentification

• Télécharge les fichiers vers https://api.koncile.ai/v1/upload_file/

• Récupère les IDs de tâches depuis la réponse

• Implémente une boucle de polling qui interroge https://api.koncile.ai/v1/fetch_tasks_results/

• Continue la vérification jusqu'à ce que le statut de la tâche change de "IN PROGRESS"

• Retourne les données extraites incluant les General fields et les Line fields

L'intervalle de polling est défini à 2 secondes entre les vérifications de statut pour équilibrer la réactivité avec la charge API.

Exemple d'implémentation

import requests
import time

API_KEY = "your-api-key"
BASE_URL = "https://api.koncile.ai/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

def upload_file(folder_id, template_id, file_path):
    with open(file_path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/upload_file/",
            headers=HEADERS,
            files={"files": f},
            params={"folder_id": folder_id, "template_id": template_id}
        )
    return response.json()["task_ids"][0]

def poll_for_results(task_id, timeout=300):
    start_time = time.time()

    while time.time() - start_time < timeout:
        response = requests.get(
            f"{BASE_URL}/fetch_tasks_results/",
            headers=HEADERS,
            params={"task_id": task_id}
        )
        result = response.json()

        if result["status"] == "DONE":
            return result
        elif result["status"] == "FAILED":
            raise Exception(f"Extraction échouée : {result['status_message']}")

        time.sleep(2)

    raise TimeoutError("Délai de polling dépassé")

# Utilisation
task_id = upload_file(folder_id=1, template_id=1, file_path="facture.pdf")
result = poll_for_results(task_id)

print(f"Document ID : {result['document_id']}")
print(f"General Fields : {result['General_fields']}")
print(f"Line Fields : {result['Line_fields']}")