215 lines
8.7 KiB
Python
215 lines
8.7 KiB
Python
"""IMAP ophalen en SMTP versturen, alleen met de standaardbibliotheek.
|
||
|
||
Beveiliging per server: 'ssl' (direct TLS, IMAP 993 / SMTP 465), 'starttls' (IMAP 143 / SMTP 587) of 'none';
|
||
leeg = afleiden uit de poort. Met smtp_sent_folder komt een kopie van verstuurde mail in die IMAP-map, voor
|
||
servers die dat zelf niet doen (Zimbra, MX Plan; Gmail doet het wel zelf).
|
||
"""
|
||
import email
|
||
import html
|
||
import imaplib
|
||
import re
|
||
import smtplib
|
||
import ssl
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from email.message import EmailMessage
|
||
from email.policy import default as policy
|
||
from email.utils import formatdate, make_msgid, parseaddr, parsedate_to_datetime
|
||
|
||
MONTHS = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
|
||
|
||
|
||
def _text(msg):
|
||
part = msg.get_body(preferencelist=('plain', 'html'))
|
||
if part is None:
|
||
return ''
|
||
try:
|
||
content = part.get_content()
|
||
except Exception:
|
||
content = (part.get_payload(decode=True) or b'').decode('utf-8', 'replace')
|
||
if part.get_content_type() == 'text/html':
|
||
content = re.sub(r'(?is)<(script|style|head).*?</\1>', '', content)
|
||
content = re.sub(r'(?i)<br\s*/?>|</p>|</div>|</tr>|</h\d>|</li>', '\n', content)
|
||
content = html.unescape(re.sub(r'<[^>]+>', '', content))
|
||
content = re.sub(r'[ \t]+\n', '\n', content)
|
||
return re.sub(r'\n{3,}', '\n\n', content).strip()
|
||
|
||
|
||
# Voorinstellingen voor Instellingen › E-mail (de pagina vult de velden in; daarna blijft alles aanpasbaar).
|
||
PROVIDERS = {
|
||
'ovh': {'label': 'OVHcloud Zimbra / MX Plan (Europa)', 'imap_host': 'imap.mail.ovh.net', 'imap_port': '993',
|
||
'imap_security': 'ssl', 'smtp_host': 'smtp.mail.ovh.net', 'smtp_port': '465', 'smtp_security': 'ssl',
|
||
'smtp_sent_folder': 'Sent'},
|
||
'ovh-ca': {'label': 'OVHcloud Zimbra / MX Plan (Amerika, Azië)', 'imap_host': 'imap.mail.ovh.ca',
|
||
'imap_port': '993', 'imap_security': 'ssl', 'smtp_host': 'smtp.mail.ovh.ca', 'smtp_port': '465',
|
||
'smtp_security': 'ssl', 'smtp_sent_folder': 'Sent'},
|
||
'gmail': {'label': 'Gmail / Google Workspace (app-wachtwoord)', 'imap_host': 'imap.gmail.com', 'imap_port': '993',
|
||
'imap_security': 'ssl', 'smtp_host': 'smtp.gmail.com', 'smtp_port': '587', 'smtp_security': 'starttls',
|
||
'smtp_sent_folder': ''},
|
||
}
|
||
SECURITY = [('', 'Automatisch (uit de poort)'), ('ssl', 'SSL/TLS'), ('starttls', 'STARTTLS'), ('none', 'Geen')]
|
||
|
||
|
||
def _security(cfg, kind, ssl_port):
|
||
sec = (cfg.get(f'{kind}_security') or '').lower()
|
||
return sec if sec in ('ssl', 'starttls', 'none') else ('ssl' if int(cfg.get(f'{kind}_port') or ssl_port) == ssl_port else 'starttls')
|
||
|
||
|
||
def _connect_imap(cfg):
|
||
host, port = cfg['imap_host'], int(cfg.get('imap_port') or 993)
|
||
sec = _security(cfg, 'imap', 993)
|
||
if sec == 'ssl':
|
||
conn = imaplib.IMAP4_SSL(host, port, ssl_context=ssl.create_default_context(), timeout=30)
|
||
else:
|
||
conn = imaplib.IMAP4(host, port, timeout=30)
|
||
if sec == 'starttls':
|
||
conn.starttls(ssl.create_default_context())
|
||
conn.login(cfg['imap_user'], cfg.get('imap_pass', ''))
|
||
return conn
|
||
|
||
|
||
def _connect_smtp(cfg):
|
||
host, port = cfg.get('smtp_host'), int(cfg.get('smtp_port') or 587)
|
||
if not host:
|
||
raise RuntimeError('Geen SMTP-server ingesteld')
|
||
sec = _security(cfg, 'smtp', 465)
|
||
if sec == 'ssl':
|
||
server = smtplib.SMTP_SSL(host, port, timeout=30, context=ssl.create_default_context())
|
||
else:
|
||
server = smtplib.SMTP(host, port, timeout=30)
|
||
if sec == 'starttls':
|
||
server.starttls(context=ssl.create_default_context())
|
||
user = cfg.get('smtp_user') or cfg.get('imap_user', '')
|
||
if user:
|
||
server.login(user, cfg.get('smtp_pass') or cfg.get('imap_pass', ''))
|
||
return server
|
||
|
||
|
||
def _quote(folder):
|
||
return '"' + folder.replace('\\', '\\\\').replace('"', '\\"') + '"'
|
||
|
||
|
||
def folders(conn):
|
||
names = []
|
||
for line in conn.list()[1] or []:
|
||
m = re.match(rb'\((?P<flags>[^)]*)\) (?:"[^"]*"|NIL) (?P<name>.+)$', line or b'')
|
||
if m and b'\\noselect' not in m.group('flags').lower():
|
||
names.append(m.group('name').decode('utf-8', 'replace').strip().strip('"'))
|
||
return names
|
||
|
||
|
||
def test(cfg):
|
||
"""Verbindingen proberen zonder iets te versturen: [(gelukt, melding)]."""
|
||
out = []
|
||
if cfg.get('imap_host'):
|
||
try:
|
||
conn = _connect_imap(cfg)
|
||
try:
|
||
names = folders(conn)
|
||
folder = cfg.get('imap_folder') or 'INBOX'
|
||
out.append((True, f"IMAP: ingelogd op {cfg['imap_host']}; mappen: {', '.join(names[:12]) or '—'}"))
|
||
if conn.select(_quote(folder), readonly=True)[0] != 'OK':
|
||
out.append((False, f'IMAP: map "{folder}" niet gevonden'))
|
||
sent = cfg.get('smtp_sent_folder')
|
||
if sent and sent not in names:
|
||
out.append((False, f'IMAP: map voor verzonden mail "{sent}" niet gevonden'))
|
||
finally:
|
||
conn.logout()
|
||
except Exception as e:
|
||
out.append((False, f"IMAP: {e or type(e).__name__}"))
|
||
else:
|
||
out.append((False, 'IMAP: geen server ingesteld'))
|
||
try:
|
||
with _connect_smtp(cfg):
|
||
out.append((True, f"SMTP: ingelogd op {cfg['smtp_host']}"))
|
||
except Exception as e:
|
||
out.append((False, f"SMTP: {e or type(e).__name__}"))
|
||
return out
|
||
|
||
|
||
def fetch(cfg, known, limit=200):
|
||
"""Haal berichten van de laatste `imap_days` dagen op die nog niet in `known` staan."""
|
||
folder = cfg.get('imap_folder') or 'INBOX'
|
||
days = int(cfg.get('imap_days') or 30)
|
||
since = datetime.now() - timedelta(days=days)
|
||
since_str = f'{since.day:02d}-{MONTHS[since.month - 1]}-{since.year}'
|
||
|
||
conn = _connect_imap(cfg)
|
||
try:
|
||
typ, _ = conn.select(_quote(folder), readonly=True)
|
||
if typ != 'OK':
|
||
raise RuntimeError(f'Map "{folder}" niet gevonden')
|
||
_, data = conn.uid('search', None, 'SINCE', since_str)
|
||
uids = data[0].split()[-limit:]
|
||
out = []
|
||
for uid in uids:
|
||
key = f'{folder}:{uid.decode()}'
|
||
if key in known:
|
||
continue
|
||
_, msgdata = conn.uid('fetch', uid, '(BODY.PEEK[])')
|
||
if not msgdata or not isinstance(msgdata[0], tuple):
|
||
continue
|
||
msg = email.message_from_bytes(msgdata[0][1], policy=policy)
|
||
name, addr = parseaddr(str(msg.get('From', '')))
|
||
try:
|
||
date = parsedate_to_datetime(msg['Date']).astimezone().strftime('%Y-%m-%d %H:%M:%S')
|
||
except Exception:
|
||
date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
out.append({
|
||
'uid': key,
|
||
'message_id': str(msg.get('Message-ID', '')),
|
||
'from_name': name,
|
||
'from_addr': addr.lower(),
|
||
'to_addr': str(msg.get('To', '')),
|
||
'subject': str(msg.get('Subject', '')) or '(geen onderwerp)',
|
||
'body': _text(msg),
|
||
'date': date,
|
||
})
|
||
return out
|
||
finally:
|
||
try:
|
||
conn.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def send(cfg, to, subject, body, in_reply_to=None, cc='', attachments=()):
|
||
"""Versturen; geeft (Message-ID, waarschuwing of None). Een mislukte kopie in Verzonden is alleen een waarschuwing.
|
||
attachments = [(bestandsnaam, bytes, mimetype)]."""
|
||
user = cfg.get('smtp_user') or cfg.get('imap_user', '')
|
||
sender = cfg.get('smtp_from') or user
|
||
msg = EmailMessage()
|
||
msg['From'] = sender
|
||
msg['To'] = to
|
||
if cc:
|
||
msg['Cc'] = cc
|
||
msg['Subject'] = subject
|
||
msg['Date'] = formatdate(localtime=True)
|
||
domain = parseaddr(sender)[1].rpartition('@')[2] or None
|
||
msg['Message-ID'] = make_msgid(domain=domain)
|
||
if in_reply_to:
|
||
msg['In-Reply-To'] = in_reply_to
|
||
msg['References'] = in_reply_to
|
||
msg.set_content(body)
|
||
for name, data, ctype in attachments:
|
||
maintype, _, subtype = (ctype or 'application/octet-stream').partition('/')
|
||
msg.add_attachment(data, maintype=maintype, subtype=subtype or 'octet-stream', filename=name)
|
||
|
||
with _connect_smtp(cfg) as server:
|
||
server.send_message(msg)
|
||
|
||
warning = None
|
||
folder = cfg.get('smtp_sent_folder')
|
||
if folder and cfg.get('imap_host'):
|
||
try:
|
||
conn = _connect_imap(cfg)
|
||
try:
|
||
typ, data = conn.append(_quote(folder), '\\Seen', imaplib.Time2Internaldate(time.time()), msg.as_bytes())
|
||
if typ != 'OK':
|
||
raise RuntimeError((data[0] or b'').decode('utf-8', 'replace') if data else typ)
|
||
finally:
|
||
conn.logout()
|
||
except Exception as e:
|
||
warning = f'Verstuurd, maar niet bewaard in "{folder}": {e}'
|
||
return msg['Message-ID'], warning
|