Laurent Chedanne
485 lines
15 KiB
Python
485 lines
15 KiB
Python
import csv
|
|
import requests
|
|
import json
|
|
import uuid
|
|
import re
|
|
import base64
|
|
import sys
|
|
import mysql.connector
|
|
from unidecode import unidecode
|
|
|
|
# T1Do1bJCthWEt8Cjcwyz!
|
|
# testovea1
|
|
|
|
##########################################
|
|
# Config
|
|
##########################################
|
|
|
|
# Récupération de la config
|
|
with open('config.json', 'r') as f:
|
|
config = json.load(f)
|
|
|
|
mydb = mysql.connector.connect(
|
|
host=config['db_host'],
|
|
user=config['db_user'],
|
|
password=config['db_password'],
|
|
database=config['db_database']
|
|
)
|
|
|
|
cursor = mydb.cursor(buffered=True)
|
|
|
|
# Envoi de la requête HTTP de type POST
|
|
url = "https://nextcloud.crealead.com.crl-01.ovea.com/index.php/apps/timemanager/api/updateObjects"
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
# Ajout de l'authentification via Basic Authentication
|
|
#username = config['nc_username']
|
|
#password = config['nc_password']
|
|
#credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
|
|
#headers["Authorization"] = f"Basic {credentials}"
|
|
|
|
##########################################
|
|
# Fonctions
|
|
##########################################
|
|
|
|
def save_db():
|
|
cursor.fetchall()
|
|
mydb.commit()
|
|
|
|
|
|
def sanitize(chain):
|
|
return re.sub("[^A-Za-z0-9 ]", "", unidecode(chain).strip().capitalize())
|
|
|
|
|
|
def sanitize_date(chain):
|
|
return re.sub("[^A-Za-z0-9 \-\:]", "", unidecode(chain).strip().capitalize())
|
|
|
|
|
|
def check_time_already_exist(task_name, project_name, client_name, user_id, dstart, hstart, dend, hend):
|
|
|
|
query = ("select task_uuid from oc_timemanager_time where user_id=\""+user_id+"\" AND status!=\"deleted\" AND start=\""+dstart+" "+hstart+"\" AND end=\""+dend+" "+hend+"\"")
|
|
cursor.execute(query)
|
|
|
|
# Test s'il existe une entrée avec les mêmes heure début et fin
|
|
for task_uuid in cursor:
|
|
|
|
query_task = ("select name,project_uuid from oc_timemanager_task where uuid=\""+task_uuid[0].decode('UTF-8')+"\" AND status!=\"deleted\"")
|
|
cursor.execute(query_task)
|
|
|
|
for query_task_data in cursor:
|
|
task_name_str = sanitize(query_task_data[0].decode('UTF-8'))
|
|
|
|
# Test nom de la tâche
|
|
if task_name_str == task_name:
|
|
project_uuid = query_task_data[1].decode('UTF-8')
|
|
|
|
query_project = ("select name,client_uuid from oc_timemanager_project where uuid=\""+project_uuid+"\" AND status!=\"deleted\"")
|
|
cursor.execute(query_project)
|
|
|
|
for query_project_data in cursor:
|
|
project_name_str = sanitize(query_project_data[0].decode('UTF-8'))
|
|
|
|
# Test nom du projet
|
|
if project_name_str == project_name:
|
|
client_uuid = query_project_data[1].decode('UTF-8')
|
|
|
|
query_client = ("select name from oc_timemanager_client where uuid=\""+client_uuid+"\" AND status!=\"deleted\"")
|
|
cursor.execute(query_client)
|
|
|
|
for query_client_data in cursor:
|
|
client_name_str = sanitize(query_client_data[0].decode('UTF-8'))
|
|
|
|
# Test nom du client
|
|
if client_name_str == client_name:
|
|
return True
|
|
|
|
return False
|
|
|
|
def task_exist_by_name(name, project_uuid, user_id):
|
|
query = ("select name from oc_timemanager_task where user_id=\""+user_id+"\" AND status!=\"deleted\" AND project_uuid=\""+project_uuid+"\"")
|
|
cursor.execute(query)
|
|
|
|
for task_name in cursor:
|
|
task_name_str = sanitize(task_name[0].decode('UTF-8'))
|
|
if name == task_name_str:
|
|
save_db()
|
|
return True
|
|
|
|
save_db()
|
|
return False
|
|
|
|
|
|
def project_exist_by_name(name, client_uuid, user_id):
|
|
query = ("select name from oc_timemanager_project where user_id=\""+user_id+"\" AND status!=\"deleted\" AND client_uuid=\""+client_uuid+"\"")
|
|
cursor.execute(query)
|
|
|
|
for project_name in cursor:
|
|
project_name_str = sanitize(project_name[0].decode('UTF-8'))
|
|
if name == project_name_str:
|
|
save_db()
|
|
return True
|
|
|
|
save_db()
|
|
return False
|
|
|
|
def client_exist_by_name(name, user_id):
|
|
query = ("select name from oc_timemanager_client where user_id=\""+user_id+"\" AND status!=\"deleted\"")
|
|
cursor.execute(query)
|
|
|
|
for client_name in cursor:
|
|
client_name_str = sanitize(client_name[0].decode('UTF-8'))
|
|
if name == client_name_str:
|
|
save_db()
|
|
return True
|
|
|
|
save_db()
|
|
return False
|
|
|
|
def get_client_uuid_by_name(client_name, user_id):
|
|
query = ("select uuid from oc_timemanager_client where name=\""+client_name+"\" AND user_id=\""+user_id+"\"")
|
|
cursor.execute(query)
|
|
|
|
row = cursor.fetchone()
|
|
save_db()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return row[0].decode('utf-8')
|
|
|
|
|
|
def get_project_uuid_by_name(project_name, client_uuid, user_id):
|
|
query = ("select uuid from oc_timemanager_project where name=\""+project_name+"\" AND client_uuid=\""+client_uuid+"\" AND user_id=\""+user_id+"\"")
|
|
cursor.execute(query)
|
|
|
|
project_uuid = cursor.fetchone()[0]
|
|
save_db()
|
|
|
|
return project_uuid.decode('utf-8')
|
|
|
|
|
|
def get_task_uuid_by_name(task_name, project_uuid, user_id):
|
|
query = ("select uuid from oc_timemanager_task where name=\""+task_name+"\" AND project_uuid=\""+project_uuid+"\" AND user_id=\""+user_id+"\"")
|
|
cursor.execute(query)
|
|
|
|
task_uuid = cursor.fetchone()[0]
|
|
save_db()
|
|
|
|
return task_uuid.decode('utf-8')
|
|
|
|
|
|
def create_task(name, project_uuid):
|
|
# print("Création de la tâche " + name)
|
|
task_created = [{
|
|
"changed": "",
|
|
"commit": "",
|
|
"created": "",
|
|
"name": name,
|
|
"project_uuid" : project_uuid,
|
|
"note": None,
|
|
"status": None,
|
|
"uuid": ""
|
|
}]
|
|
|
|
req_data = {
|
|
"lastCommit": "",
|
|
"data": {
|
|
"clients": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"projects": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"tasks": {
|
|
"created": task_created,
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"times": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
}
|
|
}
|
|
}
|
|
|
|
response = requests.post(url, data=json.dumps(req_data), headers=headers)
|
|
|
|
# Vérification de la réponse
|
|
if response.status_code == 200:
|
|
pass
|
|
#print("Tâche créée avec succès.")
|
|
#print(json.dumps(response.json(), indent=1))
|
|
else:
|
|
print("Erreur de création task")
|
|
print(response.reason)
|
|
|
|
|
|
def create_project(name, client_uuid):
|
|
# print("Création du projet " + name)
|
|
project_created = [{
|
|
"changed": "",
|
|
"commit": "",
|
|
"created": "",
|
|
"name": name,
|
|
"client_uuid" : client_uuid,
|
|
"note": None,
|
|
"status": None,
|
|
"uuid": ""
|
|
}]
|
|
|
|
req_data = {
|
|
"lastCommit": "",
|
|
"data": {
|
|
"clients": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"projects": {
|
|
"created": project_created,
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"tasks": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"times": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
}
|
|
}
|
|
}
|
|
|
|
response = requests.post(url, data=json.dumps(req_data), headers=headers)
|
|
|
|
# Vérification de la réponse
|
|
if response.status_code == 200:
|
|
pass
|
|
# print("Projet créé avec succès.")
|
|
#print(json.dumps(response.json(), indent=1))
|
|
else:
|
|
print("Erreur de création project")
|
|
print(response.reason)
|
|
|
|
def create_client(name):
|
|
# print("Création du client " + name)
|
|
clients_created = [{
|
|
"changed": "",
|
|
"city": None,
|
|
"commit": "",
|
|
"created": "",
|
|
"email": None,
|
|
"name": name,
|
|
"note": None,
|
|
"phone": None,
|
|
"postcode": None,
|
|
"street": None,
|
|
"status": None,
|
|
"uuid": "",
|
|
"web": None,
|
|
}]
|
|
|
|
req_data = {
|
|
"lastCommit": "",
|
|
"data": {
|
|
"clients": {
|
|
"created": clients_created,
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"projects": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"tasks": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"times": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
}
|
|
}
|
|
}
|
|
|
|
response = requests.post(url, data=json.dumps(req_data), headers=headers)
|
|
|
|
# Vérification de la réponse
|
|
if response.status_code == 200:
|
|
try:
|
|
resp = response.json()
|
|
# Tenter d'extraire l'UUID renvoyé par l'API (structure attendue)
|
|
client_uuid = None
|
|
if isinstance(resp, dict):
|
|
client_created = resp.get('data', {}).get('clients', {}).get('created', [])
|
|
if client_created and isinstance(client_created, list):
|
|
client_uuid = client_created[0].get('uuid')
|
|
|
|
if client_uuid:
|
|
return client_uuid
|
|
# si pas d'uuid dans la réponse, retourner None (le caller gérera)
|
|
return None
|
|
except Exception:
|
|
return None
|
|
else:
|
|
print("Erreur de création client")
|
|
print(response.reason)
|
|
return None
|
|
|
|
|
|
def insert_task_time(name, project_name, client_name, task_uuid, user_id, dstart, hstart, dend, hend):
|
|
#print("Insertion temps de la tâche " + name)
|
|
#print(dstart + " " + hstart)
|
|
#print(dend + " " + hend)
|
|
|
|
# Vérification si l'entrée existe déjà
|
|
if check_time_already_exist(name, project_name, client_name, user_id, dstart, hstart, dend, hend):
|
|
print("Existe déjà !")
|
|
return
|
|
|
|
times_created = [{
|
|
"uuid": "",
|
|
"commit": "",
|
|
"changed": "",
|
|
"created": "",
|
|
"start": dstart + " " + hstart,
|
|
"end": dend + " " + hend,
|
|
"note": "",
|
|
"paymentStatus": None,
|
|
"task_uuid": task_uuid
|
|
}]
|
|
|
|
req_data = {
|
|
"lastCommit": "",
|
|
"data": {
|
|
"clients": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"projects": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"tasks": {
|
|
"created": [],
|
|
"updated": [],
|
|
"deleted": []
|
|
},
|
|
"times": {
|
|
"created": times_created,
|
|
"updated": [],
|
|
"deleted": []
|
|
}
|
|
}
|
|
}
|
|
|
|
response = requests.post(url, data=json.dumps(req_data), headers=headers)
|
|
|
|
# Vérification de la réponse
|
|
if response.status_code == 200:
|
|
pass
|
|
#print("Temps de tâche ajouté avec succès.")
|
|
#print(json.dumps(response.json(), indent=1))
|
|
else:
|
|
print("Erreur de création task time")
|
|
print(response.reason)
|
|
|
|
|
|
|
|
##########################################
|
|
# Main
|
|
##########################################
|
|
|
|
# Pour chaque fichier csv configuré dans config
|
|
for user_info in config['calendars']:
|
|
|
|
if 'user_name' not in user_info or user_info['user_name'] == '' or 'api_password' not in user_info or user_info['api_password'] == '':
|
|
print('Il manque des informations de connexion pour' + user_info['calendar_name'] + ': ' + user_info['user_name'] + ' ' + user_info['api_password'])
|
|
continue
|
|
|
|
user_id = user_info['user_name']
|
|
print("====================")
|
|
print("User: " + user_id)
|
|
print("====================")
|
|
|
|
# On modifie le header pour utiliser les bons identifiants
|
|
username = user_info['user_name']
|
|
password = user_info['api_password']
|
|
credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
|
|
headers["Authorization"] = f"Basic {credentials}"
|
|
|
|
# Récupération des données depuis le fichier CSV
|
|
data = []
|
|
with open(user_info['csv_name'], 'r') as file:
|
|
reader = csv.reader(file, delimiter=';')
|
|
headers_csv = next(reader)
|
|
for row in reader:
|
|
data.append(dict(zip(headers_csv, row)))
|
|
|
|
for item in data:
|
|
client_name = sanitize(item['Description'])
|
|
project_name = sanitize(item['Categories'])
|
|
task_name = sanitize(item['Titre'])
|
|
|
|
if len(client_name) < 1 and len(project_name) < 1:
|
|
continue
|
|
|
|
# Description => Client
|
|
client_uuid = None
|
|
if not client_exist_by_name(client_name, user_id):
|
|
print("Création du client {}".format(client_name))
|
|
created_uuid = create_client(client_name)
|
|
if created_uuid:
|
|
client_uuid = created_uuid
|
|
else:
|
|
# tentative de récupération depuis la base au cas où l'API a créé l'entrée
|
|
client_uuid = get_client_uuid_by_name(client_name, user_id)
|
|
if not client_uuid:
|
|
print("Erreur: impossible d'obtenir l'UUID du client '{}' après création, saut.".format(client_name))
|
|
continue
|
|
else:
|
|
client_uuid = get_client_uuid_by_name(client_name, user_id)
|
|
if not client_uuid:
|
|
print("Erreur: le client '{}' existe mais son UUID est introuvable, saut.".format(client_name))
|
|
continue
|
|
|
|
# Categories => Project
|
|
if not project_exist_by_name(project_name, client_uuid, user_id):
|
|
print("Création du projet {} pour le client {}.".format(project_name, client_name))
|
|
create_project(project_name, client_uuid)
|
|
|
|
project_uuid = get_project_uuid_by_name(project_name, client_uuid, user_id)
|
|
|
|
# Title => Task
|
|
if not task_exist_by_name(task_name, project_uuid, user_id):
|
|
print("Création de la tâche '{}' dans le projet {} du client {}".format(task_name, project_name, client_name))
|
|
create_task(task_name, project_uuid)
|
|
|
|
# Ajout du temps
|
|
|
|
dstart = sanitize_date(item['Date debut'])
|
|
hstart = sanitize_date(item['Heure debut'])
|
|
dend = sanitize_date(item['Date fin'])
|
|
hend = sanitize_date(item['Heure fin'])
|
|
|
|
print("Insertion temps pour la tâche {} (le {} à {} jusqu'au {} à {}) -- (projet: {} ; client : {})".format(task_name, dstart, hstart, dend, hend, project_name, client_name))
|
|
task_uuid = get_task_uuid_by_name(task_name, project_uuid, user_id)
|
|
|
|
insert_task_time(task_name, project_name, client_name, task_uuid, user_id, dstart, hstart, dend, hend)
|
|
|
|
cursor.close()
|
|
mydb.close()
|