""" Copie les droits d'accès (UserContractAccess + UserContractStatusPermission) d'un contrat source vers un contrat cible, pour tous les utilisateurs du contrat source (ou un sous-ensemble). Comportement par défaut : - Si l'utilisateur a déjà un accès au contrat cible → son accès est ignoré (utiliser --update pour le remplacer). - L'accès au contrat source est conservé tel quel ; il n'est pas supprimé. Options : --source NUMERO Numéro du contrat source (obligatoire) --target NUMERO Numéro du contrat cible (obligatoire) --update Met à jour l'accès cible s'il existe déjà --dry-run Affiche sans écrire --user EMAIL Limite à un utilisateur précis (répétable) Exemples : python manage.py copy_contract_access --source CSC-2022-01 --target CSC-2025-01 python manage.py copy_contract_access --source CSC-2022-01 --target CSC-2025-01 --update python manage.py copy_contract_access --source CSC-2022-01 --target CSC-2025-01 --dry-run python manage.py copy_contract_access --source CSC-2022-01 --target CSC-2025-01 --user alice@example.com """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction from common.models import UserContractAccess, UserContractStatusPermission from contracts.models import Contract class Command(BaseCommand): help = "Copie les droits d'accès d'un contrat vers un autre pour tous les utilisateurs concernés." def add_arguments(self, parser): parser.add_argument( '--source', metavar='NUMERO', required=True, help='Numéro du contrat source (contract_number).', ) parser.add_argument( '--target', metavar='NUMERO', required=True, help='Numéro du contrat cible (contract_number).', ) parser.add_argument( '--update', action='store_true', default=False, help='Met à jour les accès existants au contrat cible au lieu de les ignorer.', ) parser.add_argument( '--dry-run', action='store_true', default=False, dest='dry_run', help='Affiche ce qui serait fait sans rien écrire.', ) parser.add_argument( '--user', metavar='USERNAME_OR_EMAIL', action='append', default=[], dest='users', help='Limite la copie à ces utilisateurs (répétable). Accepte username ou email.', ) # ------------------------------------------------------------------ def handle(self, *args, **options): dry_run = options['dry_run'] do_update = options['update'] identifiers = options['users'] if dry_run: self.stdout.write(self.style.WARNING('Mode DRY-RUN : aucune modification.\n')) # Résoudre les deux contrats try: source_contract = Contract.objects.get(contract_number=options['source']) except Contract.DoesNotExist: raise CommandError(f"Contrat source introuvable : {options['source']}") try: target_contract = Contract.objects.get(contract_number=options['target']) except Contract.DoesNotExist: raise CommandError(f"Contrat cible introuvable : {options['target']}") if source_contract.pk == target_contract.pk: raise CommandError("Source et cible sont le même contrat.") # Récupérer les accès source qs = UserContractAccess.objects.filter( contract=source_contract, ).select_related('user_config__user').prefetch_related('status_permissions') if identifiers: qs = qs.filter( user_config__user__username__in=identifiers ) | qs.filter( user_config__user__email__in=identifiers ) qs = qs.distinct() source_accesses = list(qs) if not source_accesses: self.stdout.write(self.style.WARNING( f'Aucun accès trouvé pour le contrat source « {options["source"]} ».' )) return self.stdout.write( f'{len(source_accesses)} utilisateur(s) à traiter : ' f'{options["source"]} → {options["target"]}\n' ) created_count = updated_count = skipped_count = 0 with transaction.atomic(): for src_uca in source_accesses: username = src_uca.user_config.user.username result = self._copy_access( src_uca, target_contract, do_update, dry_run ) if result == 'created': self.stdout.write(f' [CRÉE] {username}') created_count += 1 elif result == 'updated': self.stdout.write(f' [MAJ] {username}') updated_count += 1 else: self.stdout.write(f' [IGNORÉ] {username} (accès cible déjà présent)') skipped_count += 1 prefix = '[DRY-RUN] ' if dry_run else '' self.stdout.write('') self.stdout.write(self.style.SUCCESS( f'{prefix}Résultat — créés : {created_count} | ' f'mis à jour : {updated_count} | ignorés : {skipped_count}' )) # ------------------------------------------------------------------ def _copy_access(self, src_uca, target_contract, do_update, dry_run): """ Crée ou met à jour le UserContractAccess + UserContractStatusPermission pour le contrat cible. Retourne 'created', 'updated' ou 'skipped'. """ cfg = src_uca.user_config try: existing = UserContractAccess.objects.get( user_config=cfg, contract=target_contract, ) except UserContractAccess.DoesNotExist: existing = None if existing and not do_update: return 'skipped' if dry_run: return 'updated' if existing else 'created' # Champs à copier depuis la source access_fields = { 'can_view_posts': src_uca.can_view_posts, 'can_view_prices': src_uca.can_view_prices, 'can_view_interventions': src_uca.can_view_interventions, 'can_view_assets': src_uca.can_view_assets, 'can_create_interventions': src_uca.can_create_interventions, 'can_check_interventions': src_uca.can_check_interventions, 'can_approve': src_uca.can_approve, } if existing: for field, value in access_fields.items(): setattr(existing, field, value) existing.save() target_uca = existing # Supprimer les anciennes permissions par statut avant de les recréer target_uca.status_permissions.all().delete() label = 'updated' else: target_uca = UserContractAccess.objects.create( user_config=cfg, contract=target_contract, **access_fields, ) label = 'created' # Copier les permissions par statut UserContractStatusPermission.objects.bulk_create([ UserContractStatusPermission( user_contract=target_uca, 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 src_uca.status_permissions.all() ]) return label