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}")
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__