"""
AZALEE PATRIMOINE — Synthese quotidienne Middle Office
Version cPanel : IMAP + SMTP (sans Outlook)
Planifier via Cron Job cPanel : 0 8 * * 1-5
"""
import imaplib
import smtplib
import email
import re
import sys
import os
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.header import decode_header
from datetime import date, datetime
from pathlib import Path
from collections import defaultdict
import urllib.request
def telecharger_tracker():
"""Telecharge le tracker depuis Dropbox avant lecture."""
try:
print(" Telechargement du tracker depuis Dropbox...")
urllib.request.urlretrieve(DROPBOX_URL, CHEMIN_FICHIER)
print(" Tracker telecharge avec succes.")
return True
except Exception as e:
print(f" Erreur telechargement Dropbox : {e}")
return False
# =============================================================================
# CONFIGURATION — 6 LIGNES A REMPLIR
# =============================================================================
# Tracker telecharge automatiquement depuis Dropbox a chaque execution
DROPBOX_URL = "https://www.dropbox.com/scl/fi/zarm173eokcq3ehullo2h/tracker_azalee_2026.xlsx?rlkey=veobsj371i4iyt1oumi51et3m&dl=1"
CHEMIN_FICHIER = "/tmp/tracker_azalee_2026.xlsx"
# Compte qui LIT les emails (boite back office ou personnelle Mosbahi)
IMAP_HOST = "gavin.o2switch.net"
IMAP_USER = "middle.office@azalee-patrimoine.fr"
IMAP_PASSWORD = "cusfir-7zonBi-cybcom"
# Compte qui ENVOIE la synthese (peut etre le meme)
SMTP_HOST = "gavin.o2switch.net"
SMTP_PORT = 465 # 465 SSL ou 587 TLS selon hebergeur
SMTP_USER = "middle.office@azalee-patrimoine.fr"
SMTP_PASSWORD = "cusfir-7zonBi-cybcom"
# Destinataire de la synthese
EMAIL_DESTINATAIRE = "m.mosbahi@azalee-patrimoine.fr"
# Boite partagee a scanner en plus (laisser vide "" si inexistante)
# Mosbahi peut aussi transmettre depuis sa boite personnelle — ajouter si besoin
BOITE_PARTAGEE = ""
# Seuils alertes tracker (jours)
SEUIL_RETARD = 15
SEUIL_CRITIQUE = 20
# =============================================================================
# EMAILS COMMERCIAUX INTERNES
# =============================================================================
EMAILS_COMMERCIAUX = [
"m.mosbahi@azalee-patrimoine.fr",
"l.serre@azalee-patrimoine.fr",
"direction@azalee-patrimoine.fr",
"s.bruna@azalee-patrimoine.fr",
"benjamin.baron@hentpatrimoine.fr",
"bruna@ecop.live",
]
# =============================================================================
# PARTENAIRES — 38 COMPAGNIES PRE-CONFIGUREES
# =============================================================================
PARTENAIRES_DOMAINES = {
"ABEILLE": ["abeille-assurances.fr"],
"AESTIAM": ["aestiam.com"],
"ALPHEYS": ["alpheys.com"],
"APICIL EPARGNE": ["apicil.com", "intencial.fr"],
"APICIL INTENTIAL": ["intencial.fr"],
"APRIL PATRIMOINE": ["april.com"],
"AVIP ALLIANZ": ["allianz.fr"],
"AXA": ["axa-ws.fr"],
"AXA SECURE ADVANTAGE": ["axa-lifeeurope.com"],
"CARDIF": ["cardif.fr"],
"CORUM": ["corumlepargne.fr"],
"CREDIT FONCIER": ["creditfoncier.fr"],
"DELUBAC": ["delubac.fr"],
"EPSENS": ["epsens.com"],
"ERES": ["eres-group.com"],
"FIDUCIAL GERANCE": ["fiducial.fr"],
"KEPLER": ["keplercheuvreux.com"],
"LA MONDIALE SOFIDY": ["ag2rlamondiale.fr"],
"LYNCEUS": ["lynceus-partners.com"],
"MIMCO": ["mimco-platform.com"],
"NEUFLIZE": ["fr.abnamro.com"],
"NORTIA": ["nortia.fr"],
"ODDO": ["oddo-bhf.com"],
"ONE LIFE": ["onelife.com"],
"PAREF": ["paref.com"],
"PERIAL": ["perial.com"],
"SELENCIA": ["selencia.fr"],
"SILEX": ["silex-partners.com"],
"SOFIDY": ["sofidy.com"],
"SOGENIAL": ["sogenial.fr"],
"SWISS LIFE": ["intermediaires-sl.fr"],
"UAF LIFE": ["uaflife.fr", "uaflife.com"],
"UFG - Immobilier": ["la-francaise.com"],
"UNEP": ["unep.asso.fr"],
"VATEL": ["vatelcapital.com"],
"VIE PLUS": ["vieplus.fr", "suravenir.fr"],
"VOISIN": ["atland-voisin.com"],
}
DOMAINE_VERS_PARTENAIRE = {}
for nom, domaines in PARTENAIRES_DOMAINES.items():
for d in domaines:
DOMAINE_VERS_PARTENAIRE[d.lower()] = nom
# =============================================================================
# UTILITAIRES
# =============================================================================
def decoder_header(valeur):
if not valeur:
return ""
parts = decode_header(valeur)
result = []
for part, enc in parts:
if isinstance(part, bytes):
try:
result.append(part.decode(enc or 'utf-8', errors='replace'))
except Exception:
result.append(part.decode('latin-1', errors='replace'))
else:
result.append(str(part))
return ' '.join(result).strip()
def extraire_email(from_str):
match = re.search(r'[\w.\-+]+@[\w.\-]+', from_str or '')
return match.group(0).lower() if match else ''
def identifier_partenaire(from_str):
domain_match = re.search(r'@([\w.\-]+)', from_str.lower())
if not domain_match:
return None
domain = domain_match.group(1)
if domain in DOMAINE_VERS_PARTENAIRE:
return DOMAINE_VERS_PARTENAIRE[domain]
for d, nom in DOMAINE_VERS_PARTENAIRE.items():
if domain.endswith('.' + d):
return nom
return None
def est_commercial(from_str):
from_lower = from_str.lower()
return any(c.lower() in from_lower for c in EMAILS_COMMERCIAUX)
def formater_date(dt):
if not dt:
return "—"
if isinstance(dt, datetime):
return dt.strftime("%d/%m %H:%M")
return str(dt)
# =============================================================================
# SCAN IMAP
# =============================================================================
def scanner_imap(host, user, password, boite="INBOX"):
emails_partenaires = []
emails_commerciaux = []
try:
mail = imaplib.IMAP4_SSL(host, 993)
mail.login(user, password)
mail.select(boite)
# Tous les non-lus
status, data = mail.search(None, 'UNSEEN')
if status != 'OK' or not data[0]:
print(f" {boite} : aucun email non lu")
mail.logout()
return [], []
ids = data[0].split()
print(f" {boite} : {len(ids)} email(s) non lu(s) a analyser")
for num in ids:
try:
status, msg_data = mail.fetch(num, '(RFC822)')
if status != 'OK':
continue
raw = msg_data[0][1]
msg = email.message_from_bytes(raw)
from_raw = decoder_header(msg.get('From', ''))
subject = decoder_header(msg.get('Subject', ''))[:120]
date_str = msg.get('Date', '')
try:
from email.utils import parsedate_to_datetime
received = parsedate_to_datetime(date_str)
except Exception:
received = None
# Compter pieces jointes
pj_count = sum(
1 for part in msg.walk()
if part.get_content_disposition() == 'attachment'
)
partenaire = identifier_partenaire(from_raw)
if partenaire:
emails_partenaires.append({
"partenaire": partenaire,
"expediteur": from_raw[:60],
"objet": subject,
"date": received,
"pieces_jointes": pj_count,
"boite": boite,
})
elif est_commercial(from_raw):
# Extraire debut du corps
extrait = ""
for part in msg.walk():
if part.get_content_type() == 'text/plain':
try:
body = part.get_payload(decode=True).decode('utf-8', errors='replace')
lignes = [l.strip() for l in body.split('\n') if l.strip()][:3]
extrait = ' | '.join(lignes)[:200]
except Exception:
pass
break
emails_commerciaux.append({
"expediteur": from_raw[:60],
"objet": subject,
"date": received,
"pieces_jointes": pj_count,
"extrait": extrait,
"boite": boite,
})
except Exception as e:
continue
mail.logout()
except imaplib.IMAP4.error as e:
print(f" Erreur IMAP {boite} : {e}")
except Exception as e:
print(f" Erreur connexion {host} : {e}")
print(f" -> {len(emails_partenaires)} partenaires | {len(emails_commerciaux)} commerciaux")
return emails_partenaires, emails_commerciaux
# =============================================================================
# TRACKER EXCEL
# =============================================================================
def lire_tracker(chemin):
try:
import openpyxl
except ImportError:
print("openpyxl manquant : pip install openpyxl")
return [], [], 0
today = date.today()
actes_critique, actes_retard = [], []
total_en_cours = 0
try:
wb = openpyxl.load_workbook(chemin, data_only=True)
ws = wb["SUIVI"]
except Exception as e:
print(f" Erreur lecture tracker : {e}")
return [], [], 0
for row in ws.iter_rows(min_row=3, max_row=5000, values_only=True):
date_reception = row[0]
nom, prenom = row[1], row[2]
compagnie = row[3]
nature = row[6]
statut = row[8]
delai_cible = row[10]
date_cloture = row[13]
if not date_reception:
continue
statut_norm = (statut or '').lower().replace('é', 'e').replace('é', 'e')
if statut_norm in ("traite", "annule"):
continue
total_en_cours += 1
if isinstance(date_reception, datetime):
date_reception = date_reception.date()
if isinstance(date_cloture, datetime):
date_cloture = date_cloture.date()
ref = date_cloture if date_cloture else today
jours = (ref - date_reception).days
seuil = int(delai_cible) if delai_cible else SEUIL_RETARD
acte = {
"nom": f"{nom or ''} {prenom or ''}".strip(),
"compagnie": compagnie or "—",
"nature": nature or "—",
"statut": statut or "—",
"jours": jours,
}
if jours > SEUIL_CRITIQUE:
actes_critique.append(acte)
elif jours > seuil:
actes_retard.append(acte)
return actes_critique, actes_retard, total_en_cours
# =============================================================================
# EMAIL HTML
# =============================================================================
def _badge(n, bg):
return f"{n}"
def _sec(titre, couleur):
return f"{titre}
"
def _td(v, bold=False, color=None):
s = f"padding:7px 11px;{'font-weight:bold;' if bold else ''}{'color:'+color+';' if color else ''}"
return f"{v or ''} | "
def _pj(n):
return f"📎 {n}" if n else "—"
def _table(rows_html, headers, col_bg):
ths = "".join(f"{h} | " for h in headers)
return f""""""
def _vide():
return "Aucun element.
"
def construire_html(emails_part, emails_comm, actes_crit, actes_ret, total):
aujourd_hui = date.today().strftime("%A %d %B %Y")
nb_p = len(emails_part)
nb_c = len(emails_comm)
nb_r = len(actes_ret)
nb_k = len(actes_crit)
if nb_k > 0:
statut = f"ATTENTION — {nb_k} acte(s) critique(s)"
elif nb_r + nb_p + nb_c > 0:
statut = f"{nb_r+nb_p+nb_c} point(s) a traiter"
else:
statut = "Tout est a jour"
kpis = "".join(
f""
for v, l, c in [
(total, "ACTES EN COURS", "#1F3864"),
(nb_p, "EMAILS PARTENAIRES", "#2E5FA3"),
(nb_c, "EMAILS COMMERCIAUX", "#8B6914"),
(nb_r, "RETARDS", "#FF6600"),
(nb_k, "CRITIQUES", "#C00000"),
]
)
# Bloc partenaires
par_part = defaultdict(list)
for e in emails_part:
par_part[e['partenaire']].append(e)
html_p = _sec(f"Emails partenaires non lus {_badge(nb_p,'#2E5FA3')}", "#2E5FA3")
if par_part:
for nom, mails in sorted(par_part.items(), key=lambda x: -len(x[1])):
html_p += f"{nom} — {len(mails)} email(s)
"
rows = "".join(
f""
f"{_td(e['expediteur'])}{_td(e['objet'])}{_td(formater_date(e['date']))}{_td(_pj(e['pieces_jointes']))}
"
for i, e in enumerate(mails)
)
html_p += _table(rows, ["Expediteur","Objet","Date","PJ"], "#2E5FA3")
else:
html_p += _vide()
# Bloc commerciaux
html_c = _sec(f"Emails internes a traiter {_badge(nb_c,'#8B6914')}", "#8B6914")
if emails_comm:
rows = "".join(
f""
f"{_td(e['expediteur'],bold=True)}{_td(e['objet'])}{_td(formater_date(e['date']))}{_td(_pj(e['pieces_jointes']))}"
f"| {e.get('extrait','')} |
"
for i, e in enumerate(emails_comm)
)
html_c += _table(rows, ["De","Objet","Date","PJ","Extrait"], "#8B6914")
else:
html_c += _vide()
# Bloc tracker
def bloc_tracker(actes, couleur, titre):
h = _sec(titre, couleur)
if not actes:
return h + _vide()
rows = "".join(
f""
f"{_td(a['nom'],bold=True)}{_td(a['compagnie'])}{_td(a['nature'])}{_td(a['statut'])}"
f"{_td(str(a['jours'])+' j',bold=True,color=couleur)}
"
for i, a in enumerate(sorted(actes, key=lambda x: -x['jours']))
)
return h + _table(rows, ["Client","Compagnie","Nature","Statut","Jours"], couleur)
html_t = bloc_tracker(actes_crit, "#C00000", f"Retards critiques (>{SEUIL_CRITIQUE}j) {_badge(nb_k,'#C00000')}")
html_t += bloc_tracker(actes_ret, "#FF6600", f"Retards (delai depasse) {_badge(nb_r,'#FF6600')}")
return f"""
Azalee Patrimoine — Synthese Middle Office
{aujourd_hui}
{statut}
{kpis}
{html_p}{html_c}{html_t}
Email automatique — Ne pas repondre. Marquer les emails traites comme lus dans la messagerie.