From 946cbcae91fe42a45d39004b9b4d6bd223e70d1a Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 4 Sep 2026 09:18:07 +0200 Subject: [PATCH] feat: add bulk user creation management command with configuration cloning and email/Outlook draft generation --- .gitignore | 4 +- loko/common/admin.py | 210 +----- .../commands/create_users_from_file.py | 598 ++++++++++++++++++ .../templates/emails/user_welcome_email.txt | 17 + loko/common/tests_user_services.py | 204 ++++++ loko/common/user_services.py | 505 +++++++++++++++ loko/scripts/create_outlook_drafts.py | 215 +++++++ loko/scripts/create_users.py | 48 ++ loko/scripts/init-db.sql | 18 + loko/scripts/templates/user_welcome_email.txt | 17 + 10 files changed, 1628 insertions(+), 208 deletions(-) create mode 100644 loko/common/management/commands/create_users_from_file.py create mode 100644 loko/common/templates/emails/user_welcome_email.txt create mode 100644 loko/common/tests_user_services.py create mode 100644 loko/common/user_services.py create mode 100755 loko/scripts/create_outlook_drafts.py create mode 100755 loko/scripts/create_users.py create mode 100644 loko/scripts/init-db.sql create mode 100644 loko/scripts/templates/user_welcome_email.txt diff --git a/.gitignore b/.gitignore index 1e1e9eb..c66b6cc 100644 --- a/.gitignore +++ b/.gitignore @@ -27,7 +27,9 @@ __pycache__/ loko/.env -scripts/ +/scripts/ +!loko/scripts/ +!loko/scripts/** carl_export/ carl_export.* diff --git a/loko/common/admin.py b/loko/common/admin.py index 1844747..14c4b80 100644 --- a/loko/common/admin.py +++ b/loko/common/admin.py @@ -15,6 +15,7 @@ from .models import ( INTERVENTION_STATUS_ORDER, INTERVENTION_STATUS_CHOICES, ) from .forms import UserConfigDuplicateForm, UserConfigCopyFromForm, UserConfigAssignCompanyForm +from .user_services import copy_user_config from django.contrib.admin import helpers from contracts.models import UserCompanyTeamAccess, CompanyMember, CompanyTeam @@ -729,111 +730,7 @@ class UserConfigAdmin(admin.ModelAdmin): def _copy_config(self, source, target): """Écrase la configuration de `target` avec celle de `source` (sauf l'utilisateur lié).""" - # Champs scalaires - target.is_intern = source.is_intern - target.default_thematic = source.default_thematic - target.company = source.company - target.limit_assets_to_contracts = source.limit_assets_to_contracts - target.limit_interventions_to_contracts = source.limit_interventions_to_contracts - target.limit_shortcuts_to_contracts = source.limit_shortcuts_to_contracts - target.generic_email = source.generic_email - target.hide_intervention_map_by_default = source.hide_intervention_map_by_default - target.auto_logout_hours = source.auto_logout_hours - target.intervention_list_layout = source.intervention_list_layout - target.assets_list_layout = source.assets_list_layout - target.can_add_unique_tag_documents = source.can_add_unique_tag_documents - target.esign_can_edit = source.esign_can_edit - target.esign_can_manage_rcp = source.esign_can_manage_rcp - target.esign_can_validate_rcp = source.esign_can_validate_rcp - target.panoramax_can_upload = source.panoramax_can_upload - target.save() - # M2M directs - target.roles.set(source.roles.all()) - target.accessible_views.set(source.accessible_views.all()) - target.esign_organizations.set(source.esign_organizations.all()) - # Thématiques - target.userthematics.all().delete() - for ut in source.userthematics.prefetch_related('status_permissions').all(): - new_ut = UserThematics.objects.create( - user_config=target, - thematic=ut.thematic, - can_view_assets=ut.can_view_assets, - can_edit_assets=ut.can_edit_assets, - can_validate_assets=ut.can_validate_assets, - can_inspect=ut.can_inspect, - can_view_interventions=ut.can_view_interventions, - can_edit_interventions=ut.can_edit_interventions, - can_edit_locations=ut.can_edit_locations, - can_process_observations=ut.can_process_observations, - can_view_projects=ut.can_view_projects, - can_edit_projects=ut.can_edit_projects, - ) - UserThematicStatusPermission.objects.bulk_create([ - UserThematicStatusPermission( - user_thematic=new_ut, - status=sp.status, - can_view=sp.can_view, - can_edit=sp.can_edit, - can_change_status_to=sp.can_change_status_to, - ) - for sp in ut.status_permissions.all() - ]) - # Contrats - target.usercontractaccess_set.all().delete() - for uc in source.usercontractaccess_set.prefetch_related('status_permissions').all(): - new_uc = UserContractAccess.objects.create( - user_config=target, - contract=uc.contract, - can_view_posts=uc.can_view_posts, - can_view_prices=uc.can_view_prices, - can_view_interventions=uc.can_view_interventions, - can_view_assets=uc.can_view_assets, - can_create_interventions=uc.can_create_interventions, - can_check_interventions=uc.can_check_interventions, - can_approve=uc.can_approve, - ) - UserContractStatusPermission.objects.bulk_create([ - UserContractStatusPermission( - user_contract=new_uc, - status=sp.status, - can_view=sp.can_view, - can_edit=sp.can_edit, - can_change_status_to=sp.can_change_status_to, - ) - for sp in uc.status_permissions.all() - ]) - # Accès aux entrepôts - target.userwarehouses.all().delete() - for wa in source.userwarehouses.all(): - UserWarehouseAccess.objects.create( - user_config=target, - warehouse=wa.warehouse, - can_view=wa.can_view, - can_manage=wa.can_manage, - can_administer=wa.can_administer, - can_manage_preparation_orders=wa.can_manage_preparation_orders, - can_manage_purchase_orders=wa.can_manage_purchase_orders, - can_validate_inventory=wa.can_validate_inventory, - is_default=wa.is_default, - ) - # Accès aux emplacements - target.userlocations.all().delete() - for la in source.userlocations.all(): - UserWarehouseLocationAccess.objects.create( - user_config=target, - warehouse_location=la.warehouse_location, - can_perform_inventory=la.can_perform_inventory, - can_validate_inventory=la.can_validate_inventory, - ) - # Accès aux équipes - target.team_accesses.all().delete() - for ta in source.team_accesses.all(): - UserCompanyTeamAccess.objects.create( - user_config=target, - team=ta.team, - can_view=ta.can_view, - can_edit=ta.can_edit, - ) + return copy_user_config(source, target) def duplicate_config_action(self, request, queryset): if queryset.count() != 1: @@ -849,108 +746,7 @@ class UserConfigAdmin(admin.ModelAdmin): form = UserConfigDuplicateForm(request.POST) if form.is_valid(): new_user = form.cleaned_data["user"] - # Créer la nouvelle config - new_config = UserConfig.objects.create( - user=new_user, - is_intern=orig_config.is_intern, - default_thematic=orig_config.default_thematic, - company=orig_config.company, - limit_assets_to_contracts=orig_config.limit_assets_to_contracts, - limit_interventions_to_contracts=orig_config.limit_interventions_to_contracts, - limit_shortcuts_to_contracts=orig_config.limit_shortcuts_to_contracts, - generic_email=orig_config.generic_email, - hide_intervention_map_by_default=orig_config.hide_intervention_map_by_default, - auto_logout_hours=orig_config.auto_logout_hours, - esign_can_edit=orig_config.esign_can_edit, - esign_can_manage_rcp=orig_config.esign_can_manage_rcp, - esign_can_validate_rcp=orig_config.esign_can_validate_rcp, - ) - # Copier les rôles - new_config.roles.set(orig_config.roles.all()) - # Copier les linked_configs - new_config.linked_configs.set(orig_config.linked_configs.all()) - # Copier les vues accessibles - new_config.accessible_views.set(orig_config.accessible_views.all()) - # Copier les gestionnaires eSign - new_config.esign_organizations.set(orig_config.esign_organizations.all()) - # Copier les thématiques (UserThematics + permissions par statut) - for ut in orig_config.userthematics.prefetch_related('status_permissions').all(): - new_ut = UserThematics.objects.create( - user_config=new_config, - thematic=ut.thematic, - can_view_assets=ut.can_view_assets, - can_edit_assets=ut.can_edit_assets, - can_validate_assets=ut.can_validate_assets, - can_inspect=ut.can_inspect, - can_view_interventions=ut.can_view_interventions, - can_edit_interventions=ut.can_edit_interventions, - can_edit_locations=ut.can_edit_locations, - can_process_observations=ut.can_process_observations, - can_view_projects=ut.can_view_projects, - can_edit_projects=ut.can_edit_projects, - ) - UserThematicStatusPermission.objects.bulk_create([ - UserThematicStatusPermission( - user_thematic=new_ut, - status=sp.status, - can_view=sp.can_view, - can_edit=sp.can_edit, - can_change_status_to=sp.can_change_status_to, - ) - for sp in ut.status_permissions.all() - ]) - # Copier les contrats (UserContractAccess + permissions par statut) - for uc in orig_config.usercontractaccess_set.prefetch_related('status_permissions').all(): - new_uc = UserContractAccess.objects.create( - user_config=new_config, - contract=uc.contract, - can_view_posts=uc.can_view_posts, - can_view_prices=uc.can_view_prices, - can_view_interventions=uc.can_view_interventions, - can_view_assets=uc.can_view_assets, - can_create_interventions=uc.can_create_interventions, - can_check_interventions=uc.can_check_interventions, - can_approve=uc.can_approve, - ) - UserContractStatusPermission.objects.bulk_create([ - UserContractStatusPermission( - user_contract=new_uc, - status=sp.status, - can_view=sp.can_view, - can_edit=sp.can_edit, - can_change_status_to=sp.can_change_status_to, - ) - for sp in uc.status_permissions.all() - ]) - # Copier les accès aux entrepôts (UserWarehouseAccess) - for wa in orig_config.userwarehouses.all(): - UserWarehouseAccess.objects.create( - user_config=new_config, - warehouse=wa.warehouse, - can_view=wa.can_view, - can_manage=wa.can_manage, - can_administer=wa.can_administer, - can_manage_preparation_orders=wa.can_manage_preparation_orders, - can_manage_purchase_orders=wa.can_manage_purchase_orders, - can_validate_inventory=wa.can_validate_inventory, - is_default=wa.is_default, - ) - # Copier les accès aux emplacements (UserWarehouseLocationAccess) - for la in orig_config.userlocations.all(): - UserWarehouseLocationAccess.objects.create( - user_config=new_config, - warehouse_location=la.warehouse_location, - can_perform_inventory=la.can_perform_inventory, - can_validate_inventory=la.can_validate_inventory, - ) - # Copier les accès aux équipes (UserCompanyTeamAccess) - for ta in orig_config.team_accesses.all(): - UserCompanyTeamAccess.objects.create( - user_config=new_config, - team=ta.team, - can_view=ta.can_view, - can_edit=ta.can_edit, - ) + new_config = copy_user_config(orig_config, new_user) messages.success(request, f"Configuration dupliquée pour l'utilisateur {new_user.username} !") return redirect(f"../../{new_config.id}/change/") else: diff --git a/loko/common/management/commands/create_users_from_file.py b/loko/common/management/commands/create_users_from_file.py new file mode 100644 index 0000000..e915084 --- /dev/null +++ b/loko/common/management/commands/create_users_from_file.py @@ -0,0 +1,598 @@ +""" +Commande Django pour créer en masse des utilisateurs à partir d'un fichier (CSV, XLSX, TXT, JSON), +copier la configuration complète d'un utilisateur de référence (UserConfig et permissions), +générer des mots de passe sécurisés conformes, et exporter les fichiers textes pour les e-mails +ainsi que la charge utile pour la création de brouillons Outlook. + +Usage : + python manage.py create_users_from_file chemin/vers/utilisateurs.csv --source-user admin + python manage.py create_users_from_file chemin/vers/utilisateurs.xlsx --source-user yrheleimi --dry-run +""" + +import csv +import json +import os +import urllib.parse +from datetime import datetime +from pathlib import Path + +from django.conf import settings +from django.contrib.auth import get_user_model +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction + +from common.models import UserConfig +from common.user_services import ( + copy_user_config, + generate_secure_password, + generate_username, + read_users_file, + render_email_template, +) + +User = get_user_model() + + +class Command(BaseCommand): + help = "Crée des utilisateurs en masse, copie leur configuration et génère les emails d'accès." + + def add_arguments(self, parser): + parser.add_argument( + "file", + type=str, + help="Chemin vers le fichier contenant les utilisateurs (CSV, XLSX, TXT ou JSON).", + ) + parser.add_argument( + "--source-user", + type=str, + required=True, + dest="source_user", + help="Login (username) ou email de l'utilisateur modèle dont la configuration sera dupliquée.", + ) + parser.add_argument( + "--template", + type=str, + default=None, + help="Chemin vers un template alternatif pour le corps de l'e-mail.", + ) + parser.add_argument( + "--output-dir", + type=str, + default=None, + dest="output_dir", + help="Répertoire où enregistrer les fichiers générés (défaut : runs/user_creations/YYYYMMDD_HHMMSS/).", + ) + parser.add_argument( + "--app-name", + type=str, + default="StreetUp", + help="Nom de l'application (défaut : StreetUp).", + ) + parser.add_argument( + "--desktop-url", + type=str, + default="https://bm.streetup.be/", + help="URL de la version desktop (défaut : https://bm.streetup.be/).", + ) + parser.add_argument( + "--mobile-url", + type=str, + default="https://bm.streetup.be/mobile", + help="URL de la version mobile (défaut : https://bm.streetup.be/mobile).", + ) + parser.add_argument( + "--install-url", + type=str, + default="https://bm.streetup.be/mobile/install/", + help="URL d'aide à l'installation mobile (défaut : https://bm.streetup.be/mobile/install/).", + ) + parser.add_argument( + "--dry-run", + action="store_true", + default=False, + help="Simulation : affiche les opérations sans créer d'utilisateurs ni modifier la base de données.", + ) + parser.add_argument( + "--update-existing", + action="store_true", + default=False, + help="Si spécifié, met à jour la configuration des utilisateurs déjà existants au lieu de les ignorer.", + ) + + def handle(self, *args, **options): + input_file = options["file"] + source_identifier = options["source_user"] + template_path = options["template"] + output_dir = options["output_dir"] + app_name = options["app_name"] + desktop_url = options["desktop_url"] + mobile_url = options["mobile_url"] + install_url = options["install_url"] + dry_run = options["dry_run"] + update_existing = options["update_existing"] + + if dry_run: + self.stdout.write(self.style.WARNING("=== MODE SIMULATION (DRY-RUN) : Aucune modification en base ===")) + + # 1. Vérifier l'utilisateur source + source_user = ( + User.objects.filter(username__iexact=source_identifier).first() + or User.objects.filter(email__iexact=source_identifier).first() + ) + if not source_user: + raise CommandError(f"Utilisateur source introuvable avec l'identifiant : '{source_identifier}'") + + try: + source_config = UserConfig.objects.get(user=source_user) + except UserConfig.DoesNotExist: + raise CommandError( + f"L'utilisateur source '{source_user.username}' n'a pas de UserConfig associée à dupliquer." + ) + + self.stdout.write( + self.style.SUCCESS( + f"Utilisateur source sélectionné : {source_user.username} ({source_user.get_full_name() or source_user.email})" + ) + ) + + # 2. Résoudre le template d'email + if not template_path: + # Chercher dans templates standards + default_template_locations = [ + Path(settings.BASE_DIR) / "common" / "templates" / "emails" / "user_welcome_email.txt", + Path(settings.BASE_DIR) / "scripts" / "templates" / "user_welcome_email.txt", + Path(__file__).resolve().parent.parent.parent / "templates" / "emails" / "user_welcome_email.txt", + ] + for loc in default_template_locations: + if loc.exists(): + template_path = str(loc) + break + + if not template_path or not Path(template_path).exists(): + raise CommandError(f"Template d'e-mail introuvable. Veuillez spécifier --template.") + + self.stdout.write(f"Template d'email utilisé : {template_path}") + + # 3. Lire le fichier d'utilisateurs + try: + users_data = read_users_file(input_file) + except Exception as exc: + raise CommandError(f"Erreur lors de la lecture de '{input_file}' : {exc}") + + if not users_data: + self.stdout.write(self.style.WARNING("Aucun utilisateur trouvé dans le fichier.")) + return + + self.stdout.write(f"{len(users_data)} utilisateur(s) trouvé(s) dans le fichier.") + + # 4. Préparer le dossier de sortie + if not output_dir: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + # Dossier dans runs/user_creations + base_runs = Path(settings.BASE_DIR).parent / "runs" / "user_creations" + output_dir = str(base_runs / timestamp) + + out_path = Path(output_dir) + emails_dir = out_path / "emails" + if not dry_run: + emails_dir.mkdir(parents=True, exist_ok=True) + + existing_usernames = set(User.objects.values_list("username", flat=True)) + results_summary = [] + drafts_payload = [] + + count_created = 0 + count_updated = 0 + count_skipped = 0 + + # 5. Traiter chaque utilisateur + for idx, row in enumerate(users_data, start=1): + first_name = row.get("first_name", "").strip() + last_name = row.get("last_name", "").strip() + email = row.get("email", "").strip() + given_username = row.get("username", "").strip() + + if not email and not first_name and not last_name: + continue + + self.stdout.write(f"\n[{idx}/{len(users_data)}] Traitement de : {first_name} {last_name} <{email}>") + + # Vérifier l'existence préalable + existing_user = None + if email: + existing_user = User.objects.filter(email__iexact=email).first() + if not existing_user and given_username: + existing_user = User.objects.filter(username__iexact=given_username).first() + + if existing_user and not update_existing: + self.stdout.write( + self.style.WARNING( + f" -> Utilisateur existant ({existing_user.username}) pour '{email or given_username}'. Ignoré." + ) + ) + count_skipped += 1 + results_summary.append({ + "first_name": first_name or existing_user.first_name, + "last_name": last_name or existing_user.last_name, + "email": email or existing_user.email, + "username": existing_user.username, + "password": "[NON MODIFIÉ]", + "status": "IGNORÉ (DÉJÀ EXISTANT)", + "created_at": datetime.now().isoformat(), + "source_user": source_user.username, + }) + continue + + # Générer ou réutiliser le login + if existing_user: + username = existing_user.username + password = "[CONSERVÉ]" + user_obj = existing_user + action_label = "Mis à jour" + else: + username = given_username if given_username else generate_username(first_name, last_name, existing_usernames) + password = generate_secure_password(20) + action_label = "Créé" + + context = { + "app_name": app_name, + "desktop_url": desktop_url, + "mobile_url": mobile_url, + "install_url": install_url, + "first_name": first_name or username, + "last_name": last_name, + "username": username, + "password": password, + "email": email, + } + + subject, body = render_email_template(template_path, context) + + if not dry_run: + with transaction.atomic(): + if not existing_user: + user_obj = User.objects.create_user( + username=username, + email=email, + password=password, + first_name=first_name, + last_name=last_name, + is_active=True, + ) + count_created += 1 + else: + if first_name: + user_obj.first_name = first_name + if last_name: + user_obj.last_name = last_name + user_obj.save() + count_updated += 1 + + # Copie intégrale de la configuration + copy_user_config(source_user, user_obj) + + # Écriture du fichier texte individuel pour cet utilisateur + user_email_file = emails_dir / f"{username}.txt" + with open(user_email_file, "w", encoding="utf-8") as ef: + ef.write(f"Destinataire : {email}\n") + ef.write(f"Sujet : {subject}\n") + ef.write("--------------------------------------------------\n\n") + ef.write(body + "\n") + + else: + if not existing_user: + count_created += 1 + else: + count_updated += 1 + + self.stdout.write( + self.style.SUCCESS( + f" -> {action_label} : login = {username} | mot de passe = {password} | email = {email}" + ) + ) + + results_summary.append({ + "first_name": first_name, + "last_name": last_name, + "email": email, + "username": username, + "password": password, + "status": "CRÉÉ" if not existing_user else "MIS À JOUR", + "created_at": datetime.now().isoformat(), + "source_user": source_user.username, + }) + + # URL Outlook Web Deep Link pour composer directement + deeplink_params = urllib.parse.urlencode({ + "to": email, + "subject": subject, + "body": body, + }) + outlook_deeplink = f"https://outlook.cloud.microsoft/mail/deeplink/compose?{deeplink_params}" + + drafts_payload.append({ + "to": email, + "subject": subject, + "body": body, + "first_name": first_name, + "last_name": last_name, + "username": username, + "password": password, + "outlook_deeplink": outlook_deeplink, + }) + + # 6. Écriture du récapitulatif CSV et de la charge utile JSON + if not dry_run and results_summary: + summary_csv_path = out_path / "users_summary.csv" + with open(summary_csv_path, "w", encoding="utf-8", newline="") as cf: + fieldnames = ["first_name", "last_name", "email", "username", "password", "status", "created_at", "source_user"] + writer = csv.DictWriter(cf, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(results_summary) + + payload_json_path = out_path / "drafts_payload.json" + with open(payload_json_path, "w", encoding="utf-8") as jf: + json.dump(drafts_payload, jf, indent=2, ensure_ascii=False) + + # Génération de la page HTML interactive locale open_drafts.html + html_path = out_path / "open_drafts.html" + self._generate_html_launcher(html_path, drafts_payload, app_name) + + self.stdout.write("\n" + "=" * 60) + self.stdout.write(self.style.SUCCESS(f"Traitement terminé avec succès !")) + self.stdout.write(f"Utilisateurs créés : {count_created}") + self.stdout.write(f"Utilisateurs mis à jour : {count_updated}") + self.stdout.write(f"Utilisateurs ignorés : {count_skipped}") + self.stdout.write(f"Dossier des sorties : {out_path.resolve()}") + self.stdout.write(f" - E-mails individuels : {emails_dir.resolve()}/*.txt") + self.stdout.write(f" - Tableau CSV récap : {summary_csv_path.resolve()}") + self.stdout.write(f" - Charge utile JSON : {payload_json_path.resolve()}") + self.stdout.write(f" - Lanceur HTML : file://{html_path.resolve()}") + self.stdout.write("=" * 60) + else: + self.stdout.write("\n" + "=" * 60) + self.stdout.write(f"Simulation terminée.") + self.stdout.write(f"À créer : {count_created}") + self.stdout.write(f"À mettre à jour : {count_updated}") + self.stdout.write(f"À ignorer : {count_skipped}") + self.stdout.write("=" * 60) + + def _generate_html_launcher(self, html_path: Path, drafts: list, app_name: str): + """Génère une page HTML moderne permettant d'ouvrir les brouillons Outlook en un clic.""" + items_json = json.dumps(drafts, ensure_ascii=False) + html_content = f""" + + + + + Brouillons Outlook - Accès {app_name} + + + +
+
+

Accès créés pour {app_name}

+

+ {len(drafts)} utilisateur(s) traité(s). Cliquez sur Ouvrir dans Outlook pour pré-remplir automatiquement le brouillon dans votre messagerie. +

+
+ +
+ + +
+ +
+
+ +
Texte copié dans le presse-papier !
+ + + + +""" + with open(html_path, "w", encoding="utf-8") as f: + f.write(html_content) diff --git a/loko/common/templates/emails/user_welcome_email.txt b/loko/common/templates/emails/user_welcome_email.txt new file mode 100644 index 0000000..4ae7f1e --- /dev/null +++ b/loko/common/templates/emails/user_welcome_email.txt @@ -0,0 +1,17 @@ +Sujet : Accès {app_name} + +Bonjour {first_name} + +Voici votre login pour accéder à {app_name}. + +Version desktop : {desktop_url} +Version mobile : {mobile_url} + +{username} +{password} + +Aide pour l’installation de la partie mobile : {install_url} + +N'hésitez pas à me contacter pour toute remarque ou question. + +Bien à vous diff --git a/loko/common/tests_user_services.py b/loko/common/tests_user_services.py new file mode 100644 index 0000000..162ac3b --- /dev/null +++ b/loko/common/tests_user_services.py @@ -0,0 +1,204 @@ +""" +Tests unitaires pour les services utilisateur : génération d'identifiants, +génération de mots de passe, duplication de configuration, lecture de fichiers et rendu d'e-mails. +""" + +import tempfile +from pathlib import Path +from django.test import TestCase +from django.contrib.auth import get_user_model + +from common.models import ( + UserConfig, Thematic, Role, UserThematics, UserThematicStatusPermission, + UserContractAccess, UserContractStatusPermission, +) +from common.user_services import ( + generate_username, + generate_secure_password, + copy_user_config, + read_users_file, + render_email_template, +) +from contracts.models import Contract + +User = get_user_model() + + +class UserServicesTestCase(TestCase): + def test_generate_username_basic(self): + login = generate_username("Yassine", "Rheleimi", existing_usernames=set()) + self.assertEqual(login, "yrheleimi") + + def test_generate_username_accents_and_spaces(self): + login = generate_username("Éléonore", "De La Tour", existing_usernames=set()) + self.assertEqual(login, "edelatour") + + login2 = generate_username("Jean-François", "Müller", existing_usernames=set()) + self.assertEqual(login2, "jmuller") + + def test_generate_username_collision(self): + existing = {"yrheleimi", "yrheleimi2"} + login = generate_username("Yassine", "Rheleimi", existing_usernames=existing) + self.assertEqual(login, "yrheleimi3") + self.assertIn("yrheleimi3", existing) + + def test_generate_secure_password(self): + for _ in range(50): + pwd = generate_secure_password(20) + self.assertEqual(len(pwd), 20) + # Commence par une majuscule + self.assertTrue(pwd[0].isupper(), f"Ne commence pas par une majuscule: {pwd}") + # Uniquement lettres et chiffres + self.assertTrue(pwd.isalnum(), f"Contient des caractères spéciaux: {pwd}") + # Contient au moins un chiffre et une minuscule + self.assertTrue(any(c.isdigit() for c in pwd), f"Aucun chiffre: {pwd}") + self.assertTrue(any(c.islower() for c in pwd), f"Aucune minuscule: {pwd}") + + def test_render_email_template(self): + template = ( + "Sujet : Accès {app_name}\n\n" + "Bonjour {first_name}\n\n" + "Voici votre login pour accéder à {app_name}.\n\n" + "Version desktop : {desktop_url}\n" + "Version mobile : {mobile_url}\n\n" + "{username}\n" + "{password}\n\n" + "Aide pour l’installation de la partie mobile : {install_url}\n\n" + "N'hésitez pas à me contacter pour toute remarque ou question.\n\n" + "Bien à vous\n" + ) + ctx = { + "app_name": "StreetUp", + "first_name": "Yassine", + "username": "yrheleimi", + "password": "Password12345678901", + "desktop_url": "https://bm.streetup.be/", + "mobile_url": "https://bm.streetup.be/mobile", + "install_url": "https://bm.streetup.be/mobile/install/", + } + subject, body = render_email_template(template, ctx) + self.assertEqual(subject, "Accès StreetUp") + self.assertIn("Bonjour Yassine", body) + self.assertIn("yrheleimi", body) + self.assertIn("Password12345678901", body) + self.assertIn("toute remarque ou question", body) + + def test_read_users_file_csv(self): + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, encoding="utf-8") as f: + f.write("prenom;nom;email\n") + f.write("Yassine;Rheleimi;yrheleimi@example.com\n") + f.write("Alice;Dupont;alice@example.com\n") + f_path = f.name + + try: + users = read_users_file(f_path) + self.assertEqual(len(users), 2) + self.assertEqual(users[0]["first_name"], "Yassine") + self.assertEqual(users[0]["last_name"], "Rheleimi") + self.assertEqual(users[0]["email"], "yrheleimi@example.com") + self.assertEqual(users[1]["first_name"], "Alice") + finally: + Path(f_path).unlink(missing_ok=True) + + def test_read_users_file_txt_format(self): + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f: + f.write("Yassine Rheleimi \n") + f.write("Bob Marley \n") + f_path = f.name + + try: + users = read_users_file(f_path) + self.assertEqual(len(users), 2) + self.assertEqual(users[0]["first_name"], "Yassine") + self.assertEqual(users[0]["last_name"], "Rheleimi") + self.assertEqual(users[0]["email"], "yrheleimi@example.com") + self.assertEqual(users[1]["first_name"], "Bob") + self.assertEqual(users[1]["last_name"], "Marley") + finally: + Path(f_path).unlink(missing_ok=True) + + def test_copy_user_config(self): + # Création des utilisateurs source et cible + src_user = User.objects.create_user( + username="src_user", email="src@example.com", password="Password123" + ) + thematic = Thematic.objects.create(name_fr="Voirie", code="VOIRIE") + from datetime import date + from contracts.models import Company + + company = Company.objects.create(name="Company Test") + contract = Contract.objects.create( + company=company, + contract_number="CTR-TEST-001", + start_date=date(2026, 1, 1), + end_date=date(2026, 12, 31), + description="Contrat Test", + ) + role, _ = Role.objects.get_or_create(name="agent") + + src_config = UserConfig.objects.create( + user=src_user, + is_intern=True, + default_thematic=thematic, + limit_assets_to_contracts=True, + limit_interventions_to_contracts=True, + preferred_language="fr", + ) + src_config.roles.add(role) + + # Ajouter une thématique avec permission par statut + ut = UserThematics.objects.create( + user_config=src_config, + thematic=thematic, + can_view_interventions=True, + can_edit_interventions=True, + ) + UserThematicStatusPermission.objects.create( + user_thematic=ut, + status="in_progress", + can_view=True, + can_edit=True, + ) + + # Ajouter un accès contrat avec permission par statut + uc = UserContractAccess.objects.create( + user_config=src_config, + contract=contract, + can_view_interventions=True, + can_create_interventions=True, + ) + UserContractStatusPermission.objects.create( + user_contract=uc, + status="assigned", + can_view=True, + ) + + target_user = User.objects.create_user( + username="tgt_user", email="tgt@example.com", password="Password123" + ) + + # Duplication + target_config = copy_user_config(src_user, target_user) + + self.assertEqual(target_config.user, target_user) + self.assertTrue(target_config.is_intern) + self.assertEqual(target_config.default_thematic, thematic) + self.assertTrue(target_config.roles.filter(name="agent").exists()) + + # Vérification des thématiques dupliquées + self.assertEqual(target_config.userthematics.count(), 1) + tgt_ut = target_config.userthematics.first() + self.assertEqual(tgt_ut.thematic, thematic) + self.assertTrue(tgt_ut.can_edit_interventions) + self.assertTrue( + tgt_ut.status_permissions.filter(status="in_progress", can_edit=True).exists() + ) + + # Vérification des contrats dupliqués + self.assertEqual(target_config.usercontractaccess_set.count(), 1) + tgt_uc = target_config.usercontractaccess_set.first() + self.assertEqual(tgt_uc.contract, contract) + self.assertTrue(tgt_uc.can_create_interventions) + self.assertTrue( + tgt_uc.status_permissions.filter(status="assigned", can_view=True).exists() + ) diff --git a/loko/common/user_services.py b/loko/common/user_services.py new file mode 100644 index 0000000..e472cbd --- /dev/null +++ b/loko/common/user_services.py @@ -0,0 +1,505 @@ +""" +Services pour la gestion des utilisateurs, la génération de mots de passe / identifiants, +la copie exhaustive des configurations (UserConfig et permissions associées), +la lecture des fichiers de données (CSV, XLSX, TXT, JSON) et le rendu des e-mails. +""" + +import csv +import json +import logging +import os +import re +import secrets +import string +import unicodedata +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from django.contrib.auth import get_user_model +from django.db import transaction + +from common.models import ( + AppView, + Role, + Thematic, + UserAssetAccess, + UserConfig, + UserContractAccess, + UserContractStatusPermission, + UserThematics, + UserThematicStatusPermission, + UserWarehouseAccess, + UserWarehouseLocationAccess, +) +from contracts.models import UserCompanyTeamAccess + +logger = logging.getLogger(__name__) +User = get_user_model() + + +def clean_text_for_login(text: str) -> str: + """ + Nettoie une chaîne pour former un identifiant : + - Décompose les accents (NFKD) et retire les diacritiques + - Convertit en minuscules + - Supprime tout caractère autre que les lettres et chiffres ASCII [a-z0-9] + """ + if not text: + return "" + normalized = unicodedata.normalize("NFKD", str(text)).encode("ASCII", "ignore").decode("utf-8") + return re.sub(r"[^a-z0-9]", "", normalized.lower()) + + +def generate_username(first_name: str, last_name: str, existing_usernames: Optional[set] = None) -> str: + """ + Génère un identifiant selon la convention : + première lettre du prénom + nom en minuscules (ex: 'Yassine Rheleimi' -> 'yrheleimi'). + Gère les doublons en ajoutant un suffixe numérique (yrheleimi2, yrheleimi3, etc.). + """ + fn_clean = clean_text_for_login(first_name) + ln_clean = clean_text_for_login(last_name) + + initial = fn_clean[0] if fn_clean else "u" + base = f"{initial}{ln_clean}" + if not base: + base = "user" + + if existing_usernames is None: + existing_usernames = set(User.objects.values_list("username", flat=True)) + + candidate = base + counter = 2 + while candidate in existing_usernames: + candidate = f"{base}{counter}" + counter += 1 + + existing_usernames.add(candidate) + return candidate + + +def generate_secure_password(length: int = 20) -> str: + """ + Génère un mot de passe sécurisé répondant exactement à la règle : + - Longueur de 20 caractères + - Commence par une Majuscule [A-Z] + - Suivi uniquement de lettres (majuscules/minuscules) et de chiffres [A-Za-z0-9] + - Aucun caractère spécial + - Garantit la présence d'au moins une minuscule et un chiffre parmi les caractères restants. + """ + if length < 4: + length = 20 + + first_char = secrets.choice(string.ascii_uppercase) + remaining_pool = string.ascii_letters + string.digits + + # S'assurer d'avoir au moins une minuscule et un chiffre dans le reste + guaranteed = [ + secrets.choice(string.ascii_lowercase), + secrets.choice(string.digits), + ] + other_chars = [secrets.choice(remaining_pool) for _ in range(length - 1 - len(guaranteed))] + rest = guaranteed + other_chars + secrets.SystemRandom().shuffle(rest) + + return first_char + "".join(rest) + + +def copy_user_config(source_user_or_config: Any, target_user_or_config: Any, copy_groups: bool = True) -> UserConfig: + """ + Copie exhaustive de la configuration d'un utilisateur source vers un utilisateur cible : + - Champs scalaires de UserConfig + - Rôles, vues applicatives, gestionnaires eSign, catégories d'approbation + - Thématiques et permissions par statut (UserThematics + UserThematicStatusPermission) + - Contrats et permissions par statut (UserContractAccess + UserContractStatusPermission) + - Entrepôts (UserWarehouseAccess) et emplacements (UserWarehouseLocationAccess) + - Équipes d'entreprise (UserCompanyTeamAccess) + - Accès spécifiques aux assets (UserAssetAccess) + - Groupes Django (target_user.groups) + """ + if isinstance(source_user_or_config, UserConfig): + source_config = source_user_or_config + source_user = source_config.user + else: + source_user = source_user_or_config + source_config = UserConfig.objects.get(user=source_user) + + if isinstance(target_user_or_config, UserConfig): + target_config = target_user_or_config + target_user = target_config.user + else: + target_user = target_user_or_config + target_config, _ = UserConfig.objects.get_or_create(user=target_user) + + with transaction.atomic(): + # 1. Champs scalaires + target_config.is_intern = source_config.is_intern + target_config.company = source_config.company + target_config.default_thematic = source_config.default_thematic + target_config.default_contract = source_config.default_contract + target_config.limit_assets_to_contracts = source_config.limit_assets_to_contracts + target_config.limit_interventions_to_contracts = source_config.limit_interventions_to_contracts + target_config.limit_shortcuts_to_contracts = source_config.limit_shortcuts_to_contracts + target_config.generic_email = source_config.generic_email + target_config.regional_standard = source_config.regional_standard + target_config.preferred_language = source_config.preferred_language + target_config.hide_intervention_map_by_default = source_config.hide_intervention_map_by_default + target_config.intervention_list_layout = source_config.intervention_list_layout + target_config.assets_list_layout = source_config.assets_list_layout + target_config.auto_logout_hours = source_config.auto_logout_hours + + # Préférences JSON + target_config.intervention_list_filters = dict(source_config.intervention_list_filters or {}) + target_config.mobile_intervention_list_filters = dict(source_config.mobile_intervention_list_filters or {}) + target_config.intervention_list_sort = dict(source_config.intervention_list_sort or {}) + target_config.intervention_list_columns = ( + list(source_config.intervention_list_columns) if source_config.intervention_list_columns is not None else None + ) + target_config.observation_list_filters = dict(source_config.observation_list_filters or {}) + target_config.observation_list_sort = dict(source_config.observation_list_sort or {}) + target_config.project_list_filters = dict(source_config.project_list_filters or {}) + target_config.project_list_sort = dict(source_config.project_list_sort or {}) + target_config.control_list_filters = dict(source_config.control_list_filters or {}) + target_config.control_list_sort = dict(source_config.control_list_sort or {}) + target_config.inspection_list_filters = dict(source_config.inspection_list_filters or {}) + target_config.inspection_list_sort = dict(source_config.inspection_list_sort or {}) + target_config.reporting_preferences = dict(source_config.reporting_preferences or {}) + target_config.assets_index_preferences = dict(source_config.assets_index_preferences or {}) + target_config.calendar_preferences = ( + dict(source_config.calendar_preferences) if source_config.calendar_preferences is not None else None + ) + + # Permissions spécifiques (eSign, documents, panoramax, transport exceptionnel) + target_config.can_add_unique_tag_documents = source_config.can_add_unique_tag_documents + target_config.esign_can_edit = source_config.esign_can_edit + target_config.esign_can_manage_rcp = source_config.esign_can_manage_rcp + target_config.esign_can_validate_rcp = source_config.esign_can_validate_rcp + target_config.panoramax_can_upload = source_config.panoramax_can_upload + target_config.has_exceptional_transport_profile = source_config.has_exceptional_transport_profile + target_config.exceptional_transport_can_edit = source_config.exceptional_transport_can_edit + + # Notifications + target_config.notification_reminder_enabled = source_config.notification_reminder_enabled + target_config.notification_reminder_emails = source_config.notification_reminder_emails + target_config.notification_reminder_delay_hours = source_config.notification_reminder_delay_hours + + target_config.save() + + # 2. Relations ManyToMany directes + target_config.roles.set(source_config.roles.all()) + target_config.accessible_views.set(source_config.accessible_views.all()) + target_config.esign_organizations.set(source_config.esign_organizations.all()) + target_config.approval_categories.set(source_config.approval_categories.all()) + + # 3. Groupes Django + if copy_groups and source_user: + target_user.groups.set(source_user.groups.all()) + + # 4. Thématiques et permissions par statut + target_config.userthematics.all().delete() + for ut in source_config.userthematics.prefetch_related("status_permissions").all(): + new_ut = UserThematics.objects.create( + user_config=target_config, + thematic=ut.thematic, + can_view_assets=ut.can_view_assets, + can_edit_assets=ut.can_edit_assets, + can_validate_assets=ut.can_validate_assets, + can_inspect=ut.can_inspect, + can_view_interventions=ut.can_view_interventions, + can_edit_interventions=ut.can_edit_interventions, + can_edit_locations=ut.can_edit_locations, + can_process_observations=ut.can_process_observations, + can_view_projects=ut.can_view_projects, + can_edit_projects=ut.can_edit_projects, + ) + perms_to_create = [ + UserThematicStatusPermission( + user_thematic=new_ut, + status=sp.status, + can_view=sp.can_view, + can_edit=sp.can_edit, + can_change_status_to=sp.can_change_status_to, + ) + for sp in ut.status_permissions.all() + ] + if perms_to_create: + UserThematicStatusPermission.objects.bulk_create(perms_to_create) + + # 5. Contrats et permissions par statut + target_config.usercontractaccess_set.all().delete() + for uc in source_config.usercontractaccess_set.prefetch_related("status_permissions").all(): + new_uc = UserContractAccess.objects.create( + user_config=target_config, + contract=uc.contract, + can_view_posts=uc.can_view_posts, + can_view_prices=uc.can_view_prices, + can_view_interventions=uc.can_view_interventions, + can_view_assets=uc.can_view_assets, + can_create_interventions=uc.can_create_interventions, + can_check_interventions=uc.can_check_interventions, + can_approve=uc.can_approve, + ) + perms_to_create = [ + UserContractStatusPermission( + user_contract=new_uc, + status=sp.status, + can_view=sp.can_view, + can_edit=sp.can_edit, + can_change_status_to=sp.can_change_status_to, + ) + for sp in uc.status_permissions.all() + ] + if perms_to_create: + UserContractStatusPermission.objects.bulk_create(perms_to_create) + + # 6. Entrepôts + target_config.userwarehouses.all().delete() + for wa in source_config.userwarehouses.all(): + UserWarehouseAccess.objects.create( + user_config=target_config, + warehouse=wa.warehouse, + can_view=wa.can_view, + can_manage=wa.can_manage, + can_administer=wa.can_administer, + can_manage_preparation_orders=wa.can_manage_preparation_orders, + can_manage_purchase_orders=wa.can_manage_purchase_orders, + can_validate_inventory=wa.can_validate_inventory, + is_default=wa.is_default, + ) + + # 7. Emplacements d'entrepôt + target_config.userlocations.all().delete() + for la in source_config.userlocations.all(): + UserWarehouseLocationAccess.objects.create( + user_config=target_config, + warehouse_location=la.warehouse_location, + can_perform_inventory=la.can_perform_inventory, + can_validate_inventory=la.can_validate_inventory, + ) + + # 8. Accès équipes + target_config.team_accesses.all().delete() + for ta in source_config.team_accesses.all(): + UserCompanyTeamAccess.objects.create( + user_config=target_config, + team=ta.team, + can_view=ta.can_view, + can_edit=ta.can_edit, + ) + + # 9. Accès spécifiques aux assets + target_config.asset_accesses.all().delete() + for aa in source_config.asset_accesses.all(): + UserAssetAccess.objects.create( + user_config=target_config, + content_type=aa.content_type, + object_id=aa.object_id, + can_view=aa.can_view, + can_edit=aa.can_edit, + can_delete=aa.can_delete, + editable_fields=aa.editable_fields, + visible_fields=aa.visible_fields, + ) + + return target_config + + +def read_users_file(file_path: str) -> List[Dict[str, str]]: + """ + Lit un fichier contenant les informations des utilisateurs. + Formats supportés : + - CSV / TSV (délimiteur détecté automatiquement) + - Excel (.xlsx) + - JSON (liste d'objets) + - TXT (format 'Prénom, Nom, Email' ou 'Prénom Nom ' par ligne) + + Colonnes tolérées : + - first_name : 'prenom', 'prénom', 'first_name', 'firstname' + - last_name : 'nom', 'last_name', 'lastname' + - email : 'email', 'e-mail', 'courriel', 'mail' + - username : 'username', 'login', 'identifiant' (optionnel) + """ + path = Path(file_path) + if not path.exists(): + raise FileNotFoundError(f"Fichier introuvable : {file_path}") + + ext = path.suffix.lower() + raw_rows: List[Dict[str, Any]] = [] + + if ext == ".xlsx": + import openpyxl + + wb = openpyxl.load_workbook(filename=file_path, data_only=True) + sheet = wb.active + rows = list(sheet.iter_rows(values_only=True)) + if not rows: + return [] + headers = [str(c).strip() if c is not None else "" for c in rows[0]] + for row in rows[1:]: + if not any(row): + continue + item = {} + for h, val in zip(headers, row): + if h: + item[h] = str(val).strip() if val is not None else "" + raw_rows.append(item) + + elif ext == ".json": + with open(file_path, "r", encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, list): + raw_rows = data + elif isinstance(data, dict) and "users" in data: + raw_rows = data["users"] + else: + raise ValueError("Structure JSON non reconnue (attendu: liste d'utilisateurs ou objet avec clé 'users').") + + else: + # Fichier texte ou CSV + content = None + for encoding in ["utf-8", "utf-8-sig", "latin-1", "cp1252"]: + try: + with open(file_path, "r", encoding=encoding) as f: + content = f.read() + break + except UnicodeDecodeError: + continue + + if content is None: + raise ValueError(f"Impossible de décoder le fichier {file_path} avec un encodage supporté.") + + lines = [line.strip() for line in content.splitlines() if line.strip()] + if not lines: + return [] + + # Tenter d'abord la détection CSV + first_line = lines[0] + delimiter = None + if ";" in first_line: + delimiter = ";" + elif "\t" in first_line: + delimiter = "\t" + elif "," in first_line: + delimiter = "," + + # Vérifier si c'est un CSV avec en-têtes + reader = csv.DictReader(lines, delimiter=delimiter or ",") + has_named_headers = False + if reader.fieldnames: + normalized_fields = [clean_text_for_login(fn) for fn in reader.fieldnames if fn] + if any(key in normalized_fields for key in ["prenom", "nom", "email", "firstname", "lastname"]): + has_named_headers = True + + if has_named_headers: + for row in reader: + raw_rows.append(row) + else: + # Traitement ligne par ligne flexible + for line in lines: + # Format: Prénom Nom + bracket_match = re.search(r"^(.*?)\s*<([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)>\s*$", line) + if bracket_match: + full_name, email = bracket_match.group(1).strip(), bracket_match.group(2).strip() + parts = full_name.split(None, 1) + first_name = parts[0] if parts else "" + last_name = parts[1] if len(parts) > 1 else "" + raw_rows.append({"first_name": first_name, "last_name": last_name, "email": email}) + continue + + # Format séparé par virgule ou point-virgule ou tabulation + delims = [";", ",", "\t"] + detected_delim = next((d for d in delims if d in line), None) + if detected_delim: + parts = [p.strip() for p in line.split(detected_delim)] + if len(parts) >= 3: + # Supposer prénom, nom, email + raw_rows.append({"first_name": parts[0], "last_name": parts[1], "email": parts[2]}) + elif len(parts) == 2: + raw_rows.append({"first_name": parts[0], "last_name": "", "email": parts[1]}) + else: + # Ligne avec espaces + parts = line.split() + email_part = next((p for p in parts if "@" in p), None) + if email_part: + parts.remove(email_part) + fn = parts[0] if parts else "" + ln = " ".join(parts[1:]) if len(parts) > 1 else "" + raw_rows.append({"first_name": fn, "last_name": ln, "email": email_part}) + + # Normalisation des dictionnaires extraits + normalized_users = [] + header_mapping = { + "first_name": ["first_name", "firstname", "prenom", "prénom", "first"], + "last_name": ["last_name", "lastname", "nom", "last"], + "email": ["email", "e-mail", "courriel", "mail"], + "username": ["username", "login", "identifiant"], + } + + for raw in raw_rows: + user_dict = {} + for target_key, aliases in header_mapping.items(): + for key, val in raw.items(): + k_clean = clean_text_for_login(str(key)) + if any(clean_text_for_login(alias) == k_clean for alias in aliases): + if val is not None and str(val).strip(): + user_dict[target_key] = str(val).strip() + break + if target_key not in user_dict: + user_dict[target_key] = "" + + # Validation minimale + if user_dict.get("email") or user_dict.get("first_name") or user_dict.get("last_name"): + normalized_users.append(user_dict) + + return normalized_users + + +def render_email_template( + template_content_or_path: str, + context: Dict[str, Any], +) -> Tuple[str, str]: + """ + Rend le contenu de l'e-mail avec les variables du contexte. + Retourne (sujet, corps_texte). + """ + template_text = template_content_or_path + path = Path(template_content_or_path) + if path.exists() and path.is_file(): + with open(path, "r", encoding="utf-8") as f: + template_text = f.read() + + # Valeurs par défaut du contexte + full_context = { + "app_name": "StreetUp", + "desktop_url": "https://bm.streetup.be/", + "mobile_url": "https://bm.streetup.be/mobile", + "install_url": "https://bm.streetup.be/mobile/install/", + "first_name": "", + "last_name": "", + "username": "", + "password": "", + "email": "", + } + full_context.update(context) + + # Remplacement sécurisé des clés + rendered = template_text + for key, val in full_context.items(): + rendered = rendered.replace(f"{{{key}}}", str(val)) + + # Extraction du sujet s'il est spécifié dans le texte + lines = rendered.splitlines() + subject = f"Accès {full_context.get('app_name', 'StreetUp')}" + body_lines = [] + in_body = False + + for line in lines: + if not in_body and re.match(r"^\s*sujet\s*:\s*(.*)$", line, re.IGNORECASE): + subject_match = re.match(r"^\s*sujet\s*:\s*(.*)$", line, re.IGNORECASE) + if subject_match: + subject = subject_match.group(1).strip() + in_body = True + else: + body_lines.append(line) + + body = "\n".join(body_lines).strip() + return subject, body diff --git a/loko/scripts/create_outlook_drafts.py b/loko/scripts/create_outlook_drafts.py new file mode 100755 index 0000000..23113c1 --- /dev/null +++ b/loko/scripts/create_outlook_drafts.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +""" +Script pour la création de brouillons d'e-mails pour Outlook (PWA / Web sous Ubuntu). + +Consomme le fichier 'drafts_payload.json' (ou le dossier généré lors de la création d'utilisateurs). +Modes disponibles : + - pwa (défaut) : Ouvre directement les deep links de composition dans Chromium avec la session Outlook connectée. + - html : Ouvre la page web interactive locale 'open_drafts.html'. + - eml : Exporte des fichiers brouillons standards .eml (RFC822 avec X-Unsent: 1). + +Usage : + python scripts/create_outlook_drafts.py chemin/vers/drafts_payload.json + python scripts/create_outlook_drafts.py chemin/vers/drafts_payload.json --step + python scripts/create_outlook_drafts.py chemin/vers/drafts_payload.json --delay 3 + python scripts/create_outlook_drafts.py chemin/vers/drafts_payload.json --mode html + python scripts/create_outlook_drafts.py chemin/vers/drafts_payload.json --mode eml +""" + +import argparse +import email.message +import json +import os +import shutil +import subprocess +import sys +import time +import urllib.parse +import webbrowser +from pathlib import Path + + +def find_chromium_cmd() -> list: + """Détecte la commande Chromium disponible pour ouvrir la session PWA sous Ubuntu.""" + candidates = [ + "/snap/bin/chromium", + "chromium", + "chromium-browser", + "google-chrome", + "google-chrome-stable", + ] + for cmd in candidates: + if shutil.which(cmd): + # Vérifier si on peut ajouter le profil Default + return [cmd, "--profile-directory=Default"] + return [] + + +def create_eml_draft(to_email: str, subject: str, body: str, output_path: Path): + """Génère un fichier .eml standard marqué comme brouillon (X-Unsent: 1).""" + msg = email.message.EmailMessage() + msg["To"] = to_email + msg["Subject"] = subject + msg["X-Unsent"] = "1" # Marqueur standard de brouillon pour Outlook, Thunderbird, Apple Mail + msg.set_content(body) + + with open(output_path, "wb") as f: + f.write(msg.as_bytes()) + + +def main(): + parser = argparse.ArgumentParser( + description="Crée des brouillons d'e-mails dans Outlook PWA / Web ou génère des fichiers brouillons." + ) + parser.add_argument( + "input", + type=str, + help="Chemin vers le fichier drafts_payload.json ou le dossier de sortie de création d'utilisateurs.", + ) + parser.add_argument( + "--mode", + choices=["pwa", "html", "eml"], + default="pwa", + help="Mode de traitement : 'pwa' (ouvre dans Outlook PWA), 'html' (ouvre la page locale), 'eml' (exporte des fichiers .eml).", + ) + parser.add_argument( + "--delay", + type=float, + default=2.5, + help="Délai en secondes entre l'ouverture de chaque brouillon en mode automatique (défaut : 2.5s).", + ) + parser.add_argument( + "--step", + action="store_true", + default=False, + help="Mode pas-à-pas : demande confirmation avant d'ouvrir chaque brouillon.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + default=False, + help="Affiche les actions prévues sans ouvrir de navigateur ni créer de fichiers.", + ) + + args = parser.parse_args() + + input_path = Path(args.input) + if not input_path.exists(): + sys.stderr.write(f"Erreur : '{args.input}' est introuvable.\n") + sys.exit(1) + + # Résoudre le fichier drafts_payload.json + payload_file = input_path + if input_path.is_dir(): + payload_file = input_path / "drafts_payload.json" + if not payload_file.exists(): + sys.stderr.write(f"Erreur : Aucun fichier drafts_payload.json trouvé dans '{input_path}'.\n") + sys.exit(1) + + try: + with open(payload_file, "r", encoding="utf-8") as f: + drafts = json.load(f) + except Exception as exc: + sys.stderr.write(f"Erreur lors de la lecture de {payload_file} : {exc}\n") + sys.exit(1) + + if not isinstance(drafts, list) or not drafts: + print("Aucun brouillon trouvé dans la charge utile.") + return + + print(f"Chargement de {len(drafts)} brouillon(s) depuis {payload_file.name}...") + + # Mode HTML + if args.mode == "html": + html_file = payload_file.parent / "open_drafts.html" + if not html_file.exists(): + sys.stderr.write(f"Page HTML introuvable à {html_file}.\n") + sys.exit(1) + print(f"Ouverture de la page interactive : {html_file.resolve()}") + if not args.dry_run: + webbrowser.open(f"file://{html_file.resolve()}") + return + + # Mode EML + if args.mode == "eml": + eml_dir = payload_file.parent / "eml_drafts" + if not args.dry_run: + eml_dir.mkdir(parents=True, exist_ok=True) + print(f"Génération des fichiers .eml dans {eml_dir.resolve()}...") + for item in drafts: + username = item.get("username", "user") + out_eml = eml_dir / f"{username}_draft.eml" + print(f" -> {out_eml.name} ({item.get('to')})") + if not args.dry_run: + create_eml_draft( + to_email=item.get("to", ""), + subject=item.get("subject", ""), + body=item.get("body", ""), + output_path=out_eml, + ) + print("Fichiers .eml générés avec succès.") + return + + # Mode PWA (Défaut) + chromium_cmd = find_chromium_cmd() + if chromium_cmd: + print(f"Navigateur détecté pour la session PWA : {' '.join(chromium_cmd)}") + else: + print("Utilisation du navigateur système par défaut.") + + print("\n" + "=" * 60) + print(f"Ouverture des brouillons dans Outlook ({len(drafts)} message(s))") + print("Outlook sauvegardera automatiquement chaque message dans le dossier 'Brouillons'.") + print("=" * 60 + "\n") + + for idx, item in enumerate(drafts, start=1): + to_email = item.get("to", "") + subject = item.get("subject", "") + deeplink = item.get("outlook_deeplink") + first_name = item.get("first_name", "") + last_name = item.get("last_name", "") + + if not deeplink: + params = urllib.parse.urlencode({ + "to": to_email, + "subject": subject, + "body": item.get("body", ""), + }) + deeplink = f"https://outlook.cloud.microsoft/mail/deeplink/compose?{params}" + + print(f"[{idx}/{len(drafts)}] {first_name} {last_name} <{to_email}>") + + if args.step: + choice = input(" -> Appuyez sur Entrée pour ouvrir dans Outlook (ou 'q' pour quitter, 's' pour passer) : ").strip().lower() + if choice == "q": + print("Arrêt demandé par l'utilisateur.") + break + elif choice == "s": + print(" Passé.") + continue + + if args.dry_run: + print(f" [DRY-RUN] Ouverture de : {deeplink[:90]}...") + else: + if chromium_cmd: + try: + subprocess.Popen( + chromium_cmd + [deeplink], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except Exception: + webbrowser.open(deeplink) + else: + webbrowser.open(deeplink) + + print(" -> Brouillon ouvert dans Outlook !") + + if not args.step and idx < len(drafts) and not args.dry_run: + time.sleep(args.delay) + + print("\nTous les brouillons ont été traités.") + + +if __name__ == "__main__": + main() diff --git a/loko/scripts/create_users.py b/loko/scripts/create_users.py new file mode 100755 index 0000000..4ffd000 --- /dev/null +++ b/loko/scripts/create_users.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +""" +Script autonome pour la création d'utilisateurs, la duplication de configuration, +la génération des emails d'accès et des fichiers de brouillons. + +Utilisation : + python scripts/create_users.py chemin/vers/utilisateurs.csv --source-user admin + python scripts/create_users.py chemin/vers/utilisateurs.xlsx --source-user yrheleimi --dry-run +""" + +import os +import sys +from pathlib import Path + +# Ajouter le répertoire 'loko' au sys.path pour charger Django +current_dir = Path(__file__).resolve().parent +loko_dir = current_dir.parent if current_dir.name == "scripts" else current_dir +if str(loko_dir) not in sys.path: + sys.path.insert(0, str(loko_dir)) + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "loko.settings") + +try: + import django + + django.setup() +except Exception as e: + sys.stderr.write(f"Erreur d'initialisation de l'environnement Django : {e}\n") + sys.exit(1) + +from django.core.management import call_command + + +def main(): + if len(sys.argv) < 2 or "--help" in sys.argv or "-h" in sys.argv: + call_command("create_users_from_file", "--help") + sys.exit(0) + + # Transmettre tous les arguments reçus à la commande Django + try: + call_command("create_users_from_file", *sys.argv[1:]) + except Exception as exc: + sys.stderr.write(f"\nErreur : {exc}\n") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/loko/scripts/init-db.sql b/loko/scripts/init-db.sql new file mode 100644 index 0000000..f390f21 --- /dev/null +++ b/loko/scripts/init-db.sql @@ -0,0 +1,18 @@ +CREATE EXTENSION IF NOT EXISTS postgis; +CREATE EXTENSION IF NOT EXISTS postgis_topology; +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS unaccent; + +-- Créer la fonction unaccent_immutable pour l'insensibilité aux accents dans les requêtes/index +CREATE OR REPLACE FUNCTION public.unaccent_immutable(text) + RETURNS text AS $$ + SELECT public.unaccent('public.unaccent', $1); + $$ LANGUAGE sql IMMUTABLE PARALLEL SAFE STRICT; + +CREATE OR REPLACE FUNCTION public.unaccent_immutable(character varying) + RETURNS text AS $$ + SELECT public.unaccent('public.unaccent', $1::text); + $$ LANGUAGE sql IMMUTABLE PARALLEL SAFE STRICT; + +-- Créer le schéma reporting +CREATE SCHEMA IF NOT EXISTS reporting; diff --git a/loko/scripts/templates/user_welcome_email.txt b/loko/scripts/templates/user_welcome_email.txt new file mode 100644 index 0000000..4ae7f1e --- /dev/null +++ b/loko/scripts/templates/user_welcome_email.txt @@ -0,0 +1,17 @@ +Sujet : Accès {app_name} + +Bonjour {first_name} + +Voici votre login pour accéder à {app_name}. + +Version desktop : {desktop_url} +Version mobile : {mobile_url} + +{username} +{password} + +Aide pour l’installation de la partie mobile : {install_url} + +N'hésitez pas à me contacter pour toute remarque ou question. + +Bien à vous