import pandas as pd from datetime import datetime import gspread class DataSyncForecasting: """ Synchronise automatiquement les données de Forecasting à partir des déblocages dans Prets_Master """ def __init__(self, client, sheet_name): self.client = client self.sheet_name = sheet_name self.sh = client.open(sheet_name) def extraire_deblocages_mensuels(self): """ Extrait et agrège les déblocages par mois depuis Prets_Master Returns: DataFrame avec Date (YYYY-MM-01) et Montant_Total_Sortie """ try: ws_prets = self.sh.worksheet("Prets_Master") df_prets = pd.DataFrame(ws_prets.get_all_records()) if df_prets.empty: return pd.DataFrame(columns=['Date', 'Montant_Total_Sortie']) # Vérifier les colonnes nécessaires if 'Date_Deblocage' not in df_prets.columns or 'Montant_Capital' not in df_prets.columns: raise ValueError("Colonnes 'Date_Deblocage' ou 'Montant_Capital' manquantes dans Prets_Master") # Garder uniquement les lignes avec données valides df_prets = df_prets[ (df_prets['Date_Deblocage'].notna()) & (df_prets['Date_Deblocage'] != '') & (df_prets['Montant_Capital'].notna()) & (df_prets['Montant_Capital'] != 0) ].copy() if df_prets.empty: return pd.DataFrame(columns=['Date', 'Montant_Total_Sortie']) # Conversion des montants en numérique df_prets['Montant_Capital'] = pd.to_numeric(df_prets['Montant_Capital'], errors='coerce') # Parser les dates (format DD/MM/YYYY) df_prets['Date_Parsed'] = pd.to_datetime( df_prets['Date_Deblocage'], format='%d/%m/%Y', errors='coerce' ) # Supprimer les dates invalides df_prets = df_prets.dropna(subset=['Date_Parsed', 'Montant_Capital']) if df_prets.empty: return pd.DataFrame(columns=['Date', 'Montant_Total_Sortie']) # Créer une colonne année-mois pour le groupement df_prets['Annee_Mois'] = df_prets['Date_Parsed'].dt.to_period('M') # Grouper par mois et sommer df_mensuel = df_prets.groupby('Annee_Mois')['Montant_Capital'].sum().reset_index() # Convertir en premier jour du mois df_mensuel['Date'] = df_mensuel['Annee_Mois'].apply(lambda x: x.to_timestamp()) df_mensuel = df_mensuel.rename(columns={'Montant_Capital': 'Montant_Total_Sortie'}) # Garder uniquement Date et Montant df_mensuel = df_mensuel[['Date', 'Montant_Total_Sortie']].sort_values('Date') return df_mensuel except gspread.WorksheetNotFound: raise Exception("Feuille 'Prets_Master' introuvable") except Exception as e: raise Exception(f"Erreur lors de l'extraction : {str(e)}") def lire_forecasting_actuel(self): """ Lit les données actuelles de Forecasting Returns: DataFrame avec les données existantes """ try: ws_forecasting = self.sh.worksheet("Forecasting") df_forecasting = pd.DataFrame(ws_forecasting.get_all_records()) if df_forecasting.empty or 'Date' not in df_forecasting.columns: return pd.DataFrame(columns=['Date', 'Montant_Total_Sortie']) # Parser les dates df_forecasting['Date'] = pd.to_datetime(df_forecasting['Date'], errors='coerce') df_forecasting = df_forecasting.dropna(subset=['Date']) # Conversion montants df_forecasting['Montant_Total_Sortie'] = pd.to_numeric( df_forecasting['Montant_Total_Sortie'], errors='coerce' ).fillna(0) return df_forecasting[['Date', 'Montant_Total_Sortie']] except gspread.WorksheetNotFound: # Si la feuille n'existe pas, la créer return pd.DataFrame(columns=['Date', 'Montant_Total_Sortie']) except Exception as e: raise Exception(f"Erreur lors de la lecture de Forecasting : {str(e)}") def fusionner_donnees(self, df_existant, df_nouveau): """ Fusionne intelligemment les données existantes et nouvelles Args: df_existant: DataFrame avec données actuelles de Forecasting df_nouveau: DataFrame avec données calculées depuis Prets_Master Returns: dict avec DataFrame fusionné + statistiques """ # Convertir en période pour faciliter la comparaison df_existant['Periode'] = df_existant['Date'].dt.to_period('M') df_nouveau['Periode'] = df_nouveau['Date'].dt.to_period('M') # Identifier les mois mois_existants = set(df_existant['Periode']) mois_nouveaux = set(df_nouveau['Periode']) # Statistiques mois_a_ajouter = mois_nouveaux - mois_existants mois_a_mettre_a_jour = mois_existants & mois_nouveaux mois_conserves = mois_existants - mois_nouveaux # Créer le DataFrame fusionné # 1. Garder les mois qui ne sont pas dans Prets_Master (données manuelles historiques) df_conserves = df_existant[df_existant['Periode'].isin(mois_conserves)].copy() # 2. Mettre à jour les mois qui existent dans les deux df_mis_a_jour = df_nouveau[df_nouveau['Periode'].isin(mois_a_mettre_a_jour)].copy() # 3. Ajouter les nouveaux mois df_ajoutes = df_nouveau[df_nouveau['Periode'].isin(mois_a_ajouter)].copy() # Fusionner tout df_final = pd.concat([df_conserves, df_mis_a_jour, df_ajoutes], ignore_index=True) # Trier par date df_final = df_final.sort_values('Date').reset_index(drop=True) # Supprimer la colonne Periode df_final = df_final[['Date', 'Montant_Total_Sortie']] # Statistiques du changement stats = { 'nb_conserves': len(mois_conserves), 'nb_mis_a_jour': len(mois_a_mettre_a_jour), 'nb_ajoutes': len(mois_a_ajouter), 'total_lignes': len(df_final), 'mois_conserves': sorted([str(p) for p in mois_conserves]), 'mois_mis_a_jour': sorted([str(p) for p in mois_a_mettre_a_jour]), 'mois_ajoutes': sorted([str(p) for p in mois_a_ajouter]) # ✅ Corrigé ici } return { 'dataframe': df_final, 'stats': stats } def ecrire_forecasting(self, df_final): """ Écrit le DataFrame final dans la feuille Forecasting Args: df_final: DataFrame à écrire """ try: ws_forecasting = self.sh.worksheet("Forecasting") except gspread.WorksheetNotFound: # Créer la feuille si elle n'existe pas ws_forecasting = self.sh.add_worksheet(title="Forecasting", rows="1000", cols="10") # Vider la feuille ws_forecasting.clear() # Préparer les données pour l'écriture df_write = df_final.copy() df_write['Date'] = df_write['Date'].dt.strftime('%Y-%m-%d') df_write['Montant_Total_Sortie'] = df_write['Montant_Total_Sortie'].astype(int) # Écrire l'en-tête ws_forecasting.update('A1:B1', [['Date', 'Montant_Total_Sortie']]) # Écrire les données if not df_write.empty: data_to_write = df_write.values.tolist() ws_forecasting.update(f'A2:B{len(data_to_write) + 1}', data_to_write) def synchroniser(self): """ Fonction principale de synchronisation Returns: dict avec résultat de la synchronisation """ try: # 1. Extraire les déblocages mensuels depuis Prets_Master df_nouveau = self.extraire_deblocages_mensuels() # 2. Lire Forecasting actuel df_existant = self.lire_forecasting_actuel() # 3. Fusionner intelligemment result = self.fusionner_donnees(df_existant, df_nouveau) df_final = result['dataframe'] stats = result['stats'] # 4. Écrire dans Forecasting self.ecrire_forecasting(df_final) return { 'success': True, 'stats': stats, 'dataframe': df_final, 'message': f"Synchronisation réussie : {stats['nb_conserves']} mois conservés, {stats['nb_mis_a_jour']} mis à jour, {stats['nb_ajoutes']} ajoutés" } except Exception as e: return { 'success': False, 'error': str(e), 'message': f"Erreur lors de la synchronisation : {str(e)}" } def actualiser_forecasting_depuis_prets(client, sheet_name): """ Fonction helper pour synchroniser facilement Args: client: Client gspread sheet_name: Nom du fichier Google Sheets Returns: dict avec résultat de la synchronisation """ sync = DataSyncForecasting(client, sheet_name) return sync.synchroniser()