import mysql.connector from mysql.connector import Error import os from datetime import datetime # Charger les variables d'environnement depuis .env load_dotenv() # --- Configuration de la base de données --- DB_CONFIG = { "host": "localhost", "port": "3306", "user": "naje4034_autoconso_admin", "password": "Francisco$*1401", "database": "naje4034_autoconso_api" } DB_HOST = "localhost" DB_NAME = "autoconso-api" #"naje4034_autoconso_api" DB_USER = "root" DB_PASS = "root" DB_PORT = "8889" # --- Chemin vers votre fichier JSON des bornes de recharge --- JSON_FILE_PATH = 'bornes_recharge.json' # REMPLACEZ PAR LE VRAI NOM DE VOTRE FICHIER # --- Dictionnaire de mappage des types de prises à leurs IDs Carburant --- # Vous devrez récupérer ces IDs depuis votre table 'carburant' une fois que vous avez inséré les types VE. # Exécutez une requête SQL du type SELECT id, alias FROM carburant WHERE energy = 'ELECTRIC'; # Et remplissez ce dictionnaire. CONNECTOR_TYPE_TO_CARBURANT_ID = { "Type 2": 8, # ID de l'entrée 'Type 2' dans la table 'carburant' "Combo CCS": 9, # ID de l'entrée 'Combo CCS' dans la table 'carburant' "CHAdeMO": 10, # ID de l'entrée 'CHAdeMO' dans la table 'carburant' "prise_type_ef": 11 # ID de l'entrée 'Type E/F' si vous l'ajoutez } def get_db_connection(): """Établit une connexion à la base de données PostgreSQL.""" conn = psycopg2.connect( host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS, port=DB_PORT ) return conn def extract_address_components(full_address): """ Extrait l'adresse, le code postal et la ville d'une chaîne de caractères. Format attendu: "Adresse civique, CodePostal Ville" Ex: "93 route de Bitche, 67506 Haguenau Cedex" """ address = None postalcode = None city = None if full_address: # Tenter de trouver un code postal (5 chiffres) match = re.search(r'(\d{5})\s+(.*)', full_address) if match: # Si un code postal est trouvé: postalcode_str = match.group(1) city_part = match.group(2).strip() # Le reste après le code postal est la ville # La partie adresse est tout ce qui précède le match du code postal address_part = full_address[:match.start()].strip() # Nettoyer l'adresse si elle finit par une virgule ou un espace if address_part.endswith(','): address_part = address_part[:-1].strip() address = address_part try: postalcode = int(postalcode_str) except ValueError: postalcode = None # Au cas où ce ne serait pas un nombre valide city = city_part else: # Si aucun code postal n'est trouvé, on prend l'adresse telle quelle # Cela peut arriver pour des adresses très simples ou incomplètes address = full_address.strip() # postalcode et city restent None return address, postalcode, city def import_electric_stations(): """Importe les données des bornes de recharge depuis le JSON vers la base de données.""" try: with open(JSON_FILE_PATH, 'r', encoding='utf-8') as f: data = json.load(f) except FileNotFoundError: print(f"Erreur : Le fichier JSON '{JSON_FILE_PATH}' n'a pas été trouvé.") return except json.JSONDecodeError: print(f"Erreur : Impossible de décoder le JSON dans '{JSON_FILE_PATH}'. Vérifiez sa validité.") return conn = None try: conn = get_db_connection() cur = conn.cursor() print(f"Début de l'importation de {len(data['features'])} bornes de recharge...") for feature in data['features']: props = feature['properties'] geometry = feature['geometry'] # --- Extraction des composants de l'adresse --- full_address_string = props.get('adresse_station') extracted_address, extracted_postalcode, extracted_city = extract_address_components(full_address_string) # Utiliser les valeurs extraites, avec fallback si l'extraction échoue station_address = extracted_address if extracted_address else props.get('adresse_station') # Fallback à l'adresse complète station_postalcode = extracted_postalcode if extracted_postalcode is not None else props.get('consolidated_code_postal') station_city = extracted_city if extracted_city else props.get('consolidated_commune') # --- Mappage des données JSON vers l'entité Station --- station_data = { 'id_station': props.get('id_station_local') or props.get('id_station_itinerance'), 'id_app': props.get('id_station_itinerance'), # Ou un autre identifiant si vous en avez un pour CarLine 'name': props.get('nom_station') or props.get('nom_enseigne'), 'address': props.get('adresse_station'), 'postalcode': props.get('consolidated_code_postal'), 'city': props.get('consolidated_commune'), 'latitude': geometry['coordinates'][1], # Latitude est le 2ème élément 'longitude': geometry['coordinates'][0], # Longitude est le 1er élément 'comments': props.get('observations'), 'is_electric': True, # C'est une borne électrique 'created_at': datetime.now(), # Ou props.get('created_at') si existant et valide 'modified_at': datetime.now(), 'brand_name': props.get('nom_enseigne') or props.get('nom_operateur'), 'brand_logo': None, # À remplir si vous avez un mappage de logos 'is_active': True, # Supposons que les bornes importées sont actives 'is_valid': True, # Supposons qu'elles sont valides # Détails électriques pour le champ JSONB 'electric_details': { 'nom_amenageur': props.get('nom_amenageur'), 'contact_amenageur': props.get('contact_amenageur'), 'nom_operateur': props.get('nom_operateur'), 'contact_operateur': props.get('contact_operateur'), 'telephone_operateur': props.get('telephone_operateur'), 'implantation_station': props.get('implantation_station'), 'horaires': props.get('horaires'), 'condition_acces': props.get('condition_acces'), 'reservation': props.get('reservation') == 'true', 'accessibilite_pmr': props.get('accessibilite_pmr'), 'restriction_gabarit': props.get('restriction_gabarit'), 'station_deux_roues': props.get('station_deux_roues') == 'true', 'raccordement': props.get('raccordement'), 'num_pdl': props.get('num_pdl'), 'date_mise_en_service': props.get('date_mise_en_service'), 'observations': props.get('observations'), 'date_maj': props.get('date_maj'), 'cable_t2_attache': props.get('cable_t2_attache') == 'true', 'paiement_acte': props.get('paiement_acte') == 'true', 'paiement_cb': props.get('paiement_cb') == 'true', 'paiement_autre': props.get('paiement_autre'), 'tarification': props.get('tarification') # Peut être vide, à gérer } } # --- Insertion ou Mise à jour de la Station --- cur.execute( sql.SQL(""" INSERT INTO station ( id_station, id_app, name, address, postalcode, city, latitude, longitude, comments, created_at, modified_at, brand_name, brand_logo, is_active, is_valid, is_electric, electric_details ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) ON CONFLICT (id_station) DO UPDATE SET id_app = EXCLUDED.id_app, name = EXCLUDED.name, address = EXCLUDED.address, postalcode = EXCLUDED.postalcode, city = EXCLUDED.city, latitude = EXCLUDED.latitude, longitude = EXCLUDED.longitude, comments = EXCLUDED.comments, modified_at = EXCLUDED.modified_at, brand_name = EXCLUDED.brand_name, brand_logo = EXCLUDED.brand_logo, is_active = EXCLUDED.is_active, is_valid = EXCLUDED.is_valid, is_electric = EXCLUDED.is_electric, electric_details = EXCLUDED.electric_details; """), ( station_data['id_station'], station_data['id_app'], station_data['name'], station_data['address'], station_data['postalcode'], station_data['city'], station_data['latitude'], station_data['longitude'], station_data['comments'], station_data['created_at'], station_data['modified_at'], station_data['brand_name'], station_data['brand_logo'], station_data['is_active'], station_data['is_valid'], station_data['is_electric'], json.dumps(station_data['electric_details']) ) ) # Récupérer l'ID de la station insérée ou mise à jour cur.execute(sql.SQL("SELECT id FROM station WHERE id_station = %s;"), (station_data['id_station'],)) station_db_id = cur.fetchone()[0] # --- Mappage et insertion des Carburantlist pour les points de charge --- # Suppression des anciens points de charge pour cette station pour éviter les doublons lors de la MAJ cur.execute(sql.SQL("DELETE FROM carburantlist WHERE station_id = %s;"), (station_db_id,)) # On vérifie chaque type de prise et on insère dans carburant_list # N.B.: Le JSON d'origine ne donne pas le nombre de connecteurs par type de prise, # ni le détail par borne individuelle. Il donne "nbre_pdc" global pour la station # et des booléens pour les types de prises. Il faudrait ajuster si vos données # sont plus granulaires. Ici, on va prendre nbre_pdc comme le nombre total de connecteurs # et le répartir ou l'affecter si un seul type de prise. Sinon, on met 1 pour chaque type. nbre_pdc_total = int(props.get('nbre_pdc', 0)) # Types de prises du JSON et leur correspondance dans notre Carburant DB connector_types_in_json = { 'prise_type_2': props.get('prise_type_2') == 'true', 'prise_type_combo_ccs': props.get('prise_type_combo_ccs') == 'true', 'prise_type_chademo': props.get('prise_type_chademo') == 'true', 'prise_type_ef': props.get('prise_type_ef') == 'true', # Si vous avez ajouté cette prise comme Type E/F # 'prise_type_autre': props.get('prise_type_autre') == 'true', # Gérer si nécessaire } # Estimation du nombre de connecteurs par type si non spécifié dans le JSON original # Ici, on simplifie en mettant 1 pour chaque type trouvé, ou on divise le total si un seul type existe active_connectors_count = sum(1 for _, is_active in connector_types_in_json.items() if is_active) num_connectors_per_type = nbre_pdc_total if active_connectors_count == 1 else 1 # Simplification for json_key, is_present in connector_types_in_json.items(): if is_present: # Mappage de la clé JSON vers un alias reconnaissable pour notre dictionnaire # du Carburant_ID if json_key == 'prise_type_2': db_alias = 'Type 2' elif json_key == 'prise_type_combo_ccs': db_alias = 'Combo CCS' elif json_key == 'prise_type_chademo': db_alias = 'CHAdeMO' elif json_key == 'prise_type_ef': db_alias = 'prise_type_ef' # Assurez-vous d'avoir cet alias dans votre DB Carburant else: continue # Skip if not a recognized connector type carburant_id = CONNECTOR_TYPE_TO_CARBURANT_ID.get(db_alias) if not carburant_id: print(f"AVERTISSEMENT: ID Carburant non trouvé pour l'alias '{db_alias}'. Ignoré.") continue # Déterminer la puissance. Le JSON donne 'puissance_nominale' qui semble être pour la borne en général. # On va l'appliquer à chaque connecteur pour l'instant. power_kw = float(props.get('puissance_nominale', 0)) # Déterminer le modèle de tarification. Le JSON 'tarification' est vide, donc on déduit. pricing_model = "unknown" if props.get('gratuit') == 'true': pricing_model = "free" elif props.get('paiement_acte') == 'true' or props.get('paiement_cb') == 'true': pricing_model = "per_kwh" # Supposons au kWh si payant, à ajuster si votre source donne plus de détails # L'état de disponibilité n'est pas dans l'extrait, on met "unknown" availability = "unknown" # Ou extraire si disponible ailleurs dans le JSON cur.execute( sql.SQL(""" INSERT INTO carburantlist ( station_id, carburant_id, price, modified_at, power_kw, num_connectors, availability, pricing_model ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s ); """), ( station_db_id, carburant_id, None, # Prix non disponible directement dans l'extrait, à gérer datetime.fromisoformat(props.get('date_maj').replace('Z', '+00:00')) if props.get('date_maj') else None, power_kw, num_connectors_per_type, availability, pricing_model ) ) print(f"Importé/Mise à jour station: {station_data['name']} ({station_data['id_station']})") conn.commit() print("Importation terminée avec succès !") except (Exception, psycopg2.Error) as error: print(f"Erreur lors de l'importation: {error}") if conn: conn.rollback() # Annuler toutes les opérations en cas d'erreur finally: if conn: cur.close() conn.close() if __name__ == "__main__": # IMPORTANT: Avant d'exécuter ce script, assurez-vous d'avoir : # 1. Mis à jour votre schéma de base de données (station, carburant, carburant_list). # 2. Inséré les types de connecteurs VE (Type 2, Combo CCS, etc.) dans votre table 'carburant'. # 3. Rempli le dictionnaire 'CONNECTOR_TYPE_TO_CARBURANT_ID' avec les IDs corrects. # 4. Placé votre fichier JSON au bon endroit et mis à jour 'JSON_FILE_PATH'. # 5. Rempli votre fichier .env avec les bonnes informations de connexion à la DB. import_electric_stations()