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) client_uuid = cursor.fetchone()[0] save_db() return client_uuid.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) 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: pass #print("Client créé avec succès.") #print(json.dumps(response.json(), indent=1)) else: print("Erreur de création client") print(response.reason) 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 if not client_exist_by_name(client_name, user_id): print("Création du client {}".format(client_name)) create_client(client_name) client_uuid = get_client_uuid_by_name(client_name, user_id) # 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()