Valogeo/loko
Valogeo/loko
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add bulk user creation management command with configuration cloning and email/Outlook draft generation

Browse source
This commit is contained in:
kdeterme 2026-09-04 09:18:07 +02:00
parent 74679289a4
commit 946cbcae91
10 changed files with 1628 additions and 208 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.gitignore vendored
View file

@ -27,7 +27,9 @@ __pycache__/
loko/.env
scripts/
/scripts/
!loko/scripts/
!loko/scripts/**
carl_export/
carl_export.*

210
loko/common/admin.py
View file

@ -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:

598
loko/common/management/commands/create_users_from_file.py Normal file
View file

@ -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"""<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Brouillons Outlook - Accès {app_name}</title>
<style>
:root {{
--bg: #0f172a;
--card-bg: #1e293b;
--text: #f8fafc;
--text-muted: #94a3b8;
--primary: #38bdf8;
--primary-hover: #0284c7;
--success: #22c55e;
--border: #334155;
}}
body {{
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
background: var(--bg);
color: var(--text);
margin: 0;
padding: 30px 20px;
}}
.container {{
max-width: 960px;
margin: 0 auto;
}}
header {{
margin-bottom: 30px;
border-bottom: 1px solid var(--border);
padding-bottom: 20px;
}}
h1 {{
margin: 0 0 10px 0;
color: var(--primary);
}}
.actions-bar {{
display: flex;
gap: 12px;
margin-bottom: 25px;
flex-wrap: wrap;
}}
button, a.btn {{
background: var(--primary);
color: #0f172a;
border: none;
padding: 10px 18px;
border-radius: 8px;
font-weight: 600;
font-size: 14px;
cursor: pointer;
text-decoration: none;
display: inline-flex;
align-items: center;
gap: 8px;
transition: background 0.15s ease, transform 0.1s ease;
}}
button:hover, a.btn:hover {{
background: var(--primary-hover);
color: #fff;
}}
button.btn-secondary {{
background: var(--border);
color: var(--text);
}}
button.btn-secondary:hover {{
background: #475569;
}}
.user-card {{
background: var(--card-bg);
border: 1px solid var(--border);
border-radius: 10px;
padding: 18px 22px;
margin-bottom: 16px;
display: flex;
align-items: center;
justify-content: space-between;
gap: 20px;
flex-wrap: wrap;
}}
.user-info {{
flex: 1;
min-width: 250px;
}}
.user-info h3 {{
margin: 0 0 6px 0;
font-size: 17px;
}}
.user-meta {{
color: var(--text-muted);
font-size: 13px;
line-height: 1.5;
}}
.user-meta code {{
background: #0f172a;
padding: 2px 6px;
border-radius: 4px;
color: #38bdf8;
font-size: 13px;
}}
.card-actions {{
display: flex;
gap: 10px;
align-items: center;
}}
.badge {{
display: inline-block;
padding: 3px 8px;
border-radius: 6px;
font-size: 12px;
font-weight: 600;
background: #334155;
color: #cbd5e1;
}}
.badge.opened {{
background: #14532d;
color: #86efac;
}}
.toast {{
position: fixed;
bottom: 20px;
right: 20px;
background: var(--success);
color: #0f172a;
padding: 12px 20px;
border-radius: 8px;
font-weight: 600;
box-shadow: 0 10px 15px -3px rgba(0,0,0,0.3);
display: none;
z-index: 1000;
}}
</style>
</head>
<body>
<div class="container">
<header>
<h1>Accès créés pour {app_name}</h1>
<p style="color: var(--text-muted); margin: 0;">
{len(drafts)} utilisateur(s) traité(s). Cliquez sur <strong>Ouvrir dans Outlook</strong> pour pré-remplir automatiquement le brouillon dans votre messagerie.
</p>
</header>
<div class="actions-bar">
<button id="open-all-btn" onclick="openAllDraftsWithDelay()">
Ouvrir tous les brouillons dans Outlook (avec délai)
</button>
<button class="btn-secondary" onclick="copyAllSummary()">
Copier la liste (Logins & Mots de passe)
</button>
</div>
<div id="users-list"></div>
</div>
<div id="toast" class="toast">Texte copié dans le presse-papier !</div>
<script>
const drafts = {items_json};
const container = document.getElementById('users-list');
function renderList() {{
container.innerHTML = '';
drafts.forEach((item, index) => {{
const card = document.createElement('div');
card.className = 'user-card';
card.id = `card-${{index}}`;
card.innerHTML = `
<div class="user-info">
<h3>${{item.first_name}} ${{item.last_name}}</h3>
<div class="user-meta">
Destinataire : <strong>${{item.to}}</strong><br>
Login : <code>${{item.username}}</code> |
Mot de passe : <code>${{item.password}}</code>
</div>
</div>
<div class="card-actions">
<span class="badge" id="badge-${{index}}">En attente</span>
<button class="btn-secondary" onclick="copyText('${{item.password}}', 'Mot de passe copié !')">Copier MDP</button>
<button class="btn-secondary" onclick="copyUserEmailBody(${{index}})">Copier Message</button>
<a href="${{item.outlook_deeplink}}" target="_blank" class="btn" onclick="markOpened(${{index}})">
Ouvrir dans Outlook
</a>
</div>
`;
container.appendChild(card);
}});
}}
function markOpened(index) {{
const badge = document.getElementById(`badge-${{index}}`);
if (badge) {{
badge.textContent = 'Ouvert';
badge.classList.add('opened');
}}
}}
function copyText(text, msg = 'Texte copié !') {{
navigator.clipboard.writeText(text).then(() => showToast(msg));
}}
function copyUserEmailBody(index) {{
const item = drafts[index];
copyText(item.body, `Message pour ${{item.first_name}} copié !`);
}}
function copyAllSummary() {{
const lines = drafts.map(d => `${{d.first_name}} ${{d.last_name}} | ${{d.to}} | Login: ${{d.username}} | MDP: ${{d.password}}`);
copyText(lines.join('\\n'), 'Résumé copié !');
}}
function showToast(msg) {{
const toast = document.getElementById('toast');
toast.textContent = msg;
toast.style.display = 'block';
setTimeout(() => {{ toast.style.display = 'none'; }}, 2500);
}}
async function openAllDraftsWithDelay() {{
if (!confirm(`Ouvrir les ${{drafts.length}} brouillons l'un après l'autre dans Outlook ?`)) return;
for (let i = 0; i < drafts.length; i++) {{
window.open(drafts[i].outlook_deeplink, '_blank');
markOpened(i);
if (i < drafts.length - 1) {{
await new Promise(r => setTimeout(r, 2000));
}}
}}
}}
renderList();
</script>
</body>
</html>
"""
with open(html_path, "w", encoding="utf-8") as f:
f.write(html_content)

17
loko/common/templates/emails/user_welcome_email.txt Normal file
View file

@ -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

204
loko/common/tests_user_services.py Normal file
View file

@ -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 <yrheleimi@example.com>\n")
f.write("Bob Marley <bob@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"], "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()
)

505
loko/common/user_services.py Normal file
View file

@ -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 <email>' 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 <email@domain.com>
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

215
loko/scripts/create_outlook_drafts.py Executable file
View file

@ -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()

48
loko/scripts/create_users.py Executable file
View file

@ -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()

18
loko/scripts/init-db.sql Normal file
View file

@ -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;

17
loko/scripts/templates/user_welcome_email.txt Normal file
View file

@ -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