Spaces:
Running
Running
| """ | |
| MODULE COMMUNICATIONS — Envoi d'email en masse | |
| ================================================= | |
| Envoie un message (objet + corps) à une sélection de clients et/ou garants, | |
| choisie librement par l'utilisateur (individus, "tous les clients", "tous | |
| les garants" ou un mélange), et mesure qui l'a ouvert (core/suivi_emails.py : | |
| email HTML suivi par Brevo, une ligne par destinataire dans Suivi_Emails). | |
| L'admin (secret ADMIN_EMAIL) reçoit un récapitulatif unique plutôt qu'une | |
| copie cachée de chaque email : son ouverture fausserait les statistiques. | |
| """ | |
| import pandas as pd | |
| import streamlit as st | |
| from core import suivi_emails | |
| from core.email_client import get_admin_email | |
| from core.ui_guards import rendre_bouton_protege, liberer | |
| from core.sheets_cache import get_cached_dfs_batch | |
| def _load_recipients(_client, sheet_name): | |
| sh = _client.open(sheet_name) | |
| _dfs = get_cached_dfs_batch(sh, sheet_name, ["Clients_KYC", "Garants_KYC"]) | |
| return _dfs["Clients_KYC"], _dfs["Garants_KYC"] | |
| def _rows_to_options(df, id_col): | |
| out = {} | |
| if df.empty or id_col not in df.columns: | |
| return out | |
| for _, row in df.iterrows(): | |
| id_ = row.get(id_col, "") | |
| nom = row.get("Nom_Complet", "") | |
| email = str(row.get("Email", "")).strip() | |
| if id_ and email and "@" in email: | |
| out[f"{id_} — {nom}"] = {"id": id_, "nom": nom, "email": email} | |
| return out | |
| def show_communications_module(client, sheet_name): | |
| st.markdown("<h1>COMMUNICATIONS</h1>", unsafe_allow_html=True) | |
| envoi, stats = st.tabs(["Envoyer", "Statistiques d'ouverture"]) | |
| with envoi: | |
| _onglet_envoi(client, sheet_name) | |
| with stats: | |
| _onglet_statistiques(client, sheet_name) | |
| def _onglet_statistiques(client, sheet_name): | |
| st.caption("Qui a ouvert vos emails. Données Brevo, mises à jour toutes les heures ou à la demande.") | |
| sh = client.open(sheet_name) | |
| if st.button("ACTUALISER LES STATISTIQUES", key="comm_stats_sync"): | |
| try: | |
| with st.spinner("Lecture des événements Brevo..."): | |
| bilan = suivi_emails.synchroniser(sh) | |
| st.success(f"{bilan['campagnes']} campagne(s) synchronisée(s), {bilan['lignes_maj']} destinataire(s) mis à jour.") | |
| except Exception as exc: | |
| st.error(f"Synchronisation impossible : {exc}") | |
| try: | |
| valeurs = sh.worksheet(suivi_emails.FEUILLE).get_all_values() | |
| except Exception: | |
| valeurs = [] | |
| resume = suivi_emails.statistiques(valeurs) | |
| if resume.empty: | |
| st.info("Aucune campagne suivie pour l'instant : les statistiques apparaissent après le premier envoi.") | |
| return | |
| st.dataframe(resume, hide_index=True, use_container_width=True) | |
| st.caption("Ouverts = ouvertures détectées par le pixel. Proxy_Apple = chargement automatique par Apple Mail " | |
| "(ne prouve pas une lecture). Une personne qui bloque les images n'apparaît jamais comme « ouvert ».") | |
| libelles = {f"{r['ID_Campagne']} — {r['Objet']}": r["ID_Campagne"] for _, r in resume.iterrows()} | |
| choix = st.selectbox("Détail d'une campagne", list(libelles.keys()), key="comm_stats_campagne") | |
| entete = valeurs[0] | |
| df = pd.DataFrame([r + [""] * (len(entete) - len(r)) for r in valeurs[1:]], columns=entete) | |
| detail = df[df["ID_Campagne"] == libelles[choix]][[ | |
| "Nom", "Email", "Statut_Envoi", "Delivre", "Ouvert", "Nb_Ouvertures", "Premiere_Ouverture", | |
| "Derniere_Ouverture", "Ouverture_Proxy", "Nb_Clics", "Rebond", "Derniere_Synchro"]] | |
| st.dataframe(detail.sort_values("Ouvert", ascending=False), hide_index=True, use_container_width=True) | |
| def _onglet_envoi(client, sheet_name): | |
| st.caption("Envoi de messages par email à une sélection de clients et/ou garants, avec suivi d'ouverture.") | |
| try: | |
| df_clients, df_garants = _load_recipients(client, sheet_name) | |
| except Exception as e: | |
| st.error(f"Erreur de chargement : {e}") | |
| return | |
| opts_clients = _rows_to_options(df_clients, "ID_Client") | |
| opts_garants = _rows_to_options(df_garants, "ID_Garant") | |
| portee = st.radio( | |
| "Destinataires", | |
| ["Clients sélectionnés", "Garants sélectionnés", "Tous les clients", "Tous les garants", "Sélection mixte"], | |
| horizontal=True, | |
| ) | |
| if portee == "Clients sélectionnés": | |
| choix = st.multiselect("Choisir des clients", list(opts_clients.keys())) | |
| destinataires = [opts_clients[c] for c in choix] | |
| elif portee == "Garants sélectionnés": | |
| choix = st.multiselect("Choisir des garants", list(opts_garants.keys())) | |
| destinataires = [opts_garants[c] for c in choix] | |
| elif portee == "Tous les clients": | |
| destinataires = list(opts_clients.values()) | |
| elif portee == "Tous les garants": | |
| destinataires = list(opts_garants.values()) | |
| else: # Sélection mixte | |
| choix_c = st.multiselect("Clients", list(opts_clients.keys())) | |
| choix_g = st.multiselect("Garants", list(opts_garants.keys())) | |
| destinataires = [opts_clients[c] for c in choix_c] + [opts_garants[g] for g in choix_g] | |
| st.caption(f"{len(destinataires)} destinataire(s) sélectionné(s) — récapitulatif envoyé à {get_admin_email()}.") | |
| objet = st.text_input("Objet") | |
| message = st.text_area("Message", height=200) | |
| st.divider() | |
| if "comm_confirm" not in st.session_state: | |
| st.session_state["comm_confirm"] = False | |
| ready = bool(destinataires) and bool(objet.strip()) and bool(message.strip()) | |
| if st.button("Préparer l'envoi", disabled=not ready): | |
| st.session_state["comm_confirm"] = True | |
| if st.session_state["comm_confirm"]: | |
| st.warning(f"Vous allez envoyer cet email à {len(destinataires)} destinataire(s) :") | |
| st.dataframe(pd.DataFrame(destinataires), use_container_width=True, hide_index=True) | |
| c1, c2 = st.columns(2) | |
| with c1: | |
| if rendre_bouton_protege("CONFIRMER ET ENVOYER", cle="comm_envoi_groupe", type="primary", use_container_width=True): | |
| with st.spinner("Envoi en cours..."): | |
| id_campagne, resultats = suivi_emails.envoyer_campagne( | |
| client.open(sheet_name), destinataires, objet, message) | |
| st.session_state["comm_confirm"] = False | |
| liberer("comm_envoi_groupe") | |
| n_ok = sum(1 for r in resultats if r["ok"]) | |
| st.success(f"Campagne {id_campagne} : {n_ok}/{len(resultats)} email(s) envoyé(s). " | |
| "Les ouvertures apparaissent dans l'onglet Statistiques.") | |
| st.dataframe(pd.DataFrame(resultats), use_container_width=True, hide_index=True) | |
| with c2: | |
| if st.button("Annuler", use_container_width=True): | |
| st.session_state["comm_confirm"] = False | |
| liberer("comm_envoi_groupe") | |
| st.rerun() | |