panleicim/appointment_tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

need to test read confirmation mails

Browse Source
This commit is contained in:
PAN Lei
2026-03-29 19:13:20 +02:00
parent 9802848c5f
commit e8b0a4aae9
2 changed files with 141 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mail/imap_proxy_reader.py
+101 -13
View File
@@ -20,13 +20,14 @@ import io
import logging
import os
import re
import ssl
import socket
import ssl
from dataclasses import dataclass, field
from email.message import Message
from typing import List, Optional, Tuple
import socks
import sys
from dotenv import load_dotenv
from imapclient import IMAPClient
@@ -90,7 +91,10 @@ PROXY_TYPE_MAP = {
"HTTP": socks.HTTP,
}
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(stream=sys.stdout))
# ──────────────────────────────────────────────────────────────
@@ -209,12 +213,31 @@ class ProxyIMAPClient(IMAPClient):
Usage :
proxy = ProxyConfig(host="127.0.0.1", port=1080, proxy_type="SOCKS5")
client = ProxyIMAPClient("imap.gmail.com", proxy=proxy, use_uid=True)
client = ProxyIMAPClient("imap.gmail.com", proxy=proxy, use_uid=True,
subjects=["Confirmation", "Appointment"])
client.login("user@gmail.com", "password")
Paramètres supplémentaires
--------------------------
proxy : ProxyConfig
Configuration du proxy SOCKS/HTTP.
subjects : list[str], optional
Sujets (ou sous-chaînes) à utiliser pour filtrer les emails.
Accessibles via ``client.subjects``.
Utilisés par ``search_by_subjects()`` pour construire
automatiquement les critères IMAP SUBJECT.
"""
def __init__(self, host: str, proxy: ProxyConfig, **kwargs):
def __init__(
self,
host: str,
proxy: ProxyConfig,
subjects: Optional[List[str]] = None,
**kwargs,
):
self._proxy = proxy
# Sujets à rechercher, injectables depuis l'extérieur
self.subjects: List[str] = list(subjects) if subjects else []
super().__init__(host, **kwargs)
def _create_IMAP4(self):
@@ -228,12 +251,59 @@ class ProxyIMAPClient(IMAPClient):
timeout=getattr(self._timeout, "connect", None),
)
# Connexion non-SSL à travers le proxy (rare, mais supporté)
# On monkey-patch juste la connexion TCP
raise NotImplementedError(
"Connexion IMAP non-SSL via proxy non implémentée. "
"Utilisez ssl=True (port 993)."
)
def search_by_subjects(
self,
since: Optional[datetime.datetime] = None,
extra_criteria: Optional[List] = None,
) -> List[int]:
"""
Recherche les UIDs des emails dont le sujet correspond à l'un
des sujets stockés dans ``self.subjects``.
Si ``self.subjects`` est vide, retourne tous les messages
depuis ``since`` (sans filtre par sujet).
Paramètres
----------
since : datetime, optional
Filtre SINCE (aujourd'hui par défaut).
extra_criteria : list, optional
Critères IMAP supplémentaires à combiner (AND implicite).
Retourne
--------
list[int] — UIDs correspondants (peut être vide).
Exemple
-------
client.subjects = ["Confirmation RDV", "confirmed"]
uids = client.search_by_subjects(since=datetime.datetime.today())
"""
since = since or datetime.datetime.today()
base: List = ["SINCE", since]
if extra_criteria:
base.extend(extra_criteria)
if not self.subjects:
return self.search(base)
# Construire OR enchaîné : OR SUBJECT "A" (OR SUBJECT "B" SUBJECT "C")
# IMAPClient accepte des listes imbriquées pour les OR
def _build_or(subjects: List[str]) -> List:
if len(subjects) == 1:
return ["SUBJECT", subjects[0]]
return ["OR", ["SUBJECT", subjects[0]], _build_or(subjects[1:])]
subject_filter = _build_or(self.subjects)
# Combiner avec les critères de base (AND implicite dans IMAP)
criteria = base + subject_filter
return self.search(criteria)
# ──────────────────────────────────────────────────────────────
# Fonctions utilitaires
@@ -282,12 +352,20 @@ class ProxyMailReader:
Paramètres
----------
account : MailAccount
account : MailAccount
Identifiants du compte email.
proxy : ProxyConfig
proxy : ProxyConfig
Configuration du proxy.
timeout : float, optional
timeout : float, optional
Timeout de connexion en secondes (défaut : 30 s).
subjects : list[str], optional
Liste de sujets (ou sous-chaînes) à rechercher dans les emails.
Si None ou vide, on utilise les sujets Hermès par défaut
(VALIDATION_URL_SUBJECT_FR et VALIDATION_URL_SUBJECT_EN).
Les sujets fournis s'ajoutent aux critères par défaut (OR).
from_addresses : list[str], optional
Liste d'adresses expéditeur à accepter en complément.
Si None ou vide, on conserve uniquement "no-reply@hermes.com".
"""
def __init__(
@@ -295,10 +373,19 @@ class ProxyMailReader:
account: MailAccount,
proxy: ProxyConfig,
timeout: float = 30.0,
subjects: Optional[List[str]] = None,
from_addresses: Optional[List[str]] = None,
):
self.account = account
self.proxy = proxy
self.timeout = timeout
self._subjects = []
if subjects:
self._subjects.extend(subjects)
# Adresses expéditeur acceptées
self._from_addresses: List[str] = ["no-reply@hermes.com"]
if from_addresses:
self._from_addresses.extend(from_addresses)
# ── Connexion ────────────────────────────────────────────
@@ -311,12 +398,13 @@ class ProxyMailReader:
client = ProxyIMAPClient(
host=imap_server,
proxy=self.proxy,
subjects=self._subjects, # propagation des sujets vers le client bas niveau
use_uid=True,
ssl=True,
timeout=self.timeout,
)
client.login(self.account.login, self.account.password)
logger.info("[%s] Connecté.", self.account.login)
logger.info("[%s] Connecté. Sujets recherchés : %s", self.account.login, self._subjects)
return client
# ── Lecture des dossiers ─────────────────────────────────
@@ -343,7 +431,8 @@ class ProxyMailReader:
return results
try:
uids = client.search(["SINCE", since])
# Utilise les sujets injectés dans client pour filtrer dès la requête IMAP
uids = client.search_by_subjects(since=since)
except Exception as exc:
logger.warning("[%s] Recherche échouée dans '%s' : %s",
self.account.login, folder, exc)
@@ -366,11 +455,10 @@ class ProxyMailReader:
from_addr = em.get("From", "")
to_addr = em.get("To", self.account.login)
# Filtrer : on ne garde que les emails de validation Hermes
# Filtrer : on ne garde que les emails correspondant aux sujets/expéditeurs configurés
is_validation = (
VALIDATION_URL_SUBJECT_FR in subject
or VALIDATION_URL_SUBJECT_EN in subject
or "no-reply@hermes.com" in from_addr.lower()
any(s in subject for s in self._subjects)
or any(addr in from_addr.lower() for addr in self._from_addresses)
)
if not is_validation:
continue