#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ EmailTriage — lecture locale d'emails .eml, classification, journal anti-doublon, export texte + option push TechnofoxTodo. Usage typique (étape 1 + 2) : python3 email_triage.py --since 2026-08-01 --max-emails 10 --push-todo Cron (exemple, logs DANS le workspace uniquement) : 0 8 * * * cd /home/life/workspace/Développement/EmailTriage \\ && python3 email_triage.py --since $(date -d yesterday +\\%Y-\\%m-\\%d) --push-todo \\ >> data/cron.log 2>&1 """ from __future__ import annotations import argparse import email import hashlib import html as html_lib import imaplib import json import re import ssl import sys import urllib.error import urllib.request from dataclasses import dataclass from datetime import datetime, timedelta, timezone from email import policy from email.parser import BytesParser from email.utils import parsedate_to_datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple, Union # --------------------------------------------------------------------------- # Chemins (hôte volume OU conteneur life-box) # --------------------------------------------------------------------------- def _detect_workspace() -> Path: candidates = [ Path("/home/life/workspace"), Path("/var/lib/containers/storage/volumes/life-home/_data/workspace"), Path(__file__).resolve().parents[2], # .../workspace/Développement/EmailTriage → workspace ] for c in candidates: if (c / "Documents" / "MailExport").is_dir(): return c.resolve() # fallback: parent of this package return Path(__file__).resolve().parents[2] WORKSPACE = _detect_workspace() APP_DIR = Path(__file__).resolve().parent DATA_DIR = APP_DIR / "data" MAIL_EXPORT = WORKSPACE / "Documents" / "MailExport" PROCESSED_JSON = DATA_DIR / "processed_mails.json" # id → meta (anti-doublon) PROGRESS_TXT = DATA_DIR / "progress.txt" # liste lisible priorités / avancement LAST_RUN_JSON = DATA_DIR / "last_run.json" # résumé dernier run SKIP_CANDIDATES_JSON = DATA_DIR / "skip_candidates.json" # spam/pub à valider avant delete # TechnofoxTodo — depuis life-box utiliser host.containers.internal DEFAULT_TODO_URL = "http://host.containers.internal:8788" FALLBACK_TODO_URLS = ( "http://host.containers.internal:8788", "http://10.88.0.1:8788", "http://127.0.0.1:8788", ) # --------------------------------------------------------------------------- # Classification # --------------------------------------------------------------------------- URGENT_KW = [ "urgent", "urgence", "asap", "immédiat", "immediat", "sans délai", "sans delai", "deadline", "date limite", "action requise", "mise en demeure", "impay", "mahnung", "zahlungsaufforderung", "retard de paiement", ] PRIORITY_SENDERS = [ "g.bernert@technofox.ch", "a.praz@horizoncorpfin.ch", "jmgreindl", "damien.pittier@ubs.com", "c.rey@technofox", "@technofox.ch", "@technofox-international.com", ] CATEGORY_KW: Dict[str, List[str]] = { "Facturation": ["facture", "invoice", "paiement", "tva", "mahnung", "rechnung", "devis", "quote"], "Comptabilité": ["compta", "fiduciaire", "bilan", "audit", "écriture", "ecriture"], "Réunion": ["réunion", "reunion", "meeting", "rdv", "call", "calendly", "visio", "zoom"], "Développement": ["développement", "developpement", "feature", "bug", "sprint", "github", "deploy"], "Banque": ["e-banking", "raiffeisen", "ubs", "virements", "encaissement", "prélèvement", "prelevement"], "Juridique": ["contrat", "signature", "uptosign", "notaire", "statuts"], "RH": ["candidature", "bewerbung", "cv ", "entretien d'embauche", "salaire"], "Notification": [ "notification", "newsletter", "no-reply", "noreply", "do not reply", "automatique", "automatic", "status-update", "status update", "votre relevé", "vos derniers", "linkedin", "messages-noreply@", "wordpress@", "extensions ont été mises à jour", "receipt from", "stripe.com", "erforderliche aktion", "bestätige den zugang", ], "Spam/Pub": [ "unsubscribe", "promotion", "webinar", "offre exclusive", "soldes", "free trial", "marketing", "newsletter commerciale", "lenovo@", "ecomm.", "business.lenovo", "flexible laptops", "skalierung von ki", "schaffen sie die richtige", ], } ACTION_BY_CAT = { "Facturation": "Vérifier facture / paiement dans Dolibarr et noter l'échéance.", "Comptabilité": "Préparer ou classer les pièces pour la fiduciaire.", "Réunion": "Confirmer présence ou proposer des créneaux (Calendly).", "Développement": "Évaluer la demande et créer/mettre à jour une carte technique.", "Banque": "Contrôler le mouvement bancaire et rapprocher si besoin.", "Juridique": "Relire le document / finaliser la signature électronique.", "RH": "Traiter la candidature ou la demande RH.", "Notification": "Archiver — aucune action métier requise.", "Spam/Pub": "Candidat suppression (spam/pub) — attendre validation Cédric.", "Autre": "Évaluer manuellement et décider de l'action.", } # Priorité API Todo : P0 / P1 / P2 PRIO_MAP = { "URGENT": "P0", "HAUTE": "P1", "MOYENNE": "P2", "INFO": "P2", "SKIP": "P2", } @dataclass class TriageItem: mail_id: str filepath: str title: str description: str priority: str category: str tag: str deadline: str process: str action_type: str # mail | admin | finance | other sender: str date_iso: str skip_candidate: bool = False skip_reason: str = "" todo_id: Optional[int] = None status: str = "new" # new | processed | skipped_duplicate # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def ensure_data_dir() -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) def load_json(path: Path, default: Any) -> Any: if not path.exists(): return default try: with path.open("r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError): return default def save_json(path: Path, data: Any) -> None: ensure_data_dir() tmp = path.with_suffix(path.suffix + ".tmp") with tmp.open("w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") tmp.replace(path) def mail_id_for(path: Union[Path, str], subject: str, sender: str, date_iso: str) -> str: """Stable id: filename/uid + hash of subject/sender/date.""" name = path.name if isinstance(path, Path) else str(path).rsplit("/", 1)[-1] raw = f"{name}|{subject}|{sender}|{date_iso}" return hashlib.sha256(raw.encode("utf-8", errors="replace")).hexdigest()[:16] def load_mail_conf(path: Path) -> Dict[str, str]: """Parse simple key=value conf (host/user/pass[/port/mailbox]).""" cfg: Dict[str, str] = {} text = path.read_text(encoding="utf-8", errors="replace") for line in text.splitlines(): line = line.strip() if not line or line.startswith("#"): continue if "=" not in line: continue k, v = line.split("=", 1) k = k.strip().lower() v = v.split("#", 1)[0].strip().strip('"').strip("'") if k in ("password", "pwd", "secret"): k = "pass" cfg[k] = v if not cfg.get("host") or not cfg.get("user") or not cfg.get("pass"): raise ValueError(f"Conf IMAP incomplète (host/user/pass requis): {path}") cfg.setdefault("port", "993") cfg.setdefault("mailbox", "INBOX") return cfg def resolve_mail_conf(explicit: Optional[str] = None) -> Path: """Trouve un fichier conf IMAP sous workspace/config/mail/.""" if explicit: p = Path(explicit).expanduser() if not p.is_file(): # relative to workspace p2 = WORKSPACE / explicit if p2.is_file(): return p2.resolve() raise FileNotFoundError(f"Conf IMAP introuvable: {explicit}") return p.resolve() candidates = [ WORKSPACE / "config" / "mail" / "crey_technofoxusa.inc.conf", WORKSPACE / "config" / "mail" / "crey_technofox_ch.conf", Path.home() / ".config" / "technofox" / "mail.conf", ] for c in candidates: if c.is_file(): return c.resolve() raise FileNotFoundError( "Aucune conf IMAP trouvée. Passe --imap-config config/mail/crey_technofoxusa.inc.conf" ) def _imap_since_criterion(since_dt: datetime) -> str: # IMAP SINCE day (inclusive), English month abbr return since_dt.strftime("%d-%b-%Y") def fetch_imap_candidates( conf_path: Path, since_dt: datetime, max_emails: Optional[int], unseen_only: bool = False, ) -> List[Dict[str, Any]]: """ Récupère des mails via IMAP et renvoie la même structure que parse_eml(). path = chemin logique imap://user/mailbox/uid """ cfg = load_mail_conf(conf_path) host = cfg["host"] port = int(cfg.get("port") or 993) user = cfg["user"] password = cfg["pass"] mailbox = cfg.get("mailbox") or "INBOX" print(f"[INFO] IMAP : {user} @ {host}:{port} ({mailbox})") print(f"[INFO] IMAP conf : {conf_path}") ctx = ssl.create_default_context() candidates: List[Dict[str, Any]] = [] with imaplib.IMAP4_SSL(host, port, ssl_context=ctx) as imap: typ, _ = imap.login(user, password) if typ != "OK": raise RuntimeError(f"IMAP login failed: {typ}") # EXAMINE (readonly=True) : boîte en lecture seule — ne modifie pas les flags. # Ne JAMAIS utiliser SELECT en écriture ni STORE +FLAGS (\Seen). typ, _ = imap.select(mailbox, readonly=True) if typ != "OK": raise RuntimeError(f"IMAP examine {mailbox!r} failed: {typ}") print("[INFO] IMAP mode : read-only (EXAMINE) — mails restent non lus côté boîte") criteria = [] if unseen_only: criteria.append("UNSEEN") criteria.append(f'SINCE {_imap_since_criterion(since_dt)}') search_str = "(" + " ".join(criteria) + ")" typ, data = imap.search(None, search_str) if typ != "OK" or not data or not data[0]: print(f"[INFO] IMAP search {search_str!r} → 0 message") return [] uids = data[0].split() # newest last in IMAP sequence often; reverse for newest first uids = list(reversed(uids)) if max_emails is not None: uids = uids[:max_emails] print(f"[INFO] IMAP fetch : {len(uids)} message(s) via BODY.PEEK[] (sans flag \\Seen)") for uid in uids: # BODY.PEEK[] = même contenu que BODY[] / RFC822 mais NE pose PAS \Seen # (même si un serveur ignore EXAMINE — double protection) typ, msg_data = imap.fetch(uid, "(BODY.PEEK[])") if typ != "OK" or not msg_data or not msg_data[0]: continue raw = msg_data[0] if isinstance(raw, tuple) and len(raw) >= 2: raw_bytes = raw[1] else: continue if not isinstance(raw_bytes, (bytes, bytearray)): continue try: msg = BytesParser(policy=policy.default).parsebytes(raw_bytes) except Exception as e: print(f"[WARN] parse IMAP uid={uid!r}: {e}", file=sys.stderr) continue subject = decode_header_value(msg.get("subject")) or "(Sans sujet)" sender = decode_header_value(msg.get("from")) or "(inconnu)" # synthetic path for id + logging logical = Path(f"imap://{user}/{mailbox}/{uid.decode() if isinstance(uid, bytes) else uid}") # reuse date helpers with a fake name prefix YYYYMMDD if possible date_iso, dt = parse_email_date(msg, Path(f"{since_dt.strftime('%Y%m%d')}-imap-{uid}.eml")) body = extract_body(msg) msg_id = decode_header_value(msg.get("message-id")) or "" candidates.append( { "path": logical, "subject": subject, "sender": sender, "date_iso": date_iso, "dt": dt, "body": body, "message_id": msg_id, "imap_uid": uid.decode() if isinstance(uid, bytes) else str(uid), "source": "imap", } ) return candidates def parse_date_from_filename(name: str) -> Optional[datetime]: m = re.match(r"^(\d{8})", name) if not m: return None try: return datetime.strptime(m.group(1), "%Y%m%d") except ValueError: return None def parse_email_date(msg: email.message.Message, filepath: Path) -> Tuple[str, Optional[datetime]]: raw = msg.get("date") or "" dt: Optional[datetime] = None if raw: try: dt = parsedate_to_datetime(raw) if dt.tzinfo is not None: dt = dt.astimezone(timezone.utc).replace(tzinfo=None) except (TypeError, ValueError, IndexError, OverflowError): dt = None if dt is None: dt = parse_date_from_filename(filepath.name) if dt is None: # mtime fallback try: dt = datetime.fromtimestamp(filepath.stat().st_mtime) except OSError: dt = None iso = dt.strftime("%Y-%m-%d") if dt else "" return iso, dt def clean_text(text: str) -> str: """Decode HTML entities and normalize whitespace / accents display.""" if not text: return "" # Named + numeric entities: é à   ' … t = html_lib.unescape(text) # Residual tags if any t = re.sub(r"", "\n", t, flags=re.I) t = re.sub(r"", "\n", t, flags=re.I) t = re.sub(r"<[^>]+>", " ", t) # Common mojibake leftovers t = t.replace("\xa0", " ").replace("\u200b", "") t = re.sub(r"[ \t]+", " ", t) t = re.sub(r"\n{3,}", "\n\n", t) return t.strip() def extract_body(msg: email.message.Message) -> str: body = "" if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() == "text/plain": try: payload = part.get_payload(decode=True) if payload: charset = part.get_content_charset() or "utf-8" body = payload.decode(charset, errors="replace") break except Exception: continue if not body: for part in msg.walk(): if part.get_content_type() == "text/html": try: payload = part.get_payload(decode=True) if payload: charset = part.get_content_charset() or "utf-8" raw_html = payload.decode(charset, errors="replace") body = clean_text(raw_html) break except Exception: continue else: try: payload = msg.get_payload(decode=True) charset = msg.get_content_charset() or "utf-8" if payload: body = payload.decode(charset, errors="replace") else: body = str(msg.get_payload() or "") if (msg.get_content_type() or "").lower() == "text/html": body = clean_text(body) except Exception: body = str(msg.get_payload() or "") return clean_text(body) def decode_header_value(val: Optional[str]) -> str: if not val: return "" return clean_text(str(val).replace("\n", " ").replace("\r", " ")) def summarize(body: str, max_len: int = 280) -> str: body = clean_text(body) if not body: return "(corps vide)" lines = [] for line in body.splitlines(): s = line.strip() if len(s) < 4: continue low = s.lower() if any( x in low for x in ( "sent from my", "envoyé de mon", "unsubscribe", "click here", "merci de ne pas répondre", "ne pas répondre à cet e-mail", "do not reply", "aucun traitement de", ) ): continue if s.startswith(">") or s.startswith("--"): continue lines.append(s) if len(lines) >= 4: break text = " ".join(lines) if lines else body[:max_len].replace("\n", " ") text = re.sub(r"\s+", " ", text).strip() if len(text) > max_len: return text[: max_len - 3] + "..." return text def extract_deadline(text: str) -> str: """Heuristic ISO date YYYY-MM-DD if present in subject/body.""" text = clean_text(text) m = re.search(r"\b(20\d{2})-(\d{2})-(\d{2})\b", text) if m: return f"{m.group(1)}-{m.group(2)}-{m.group(3)}" m = re.search(r"\b(\d{1,2})[./](\d{1,2})[./](20\d{2})\b", text) if m: d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3)) try: return datetime(y, mo, d).strftime("%Y-%m-%d") except ValueError: pass return "" # Jours ajoutés à la date du mail (ou aujourd'hui) si aucune échéance dans le texte INDICATIVE_DEADLINE_DAYS = { "URGENT": 1, "HAUTE": 3, "MOYENNE": 7, "INFO": 14, "SKIP": 0, } def indicative_deadline( priority: str, email_date_iso: str, extracted: str, ) -> str: """Préfère une date trouvée dans le mail ; sinon échéance indicative par priorité.""" if extracted: return extracted days = INDICATIVE_DEADLINE_DAYS.get(priority, 7) if days <= 0: return "" base = datetime.now() if email_date_iso: try: base = datetime.strptime(email_date_iso, "%Y-%m-%d") except ValueError: pass # ne pas dater dans le passé pour une tâche à faire today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) start = max(base.replace(hour=0, minute=0, second=0, microsecond=0), today) return (start + timedelta(days=days)).strftime("%Y-%m-%d") def extract_sender_name(sender: str) -> str: """'Patrice Ledermann ' → 'Patrice'. Ne pas passer par clean_text (qui retire les balises/angles ). """ sender = (sender or "").replace("\n", " ").replace("\r", " ").strip() sender = html_lib.unescape(sender) # Display name before m = re.search(r'^["\']?([^<"\']+?)["\']?\s*<', sender) if m: name = m.group(1).strip().strip('"').strip("'") parts = [p for p in re.split(r"\s+", name) if p and "@" not in p] if parts: if parts[0].lower() not in ("support", "noreply", "no-reply", "info", "mail"): return parts[0] m = re.search(r"([\w.+-]+)@", sender) if m: local = m.group(1) if local.lower() not in ("noreply", "no-reply", "bounce", "mailer-daemon"): return local.split(".")[0].capitalize() return "" def needs_reply(subject: str, sender: str, body: str, category: str) -> Tuple[bool, str]: """ Détecte si une réponse mail est attendue. Returns (needs_reply, reason). """ subject_l = subject.lower() sender_l = sender.lower() body_l = body.lower()[:3000] blob = f"{subject_l}\n{body_l}" # Explicit no-reply no_reply_markers = ( "ne pas répondre", "ne pas repondre", "do not reply", "no-reply", "noreply", "donotreply", "no reply", "merci de ne pas répondre", "this mailbox is not monitored", "ne répondez pas", ) if any(m in blob or m in sender_l for m in no_reply_markers): return False, "expéditeur / corps indique de ne pas répondre" if any(x in sender_l for x in ("noreply@", "no-reply@", "bounce@", "mailer-daemon")): # factures eBill etc. : notification, pas de réponse return False, "adresse technique (noreply/bounce)" if category in ("Spam/Pub", "Notification"): return False, "catégorie sans réponse attendue" # Strong signals that a reply is expected reply_markers = ( "pouvez-vous", "pourriez-vous", "merci de me", "merci de votre retour", "votre retour", "en attente de votre", "waiting for your", "please confirm", "please reply", "could you", "can you", "would you", "qu'en pensez-vous", "qu en pensez", "disponibilités", "disponibilites", "confirmer", "confirmation", "rdv", "réunion", "reunion", "meeting", "as soon as possible", "dans l'attente", "dans l attente", "bien à vous", # often end of human mail expecting reply ) human_question = "?" in subject or body_l.count("?") >= 1 re_thread = bool(re.match(r"^(re|aw|sv|rif|ref)\s*:", subject_l.strip())) if any(m in blob for m in reply_markers) or human_question or re_thread: # Real person domain (not pure automation) if "@" in sender_l and not any( x in sender_l for x in ("newsletter", "news@", "marketing@", "notif") ): return True, "demande / question / fil de discussion détecté" # Facturation from human (not eBill auto) if category == "Facturation" and "facture" in blob and "ebill" not in blob: if re_thread or human_question: return True, "facture avec échange possible" if category == "Réunion": return True, "contexte réunion / RDV" if category in ("Juridique", "RH", "Développement") and ( human_question or re_thread or "coordonnées" in subject_l or "coordonnees" in subject_l ): return True, f"mail {category} nécessitant un retour" return False, "pas de signal de réponse attendue" def draft_reply( subject: str, sender: str, body: str, category: str, priority: str, summary: str, ) -> str: """Brouillon de réponse FR adapté au contexte.""" name = extract_sender_name(sender) or "Madame, Monsieur" subj = subject if subject.lower().startswith("re:") else f"Re: {subject}" summary_short = summary[:180] if summary else subject[:80] if category == "Réunion": body_txt = ( f"Bonjour {name},\n\n" f"Merci pour votre message concernant « {subject[:80]} ».\n\n" "Je vous propose les créneaux suivants (heure suisse) :\n" "- Demain matin (9h–12h)\n" "- Après-demain après-midi (14h–17h)\n" "- En début de semaine prochaine\n\n" "Indiquez-moi ce qui vous arrange le mieux, ou proposez une autre plage.\n\n" "Bien cordialement,\n" "Cédric Rey\n" "Technofox" ) elif category == "Facturation": body_txt = ( f"Bonjour {name},\n\n" f"Merci pour votre message au sujet de « {subject[:80]} ».\n\n" "Je prends note et vérifie le dossier côté comptabilité / Dolibarr. " "Je reviens vers vous rapidement avec confirmation ou les éléments manquants.\n\n" "Bien cordialement,\n" "Cédric Rey\n" "Technofox" ) elif category == "Juridique": body_txt = ( f"Bonjour {name},\n\n" f"Merci pour votre message (« {subject[:80]} »).\n\n" "J'ai bien reçu les éléments. Je les examine et vous confirme la suite " "(signature / documents / prochaines étapes) dans les meilleurs délais.\n\n" "Bien cordialement,\n" "Cédric Rey\n" "Technofox" ) elif "defective" in subject.lower() or "retour" in subject.lower() or "réclamation" in body.lower(): body_txt = ( f"Bonjour {name},\n\n" f"Merci pour votre message concernant « {subject[:80]} ».\n\n" "Nous prenons votre retour au sérieux. Pouvez-vous nous préciser :\n" "- la référence commande / livraison\n" "- le problème constaté\n" "- photos ou rapport si disponibles\n\n" "Dès réception, nous organisons le traitement (échange / avoir / suite).\n\n" "Bien cordialement,\n" "Cédric Rey\n" "Technofox" ) else: body_txt = ( f"Bonjour {name},\n\n" f"Merci pour votre e-mail au sujet de « {subject[:80]} ».\n\n" f"J'ai bien pris connaissance de votre message" + (f" ({summary_short})" if summary_short else "") + ".\n" "Je reviens vers vous rapidement avec une réponse détaillée.\n\n" "Bien cordialement,\n" "Cédric Rey\n" "Technofox" ) if priority == "URGENT": body_txt = body_txt.replace( "Je reviens vers vous rapidement", "Je traite ce point en priorité et reviens vers vous dès que possible", 1, ) return f"Brouillon de réponse\nObjet : {subj}\n\n{body_txt}" def build_process( category: str, priority: str, subject: str, body: str, sender: str, summary: str, ) -> Tuple[str, str]: """ Returns (process_text, action_type). Si une réponse est utile → brouillon mail + action_type=mail. """ base = ACTION_BY_CAT.get(category, ACTION_BY_CAT["Autre"]) reply, reason = needs_reply(subject, sender, body, category) if reply: draft = draft_reply(subject, sender, body, category, priority, summary) process = ( f"Réponse attendue ({reason}).\n\n" f"{draft}\n\n" f"Tâche associée : {base}" ) return process, "mail" # Pas de réponse : tâche pure if category == "Facturation": process = ( f"Tâche (pas de réponse mail — {reason}) :\n" f"- {base}\n" f"- Sujet : {subject[:100]}\n" f"- Payer / classer / archiver selon statut dans e-banking" ) return process, "finance" if priority == "URGENT": return f"URGENT — {base}\nSujet : {subject[:120]}", "admin" return f"{base}\n(Contexte : aucune réponse mail requise — {reason})", ( "finance" if category in ("Facturation", "Banque") else "call" if category == "Réunion" else "admin" ) def classify(subject: str, sender: str, body: str) -> Tuple[str, str, bool, str, str]: """ Returns: priority, category, skip_candidate, skip_reason, action_type """ blob = f"{subject}\n{sender}\n{body[:4000]}".lower() sender_l = sender.lower() # Fast-path skip by sender domain (avant catégories métier) skip_sender_bits = ( "linkedin.com", "lenovo.com", "ecomm.", "noreply@", "no-reply@", "bounce@", "wordpress@", "stripe.com", "mailchimp", "sendgrid", ) force_skip = any(b in sender_l for b in skip_sender_bits) # Wingo invoices are real bills — don't force-skip bounce@wingo if "wingo" in sender_l and "facture" in blob: force_skip = False # Skip categories: match on subject+sender only (body footers often say "unsubscribe") head = f"{subject}\n{sender}".lower() category = "Autre" for cat in ("Spam/Pub", "Notification"): kws = CATEGORY_KW[cat] if any(kw.lower() in head for kw in kws): category = cat break else: for cat, kws in CATEGORY_KW.items(): if cat in ("Spam/Pub", "Notification"): continue if any(kw.lower() in blob for kw in kws): category = cat break if force_skip and category not in ("Facturation", "Banque", "Juridique", "RH"): if any(x in sender_l for x in ("linkedin", "wordpress", "stripe", "noreply", "no-reply")): category = "Notification" if "linkedin" in sender_l or "wordpress" in sender_l else "Spam/Pub" elif any(x in sender_l for x in ("lenovo", "ecomm", "newsletter", "marketing")): category = "Spam/Pub" # sinon: vrai expéditeur humain → ne pas forcer le skip skip = category in ("Spam/Pub", "Notification") skip_reason = "" if category == "Spam/Pub": skip_reason = "Détecté comme pub/spam (mots-clés / expéditeur)" elif category == "Notification": skip_reason = "Notification automatique / newsletter" # Priority known = any(p.lower() in sender_l for p in PRIORITY_SENDERS) urgent = any(kw in blob for kw in URGENT_KW) if skip: priority = "INFO" elif urgent and (known or category in ("Facturation", "Juridique", "Banque")): priority = "URGENT" elif category in ("Facturation", "Banque", "Juridique") or known: priority = "HAUTE" if urgent or "impay" in blob or "retard" in blob else "MOYENNE" elif category in ("Réunion", "Comptabilité", "Développement", "RH"): priority = "MOYENNE" else: priority = "INFO" # action type for Todo if category == "Facturation" or category == "Banque": action_type = "finance" elif category == "Réunion": action_type = "call" elif "répond" in blob or "re:" in subject.lower()[:4]: action_type = "mail" else: action_type = "admin" if not skip else "other" return priority, category, skip, skip_reason, action_type def iter_eml_files(mail_root: Path) -> Iterable[Path]: if not mail_root.is_dir(): return [] return sorted(mail_root.rglob("*.eml"), key=lambda p: p.name, reverse=True) def parse_eml(path: Path) -> Optional[Dict[str, Any]]: try: with path.open("rb") as f: msg = BytesParser(policy=policy.default).parse(f) except OSError as e: print(f"[WARN] lecture impossible {path}: {e}", file=sys.stderr) return None subject = decode_header_value(msg.get("subject")) or "(Sans sujet)" sender = decode_header_value(msg.get("from")) or "(inconnu)" date_iso, dt = parse_email_date(msg, path) body = extract_body(msg) return { "path": path, "subject": subject, "sender": sender, "date_iso": date_iso, "dt": dt, "body": body, } def resolve_todo_base(explicit: Optional[str] = None) -> Optional[str]: urls = [] if explicit: urls.append(explicit.rstrip("/")) urls.extend(FALLBACK_TODO_URLS) for base in urls: try: req = urllib.request.Request(base + "/api/cards", method="GET") with urllib.request.urlopen(req, timeout=3) as resp: if 200 <= resp.status < 300: return base except Exception: continue return None def push_todo_card(base: str, item: TriageItem) -> Optional[int]: # Une seule étiquette lisible (pas de JSON / pas de doublon category+tag) tag_parts = [] for t in (item.category, item.tag): t = (t or "").strip() if t and t not in tag_parts: tag_parts.append(t) tags_str = ", ".join(tag_parts) payload = { "title": item.title[:200], "description": item.description, "summary": item.description[:300], "action_proposal": item.process, "action_type": item.action_type if item.action_type in ( "mail", "call", "admin", "finance", "code", "other" ) else "other", "priority": PRIO_MAP.get(item.priority, "P2"), "tags": tags_str, "due_date": item.deadline or "", "column_id": "backlog", } data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( base + "/api/cards", data=data, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: body = json.loads(resp.read().decode("utf-8")) if body.get("ok"): return int(body.get("id") or 0) or None print(f"[WARN] Todo API: {body}", file=sys.stderr) except urllib.error.HTTPError as e: print(f"[WARN] Todo HTTP {e.code}: {e.read()[:300]!r}", file=sys.stderr) except Exception as e: print(f"[WARN] Todo push failed: {e}", file=sys.stderr) return None def write_progress_txt(items: List[TriageItem], processed: Dict[str, Any]) -> None: ensure_data_dir() lines = [ f"# EmailTriage progress — {datetime.now().strftime('%Y-%m-%d %H:%M')}", f"# Workspace: {WORKSPACE}", f"# Total entries in processed log: {len(processed)}", "", "## Dernier run (par priorité)", "", ] order = {"URGENT": 0, "HAUTE": 1, "MOYENNE": 2, "INFO": 3, "SKIP": 4} sorted_items = sorted(items, key=lambda x: (order.get(x.priority, 9), x.date_iso or "", x.title)) for it in sorted_items: flag = "SKIP?" if it.skip_candidate else "TASK" lines.append( f"- [{it.priority:7}] [{flag}] {it.date_iso} | {it.category:14} | {it.title[:70]}" ) lines.append(f" id={it.mail_id} file={Path(it.filepath).name}") if it.deadline: lines.append(f" deadline={it.deadline}") lines.append(f" process: {it.process.splitlines()[0][:100]}") if it.todo_id: lines.append(f" todo_card_id={it.todo_id}") lines.append("") # open priorities still in log lines.append("## Journal (statuts récents, 30 max)") recent = sorted( processed.values(), key=lambda x: x.get("processed_at", ""), reverse=True, )[:30] for e in recent: lines.append( f"- {e.get('processed_at','?')[:16]} [{e.get('priority','?')}] " f"{e.get('status','?')} {e.get('title','')[:60]}" ) PROGRESS_TXT.write_text("\n".join(lines) + "\n", encoding="utf-8") def run( since: str, max_emails: Optional[int], push_todo: bool, todo_url: Optional[str], dry_run: bool, include_skip: bool, source: str = "local", imap_config: Optional[str] = None, unseen_only: bool = False, ) -> int: ensure_data_dir() print(f"[INFO] Workspace : {WORKSPACE}") print(f"[INFO] Source : {source}") print(f"[INFO] Data dir : {DATA_DIR}") print(f"[INFO] Since : {since}") print(f"[INFO] Max emails : {max_emails or '∞'}") print(f"[INFO] Push Todo : {push_todo}") try: since_dt = datetime.strptime(since, "%Y-%m-%d") except ValueError: print("[ERROR] --since doit être YYYY-MM-DD", file=sys.stderr) return 2 processed: Dict[str, Any] = load_json(PROCESSED_JSON, {}) if not isinstance(processed, dict): processed = {} # Collect candidates (IMAP live OU export .eml local) candidates: List[Dict[str, Any]] = [] if source == "imap": try: conf_path = resolve_mail_conf(imap_config) candidates = fetch_imap_candidates( conf_path, since_dt=since_dt, max_emails=max_emails, unseen_only=unseen_only, ) except Exception as e: print(f"[ERROR] IMAP: {e}", file=sys.stderr) return 2 else: print(f"[INFO] MailExport : {MAIL_EXPORT}") if not MAIL_EXPORT.is_dir(): # aussi Doc/MailExport si rename Documents→Doc alt = WORKSPACE / "Doc" / "MailExport" if alt.is_dir(): mail_root = alt print(f"[INFO] MailExport : {mail_root} (Doc/)") else: print(f"[ERROR] MailExport introuvable: {MAIL_EXPORT}", file=sys.stderr) return 2 else: mail_root = MAIL_EXPORT for path in iter_eml_files(mail_root): meta = parse_eml(path) if not meta: continue dt = meta["dt"] if dt is not None and dt.date() < since_dt.date(): continue fd = parse_date_from_filename(path.name if isinstance(path, Path) else Path(str(path)).name) if fd is not None and fd.date() < since_dt.date(): continue meta["source"] = "local" candidates.append(meta) def sort_key(m: Dict[str, Any]): d = m.get("dt") or datetime.min return d candidates.sort(key=sort_key, reverse=True) if max_emails is not None: candidates = candidates[:max_emails] print(f"[INFO] Candidats après filtre: {len(candidates)}") todo_base = None if push_todo and not dry_run: todo_base = resolve_todo_base(todo_url) if not todo_base: print( "[WARN] TechnofoxTodo injoignable " "(essayé host.containers.internal / 10.88.0.1 / 127.0.0.1). " "Continuer sans push cartes.", file=sys.stderr, ) else: print(f"[INFO] Todo API : {todo_base}") items: List[TriageItem] = [] skip_batch: List[Dict[str, Any]] = [] new_count = 0 dup_count = 0 for meta in candidates: path = meta["path"] # Path (.eml) ou Path logique imap://… subject = meta["subject"] sender = meta["sender"] date_iso = meta["date_iso"] body = meta["body"] # Préférer Message-ID IMAP pour anti-doublon stable if meta.get("message_id"): mid = hashlib.sha256( meta["message_id"].encode("utf-8", errors="replace") ).hexdigest()[:16] else: mid = mail_id_for(path, subject, sender, date_iso) if mid in processed: dup_count += 1 items.append( TriageItem( mail_id=mid, filepath=str(path), title=subject[:200], description=summarize(body), priority=processed[mid].get("priority", "INFO"), category=processed[mid].get("category", "Autre"), tag=processed[mid].get("category", "Autre"), deadline=processed[mid].get("deadline", ""), process=processed[mid].get("process", ""), action_type=processed[mid].get("action_type", "other"), sender=sender, date_iso=date_iso, skip_candidate=bool(processed[mid].get("skip_candidate")), skip_reason=processed[mid].get("skip_reason", ""), todo_id=processed[mid].get("todo_id"), status="skipped_duplicate", ) ) continue priority, category, skip, skip_reason, _action_hint = classify( subject, sender, body ) if skip: priority = "INFO" desc = summarize(body) extracted_dl = extract_deadline(f"{subject}\n{body[:2000]}") deadline = indicative_deadline(priority, date_iso, extracted_dl) process, action_type = build_process( category, priority, subject, body, sender, desc ) title = subject[:200] item = TriageItem( mail_id=mid, filepath=str(path), title=title, description=desc, priority=priority, category=category, tag=category, deadline=deadline, process=process, action_type=action_type, sender=sender, date_iso=date_iso, skip_candidate=skip, skip_reason=skip_reason, status="new", ) if skip and not include_skip: # still record as processed-for-scan but not as Todo task item.status = "skip_candidate" skip_batch.append( { "mail_id": mid, "filepath": str(path), "title": title, "sender": sender, "date": date_iso, "reason": skip_reason, "category": category, } ) elif push_todo and todo_base and not dry_run and not skip: tid = push_todo_card(todo_base, item) item.todo_id = tid item.status = "processed" if tid else "processed_no_todo" else: item.status = "processed" if not dry_run else "dry_run" if not dry_run: processed[mid] = { "mail_id": mid, "filepath": str(path), "title": title, "sender": sender, "date": date_iso, "priority": item.priority, "category": category, "deadline": deadline, "process": process, "action_type": action_type, "skip_candidate": skip, "skip_reason": skip_reason, "todo_id": item.todo_id, "status": item.status, "processed_at": datetime.now().isoformat(timespec="seconds"), } new_count += 1 items.append(item) if not dry_run: save_json(PROCESSED_JSON, processed) # merge skip candidates file prev_skip = load_json(SKIP_CANDIDATES_JSON, []) if not isinstance(prev_skip, list): prev_skip = [] # de-dupe by mail_id by_id = {x.get("mail_id"): x for x in prev_skip if isinstance(x, dict)} for s in skip_batch: by_id[s["mail_id"]] = s save_json(SKIP_CANDIDATES_JSON, list(by_id.values())) write_progress_txt(items, processed) save_json( LAST_RUN_JSON, { "finished_at": datetime.now().isoformat(timespec="seconds"), "since": since, "max_emails": max_emails, "candidates": len(candidates), "new": new_count, "duplicates": dup_count, "skip_candidates": len(skip_batch), "push_todo": push_todo, "todo_base": todo_base, }, ) # Console report print("\n" + "=" * 72) print("RÉSULTAT TRI EMAIL") print("=" * 72) for it in items: if it.status == "skipped_duplicate": continue mark = "SKIP?" if it.skip_candidate else "OK" print(f"\n[{mark}] [{it.priority}] {it.date_iso} — {it.title[:70]}") print(f" cat={it.category} from={it.sender[:50]}") print(f" résumé: {it.description[:120]}") print(f" process: {it.process.splitlines()[0][:100]}") if it.deadline: print(f" deadline: {it.deadline}") if it.todo_id: print(f" → carte Todo id={it.todo_id}") if it.skip_candidate: print(f" → candidat suppression: {it.skip_reason}") print("\n" + "-" * 72) print(f"Nouveaux traités : {new_count}") print(f"Doublons ignorés : {dup_count}") print(f"Candidats skip : {len(skip_batch)} (voir data/skip_candidates.json)") print(f"Journal : {PROCESSED_JSON}") print(f"Progress texte : {PROGRESS_TXT}") if dry_run: print("(dry-run : aucune écriture journal / Todo)") print("=" * 72) return 0 def main(argv: Optional[List[str]] = None) -> int: p = argparse.ArgumentParser( description="Triage emails (local .eml OU IMAP) → journal + TechnofoxTodo /api/cards", ) p.add_argument( "--since", default="2026-08-01", help="Ne traiter que les mails à partir de cette date (YYYY-MM-DD)", ) p.add_argument( "--max-emails", type=int, default=None, help="Limiter au N plus récents (ex: 10 pour le test)", ) p.add_argument( "--source", choices=("local", "imap"), default="local", help="local = Documents/MailExport .eml ; imap = boîte live (config/mail/*.conf)", ) p.add_argument( "--imap-config", default=None, help="Fichier conf IMAP (ex: config/mail/crey_technofoxusa.inc.conf)", ) p.add_argument( "--unseen-only", action="store_true", help="IMAP: uniquement les UNSEEN (non lus)", ) p.add_argument( "--push-todo", action="store_true", help="Créer une carte TechnofoxTodo pour chaque mail utile (pas les skip)", ) p.add_argument( "--todo-url", default=None, help="Base URL Todo (défaut: auto host.containers.internal:8788 …)", ) p.add_argument( "--dry-run", action="store_true", help="Analyser sans écrire journal ni Todo", ) p.add_argument( "--include-skip", action="store_true", help="Pousser aussi spam/notifications en Todo (déconseillé)", ) p.add_argument( "--list-skips", action="store_true", help="Afficher les candidats suppression en attente de validation", ) args = p.parse_args(argv) if args.list_skips: ensure_data_dir() skips = load_json(SKIP_CANDIDATES_JSON, []) print(json.dumps(skips, indent=2, ensure_ascii=False)) print(f"\n{len(skips)} candidat(s) — aucune suppression auto (validation manuelle).") return 0 # Raccourci: si --imap-config est fourni, source=imap par défaut source = args.source if args.imap_config and source == "local": source = "imap" return run( since=args.since, max_emails=args.max_emails, push_todo=args.push_todo, todo_url=args.todo_url, dry_run=args.dry_run, include_skip=args.include_skip, source=source, imap_config=args.imap_config, unseen_only=args.unseen_only, ) if __name__ == "__main__": sys.exit(main())