from enum import Enum from typing import Dict, List, Optional, Tuple, Union from ..models import Administre, CustomUser, Decision, DecisionChoices, Administres_Pams from ..models import StatutPamChoices as StatutPam from .logging import get_logger from .permissions import KEY_WRITE, Profiles, get_profiles_by_adm logger = get_logger(__name__) # clé d'une valeur de l'arbre de décision : choix disponibles KEY_CHOICES = 'choices' # clé d'une valeur de l'arbre de décision : profils habilités à modifier le statut KEY_PROFILES = 'editable_by' # clé d'un élément de "get_available_decisions" : pour une création KEY_CREATE = 'creation' # clé d'un élément de "get_available_decisions" : pour une mise à jour KEY_UPDATE = 'update' class ExtraDecisions(str, Enum): """ décisions en plus """ EMPTY = '' def __repr__(self): return self.__str__() def __init_decisions(): D = DecisionChoices EX = ExtraDecisions P = Profiles return { # === commun === None: { KEY_PROFILES: (P.FILIERE, P.BVT), KEY_CHOICES: (D.PROPOSITION_FE, D.HME_PROPOSITION_VIVIER) }, D.REMIS_A_DISPOSITION: { KEY_PROFILES: (P.FILIERE,), KEY_CHOICES: (EX.EMPTY,), }, # === en métropole === D.PROPOSITION_FE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL), KEY_CHOICES: (D.DIALOGUE_EN_COURS, D.FOREMP_EN_COURS) }, D.DIALOGUE_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL,), KEY_CHOICES: (D.DIALOGUE_TERMINE, D.DIALOGUE_INFRUCTUEUX) }, D.DIALOGUE_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL,), KEY_CHOICES: (D.FOREMP_EN_COURS,) }, D.DIALOGUE_INFRUCTUEUX: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL,), KEY_CHOICES: (D.FOREMP_EN_COURS, D.REMIS_A_DISPOSITION) }, D.FOREMP_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL,), KEY_CHOICES: (D.FOREMP_TERMINE,) }, D.FOREMP_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL,), KEY_CHOICES: (D.PREPOSITIONNE, D.REMIS_A_DISPOSITION) }, D.PREPOSITIONNE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.POSITIONNE, D.REMIS_A_DISPOSITION) }, D.POSITIONNE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.OMIP_EN_COURS, D.OMI_EN_COURS, D.REMIS_A_DISPOSITION) }, D.OMIP_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.OMIP_TERMINE, D.REMIS_A_DISPOSITION), }, D.OMIP_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.ATTENTE_AVIONAGE, D.OMI_EN_COURS, D.REMIS_A_DISPOSITION), }, D.ATTENTE_AVIONAGE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.OMI_EN_COURS,), }, D.OMI_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.OMI_ACTIVE, D.REMIS_A_DISPOSITION), }, D.OMI_ACTIVE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.OMI_ANNULE,), }, D.OMI_ANNULE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.REMIS_A_DISPOSITION,), }, # === hors métropole (HME) === D.HME_PROPOSITION_VIVIER: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_DIALOGUE_INITIE,) }, D.HME_ETUDE_DESISTEMENT: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_DESISTEMENT,) }, D.HME_DESISTEMENT: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.REMIS_A_DISPOSITION,) }, D.HME_DIALOGUE_INITIE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_DIALOGUE_EN_COURS,) }, D.HME_DIALOGUE_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_DIALOGUE_TERMINE, D.HME_DIALOGUE_INFRUCTUEUX) }, D.HME_DIALOGUE_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_PREPOSITIONNE,) }, D.HME_DIALOGUE_INFRUCTUEUX: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_ETUDE_DESISTEMENT, D.HME_PREPOSITIONNE, D.REMIS_A_DISPOSITION) }, D.HME_FOREMP_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_FOREMP_TERMINE, D.HME_OMI_EN_COURS) }, D.HME_FOREMP_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_OMIP_EN_COURS, D.HME_OMI_EN_COURS) }, D.HME_PREPOSITIONNE: { KEY_PROFILES: (P.HME,), KEY_CHOICES: (D.HME_VALIDATION_EXPERT, D.HME_REFUS_EXPERT) }, D.HME_VALIDATION_EXPERT: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_POSITIONNE,) }, D.HME_POSITIONNE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_OMIP_EN_COURS, D.HME_OMI_EN_COURS, D.HME_FOREMP_EN_COURS) }, D.HME_REFUS_EXPERT: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_PREPOSITIONNE, D.REMIS_A_DISPOSITION), }, D.HME_OMIP_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_OMIP_TERMINE,), }, D.HME_OMIP_TERMINE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_ATTENTE_AVIONAGE, D.HME_OMI_EN_COURS), }, D.HME_ATTENTE_AVIONAGE: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_OMI_EN_COURS,), }, D.HME_OMI_EN_COURS: { KEY_PROFILES: (P.PCP, P.PCP_ACTUEL, P.PCP_FUTUR), KEY_CHOICES: (D.HME_OMI_ACTIVE,), }, D.HME_OMI_ACTIVE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.HME_OMI_ANNULE,), }, D.HME_OMI_ANNULE: { KEY_PROFILES: (P.PCP, P.PCP_FUTUR,), KEY_CHOICES: (D.REMIS_A_DISPOSITION,), }, } # enchaînement des décisions DECISIONS = __init_decisions() def get_all_decisions() -> Tuple[DecisionChoices]: """ Renvoie tous les statuts de décisions possibles. Une décision vide correspondrait à une absence de décision et n'est donc pas présente. :return: toutes les décisions :rtype: Tuple[DecisionChoices] """ return tuple(DecisionChoices) def get_available_decisions( administres: Union[List[Administres_Pams], Tuple[Administres_Pams]], user: CustomUser = None, profiles_by_adm: Dict[int, Tuple[Profiles]] = None ) -> Dict[int, Dict[str, Tuple[Union[DecisionChoices, ExtraDecisions]]]]: """ Renvoie les décisions disponibles pour l'utilisateur s'il modifie le statut de décision des administrés donnés. :param administres: Administres_Pams :type administres: Union[List[Administres_Pams], Tuple[Administres_Pams]] :param user: utilisateur :type user: class:`CustomUser`, optional :param profiles_by_adm: profils pour chaque administré (voir get_profiles_by_adm) :type profiles_by_adm: Dict[int, Tuple[Profiles]], optional :return: dictionnaire de dictionnaires {: {: , : } } :rtype: Dict[int, Dict[str, Tuple[Union[DecisionChoices, ExtraDecisions]]]] """ pam_without_decision = tuple(x for x in StatutPam if not x.dec_enabled) pam_with_decision = tuple(x for x in StatutPam if x.dec_enabled) result = {} # restrictions : dictionnaire de dictionnaires {: } restrictions_create_by_adm = {} restrictions_update_by_adm = {} adm_to_process = [] for adm in administres: adm_id = adm.pk pam = adm.a_statut_pam_annee if pam in pam_without_decision: # le statut PAM ne permet aucun choix result[adm_id] = {KEY_CREATE: (), KEY_UPDATE: ()} elif pam in pam_with_decision: # le statut PAM active l'arbre de décisions adm_to_process.append(adm) statut_decision = adm.decision.de_decision or None if hasattr(adm, Administres_Pams.Cols.REL_DECISION) else None restrictions_update_by_adm[adm_id] = DECISIONS.get(statut_decision) or {} else: logger.info('statut PAM non géré pour les décisions possibles : %s', pam) result[adm_id] = {KEY_CREATE: (), KEY_UPDATE: ()} if adm_to_process: default_restrictions_create = DECISIONS.get(None) or {} def get_decisions(profiles, restrictions) -> Tuple[Union[DecisionChoices, ExtraDecisions]]: allowed_profiles = restrictions.get(KEY_PROFILES) or () choices = restrictions.get(KEY_CHOICES) decisions = () if choices and any(p in allowed_profiles for p in profiles): decisions = tuple(choices) return decisions _profiles_by_adm = profiles_by_adm or get_profiles_by_adm(user, *adm_to_process) for adm in adm_to_process: adm_id = adm.pk profiles = (_profiles_by_adm.get(adm_id) or {}).get(KEY_WRITE) or () result.setdefault(adm_id, { KEY_CREATE: get_decisions(profiles, restrictions_create_by_adm.get(adm_id) or default_restrictions_create), KEY_UPDATE: get_decisions(profiles, restrictions_update_by_adm.get(adm_id) or {}) }) return result