""" Ce fichier permet de gerer les modèles de planning Un modele de planning est defini par : - modele_session_formation_sequence_header : entete avec le nom, public pour perso, etc - modele_session_formation_sequence_detail : detail (contenu de la collection modele_session_formation Pour rappeler et appliquer un planning modèle, on va se baser sur 2 variables: - la date_debut d'application du planning - l'application du calendrier (date et jours travaillés) """ import ast import bson import pymongo from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import prj_common as mycommon import secrets import inspect import sys, os import csv import pandas as pd from pymongo import ReturnDocument import GlobalVariable as MYSY_GV from math import isnan import GlobalVariable as MYSY_GV from datetime import timedelta import ela_index_bdd_classes as eibdd import email_mgt as email import jinja2 from flask import send_file from xhtml2pdf import pisa from email.message import EmailMessage from email.mime.text import MIMEText from email import encoders import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders import base_config_modele_journee as base_config_modele_journee """ Ajout d'un planning modele """ def Add_Sequence_Planning_Model(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token','session_id', 'code', 'nom', 'is_public', 'comment'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'session_id', 'code', 'is_public',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner new_data = diction del diction['token'] # Initialisation des champs non envoyés à vide for val in field_list: if val not in diction.keys(): new_data[str(val)] = "" new_data['valide'] = "1" new_data['locked'] = "0" new_data['date_create'] = str(datetime.now()) new_data['date_update'] = str(datetime.now()) new_data['update_by'] = str(my_partner['_id']) new_data['partner_owner_recid'] = str(my_partner['recid']) if( "token" in new_data.keys() ): del new_data['token'] print("new_data = ", new_data) planning_model_header_inserted_id = MYSY_GV.dbname['planning_model_header'].insert_one(new_data).inserted_id if (not planning_model_header_inserted_id): mycommon.myprint( " Impossible d'ajouter le modèle de planning (2) ") return False, " Impossible d'ajouter le modèle de planning (2) " """ Apres ajout de l'entete, creation des lignes dans la collection 'planning_model_header_line' """ for New_retVal in MYSY_GV.dbname['session_formation_sequence'].find({'session_id':str(diction['session_id']), 'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0'}): user = New_retVal del user['_id'] if( "update_by" in user.keys() ): del user['update_by'] user['planning_model_header_id'] = str(planning_model_header_inserted_id) user['valide'] = "1" user['locked'] = "0" user['date_update'] = str(datetime.now()) user['update_by'] = str(my_partner['_id']) user['partner_owner_recid'] = str(my_partner['recid']) planning_model_line_inserted_id = MYSY_GV.dbname['planning_model_line'].insert_one(user).inserted_id return True, " Le modèle de planning a été correctement ajouté" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible d'ajouter le modèle de planning " """ Mise à jour d'un modele de planning """ def Delete_Sequence_Planning_Model(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token','_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token','_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner """ Verifier la validité du modele de planning """ is_valide_modele_planning = MYSY_GV.dbname['planning_model_header'].count_documents({'_id':ObjectId(str(diction['_id'])), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( is_valide_modele_planning != 1 ): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant du modèle de planning est invalide ") return False, " L'identifiant du modèle de planning est invalide " """ Suppression des lignes """ deleted_data_lines_qry = {'planning_model_header_id': str(diction['_id']), 'partner_owner_recid': str(my_partner['recid']), } # print(" ### deleted_data_qry = ", deleted_data_qry) deleted_data = MYSY_GV.dbname['planning_model_line'].delete_many(deleted_data_lines_qry) deleted_data_header_qry = {'_id': ObjectId(str(diction['_id'])), 'partner_owner_recid': str(my_partner['recid']), } # print(" ### deleted_data_qry = ", deleted_data_qry) deleted_header_data = MYSY_GV.dbname['planning_model_header'].delete_many(deleted_data_header_qry) return True, " Le modèle de planning a été correctement supprimé" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de supprimer le modèle de planning " """ Recuperer la liste des modeles de planning,(SEULEMENT LES ENTETES) """ def Get_List_Sequence_Planning_Model_Header(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token',] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner data_header_qry = {'partner_owner_recid': str(my_partner['recid']), 'valide':'1', 'locked':'0'} RetObject = [] val_tmp = 0 for retval in MYSY_GV.dbname['planning_model_header'].find(data_header_qry): user = retval user['id'] = str(val_tmp) val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récuperer le modèle de planning " """ Recuperer la liste des modeles de planning """ def Get_Given_Sequence_Planning_Model(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', '_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', '_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner """ Verifier la validité du modele de planning """ is_valide_modele_planning = MYSY_GV.dbname['planning_model_header'].count_documents( {'_id': ObjectId(str(diction['_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) if (is_valide_modele_planning != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant du modèle de planning est invalide ") return False, " L'identifiant du modèle de planning est invalide " data_header_qry = {'_id': ObjectId(str(diction['_id'])), 'partner_owner_recid': str(my_partner['recid']), } header_data = [] for retval in MYSY_GV.dbname['planning_model_header'].find(data_header_qry): user = retval user['id'] = str(val_tmp) val_tmp = val_tmp + 1 header_data.append(user) data_lines_qry = {'planning_model_header_id': str(diction['_id']), 'partner_owner_recid': str(my_partner['recid']), } lines_data = [] for retval in MYSY_GV.dbname['planning_model_line'].find(data_lines_qry): user = retval user['id'] = str(val_tmp) val_tmp = val_tmp + 1 lines_data.append(user) RetObject = {} RetObject['header_data'] = mycommon.JSONEncoder().encode(header_data) RetObject['lines_data'] = mycommon.JSONEncoder().encode(lines_data) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récuperer le modèle de planning " """ Recuperer la liste des modeles de planning avec les options suivantes : - la date_debut d'application du planning - l'application du calendrier (date et jours travaillés) """ def Get_Given_Sequence_Planning_Model_With_Option(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', '_id', 'date_debut', 'jounree_modele_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', '_id', 'date_debut', 'jounree_modele_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner """ Si le "jounree_modele_id" est vide, cela veut dire qu'il faut aller chercher le modele par default """ jounree_modele_id = "" if (str(diction['jounree_modele_id']) == ""): jounree_modele_data = MYSY_GV.dbname['base_config_modele_journee'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0'}) if (jounree_modele_data is None): # On ne trouve pas de modele par defaut pour ce client. On va aller le chercher le modele par defaut de MySy jounree_modele_data_mysy = MYSY_GV.dbname['base_config_modele_journee'].find_one( {'partner_owner_recid': "default", 'valide': '1', 'locked': '0'}) if (jounree_modele_data_mysy is None): mycommon.myprint( str(inspect.stack()[0][3]) + " - Aucun modèle de journée n'est configuré dans le système ") return False, " Aucun modèle de journée n'est configuré dans le système " jounree_modele_id = str(jounree_modele_data_mysy['_id']) else: jounree_modele_id = str(jounree_modele_data['_id']) else: jounree_modele_id = str(diction['jounree_modele_id']) if (str(jounree_modele_id).strip() == ""): mycommon.myprint( str(inspect.stack()[0][3]) + " - Aucun modèle de journée n'est configuré dans le système (2) ") return False, " Aucun modèle de journée n'est configuré dans le système (2) " # Verification de l'existence du modele de journée is_existe_valide_journee_model_id = MYSY_GV.dbname["base_config_modele_journee"].count_documents( {'_id': ObjectId(str(jounree_modele_id)), 'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0'}) if (is_existe_valide_journee_model_id <= 0): mycommon.myprint( str(inspect.stack()[0][3]) + " - L'identifiant du modèle de journee n'est pas valide ") return False, " L'identifiant du modèle de journee n'est pas valide " # Recuperation des data de la journée modele is_existe_valide_journee_model_data = MYSY_GV.dbname["base_config_modele_journee"].find_one( {'_id': ObjectId(str(jounree_modele_id)), 'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0'}) """ Verifier la validité du modele de planning """ is_valide_modele_planning = MYSY_GV.dbname['planning_model_header'].count_documents( {'_id': ObjectId(str(diction['_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) if (is_valide_modele_planning != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant du modèle de planning est invalide ") return False, " L'identifiant du modèle de planning est invalide " # Controle de cohérence sur les dates date_debut = "" if( "date_debut" in diction.keys() and diction['date_debut']): local_status = mycommon.CheckisDate(str(diction['date_debut'])[0:10]) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " La date de debut n'est pas au format jj/mm/aaaa ") return False, " La date de debut n'est pas au format jj/mm/aaaa " date_debut = str(diction['date_debut'])[0:10] jounree_modele_id = "" if( "jounree_modele_id" in diction.keys() and diction['jounree_modele_id']): jounree_modele_id = diction['jounree_modele_id'] data_header_qry = {'_id': ObjectId(str(diction['_id'])), 'partner_owner_recid': str(my_partner['recid']), } header_data = [] for retval in MYSY_GV.dbname['planning_model_header'].find(data_header_qry): user = retval header_data.append(user) """ data_lines_qry = {'planning_model_header_id': str(diction['_id']), 'partner_owner_recid': str(my_partner['recid']), } """ data_lines_qry = ([ {"$addFields": { "mysy_sequence_start": { '$dateFromString': { 'dateString': '$sequence_start', 'format': "%d/%m/%Y %H:%M" } } } }, {'$match': {'planning_model_header_id': str(diction['_id']), 'partner_owner_recid': str(my_partner['recid']), } }, { "$sort": {"mysy_sequence_start": 1}} ]) lines_data = [] for retval in MYSY_GV.dbname['planning_model_line'].aggregate(data_lines_qry): user = retval lines_data.append(user) """ Traiter l'application des options au calendrier """ print(" jounree_modele_id = ", jounree_modele_id, " date_debut = ", date_debut) if( jounree_modele_id and date_debut ): mytoday = datetime.today().strftime("%d/%m/%Y") decalage_planning = (datetime.strptime(str(date_debut), '%d/%m/%Y') - datetime.strptime(str(mytoday), '%d/%m/%Y')).days print(" ### ce planning doit etre appliqué avec un de calage de "+str(decalage_planning)+" jours ") for planning_line in lines_data: planning_line_sequence_start = str(planning_line['sequence_start']) local_status = mycommon.CheckisDate_Hours(planning_line_sequence_start) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " La date de debut de séquence " + str( planning_line_sequence_start) + " n'est pas au format 'jj/mm/aaaa hh:mm' ") return False, " La date de debut de séquence " + str( planning_line_sequence_start) + " n'est pas au format 'jj/mm/aaaa hh:mm'" planning_line_sequence_start_datetime = datetime.strptime(str(planning_line_sequence_start).strip(), '%d/%m/%Y %H:%M') planning_line_sequence_end = str(planning_line['sequence_end']) local_status = mycommon.CheckisDate_Hours(planning_line_sequence_end) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " La date de debut de séquence " + str( planning_line_sequence_end) + " n'est pas au format 'jj/mm/aaaa hh:mm' ") return False, " La date de debut de séquence " + str( planning_line_sequence_end) + " n'est pas au format 'jj/mm/aaaa hh:mm'" planning_line_sequence_end_datetime = datetime.strptime(str(planning_line_sequence_end).strip(), '%d/%m/%Y %H:%M') planning_line_sequence_date = planning_line_sequence_start_datetime.date() duree_sequence = planning_line_sequence_end_datetime - planning_line_sequence_start_datetime new_planning_line_sequence_start_datetime = planning_line_sequence_start_datetime + timedelta(days=decalage_planning) new_planning_line_sequence_end_datetime = new_planning_line_sequence_start_datetime+ timedelta(seconds=duree_sequence.seconds) print(" init_planning_line_sequence_start_datetime = ", planning_line_sequence_start_datetime) print(" init_planning_line_sequence_end_datetime = ", planning_line_sequence_end_datetime) print(" declage en jour = ", decalage_planning) print(" new_planning_line_sequence_start_datetime = ", new_planning_line_sequence_start_datetime) print(" new_planning_line_sequence_end_datetime = ", new_planning_line_sequence_end_datetime) date_jour = str(new_planning_line_sequence_start_datetime.strftime("%Y-%m-%d")) jour_name = new_planning_line_sequence_start_datetime.strftime("%A") print(" date_jour = ", date_jour) print(" jour_name = ", jour_name) is_off_day_of_week = "0" if (str(jour_name) in is_existe_valide_journee_model_data.keys()): is_off_day_of_week = is_existe_valide_journee_model_data[str(jour_name)]['is_off_day'] if (date_jour not in is_existe_valide_journee_model_data['date_off'] and is_off_day_of_week != "1"): print(" DATE / JOUR OKKKKK") while (date_jour in is_existe_valide_journee_model_data['date_off'] or is_off_day_of_week == "1"): print(str(date_jour)+" est KOOOO ") print(str(jour_name) + " est KOOOO ") decalage_planning = decalage_planning + 1 new_planning_line_sequence_start_datetime = planning_line_sequence_start_datetime + timedelta( days=decalage_planning) new_planning_line_sequence_end_datetime = new_planning_line_sequence_start_datetime + timedelta( seconds=duree_sequence.seconds) print(" new_planning_line_sequence_start_datetime = ", new_planning_line_sequence_start_datetime) print(" new_planning_line_sequence_end_datetime = ", new_planning_line_sequence_end_datetime) date_jour = str(new_planning_line_sequence_start_datetime.strftime("%Y-%m-%d")) jour_name = new_planning_line_sequence_start_datetime.strftime("%A") if (str(jour_name) in is_existe_valide_journee_model_data.keys()): is_off_day_of_week = is_existe_valide_journee_model_data[str(jour_name)]['is_off_day'] RetObject = {} RetObject['header_data'] = mycommon.JSONEncoder().encode(header_data) RetObject['lines_data'] = mycommon.JSONEncoder().encode(lines_data) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récuperer le modèle de planning "