Mail Fetch

Inputs

Outputs

Source

python

inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

import imaplib
import re
import socket

def connect():
    host = str(inp.get("imap_host") or "imap.gmail.com")
    port = int(inp.get("imap_port") or 993)
    username = str(inp.get("username") or "").strip()
    # mail_password since 0.2.0; gmail_app_password is the 0.1.x name, read so
    # an agent built on 0.1.x still runs where 0.2.0 is installed.
    password = inp.get("mail_password") or inp.get("gmail_app_password")
    if not username:
        raise ValueError("username is missing; give the mailbox address.")
    if not password:
        raise ValueError("mail_password is missing; reference the stored secret mail-password.")
    if re.search(r"(^|\.)(outlook\.office365\.com|outlook\.com|office365\.com)$", host, re.I):
        raise PermissionError(f"{host} does not take a password over IMAP: Microsoft 365 and Outlook.com sign in with OAuth, which this node does not do.")
    use_ssl = bool(inp.get("use_ssl", True))
    try:
        box = (imaplib.IMAP4_SSL if use_ssl else imaplib.IMAP4)(host, port, timeout=30)
    except (OSError, socket.error) as exc:
        raise ConnectionError(f"Could not connect to {host}:{port}: {exc}")
    try:
        box.login(username, str(password))
    except imaplib.IMAP4.error as exc:
        box.logout()
        raise PermissionError(f"The mail server refused the login for {username}: {exc}")
    return box

def select(box, readonly):
    folder = str(inp.get("folder") or "INBOX")
    # imaplib quotes nothing: "[Gmail]/All Mail" has to arrive in quotes.
    status, data = box.select(quote(folder), readonly=readonly)
    if status != "OK":
        box.logout()
        raise RuntimeError(f"The folder {folder!r} does not exist on the server: {data}")
    return folder

def quote(text):
    return '"' + str(text).replace("\\", "").replace('"', "") + '"'

import email
import email.policy
import email.utils
import datetime
import os
import re
from html.parser import HTMLParser

class _Text(HTMLParser):
    BLOCK = {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6", "table", "ul", "ol", "blockquote", "pre"}
    def __init__(self):
        super().__init__(); self.parts = []; self.skip = 0
    def handle_starttag(self, tag, attrs):
        if tag in ("script", "style", "head"): self.skip += 1
        if tag in self.BLOCK: self.parts.append("\n")
    def handle_endtag(self, tag):
        if tag in ("script", "style", "head"): self.skip = max(0, self.skip - 1)
        if tag in self.BLOCK: self.parts.append("\n")
    def handle_data(self, data):
        if not self.skip: self.parts.append(data)

def html_to_text(html):
    p = _Text(); p.feed(html); p.close()
    lines = [re.sub(r"[ \t\r\f\v]+", " ", l).strip() for l in "".join(p.parts).split("\n")]
    return "\n".join(l for l in lines if l)

def body_text(msg):
    part = msg.get_body(preferencelist=("plain",))
    if part is not None:
        text = part.get_content().strip()
        if text:
            return text
    part = msg.get_body(preferencelist=("html",))
    if part is not None:
        return html_to_text(part.get_content())
    return ""

def criteria(query):
    """unseen since:7 from:acme subject:invoice|receipt text:word → IMAP SEARCH keys.

    A planner writes the same thing several ways (measured 2026-10-06:
    "since:1d", "since:yesterday", "subject:invoice OR subject:receipt"), so
    a day count takes a d/w suffix or a word, and OR between two words is
    the same as | inside one.
    """
    words = re.findall(r'[a-z_]+:"[^"]*"|\S+', str(query or "").strip(), flags=re.I)
    # "a OR b" → "a|b" when both sides name the same field.
    merged = []
    for word in words:
        if word.upper() == "OR" and merged:
            merged.append("OR")
        elif merged and merged[-1] == "OR":
            merged.pop()
            left = merged.pop() if merged else ""
            lf, _, lv = left.partition(":"); rf, _, rv = word.partition(":")
            if lf.lower() == rf.lower() and lv and rv:
                merged.append(f"{lf}:{lv}|{rv}")
            else:
                merged.extend([left, word])
        else:
            merged.append(word)
    keys = []
    for word in merged:
        name, _, value = word.partition(":")
        name = name.lower(); value = value.strip('"')
        if not _:
            if name in ("unseen", "unread"): keys.append("UNSEEN")
            elif name in ("all", ""): keys.append("ALL")
            elif name == "seen": keys.append("SEEN")
            elif name == "or": continue
            else: keys.extend(["TEXT", quote(word)])
            continue
        if name == "since":
            v = value.lower()
            m = re.fullmatch(r"(\d+)\s*(d|day|days|w|week|weeks)?", v)
            if m:
                days = int(m.group(1)) * (7 if (m.group(2) or "d").startswith("w") else 1)
            elif v in ("today",):
                days = 0
            elif v in ("yesterday",):
                days = 1
            else:
                raise ValueError(f"since: takes a number of days, such as since:7, not {value!r}.")
            day = (datetime.date.today() - datetime.timedelta(days=days)).strftime("%d-%b-%Y")
            keys.extend(["SINCE", day])
            continue
        field = {"from": "FROM", "subject": "SUBJECT", "to": "TO", "text": "TEXT", "body": "BODY"}.get(name)
        if not field:
            raise ValueError(f"Unknown query word {word!r}; use unseen, since:<days>, from:, to:, subject: or text:.")
        options = [o for o in value.split("|") if o.strip()]
        if not options:
            raise ValueError(f"{name}: needs a value.")
        expr = [field, quote(options[-1])]
        for option in reversed(options[:-1]):
            expr = ["OR", field, quote(option)] + expr
        keys.extend(expr)
    return keys or ["ALL"]

keys = criteria(inp.get("query", "unseen"))
limit = int(inp.get("limit", 20))
if limit < 1:
    raise ValueError("limit must be at least 1.")
mark_seen = bool(inp.get("mark_seen", False))
save = bool(inp.get("save_attachments", False))
attachments_dir = str(inp.get("attachments_dir") or "mail/attachments")

def safe_name(name):
    name = os.path.basename(str(name or "attachment")).strip() or "attachment"
    return re.sub(r"[^A-Za-z0-9._-]+", "_", name)

def save_attachments(uid, msg):
    saved = []
    if not save:
        return saved
    os.makedirs(attachments_dir, exist_ok=True)
    for part in msg.iter_attachments():
        data = part.get_payload(decode=True)
        if data is None:
            continue
        path = os.path.join(attachments_dir, f"{uid}-{safe_name(part.get_filename())}")
        with open(path, "wb") as handle_:
            handle_.write(data)
        saved.append({"name": part.get_filename() or "", "path": path, "bytes": len(data)})
    return saved

box = connect()
try:
    folder = select(box, readonly=not mark_seen)
    status, data = box.uid("SEARCH", None, *keys)
    if status != "OK":
        raise RuntimeError(f"The server refused the search {' '.join(keys)}: {data}")
    uids = [u.decode() for u in (data[0] or b"").split()]
    uids = list(reversed(uids))[:limit]
    messages = []
    for uid in uids:
        status, data = box.uid("FETCH", uid, "(BODY.PEEK[])" if not mark_seen else "(BODY[])")
        raw = next((item[1] for item in data if isinstance(item, tuple)), None)
        if status != "OK" or raw is None:
            raise RuntimeError(f"The server did not return message {uid}.")
        msg = email.message_from_bytes(raw, policy=email.policy.default)
        try:
            date = email.utils.parsedate_to_datetime(msg.get("Date", "")).isoformat()
        except Exception:
            date = str(msg.get("Date", ""))
        messages.append({
            "uid": uid,
            "from": str(msg.get("From", "")),
            "to": str(msg.get("To", "")),
            "subject": str(msg.get("Subject", "")),
            "date": date,
            "text": body_text(msg),
            "has_attachments": any(True for _ in msg.iter_attachments()),
            "attachments": save_attachments(uid, msg),
        })
        if mark_seen:
            box.uid("STORE", uid, "+FLAGS.SILENT", "(\\Seen)")
finally:
    try:
        box.logout()
    except Exception:
        pass

out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out["messages"] = messages
out["uids"] = [m["uid"] for m in messages]
out["count"] = len(messages)
out["folder"] = folder
out["attachment_paths"] = [a["path"] for m in messages for a in m["attachments"]]

Tests

Requires: python:3.9

• unseen

  The unread messages, newest first; HTML becomes text and an attachment is counted.

• subject-either

  subject:invoice|receipt matches either word, in any folder state.

• from-since-limit

  from: narrows to a sender; since: and limit: cut the list.

• save-attachments

  With save_attachments the receipt's PDF is written and its path returned.

• planner-spellings

  The forms a planner writes: a OR b between fields, since: with a unit or a word.

• old-field-name

  The 0.1.x name of the password is still read, so an agent built on it still runs.

• microsoft-365

  Microsoft 365 takes no password over IMAP, and the refusal says so before connecting.

• bad-password

  A refused login says so and names the account.

• no-password

  A missing password is refused before connecting.

• bad-query

  An unknown query word is refused before connecting.