diff --git a/app/scodoc/sco_import_users.py b/app/scodoc/sco_import_users.py index a1e2f77a..81ba3a4e 100644 --- a/app/scodoc/sco_import_users.py +++ b/app/scodoc/sco_import_users.py @@ -85,7 +85,16 @@ def generate_excel_sample(): def import_excel_file(datafile): - "Create users from Excel file" + """ + Import scodoc users from Excel file. + This method: + * checks that the current_user has the ability to do so (at the moment only a SuperAdmin). He may thereoff import users with any well formed role into any deprtment (or all) + * Once the check is done ans successfull, build the list of users (does not check the data) + * call :func:`import_users` to actually do the job + history: scodoc7 with no SuperAdmin every Admin_XXX could import users. + :param datafile: the stream from to the to be imported + :return: same as import users + """ # Check current user privilege auth_dept = current_user.dept auth_name = str(current_user) @@ -93,7 +102,7 @@ def import_excel_file(datafile): raise AccessDenied("invalid user (%s) must be SuperAdmin" % auth_name) # Récupération des informations sur l'utilisateur courant log("sco_import_users.import_excel_file by %s" % auth_name) - + # Read the data from the stream exceldata = datafile.read() if not exceldata: raise ScoValueError("Ficher excel vide ou invalide") @@ -116,42 +125,55 @@ def import_excel_file(datafile): "colonnes incorrectes (on attend %d, et non %d)
(colonnes manquantes: %s, colonnes invalides: %s)" % (len(TITLES), len(fs), list(cols.keys()), unknown) ) - # ok, same titles... - U = [] + # ok, same titles... : build the list of dictionaries + users = [] for line in data[1:]: d = {} for i in range(len(fs)): d[fs[i]] = line[i] - U.append(d) + users.append(d) - return import_users(U, auth_dept=auth_dept) + return import_users(users) -def import_users(users, auth_dept=""): - """Import des utilisateurs: - Pour chaque utilisateur à créer: - - vérifier données - - générer mot de passe aléatoire - - créer utilisateur et mettre le mot de passe - - envoyer mot de passe par mail - - En cas d'erreur: supprimer tous les utilisateurs que l'on vient de créer. +def import_users(users): """ + Import users from a list of users_descriptors. - def append_msg(msg): - msg_list.append("Ligne %s : %s" % (line, msg)) + descriptors are dictionaries hosting users's data. + The operation is atomic (all the users are imported or none) + + :param users: list of descriptors to be imported + + :return: a tuple that describe the result of the import: + * ok: import ok or aborted + * messages: the list of messages + * the # of users created + """ + """ Implémentation: + Pour chaque utilisateur à créer: + * vérifier données (y compris que le même nom d'utilisateur n'est pas utilisé plusieurs fois) + * générer mot de passe aléatoire + * créer utilisateur et mettre le mot de passe + * envoyer mot de passe par mail + Les utilisateurs à créer sont stockés dans un dictionnaire. + L'ajout effectif ne se fait qu'en fin de fonction si aucune erreur n'a été détectée + """ if len(users) == 0: import_ok = False - msg_list = ["Feuilles vide ou illisible"] + msg_list = ["Feuille vide ou illisible"] else: - created = [] # liste de uid créés + created = {} # liste de uid créés msg_list = [] - line = 1 # satr from excel line #2 + line = 1 # start from excel line #2 import_ok = True + + def append_msg(msg): + msg_list.append("Ligne %s : %s" % (line, msg)) + try: for u in users: - user_ok = True line = line + 1 user_ok, msg = sco_users.check_modif_user( 0, @@ -175,7 +197,7 @@ def import_users(users, auth_dept=""): "identifiant '%s' invalide (pas d'accents ni de caractères spéciaux)" % u["user_name"] ) - elif len(u["user_name"]) > 64: + if len(u["user_name"]) > 64: user_ok = False append_msg( "identifiant '%s' trop long (64 caractères)" % u["user_name"] @@ -189,29 +211,41 @@ def import_users(users, auth_dept=""): if len(u["email"]) > 120: user_ok = False append_msg("email '%s' trop long (120 caractères)" % u["email"]) + # check that tha same user_name has not already been described in this import + if u["user_name"] in created.keys(): + user_ok = False + append_msg( + "l'utilisateur '%s' a déjà été décrit ligne %s" + % (u["user_name"], created[u["user_name"]]["line"]) + ) # check département if u["dept"] != "": dept = Departement.query.filter_by(acronym=u["dept"]).first() if dept is None: user_ok = False append_msg("département '%s' inexistant" % u["dept"]) + # check roles / ignore whitespaces around roles / build roles_string + # roles_string (expected by User) appears as column 'roles' in excel file + roles_list = [] for role in u["roles"].split(","): try: - _, _ = UserRole.role_dept_from_string(role) + _, _ = UserRole.role_dept_from_string(role.strip()) + roles_list.append(role.strip()) except ScoValueError as value_error: user_ok = False - append_msg("role : %s " % role) - # Création de l'utilisateur (via SQLAlchemy) + append_msg("role %s : %s" % (role, value_error)) + u["roles_string"] = ",".join(roles_list) if user_ok: - created.append(u) + u["line"] = line + created[u["user_name"]] = u else: import_ok = False except ScoValueError as value_error: - log("import_users: exception: abort create %s" % str(created)) + log("import_users: exception: abort create %s" % str(created.keys())) raise ScoValueError(msg) # re-raise exception if import_ok: - for u in created: - u["roles_string"] = u["roles"] + for u in created.values(): + # Création de l'utilisateur (via SQLAlchemy) user = User() user.from_dict(u, new_user=True) db.session.add(user)