Init

.gitignore

@@ -0,0 +1,3 @@
+env/*
+config.json
+config.json.orig

crealead-caldav.py

@@ -0,0 +1,209 @@
+# -*- coding: utf-8 -*-
+"""
+Script permettant de générer des fichiers CSV à partir d'agenda Nextcloud 
+contenant un récapitulatif des heures de travail effectué
+
+Pour Crealead
+
+@author: Jordan
+@date: 17/11/2022
+
+@TODO: Améliorer la structure du code, pour l'instant c'est un peu moche
+@TODO: Ajouter des paramètres pour le mois, et le verbose pour avoir des logs
+@TOOD: README sur l'utilisation du script
+"""
+
+import sys
+import caldav
+import json
+import csv
+import calendar
+from datetime import date, datetime, timezone
+import pytz
+
+
+class EventRow():
+
+  def __init__(self):
+    self.title = ''
+    self.start_date = ''
+    self.start_time = ''
+    self.end_date = ''
+    self.end_time = ''
+    self.categories = ''
+    self.location = ''
+    self.description = ''
+
+  def set_title(self, title):
+    self.title = title
+
+  def set_start(self, start):
+    self.start_date = start.date()
+    self.start_time = start.time()
+
+  def set_end(self, end):
+    self.end_date = end.date()
+    self.end_time = end.time()
+
+  def set_categories(self, categories):
+    self.categories = categories
+
+  def set_location(self, location):
+    self.location = location
+
+  def set_description(self, description):
+    self.description = description
+
+  def get_csv_row(self):
+    """Return CSV row"""
+    return [self.title,self.start_date,self.start_time,self.end_date,self.end_time,self.categories,self.location,self.description]
+
+def load_json_config(filename):
+  """Return the config from JSON file"""
+  with open(filename, 'r') as f:
+    config_json = json.load(f)
+    return config_json
+
+
+def init_connection(config):
+  """Return all the calendars from the server"""
+  
+  dav_client = caldav.DAVClient(url=config['caldav_url'], username=config['username'], password=config['password'])
+
+  try:
+    dav_connection = dav_client.principal()
+  except Exception as e:
+    print("Une erreur a eu lieu pendant la connexion : ", e)
+    sys.exit()
+  
+  calendars_server = dav_connection.calendars()
+
+  if not calendars_server:
+    print("Aucun calendrier disponible sur le compte !")
+    sys.exit()
+
+  return calendars_server
+
+
+def link_calendars(config, calendars_server):
+  """Do the link between calendar we search (from the config)
+  and calendar available on the remote server"""
+
+  calendars_find = []
+  for calendar_config in config['calendars']:
+    cpt = 0 
+    for calendar_server in calendars_server:
+      if calendar_config['calendar_name'] in calendar_server.name:
+        cpt += 1
+        calendar_find = calendar_config
+        calendar_find['calendar_server'] = calendar_server
+        calendars_find.append(calendar_find)
+  
+    if not cpt:
+      print("Le calendrier nommé " + calendar_config['calendar_name'] + " pour le fichier " + calendar_config['csv_name'] + " n'est pas trouvé sur le serveur !")
+      sys.exit()
+    elif cpt >= 2:
+      print("Il y a plusieurs calendriers sur le serveur correspondant au calendrier nommé " + calendar_config['calendar_name'] + " pour le fichier " + calendar_config['csv_name'] + " !")
+      print(calendars_find)
+      sys.exit()
+
+  return calendars_find
+
+
+def split_data(data):
+  """Return string splitted by ';' separator"""
+  return data.split(':')
+
+
+def datetime_format(date_str):
+  """Return date formated"""
+  try: # Nextcloud Format
+   date_formated = datetime.strptime(date_str, '%Y%m%dT%H%M%SZ')
+  except: # Zimbra format
+   date_formated = datetime.strptime(date_str, '%Y%m%d')
+  paris_tz = pytz.timezone('Europe/Paris')
+  date_formated = date_formated.replace(tzinfo=timezone.utc).astimezone(paris_tz)
+  #date_formated = date_formated.astimezone(paris_tz)
+  return date_formated
+
+
+def get_all_events(config, calendars_find):
+  """Return row with the specific information we need to
+  write it in CSV"""
+
+  calendars_to_write = [] 
+
+  for calendar_find in calendars_find:
+    print(calendar_find['csv_name'], "...")
+    events_fetched = calendar_find['calendar_server'].date_search(start=datetime(config['year_start'], config['month_start'], config['day_start']),
+                                                                  end=datetime(config['year_end'], config['month_end'], config['day_end']),
+                                                                  expand=True)
+    event_rows = []
+  
+    for event in events_fetched:
+      event_fields = list(map(split_data, event.data.splitlines()))
+
+      # Fix problem #2023022110004936
+      has = False 
+      for sub in event_fields:
+          if 'SUMMARY' in sub:
+              has = True
+              break
+      if not has:
+          continue 
+
+      event_row = EventRow()
+
+      for field in ['SUMMARY', 'DTSTART', 'DTEND', 'CATEGORIES', 'LOCATION', 'DESCRIPTION']:
+        for event_field in event_fields:
+          if field in event_field[0]:
+            if field == "SUMMARY":
+              event_row.set_title(event_field[1])
+            elif field == "DTSTART":
+              date = datetime_format(event_field[1])
+              event_row.set_start(date)
+            elif field == "DTEND":
+              date = datetime_format(event_field[1])
+              event_row.set_end(date)
+            elif field == "CATEGORIES":
+              event_row.set_categories(event_field[1])
+            elif field == "LOCATION":
+              event_row.set_location(event_field[1])
+            elif field == "DESCRIPTION":
+              event_row.set_description(event_field[1])
+      event_rows.append(event_row)
+
+    sorted_rows = sorted(event_rows, key=lambda l: datetime.strptime(str(l.start_date)+'T'+str(l.start_time), "%Y-%m-%dT%H:%M:%S"))
+    calendars_to_write.append({'csv_name' : calendar_find['csv_name'],
+                               'line' : sorted_rows})
+
+  return calendars_to_write
+
+
+def write_csv(filename, rows):
+  """Write CSV file"""
+  with open(filename, 'w') as file:
+    writer = csv.writer(file, delimiter=';')
+    csv_title= ['Titre', 'Date debut', 'Heure debut', 'Date fin', 'Heure fin', 'Categories', 'Emplacement', 'Description']
+    writer.writerow(csv_title)
+    for row in rows:
+      writer.writerow(row.get_csv_row())
+
+
+if __name__ == "__main__":
+
+  # Load config
+  config = load_json_config("./config.json")
+
+  # Init connection with server
+  calendars_server = init_connection(config)
+
+  # Link calendars in config with calendar from server 
+  calendars_find = link_calendars(config, calendars_server)
+
+  # Get all events
+  calendars_to_write = get_all_events(config, calendars_find)
+
+  # Write CSV
+  for calendar_to_write in calendars_to_write:
+    write_csv(calendar_to_write['csv_name'], calendar_to_write['line'])

csv-to-timemanager.py

@@ -0,0 +1,456 @@
+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.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()

exec-csv-to-timemanager.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+source env/bin/activate
+python3 csv-to-timemanager.py

exec-export-caldav.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+source env/bin/activate
+python3 crealead-caldav.py

requirements.txt

@@ -0,0 +1,17 @@
+caldav==0.11.0
+certifi==2022.9.24
+charset-normalizer==2.1.1
+icalendar==5.0.2
+idna==3.4
+lxml==4.9.1
+python-dateutil==2.8.2
+pytz==2022.6
+pytz-deprecation-shim==0.1.0.post0
+recurring-ical-events==1.1.0b0
+requests==2.28.1
+six==1.16.0
+tzdata==2022.6
+tzlocal==4.2
+urllib3==1.26.12
+vobject==0.9.6.1
+x-wr-timezone==0.0.5