import pymysql import logging from datetime import datetime # Configuration des logs logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('insert_weather.log'), logging.StreamHandler() ] ) # Chemin vers le fichier CSV file_path = "donnees-synop-essentielles-omm.csv" def load_and_clean_data(file_path): """Charge et nettoie les données du fichier CSV.""" try: logging.info("Chargement du fichier CSV...") data = [] with open(file_path, 'r', encoding='utf-8') as file: headers = file.readline().strip().split(';') for line in file: values = line.strip().split(';') if len(values) == len(headers): row = dict(zip(headers, values)) data.append(row) logging.info(f"Fichier CSV chargé avec succès. Nombre de lignes : {len(data)}") return data except Exception as e: logging.error(f"Erreur lors du chargement du fichier CSV : {str(e)}") exit(1) def transform_data(data): """Nettoie et transforme les données.""" transformed_data = [] for row in data: try: # Extraire latitude et longitude coordonnees = row.get('Coordonnees', '').split(', ') latitude = float(coordonnees[0]) if len(coordonnees) > 0 and coordonnees[0] else None longitude = float(coordonnees[1]) if len(coordonnees) > 1 and coordonnees[1] else None transformed_row = { 'latitude': latitude, 'longitude': longitude, 'commune_name': row.get('communes (name)', ''), 'commune_code': row.get('communes (code)', ''), 'Date': row.get('Date', ''), 'Température (°C)': float(row.get('Température (°C)', 0)) if row.get('Température (°C)') else None, 'Humidité': float(row.get('Humidité', 0)) if row.get('Humidité') else None, 'Précipitations dans la dernière heure': float(row.get('Précipitations dans la dernière heure', 0)) if row.get('Précipitations dans la dernière heure') else None, 'Vitesse du vent moyen 10 mn': float(row.get('Vitesse du vent moyen 10 mn', 0)) if row.get('Vitesse du vent moyen 10 mn') else None } transformed_data.append(transformed_row) except Exception as e: logging.warning(f"Ligne ignorée lors du nettoyage : {str(e)}") continue logging.info(f"Données nettoyées et transformées avec succès. Nombre de lignes : {len(transformed_data)}") return transformed_data def insert_data_into_db(transformed_data): """Insère les données dans la base de données MySQL.""" try: logging.info("Connexion à la base de données MySQL...") db_config = { 'host': 'localhost', 'user': 'root', 'password': '_z7b6CJc', 'database': 'immo_new', 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor } connection = pymysql.connect(**db_config) # Désactiver l'autocommit pour contrôler les transactions manuellement connection.autocommit(False) cursor = connection.cursor() # Récupérer les insee_code valides depuis la table cities cursor.execute("SELECT insee_code FROM cities;") valid_insee_codes = {row['insee_code'] for row in cursor.fetchall()} logging.info(f"Codes INSEE valides dans 'cities' : {len(valid_insee_codes)} codes") logging.info("Insertion des données dans la table weather_data...") inserted_lines = 0 skipped_lines = 0 for index, row in enumerate(transformed_data, start=1): try: if not row["commune_code"] or str(row["commune_code"]).strip() == "": skipped_lines += 1 logging.warning(f"Ligne {index} ignorée : insee_code vide ou NULL") continue if row["commune_code"] not in valid_insee_codes: skipped_lines += 1 logging.warning(f"Ligne {index} ignorée : insee_code {row['commune_code']} introuvable dans 'cities'") continue # Convertir la date en objet datetime try: forecast_timestamp = datetime.strptime(row["Date"], "%Y-%m-%dT%H:%M:%S%z") except ValueError as e: skipped_lines += 1 logging.warning(f"Ligne {index} ignorée : format de date invalide ({row['Date']}) - {str(e)}") continue # Vérifier latitude et longitude try: latitude = float(row["latitude"]) longitude = float(row["longitude"]) except (ValueError, TypeError) as e: skipped_lines += 1 logging.warning(f"Ligne {index} ignorée : latitude ou longitude invalide - {str(e)}") continue # Vérifier les valeurs numériques try: avg_temperature = float(row["Température (°C)"]) if row["Température (°C)"] is not None else None humidity = float(row["Humidité"]) if row["Humidité"] is not None else None precipitation = float(row["Précipitations dans la dernière heure"]) if row["Précipitations dans la dernière heure"] is not None else None wind_speed = float(row["Vitesse du vent moyen 10 mn"]) if row["Vitesse du vent moyen 10 mn"] is not None else None except (ValueError, TypeError) as e: skipped_lines += 1 logging.warning(f"Ligne {index} ignorée : valeur numérique invalide - {str(e)}") continue # Préparer les valeurs pour l'insertion values = [ forecast_timestamp, latitude, longitude, None, # forecast_base avg_temperature, None, # min_temperature None, # max_temperature humidity, precipitation, wind_speed, None, # solar_radiation row["commune_name"], row["commune_code"] ] # Construire la requête SQL query = """ INSERT INTO weather_data ( forecast_timestamp, latitude, longitude, forecast_base, avg_temperature, min_temperature, max_temperature, humidity, precipitation, wind_speed, solar_radiation, commune, insee_code ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(query, values) connection.commit() # Valider chaque insertion individuellement inserted_lines += 1 logging.info(f"Ligne {index} insérée avec succès.") except Exception as e: connection.rollback() # Annuler uniquement cette insertion skipped_lines += 1 logging.error(f"Erreur lors de l'insertion de la ligne {index} : {str(e)}") continue logging.info(f"Insertion terminée. Lignes insérées : {inserted_lines}, Lignes ignorées : {skipped_lines}") except Exception as e: logging.error(f"Erreur globale lors de l'insertion des données : {str(e)}") finally: cursor.close() connection.close() logging.info("Connexion à la base de données fermée.") def main(): # Charger et nettoyer les données data = load_and_clean_data(file_path) transformed_data = transform_data(data) # Insérer les données dans la base de données insert_data_into_db(transformed_data) if __name__ == "__main__": main()