v_9.3
This commit is contained in:
jpmvaz
@@ -0,0 +1,145 @@
|
||||
"""Outbound email via Purelymail, and template rendering.
|
||||
|
||||
Preconfigured for Purelymail — only the account's login and password are needed:
|
||||
|
||||
MAIL_USER=you@yourdomain.com (or you@purelymail.com)
|
||||
MAIL_PASSWORD=... (use an App Password if 2FA is enabled)
|
||||
|
||||
Every message is sent through SMTP (smtp.purelymail.com:465, SSL/TLS) and a
|
||||
copy is saved to the account's Sent folder over IMAP
|
||||
(imap.purelymail.com:993, SSL/TLS). Every attempt — success or failure — is
|
||||
recorded in the mail_log table shown on the admin "Mail status" page.
|
||||
"""
|
||||
import html as html_mod
|
||||
import imaplib
|
||||
import os
|
||||
import smtplib
|
||||
import time
|
||||
from email.message import EmailMessage
|
||||
from email.utils import make_msgid
|
||||
|
||||
import db
|
||||
|
||||
SMTP_HOST = "smtp.purelymail.com"
|
||||
SMTP_PORT = 465 # SSL/TLS
|
||||
IMAP_HOST = "imap.purelymail.com"
|
||||
IMAP_PORT = 993 # SSL/TLS
|
||||
|
||||
SENT_FOLDER_CANDIDATES = ("Sent", "INBOX.Sent", "Sent Messages", "Sent Items")
|
||||
|
||||
|
||||
def render(template: str, **vars) -> str:
|
||||
"""Safe {placeholder} substitution — unknown placeholders are left as-is."""
|
||||
out = template
|
||||
for k, v in vars.items():
|
||||
out = out.replace("{" + k + "}", str(v))
|
||||
return out
|
||||
|
||||
|
||||
def _save_to_sent(user, password, msg) -> str:
|
||||
"""Append a copy of the message to the account's Sent folder. Returns a status string."""
|
||||
try:
|
||||
with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) as imap:
|
||||
imap.login(user, password)
|
||||
raw = msg.as_bytes()
|
||||
stamp = imaplib.Time2Internaldate(time.time())
|
||||
last_err = ""
|
||||
for folder in SENT_FOLDER_CANDIDATES:
|
||||
status, resp = imap.append(folder, r"(\Seen)", stamp, raw)
|
||||
if status == "OK":
|
||||
return f"saved to '{folder}'"
|
||||
last_err = (resp[0] or b"").decode("utf-8", "replace") if resp else ""
|
||||
# No known folder worked — create "Sent" and try once more.
|
||||
imap.create("Sent")
|
||||
status, resp = imap.append("Sent", r"(\Seen)", stamp, raw)
|
||||
if status == "OK":
|
||||
return "saved to 'Sent' (folder created)"
|
||||
return f"failed: {last_err or 'could not append to any Sent folder'}"
|
||||
except Exception as e:
|
||||
return f"failed: {e}"
|
||||
|
||||
|
||||
def _record(request_id, recipients, subject, status, detail, sent_copy):
|
||||
try:
|
||||
con = db.get_db()
|
||||
con.execute(
|
||||
"INSERT INTO mail_log (at, request_id, recipients, subject, status, detail, sent_copy)"
|
||||
" VALUES (?,?,?,?,?,?,?)",
|
||||
(db.now(), request_id, ", ".join(recipients), subject, status, detail, sent_copy))
|
||||
con.commit()
|
||||
con.close()
|
||||
except Exception as e:
|
||||
print(f"[mail] Could not record mail log entry: {e}")
|
||||
|
||||
|
||||
def send_mail(to_addrs, subject, body, request_id=None, attachments=None, inline_images=None):
|
||||
"""Send an email through Purelymail, keep a copy in the Sent folder, and
|
||||
record the attempt in the mail log. Returns (ok, detail).
|
||||
|
||||
attachments: optional list of (filename, bytes, maintype, subtype).
|
||||
inline_images: optional dict {marker_name: (bytes, subtype)}. When given,
|
||||
an HTML alternative is built where each "{image:marker_name}"
|
||||
in the body becomes an embedded <img>; the plain-text part
|
||||
shows "[image: marker_name]" instead.
|
||||
"""
|
||||
if isinstance(to_addrs, str):
|
||||
to_addrs = [to_addrs]
|
||||
to_addrs = [a for a in to_addrs if a]
|
||||
if not to_addrs:
|
||||
return False, "no recipients"
|
||||
|
||||
user = os.environ.get("MAIL_USER", "")
|
||||
password = os.environ.get("MAIL_PASSWORD", "")
|
||||
|
||||
msg = EmailMessage()
|
||||
msg["From"] = user or "approvalflow@localhost"
|
||||
msg["To"] = ", ".join(to_addrs)
|
||||
msg["Subject"] = subject
|
||||
|
||||
if inline_images:
|
||||
text = body
|
||||
for name in inline_images:
|
||||
text = text.replace("{image:" + name + "}", f"[image: {name}]")
|
||||
msg.set_content(text)
|
||||
|
||||
html = html_mod.escape(body).replace("\n", "<br>\n")
|
||||
cids = {}
|
||||
for name in inline_images:
|
||||
cid = make_msgid(domain="martinhal.approvalflow")
|
||||
cids[name] = cid
|
||||
html = html.replace("{image:" + name + "}",
|
||||
f'<img src="cid:{cid[1:-1]}" alt="{html_mod.escape(name)}" '
|
||||
'style="max-width:100%">')
|
||||
msg.add_alternative(
|
||||
f'<div style="font-family:sans-serif;font-size:15px;line-height:1.5">{html}</div>',
|
||||
subtype="html")
|
||||
html_part = msg.get_payload()[-1]
|
||||
for name, (data, subtype) in inline_images.items():
|
||||
html_part.add_related(data, maintype="image", subtype=subtype, cid=cids[name])
|
||||
else:
|
||||
msg.set_content(body)
|
||||
|
||||
for filename, data, maintype, subtype in (attachments or []):
|
||||
msg.add_attachment(data, maintype=maintype, subtype=subtype, filename=filename)
|
||||
|
||||
if not user:
|
||||
# Dev mode: print instead of sending so the app is testable without an account.
|
||||
att = "".join(f"\n[attachment: {f} ({len(b)} bytes)]" for f, b, *_ in (attachments or []))
|
||||
img = "".join(f"\n[inline image: {n} ({len(d)} bytes, {s})]" for n, (d, s) in (inline_images or {}).items())
|
||||
print(f"[mail:DEV] To: {msg['To']}\nSubject: {subject}\n{body}{att}{img}\n{'-'*50}")
|
||||
_record(request_id, to_addrs, subject, "dev",
|
||||
"MAIL_USER not set — printed to console", "skipped (dev mode)")
|
||||
return True, "dev mode (MAIL_USER not set) — printed to console"
|
||||
|
||||
try:
|
||||
with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=30) as server:
|
||||
server.login(user, password)
|
||||
server.send_message(msg)
|
||||
except Exception as e:
|
||||
detail = f"SMTP error: {e}"
|
||||
_record(request_id, to_addrs, subject, "error", detail, "skipped (send failed)")
|
||||
return False, detail
|
||||
|
||||
sent_copy = _save_to_sent(user, password, msg)
|
||||
_record(request_id, to_addrs, subject, "sent", "", sent_copy)
|
||||
return True, f"sent to {', '.join(to_addrs)}"
|
||||
Reference in New Issue
Block a user