""" Commande Django pour importer les stations depuis le fichier Excel. Usage: python manage.py import_stations /path/to/excel/file.xlsx python manage.py import_stations /path/to/excel/file.xlsx --update """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction from bilal.models import Station import pandas as pd import os class Command(BaseCommand): help = 'Importe les stations Bilal Oil depuis un fichier Excel' def add_arguments(self, parser): parser.add_argument( 'excel_file', type=str, help='Chemin vers le fichier Excel contenant les données des stations' ) parser.add_argument( '--update', action='store_true', help='Met à jour les stations existantes au lieu de les ignorer', ) parser.add_argument( '--dry-run', action='store_true', help='Simule l\'import sans créer les enregistrements', ) def handle(self, *args, **options): excel_file = options['excel_file'] update_existing = options['update'] dry_run = options['dry_run'] if not os.path.exists(excel_file): raise CommandError(f'Le fichier "{excel_file}" n\'existe pas.') self.stdout.write( self.style.SUCCESS('🚀 IMPORT DES STATIONS BILAL OIL') ) self.stdout.write('=' * 60) if dry_run: self.stdout.write( self.style.WARNING('⚠️ MODE SIMULATION - Aucune donnée ne sera créée') ) try: # Lire le fichier Excel self.stdout.write(f'📖 Lecture du fichier Excel : {excel_file}') df = pd.read_excel(excel_file, skiprows=2) # Renommer les colonnes df.columns = ['NUMERO', 'REGION', 'DEPARTEMENT', 'COMMUNE', 'SITE', 'ETAT'] self.stdout.write(f'✅ {len(df)} stations trouvées dans le fichier') # Mapping des coordonnées GPS coordinates_mapping = self._get_coordinates_mapping() stations_created = 0 stations_updated = 0 stations_skipped = 0 errors = [] with transaction.atomic(): for idx, row in df.iterrows(): try: result = self._process_station_row( row, coordinates_mapping, update_existing, dry_run ) if result == 'created': stations_created += 1 elif result == 'updated': stations_updated += 1 elif result == 'skipped': stations_skipped += 1 except Exception as e: error_msg = f'Erreur ligne {idx + 3}: {str(e)}' errors.append(error_msg) self.stdout.write( self.style.ERROR(f' ❌ {error_msg}') ) if dry_run: # Annuler la transaction en mode simulation transaction.set_rollback(True) # Afficher le résumé self._display_summary( len(df), stations_created, stations_updated, stations_skipped, errors, dry_run ) except Exception as e: raise CommandError(f'Erreur lors de l\'import: {str(e)}') def _process_station_row(self, row, coordinates_mapping, update_existing, dry_run): """Traite une ligne de données de station.""" # Construire le nom de la station station_name = f"Bilal Oil {row['SITE']}" # Construire l'adresse adresse = f"{row['SITE']}, {row['COMMUNE']}, {row['DEPARTEMENT']}, {row['REGION']}" # Récupérer les coordonnées si disponibles site_key = row['SITE'].upper() latitude, longitude = coordinates_mapping.get(site_key, (None, None)) # Déterminer l'état de la station est_active = row['ETAT'] == 'EN SERVICE' station_data = { 'nom': station_name, 'adresse': adresse, 'ville': row['COMMUNE'], 'region': row['REGION'], 'latitude': latitude, 'longitude': longitude, 'telephone': '', 'email': '', 'horaires': '24h/24 - 7j/7', 'est_active': est_active, 'date_ouverture': None, } self.stdout.write(f"📍 Station {int(row['NUMERO']):2d}: {station_name}") self.stdout.write(f" 📍 {adresse}") if latitude and longitude: self.stdout.write(f" 🌍 GPS: {latitude}, {longitude}") else: self.stdout.write(f" ⚠️ GPS: Coordonnées non disponibles") if dry_run: self.stdout.write(f" 🔄 [SIMULATION] Station analysée") return 'simulated' # Vérifier si la station existe déjà existing_station = Station.objects.filter(nom=station_name).first() if existing_station: if update_existing: # Mettre à jour la station existante for key, value in station_data.items(): if value is not None: setattr(existing_station, key, value) existing_station.save() self.stdout.write( self.style.WARNING(f" 🔄 Station mise à jour") ) return 'updated' else: self.stdout.write( self.style.WARNING(f" ⏭️ Station existante ignorée (utilisez --update pour mettre à jour)") ) return 'skipped' else: # Créer une nouvelle station Station.objects.create(**station_data) self.stdout.write( self.style.SUCCESS(f" ✅ Station créée") ) return 'created' def _get_coordinates_mapping(self): """Retourne le mapping des coordonnées GPS.""" return { 'BAMBILOR': (14.7167, -17.0833), 'POUT': (14.7500, -17.0500), 'BOUKHOU': (14.6833, -16.9833), 'SANDIARA': (14.6167, -17.0167), 'THIENEBA': (14.7833, -16.9167), 'KHAIRA': (14.8500, -15.8833), 'NDAMATOU': (14.8667, -15.8667), 'SADIO': (14.8167, -15.8500), 'AFFE': (15.3500, -15.0833), 'KAHI': (14.1167, -15.5833), 'KOUMPENTOUM': (13.9667, -14.5833), 'MERETO': (13.9500, -14.5500), 'KOUTIABA-WOLOF': (13.9833, -14.5667), 'BAMBA THIALEME': (13.9333, -14.5333), 'TAMBACOUNDA': (13.7667, -13.6667), 'KIDIRA': (14.4500, -12.2167), 'DIANKE-MAKHA': (13.6500, -12.8833), 'BANTACO': (12.5500, -12.1833), 'KHARAKHENA': (12.7167, -12.0333), 'MOUSSALA': (12.7000, -12.0500), 'GUEMEDIE': (12.6833, -12.0667), 'MISSIRAH-SIRIMINA': (12.7333, -12.1167), 'SAINSOUTOU': (12.7500, -12.1000), 'SABADOLA': (12.8167, -12.1333), 'KEDOUGOU': (12.5500, -12.1833), 'SARE-KEMO': (12.8833, -14.9500), 'SARE-YOBA': (12.8667, -14.9333), 'SARE-BIDJI': (12.9000, -14.9167), 'CARREFOUR NDIAYE': (12.7500, -15.5167), 'DIANNAH MALARY': (12.6500, -15.3000), 'GOUDOMP': (12.6167, -15.1333), 'SAMINE': (12.5833, -15.0833), 'DIANNAH-KAFOUTINE': (12.4167, -16.6167), 'MPACK': (12.3833, -16.5500), } def _display_summary(self, total, created, updated, skipped, errors, dry_run): """Affiche le résumé de l'import.""" self.stdout.write() self.stdout.write('=' * 60) if dry_run: self.stdout.write( self.style.SUCCESS('📊 RÉSUMÉ DE LA SIMULATION') ) else: self.stdout.write( self.style.SUCCESS('📊 RÉSUMÉ DE L\'IMPORT') ) self.stdout.write(f' Total dans le fichier: {total} stations') if not dry_run: self.stdout.write(f' Stations créées: {created}') self.stdout.write(f' Stations mises à jour: {updated}') self.stdout.write(f' Stations ignorées: {skipped}') if errors: self.stdout.write(f' Erreurs: {len(errors)}') for error in errors: self.stdout.write(f' - {error}') self.stdout.write('=' * 60) if dry_run: self.stdout.write( self.style.WARNING('⚠️ Mode simulation terminé - Aucune donnée créée') ) self.stdout.write( self.style.SUCCESS('💡 Utilisez la commande sans --dry-run pour effectuer l\'import réel') ) elif created > 0 or updated > 0: self.stdout.write( self.style.SUCCESS(f'🎉 Import terminé avec succès!') ) else: self.stdout.write( self.style.WARNING('⚠️ Aucune station créée ou mise à jour') )