Using Api to create tasks

I’m trying to create a script to import a cdv file directly into task …

In my CSV file, I put the Number of a Project, time start, duration, and description …

The python Script get the list of project, find out the project ID, than ask to create a task with those information…

When I run the script, a new task is created, but it’s empty… None of the informations in putted in it goes into the task…

What did I missed ???

Here’s my Python Code !

import csv
import requests
import json

# URL de l'API REST
API_URL = 'MYSERVER-URL/api/v1'

# Fichier CSV à lire
CSV_FILE = 'data.csv'

# En-têtes du fichier CSV
HEADERS = ['project_code','date', 'description', 'hours_worked', 'project_id']

# Les informations d'authentification pour accéder à l'API
API_KEY = 'MY_API_KEY'

def get_project_id(code):
    # Interroge l'API pour obtenir les informations sur les projets
    headers = {'X-API-Token': API_KEY}
    
    params = {
        'page': 1,
        'per_page': 9999
    }

    
    response = requests.get(f'{API_URL}/projects', headers=headers, params=params)
    print(response.text)
     # Vérifie si la requête a réussi
    if response.status_code == 200:
        # Charge les données JSON retournées
        projects = response.json()['data']
        #print(projects.text)
        # Recherche le projet avec le code de projet correspondant
        for project in projects:
            #print(project['number'])
            
            if project['number'] == code:
                return project['id']
			
    return None
    
# Lit les lignes du fichier CSV
with open(CSV_FILE, 'r') as file:
    reader = csv.DictReader(file, fieldnames=HEADERS)

    # Pour chaque ligne du fichier CSV
    for row in reader:
        # Recherche l'ID de projet correspondant
        project_id = get_project_id(row['project_code'])
        if not project_id:
            print(f"Failed to find project with code '{row['project_code']}'")
            continue
        print(project_id)
        # Convertit la ligne en dictionnaire JSON
        data = {
             'task': {
                'start_time': row['date'],
                'client_id': row['project_code'],
                'description': 'allo',
                'duration': float(row['hours_worked']),
                'project_id': project_id,
            }
        }
        print(data)
        # Envoie la donnée à l'API REST
        headers = {'X-API-Token': API_KEY}
        response = requests.post(f'{API_URL}/tasks', headers=headers, json=data)

        # Vérifie si la requête a réussi
        if response.status_code == 200:
            print(f"Successfully sent data for date '{row['date']}'")
            print({response.text})
        else:
            print(f"Failed to send data for date '{row['date']}': {response.text}")

Some precision… Bellow, what I submit and what the API return me

submit :
{'task': {'start_time': '2023-01-13', 'description': 'allo', 'duration': 1.0, 'project_id': '4y1aKrReQG'}}

Answer :

Successfully sent data for date '2023-01-13'
{'{\n    "data": {\n        "id": "4open14d7A",\n        "user_id": "VolejRejNm",\n        "assigned_user_id": "",\n        "number": "T-2023-1370",\n        "description": "",\n        "duration": 0,\n        "rate": 0,\n        "created_at": 1675985600,\n        "updated_at": 1675985600,\n        "archived_at": 0,\n        "invoice_id": "",\n        "client_id": "",\n        "project_id": "",\n        "is_deleted": false,\n        "time_log": "[]",\n        "is_running": false,\n        "custom_value1": "",\n        "custom_value2": "",\n        "custom_value3": "",\n        "custom_value4": "",\n        "status_id": "",\n        "status_sort_order": 0,\n        "is_date_based": false,\n        "status_order": null,\n        "documents": []\n    }\n}'}

Hi,

start_time isn’t a valid field. You can either set the time_log array (as the app does) or pass a time_details array (as is used by Zapier).

I finally got something that work with this code Bellow…

If it can help someone to create his own script… It’s pretty easy to adapt …

import csv
import requests
import json
import datetime
import time
import sys
import os
from datetime import datetime
from pathlib import Path

# URL de l'API REST
API_URL = 'https://your.invoiceninja.domain/api/v1'

if len(sys.argv) > 1:
    file_path = sys.argv[1]
    if os.path.exists(file_path):
       CSV_FILE = file_path
    else:
        print(f"File not found: {file_path}")
else:
    print("Usage: python script.py file_path")
    sys.exit()


# En-têtes du fichier CSV
HEADERS = ['project_code','date', 'description', 'hours_worked', 'project_id']

# Les informations d'authentification pour accéder à l'API
API_KEY = 'YOUR_API_KEY'

# Chemin pour le Log File
path_log_file = Path(CSV_FILE).parent

# Ouvrir un fichier log
log_file = open(f'{path_log_file}/transmission-log.log', "w")

# Rediriger la sortie standard vers le fichier log
sys.stdout = log_file

def get_project_id(code):
    # Interroge l'API pour obtenir les informations sur les projets
    headers = {'X-API-Token': API_KEY}
    
    params = {
        'page': 1,
        'per_page': 9999
    }

    
    response = requests.get(f'{API_URL}/projects', headers=headers, params=params)
    #print(response.text)
     # Vérifie si la requête a réussi
    if response.status_code == 200:
        # Charge les données JSON retournées
        projects = response.json()['data']
        #print(projects.text)
        # Recherche le projet avec le code de projet correspondant
        for project in projects:
            #print(project['number'])
            
            if project['number'] == code:
                return project['id']

    return None
    
def get_client_id(code):
    # Interroge l'API pour obtenir les informations sur les projets
    headers = {'X-API-Token': API_KEY}
    
    params = {
        'page': 1,
        'per_page': 9999
    }

    
    response = requests.get(f'{API_URL}/clients', headers=headers, params=params)
    #print(response.text)
     # Vérifie si la requête a réussi
    if response.status_code == 200:
        # Charge les données JSON retournées
        clients = response.json()['data']
        #print(clients)
        # Recherche le projet avec le code de projet correspondant
        for client in clients:
            #print(client['number'])
            
            if client['number'] == code:
                return client['id']

    return None
    
# Lit les lignes du fichier CSV
with open(CSV_FILE, 'r') as file:
    reader = csv.DictReader(file, fieldnames=HEADERS)

    # Pour chaque ligne du fichier CSV
    for index, row in enumerate(reader):
        #on supprime le caractère UTF-8 s'il est présent
        row['project_code'] = row['project_code'].lstrip("\ufeff")
        
        print(" ")
        print(f"Traitement de la ligne :", index)
        # Convertit la ligne en dictionnaire JSON
        format = "%Y-%m-%d %H:%M"
        date_debut = datetime.strptime(row['date'], format)
        date_debut_timestamp = int(date_debut.timestamp())
        duree = float(row['hours_worked']) * 60 * 60
        date_fin_timestamp = date_debut_timestamp + int(duree)
        time_log_var = [[date_debut_timestamp,date_fin_timestamp]]
        
        if row['project_code'].startswith("PR-"):
    
            # Recherche l'ID de projet correspondant
            project_id = get_project_id(row['project_code'])
        
            if not project_id:
                print(f"Aucun projet existant avec ce numéro '{row['project_code']}'")
                print(row)
                print(data)
                continue

            data = {
                'description': row['description'],
                'project_id': project_id,
                'time_log': json.dumps(list(time_log_var)),
            }
        
        elif row['project_code'].startswith("C-"):
              # Recherche l'ID de client correspondant
            client_id = get_client_id(row['project_code'])
            
            if not client_id:
                print(f"Aucun client existant avec ce numéro '{row['project_code']}'")
                print(row)
                print(data)
                continue
                
            data = {
                'description': row['description'],
                'client_id': client_id,
                'time_log': json.dumps(list(time_log_var)),
            }
        
        else : 
            print(f"Ne peut identifier a quoi assigner a partir de ce numero '{row['project_code']}'")
            print(row)
            print(data)
            continue
             
        # Envoie la donnée à l'API REST
        headers = {'X-API-Token': API_KEY}
        response = requests.post(f'{API_URL}/tasks', headers=headers, json=data)

        # Vérifie si la requête a réussi
        if response.status_code == 200:
            print(f"Information Transmise a Jarvis! Ligne: '{index}'")
        else:
            print(f"Erreur lors de la transmission a Jarvis '{index}': {response.text}")
            print(row)
            print(data)
            print(response)
            
# Fermer le fichier log
log_file.close()

# Rediriger la sortie standard vers la console
sys.stdout = sys.__stdout__
2 Likes