Spaces:
Sleeping
Sleeping
| """ | |
| What happened to a message after the mailbox's server accepted it. | |
| SMTP only tells us the provider took the message. A provider may then hold it for an hour or more | |
| (traffic, throttling) and a person reading "sent" cannot tell that from a lost message. The | |
| mailbox's IMAP side shows what comes back: delay notices, bounce notices and replies. This module | |
| reads those and records one state per recipient: | |
| accepted The provider took it. Nothing has come back. This is all SMTP can prove. | |
| delayed The provider (or the next server) says it is still trying. | |
| bounced A failure notice came back for this message. | |
| replied The person answered. | |
| IMAP cannot prove a message reached an inbox, so nothing here says "delivered". After a day with | |
| nothing reported, the screen says "no problems reported", which is what is actually known. | |
| """ | |
| from __future__ import annotations | |
| import email | |
| import imaplib | |
| import logging | |
| import re | |
| import ssl | |
| from datetime import datetime, timedelta, timezone | |
| from email import policy | |
| from backend import firestore_client as db | |
| log = logging.getLogger(__name__) | |
| ACCEPTED, DELAYED, BOUNCED, REPLIED = "accepted", "delayed", "bounced", "replied" | |
| # A later state replaces an earlier one; a bounce is never overwritten by a delay notice. | |
| _RANK = {ACCEPTED: 0, DELAYED: 1, REPLIED: 2, BOUNCED: 3} | |
| WATCH_DAYS = 3 | |
| RECHECK_MINUTES = 10 | |
| QUIET_AFTER_HOURS = 24 | |
| MAX_MESSAGES_READ = 400 | |
| IMAP_PORT = 993 | |
| _ID = re.compile(r"<([^<>\s]+)>") | |
| _DELAY_WORDS = re.compile(r"delay|still trying|will keep trying|not yet delivered|warning", re.I) | |
| _BOUNCE_WORDS = re.compile(r"undeliver|failure|failed|returned|bounce|could not be delivered|couldn't be delivered|rejected|no such user", re.I) | |
| _BOUNCE_SENDERS = re.compile(r"mailer-daemon|postmaster|mail delivery", re.I) | |
| def imap_details(credentials: dict) -> dict | None: | |
| """The IMAP server and sign-in saved for a mailbox, or None when none was given. | |
| The IMAP user and password default to the SMTP ones, which is usual for one mailbox.""" | |
| host = str(credentials.get("imap_host") or "").strip() | |
| if not host: | |
| return None | |
| return { | |
| "host": host, | |
| "port": int(credentials.get("imap_port") or IMAP_PORT), | |
| "username": str(credentials.get("imap_username") or credentials.get("username") or "").strip(), | |
| "password": "".join(str(credentials.get("imap_password") or credentials.get("password") or "").split()), | |
| } | |
| def _ids_in(text: str) -> set[str]: | |
| return {m.group(1) for m in _ID.finditer(text or "")} | |
| def _text_of(msg) -> str: | |
| """Every readable part of a notice, including the returned original, for finding our ids.""" | |
| out = [] | |
| for part in msg.walk(): | |
| if part.get_content_type() in ("message/rfc822", "text/rfc822-headers", "message/delivery-status"): | |
| for sub in part.iter_parts() if part.is_multipart() else []: | |
| out.append(str(sub)) | |
| elif part.get_content_maintype() == "text": | |
| try: | |
| out.append(part.get_content()) | |
| except (LookupError, UnicodeDecodeError): | |
| payload = part.get_payload(decode=True) or b"" | |
| out.append(payload.decode("utf-8", errors="replace")) | |
| return "\n".join(out) | |
| def classify(raw: bytes, known_ids: set[str]) -> list[tuple[str, str]]: | |
| """Reads one message from the mailbox and returns [(our message id, state)] for each of our | |
| messages it is about. Ordinary mail that is about none of them returns [].""" | |
| try: | |
| msg = email.message_from_bytes(raw, policy=policy.default) | |
| except Exception: # a malformed message is skipped, never fatal | |
| return [] | |
| subject = str(msg.get("Subject", "")) | |
| sender = str(msg.get("From", "")) | |
| notice = ( | |
| msg.get_content_type() == "multipart/report" | |
| or bool(_BOUNCE_SENDERS.search(sender)) | |
| or bool(re.search(r"delivery status|undeliver|returned mail|failure notice", subject, re.I)) | |
| ) | |
| if notice: | |
| body = _text_of(msg) | |
| mentioned = _ids_in(body + "\n" + subject) & known_ids | |
| if not mentioned: | |
| return [] | |
| action = re.search(r"^Action:\s*(\w+)", body, re.I | re.M) | |
| verdict = (action.group(1).lower() if action else "") | |
| if verdict == "delayed": | |
| state = DELAYED | |
| elif verdict == "failed": | |
| state = BOUNCED | |
| elif _BOUNCE_WORDS.search(subject): | |
| state = BOUNCED | |
| elif _DELAY_WORDS.search(subject): | |
| state = DELAYED | |
| else: | |
| state = BOUNCED | |
| return [(mid, state) for mid in mentioned] | |
| # Out-of-office and other automatic answers are not a person replying. | |
| auto = str(msg.get("Auto-Submitted", "no")).lower() | |
| if auto != "no" or str(msg.get("Precedence", "")).lower() in ("bulk", "junk", "auto_reply"): | |
| return [] | |
| replied_to = _ids_in(str(msg.get("In-Reply-To", "")) + " " + str(msg.get("References", ""))) & known_ids | |
| return [(mid, REPLIED) for mid in replied_to] | |
| class ImapInbox: | |
| """A short read-only visit to a mailbox's IMAP server.""" | |
| def __init__(self, details: dict): | |
| self._imap = imaplib.IMAP4_SSL(details["host"], details["port"], ssl_context=ssl.create_default_context(), timeout=20) | |
| self._imap.login(details["username"], details["password"]) | |
| def recent(self, since: datetime, limit: int = MAX_MESSAGES_READ) -> list[bytes]: | |
| """Raw messages that arrived since `since`, newest first, from the inbox and junk folders.""" | |
| out: list[bytes] = [] | |
| for folder in self._folders(): | |
| status, _ = self._imap.select(f'"{folder}"', readonly=True) | |
| if status != "OK": | |
| continue | |
| status, data = self._imap.search(None, "SINCE", since.strftime("%d-%b-%Y")) | |
| if status != "OK" or not data or not data[0]: | |
| continue | |
| for num in reversed(data[0].split()[-limit:]): | |
| status, parts = self._imap.fetch(num, "(BODY.PEEK[])") | |
| if status == "OK" and parts and isinstance(parts[0], tuple): | |
| out.append(parts[0][1]) | |
| if len(out) >= limit: | |
| return out | |
| return out | |
| def _folders(self) -> list[str]: | |
| names = ["INBOX"] | |
| status, lines = self._imap.list() | |
| if status == "OK": | |
| for line in lines or []: | |
| text = line.decode(errors="replace") if isinstance(line, bytes) else str(line) | |
| if "\\junk" in text.lower(): | |
| name = text.rsplit(' "', 1)[-1].strip('"') if ' "' in text else text.split()[-1] | |
| names.append(name.strip('"')) | |
| return names | |
| def close(self) -> None: | |
| try: | |
| self._imap.logout() | |
| except (imaplib.IMAP4.error, OSError): | |
| pass | |
| def open_inbox(details: dict) -> ImapInbox: | |
| """Separate so tests can stand in for a mailbox.""" | |
| return ImapInbox(details) | |
| def _watch_path(org_id: str, send_id: str) -> str: | |
| return f"watch_sends/{org_id}__{send_id}" | |
| def watch(org_id: str, send_id: str) -> None: | |
| """Asks for this send's recipients to be checked for a few days.""" | |
| until = (datetime.now(timezone.utc) + timedelta(days=WATCH_DAYS)).isoformat() | |
| db.set_doc(_watch_path(org_id, send_id), {"org_id": org_id, "send_id": send_id, "until": until}) | |
| def shown_state(record: dict, now: datetime | None = None, tracking: bool = True) -> dict: | |
| """What to show for one recipient: {state, label, note}. `tracking` is False when the mailbox | |
| has no IMAP details, which is said plainly instead of implying all is well.""" | |
| now = now or datetime.now(timezone.utc) | |
| status = record.get("status") | |
| if status != "sent": | |
| return {"state": status, "label": None, "note": None} | |
| state = record.get("delivery") or ACCEPTED | |
| if state == BOUNCED: | |
| return {"state": state, "label": "Bounced", "note": record.get("delivery_note") or "The receiving server sent it back."} | |
| if state == REPLIED: | |
| return {"state": state, "label": "Replied", "note": None} | |
| if state == DELAYED: | |
| return {"state": state, "label": "Delayed at your provider", "note": "Your provider or the receiving server is still trying. This is not an error."} | |
| if not tracking: | |
| return {"state": state, "label": "Accepted by your provider", "note": "Add the IMAP details to this mailbox to see delays, bounces and replies."} | |
| try: | |
| age = now - datetime.fromisoformat(record.get("at")) | |
| except (TypeError, ValueError): | |
| age = timedelta(0) | |
| if age >= timedelta(hours=QUIET_AFTER_HOURS) and record.get("delivery_checked"): | |
| return {"state": state, "label": "Accepted, no problems reported", "note": None} | |
| return {"state": state, "label": "Accepted by your provider", "note": "Providers sometimes hold mail for an hour or more before passing it on."} | |
| def check_send(org_id: str, send_id: str, now: datetime | None = None) -> dict: | |
| """Reads the mailbox's IMAP side once and updates each recipient of this send. Returns | |
| {checked, bounced, delayed, replied, reason}. Never raises for a mailbox problem.""" | |
| from backend import mailboxes, sending # imported here: both import this module's neighbours | |
| now = now or datetime.now(timezone.utc) | |
| send = sending.get_send(org_id, send_id) | |
| records = send.get("per_recipient") or {} | |
| wanted = {r["reference"].strip("<>"): row_id for row_id, r in records.items() if r.get("status") == "sent" and r.get("reference")} | |
| result = {"checked": False, "bounced": 0, "delayed": 0, "replied": 0, "reason": ""} | |
| if not wanted: | |
| result["reason"] = "nothing_to_check" | |
| return result | |
| try: | |
| credentials = mailboxes.get_decrypted_credentials(org_id, send["mailbox_connection_id"]) | |
| except Exception: | |
| result["reason"] = "mailbox_missing" | |
| return result | |
| details = imap_details(credentials) | |
| if details is None: | |
| result["reason"] = "imap_missing" | |
| return result | |
| try: | |
| started = min(datetime.fromisoformat(r["at"]) for r in records.values() if r.get("at")) | |
| except ValueError: | |
| started = now - timedelta(days=1) | |
| inbox = None | |
| try: | |
| inbox = open_inbox(details) | |
| raws = inbox.recent(started - timedelta(days=1)) | |
| except (imaplib.IMAP4.error, OSError) as exc: | |
| log.info("IMAP check failed for %s: %r", send_id, exc) | |
| result["reason"] = "imap_failed" | |
| return result | |
| finally: | |
| if inbox is not None: | |
| inbox.close() | |
| found: dict[str, str] = {} | |
| for raw in raws: | |
| for mid, state in classify(raw, set(wanted)): | |
| if _RANK[state] > _RANK.get(found.get(mid, ACCEPTED), 0): | |
| found[mid] = state | |
| stamp = now.isoformat() | |
| for mid, row_id in wanted.items(): | |
| record = dict(records[row_id]) | |
| record["delivery_checked"] = stamp | |
| state = found.get(mid) | |
| if state and _RANK[state] >= _RANK.get(record.get("delivery") or ACCEPTED, 0): | |
| record["delivery"] = state | |
| record["delivery_at"] = stamp | |
| result[state] = result.get(state, 0) + 1 | |
| records[row_id] = record | |
| db.update_doc(f"orgs/{org_id}/sends/{send_id}", {"per_recipient": records}) | |
| result["checked"] = True | |
| return result | |
| def run_due(now: datetime | None = None, budget_seconds: float = 60) -> dict: | |
| """Checks watched sends that have not been looked at for a few minutes. Real mailboxes only.""" | |
| import time | |
| from backend import mailer_client # here: mailer_client imports the SMTP mailer, which imports this module | |
| if not mailer_client.real_send_enabled(): | |
| return {"checked": 0} | |
| now = now or datetime.now(timezone.utc) | |
| started = time.monotonic() | |
| done = 0 | |
| for path, doc in list(db.query_prefix("watch_sends/").items()): | |
| if time.monotonic() - started > budget_seconds: | |
| break | |
| if datetime.fromisoformat(doc["until"]) < now: | |
| db.delete_doc(path) | |
| continue | |
| last = doc.get("last_check") | |
| if last and now - datetime.fromisoformat(last) < timedelta(minutes=RECHECK_MINUTES): | |
| continue | |
| try: | |
| check_send(doc["org_id"], doc["send_id"], now) | |
| except Exception: # one bad send must not stop the rest | |
| log.exception("Delivery check failed for %s", doc.get("send_id")) | |
| db.update_doc(path, {"last_check": now.isoformat()}) | |
| done += 1 | |
| return {"checked": done} | |