""" Commande Django pour pré-remplir les permissions par statut (UserThematicStatusPermission et UserContractStatusPermission) à partir des rôles et flags déjà présents dans la base. Même logique que la migration 0055, mais utilisable à tout moment sur n'importe quel serveur, avec des options de contrôle. Usage : # Crée les lignes manquantes (ne touche pas aux lignes existantes) python manage.py fill_status_permissions # Met également à jour les lignes existantes selon les rôles python manage.py fill_status_permissions --update # Efface tout et recrée depuis zéro (demande confirmation) python manage.py fill_status_permissions --reset # Aperçu sans rien écrire python manage.py fill_status_permissions --dry-run # Limité à un seul utilisateur python manage.py fill_status_permissions --user kevin@example.com # Combinaisons python manage.py fill_status_permissions --update --user kevin@example.com --dry-run """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction from common.models import UserThematics, UserContractAccess from common.models import UserThematicStatusPermission, UserContractStatusPermission # --------------------------------------------------------------------------- # Référentiels rôles → statuts (sans 'archived', supprimé en migration 0057) # --------------------------------------------------------------------------- VIEW_ROLE_STATUS_PERMISSIONS = { 'admin': [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'manager': [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'controller': [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'operator': [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'editor': [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'external_manager': [ 'in_preparation', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'technician': [ 'in_preparation', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', ], 'observer': [ 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], 'viewer': [ 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ], } ROLE_TRANSITIONS = { 'manager': [ ('in_preparation', 'to_be_approved'), ('in_preparation', 'to_be_planned'), ('in_preparation', 'to_be_processed'), ('in_preparation', 'canceled'), ('to_be_approved', 'to_be_planned'), ('to_be_approved', 'to_be_processed'), ('to_be_approved', 'canceled'), ('to_be_planned', 'to_be_processed'), ('to_be_planned', 'canceled'), ('to_be_processed','canceled'), ('processed', 'to_be_corrected'), ('processed', 'validated'), ('corrected', 'validated'), ('invoiced', 'closed'), ], 'technician': [ ('in_preparation', 'to_be_approved'), ('in_preparation', 'to_be_planned'), ('in_preparation', 'to_be_processed'), ('in_preparation', 'canceled'), ('to_be_processed','assigned'), ('assigned', 'in_progress'), ('assigned', 'to_be_processed'), ('to_be_processed','in_progress'), ('in_progress', 'on_pause'), ('in_progress', 'finished'), ('on_pause', 'in_progress'), ('on_pause', 'finished'), ], 'external_manager': [ ('in_preparation', 'to_be_approved'), ('in_preparation', 'to_be_planned'), ('in_preparation', 'to_be_processed'), ('in_preparation', 'canceled'), ('to_be_approved', 'canceled'), ('to_be_planned', 'to_be_processed'), ('to_be_planned', 'canceled'), ('to_be_processed','assigned'), ('to_be_processed','in_progress'), ('to_be_processed','to_be_planned'), ('to_be_processed','canceled'), ('assigned', 'in_progress'), ('assigned', 'to_be_processed'), ('assigned', 'canceled'), ('in_progress', 'on_pause'), ('in_progress', 'finished'), ('in_progress', 'canceled'), ('on_pause', 'in_progress'), ('on_pause', 'finished'), ('on_pause', 'to_be_planned'), ('on_pause', 'to_be_processed'), ('on_pause', 'canceled'), ('finished', 'processed'), ('finished', 'to_be_processed'), ('finished', 'on_pause'), ('finished', 'canceled'), ('to_be_corrected','corrected'), ], 'controller': [ ('in_preparation', 'to_be_approved'), ('in_preparation', 'to_be_planned'), ('in_preparation', 'to_be_processed'), ('to_be_approved', 'to_be_planned'), ('to_be_approved', 'to_be_processed'), ('to_be_approved', 'canceled'), ('to_be_planned', 'to_be_processed'), ('to_be_planned', 'canceled'), ('to_be_processed','canceled'), ('processed', 'validated'), ('processed', 'to_be_corrected'), ('corrected', 'to_be_corrected'), ('corrected', 'validated'), ], 'operator': [ ('in_preparation', 'to_be_approved'), ('in_preparation', 'to_be_planned'), ('in_preparation', 'to_be_processed'), ('in_preparation', 'canceled'), ('to_be_approved', 'to_be_planned'), ('to_be_approved', 'to_be_processed'), ('to_be_approved', 'canceled'), ('to_be_planned', 'to_be_processed'), ('to_be_planned', 'canceled'), ('to_be_processed','canceled'), ], } ROLE_EDIT_STATUSES = { 'admin': list(VIEW_ROLE_STATUS_PERMISSIONS['admin']), 'manager': ['in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced'], 'controller': ['in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'processed', 'to_be_corrected', 'corrected', 'validated'], 'external_manager': ['in_preparation', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'to_be_corrected', 'corrected'], 'technician': ['in_preparation', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished'], 'operator': ['in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed'], } ALL_STATUSES = [ 'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished', 'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled', ] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _build_view_statuses(roles): result = set() for role in roles: result.update(VIEW_ROLE_STATUS_PERMISSIONS.get(role, [])) return result def _build_transition_targets(roles): result = set() for role in roles: if role == 'admin': return set(ALL_STATUSES) for (_src, dst) in ROLE_TRANSITIONS.get(role, []): result.add(dst) return result def _build_edit_statuses(roles, can_edit_flag): if not can_edit_flag: return set() result = set() for role in roles: result.update(ROLE_EDIT_STATUSES.get(role, [])) return result def _compute_thematic_rows(ut): """Retourne la liste de dicts {status, can_view, can_edit, can_change_status_to} pour un UserThematics.""" roles = list(ut.user_config.roles.values_list('name', flat=True)) view_statuses = _build_view_statuses(roles) trans_targets = _build_transition_targets(roles) if ut.can_edit_interventions else set() edit_statuses = _build_edit_statuses(roles, ut.can_edit_interventions) all_involved = view_statuses | trans_targets | edit_statuses return [ { 'status': status, 'can_view': status in view_statuses, 'can_edit': status in edit_statuses, 'can_change_status_to': status in trans_targets, } for status in all_involved ] def _compute_contract_rows(uca): """Retourne la liste de dicts {status, can_view, can_edit, can_change_status_to} pour un UserContractAccess.""" roles = list(uca.user_config.roles.values_list('name', flat=True)) view_statuses = _build_view_statuses(roles) trans_targets = _build_transition_targets(roles) edit_roles = {'admin', 'manager', 'controller', 'external_manager', 'technician'} can_edit_flag = bool(set(roles) & edit_roles) edit_statuses = _build_edit_statuses(roles, can_edit_flag) if uca.can_approve: trans_targets.update(['to_be_planned', 'to_be_processed', 'canceled']) if uca.can_create_interventions: view_statuses.add('in_preparation') edit_statuses.add('in_preparation') all_involved = view_statuses | trans_targets | edit_statuses return [ { 'status': status, 'can_view': status in view_statuses, 'can_edit': status in edit_statuses, 'can_change_status_to': status in trans_targets, } for status in all_involved ] # --------------------------------------------------------------------------- # Commande # --------------------------------------------------------------------------- class Command(BaseCommand): help = ( 'Pré-remplit UserThematicStatusPermission et UserContractStatusPermission ' 'à partir des rôles et flags existants.' ) def add_arguments(self, parser): parser.add_argument( '--update', action='store_true', default=False, help='Met à jour les lignes existantes en plus de créer les manquantes.', ) parser.add_argument( '--reset', action='store_true', default=False, help='Supprime toutes les lignes existantes puis les recrée. ' 'Demande une confirmation interractive sauf si --no-input est fourni.', ) parser.add_argument( '--dry-run', action='store_true', default=False, dest='dry_run', help='Affiche ce qui serait fait sans écrire en base.', ) parser.add_argument( '--user', metavar='EMAIL', default=None, help='Limite le traitement à un seul utilisateur (email).', ) parser.add_argument( '--no-input', '--noinput', action='store_true', default=False, dest='no_input', help='Ne pose pas de question de confirmation (utile pour les scripts automatisés).', ) # ------------------------------------------------------------------ def handle(self, *args, **options): dry_run = options['dry_run'] do_reset = options['reset'] do_update = options['update'] user_email = options['user'] no_input = options['no_input'] if dry_run: self.stdout.write(self.style.WARNING('Mode DRY-RUN : aucune modification ne sera effectuée.\n')) if do_reset and not dry_run: if not no_input: confirm = input( 'ATTENTION : --reset va supprimer toutes les lignes de permissions par statut ' 'et les recréer depuis zéro.\nContinuer ? [o/N] ' ) if confirm.strip().lower() not in ('o', 'oui', 'y', 'yes'): self.stdout.write(self.style.WARNING('Opération annulée.')) return # Filtrage optionnel par utilisateur ut_qs = UserThematics.objects.select_related('user_config').prefetch_related('user_config__roles') uca_qs = UserContractAccess.objects.select_related('user_config').prefetch_related('user_config__roles') if user_email: ut_qs = ut_qs.filter(user_config__user__email=user_email) uca_qs = uca_qs.filter(user_config__user__email=user_email) if not ut_qs.exists() and not uca_qs.exists(): raise CommandError(f"Aucun utilisateur trouvé avec l'email « {user_email} ».") with transaction.atomic(): t_created, t_updated, t_skipped = self._process_thematics(ut_qs, do_reset, do_update, dry_run) c_created, c_updated, c_skipped = self._process_contracts(uca_qs, do_reset, do_update, dry_run) # Résumé prefix = '[DRY-RUN] ' if dry_run else '' self.stdout.write('') self.stdout.write(self.style.SUCCESS( f'{prefix}Thématiques — créées: {t_created} | mises à jour: {t_updated} | ignorées: {t_skipped}' )) self.stdout.write(self.style.SUCCESS( f'{prefix}Contrats — créées: {c_created} | mises à jour: {c_updated} | ignorées: {c_skipped}' )) # ------------------------------------------------------------------ def _process_thematics(self, qs, do_reset, do_update, dry_run): created = updated = skipped = 0 if do_reset and not dry_run: deleted, _ = UserThematicStatusPermission.objects.filter( user_thematic__in=qs.values_list('pk', flat=True) ).delete() self.stdout.write(f' Suppression de {deleted} lignes thématiques existantes.') for ut in qs.filter(can_view_interventions=True).iterator(chunk_size=500): rows = _compute_thematic_rows(ut) if not rows: continue existing_map = {} if not do_reset: existing_map = { p.status: p for p in UserThematicStatusPermission.objects.filter(user_thematic=ut) } for row in rows: status = row['status'] if status in existing_map: perm = existing_map[status] needs_update = ( perm.can_view != row['can_view'] or perm.can_edit != row['can_edit'] or perm.can_change_status_to != row['can_change_status_to'] ) if needs_update and do_update: if dry_run: self.stdout.write( f' [MAJ thém.] ut={ut.pk} statut={status} ' f'view={row["can_view"]} edit={row["can_edit"]} ' f'transition={row["can_change_status_to"]}' ) else: perm.can_view = row['can_view'] perm.can_edit = row['can_edit'] perm.can_change_status_to = row['can_change_status_to'] perm.save(update_fields=['can_view', 'can_edit', 'can_change_status_to']) updated += 1 else: skipped += 1 else: if dry_run: self.stdout.write( f' [CRÉE thém.] ut={ut.pk} statut={status} ' f'view={row["can_view"]} edit={row["can_edit"]} ' f'transition={row["can_change_status_to"]}' ) else: UserThematicStatusPermission.objects.get_or_create( user_thematic=ut, status=status, defaults={ 'can_view': row['can_view'], 'can_edit': row['can_edit'], 'can_change_status_to': row['can_change_status_to'], }, ) created += 1 return created, updated, skipped # ------------------------------------------------------------------ def _process_contracts(self, qs, do_reset, do_update, dry_run): created = updated = skipped = 0 if do_reset and not dry_run: deleted, _ = UserContractStatusPermission.objects.filter( user_contract__in=qs.values_list('pk', flat=True) ).delete() self.stdout.write(f' Suppression de {deleted} lignes contrat existantes.') for uca in qs.filter(can_view_interventions=True).iterator(chunk_size=500): rows = _compute_contract_rows(uca) if not rows: continue existing_map = {} if not do_reset: existing_map = { p.status: p for p in UserContractStatusPermission.objects.filter(user_contract=uca) } for row in rows: status = row['status'] if status in existing_map: perm = existing_map[status] needs_update = ( perm.can_view != row['can_view'] or perm.can_edit != row['can_edit'] or perm.can_change_status_to != row['can_change_status_to'] ) if needs_update and do_update: if dry_run: self.stdout.write( f' [MAJ cont.] uca={uca.pk} statut={status} ' f'view={row["can_view"]} edit={row["can_edit"]} ' f'transition={row["can_change_status_to"]}' ) else: perm.can_view = row['can_view'] perm.can_edit = row['can_edit'] perm.can_change_status_to = row['can_change_status_to'] perm.save(update_fields=['can_view', 'can_edit', 'can_change_status_to']) updated += 1 else: skipped += 1 else: if dry_run: self.stdout.write( f' [CRÉE cont.] uca={uca.pk} statut={status} ' f'view={row["can_view"]} edit={row["can_edit"]} ' f'transition={row["can_change_status_to"]}' ) else: UserContractStatusPermission.objects.get_or_create( user_contract=uca, status=status, defaults={ 'can_view': row['can_view'], 'can_edit': row['can_edit'], 'can_change_status_to': row['can_change_status_to'], }, ) created += 1 return created, updated, skipped