Scripts for email communication with Codex

scripts/README.md

@@ -21,6 +21,18 @@ HTTP MCP endpoints and usage notes for controlling imagej-elphel via MCP.
 
 Path: `scripts/mcp-http-howto.md`
 
+### email_send.py
+
+Send SMTP email via mail.elphel.com using env vars for credentials.
+
+Path: `scripts/email_send.py`
+
+### email_fetch.py
+
+Fetch email via IMAP (mail.elphel.com) and save messages to files.
+
+Path: `scripts/email_fetch.py`
+
 ---
 
 ## mcp_filter_list_by_model_version.py

scripts/catalog.json

@@ -5,6 +5,16 @@
       "name": "mcp-http-howto.md",
       "path": "scripts/mcp-http-howto.md",
       "purpose": "HTTP MCP endpoints and usage notes for controlling imagej-elphel via MCP."
+    },
+    {
+      "name": "email_send.py",
+      "path": "scripts/email_send.py",
+      "purpose": "Send SMTP email via mail.elphel.com using env vars for credentials."
+    },
+    {
+      "name": "email_fetch.py",
+      "path": "scripts/email_fetch.py",
+      "purpose": "Fetch email via IMAP (mail.elphel.com) and save messages to files."
     }
   ],
   "scripts": [

scripts/email_fetch.py

+#!/usr/bin/env python3
+import email
+import imaplib
+import os
+from email.header import decode_header
+from pathlib import Path
+
+IMAP_HOST = os.environ.get("ELPHEL_IMAP_HOST", "mail.elphel.com")
+IMAP_PORT = int(os.environ.get("ELPHEL_IMAP_PORT", "993"))
+IMAP_USER = os.environ.get("ELPHEL_IMAP_USER", "codex@elphel.com")
+IMAP_PASS = os.environ.get("ELPHEL_IMAP_PASS")
+MAILBOX = os.environ.get("ELPHEL_IMAP_BOX", "INBOX")
+OUT_DIR = Path(os.environ.get("EMAIL_OUT_DIR", "attic/email_inbox"))
+SEARCH = os.environ.get("EMAIL_SEARCH", "ALL")
+
+
+def _decode(value):
+    if value is None:
+        return ""
+    parts = decode_header(value)
+    out = ""
+    for text, enc in parts:
+        if isinstance(text, bytes):
+            out += text.decode(enc or "utf-8", errors="replace")
+        else:
+            out += text
+    return out
+
+
+def _safe_subject(subject):
+    return "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in subject)[:80] or "message"
+
+
+def _write_index(index_path, values):
+    line = ",".join(v.replace(",", " ") for v in values)
+    index_path.parent.mkdir(parents=True, exist_ok=True)
+    with index_path.open("a", encoding="utf-8") as f:
+        f.write(line + "\n")
+
+
+def fetch():
+    if not IMAP_PASS:
+        raise SystemExit("Set ELPHEL_IMAP_PASS")
+    with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) as imap:
+        imap.login(IMAP_USER, IMAP_PASS)
+        imap.select(MAILBOX)
+        status, data = imap.search(None, SEARCH)
+        if status != "OK":
+            raise SystemExit("IMAP search failed")
+        ids = data[0].split()
+        for idx, msg_id in enumerate(ids, start=1):
+            status, msg_data = imap.fetch(msg_id, "(RFC822)")
+            if status != "OK":
+                continue
+            raw = msg_data[0][1]
+            msg = email.message_from_bytes(raw)
+            subject = _decode(msg.get("Subject"))
+            date = _decode(msg.get("Date"))
+            from_ = _decode(msg.get("From"))
+            to_ = _decode(msg.get("To"))
+            cc_ = _decode(msg.get("Cc"))
+            date_dir = OUT_DIR / (date[:10].replace(" ", "_") if date else "unknown-date")
+            date_dir.mkdir(parents=True, exist_ok=True)
+            seq = f"{idx:04d}"
+            fname = f"{seq}_{_safe_subject(subject)}.eml"
+            out_path = date_dir / fname
+            out_path.write_bytes(raw)
+            meta_path = date_dir / f"{seq}_meta.txt"
+            meta_path.write_text(
+                f"From: {from_}\nTo: {to_}\nCc: {cc_}\nDate: {date}\nSubject: {subject}\n",
+                encoding="utf-8",
+            )
+            _write_index(OUT_DIR / "index.csv", [date, from_, to_, cc_, subject, str(out_path)])
+
+
+if __name__ == "__main__":
+    fetch()

scripts/email_send.py

+#!/usr/bin/env python3
+import os
+import smtplib
+import email.utils
+from email.message import EmailMessage
+from pathlib import Path
+
+SMTP_HOST = os.environ.get("ELPHEL_SMTP_HOST", "mail.elphel.com")
+SMTP_PORT = int(os.environ.get("ELPHEL_SMTP_PORT", "465"))
+SMTP_USER = os.environ.get("ELPHEL_SMTP_USER", "codex@elphel.com")
+SMTP_PASS = os.environ.get("ELPHEL_SMTP_PASS")
+OUT_DIR = os.environ.get("EMAIL_OUT_DIR", "attic/email_outbox")
+
+
+def _safe_subject(subject):
+    return "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in subject)[:80] or "message"
+
+
+def _write_index(index_path, values):
+    line = ",".join(v.replace(",", " ") for v in values)
+    index_path.parent.mkdir(parents=True, exist_ok=True)
+    with index_path.open("a", encoding="utf-8") as f:
+        f.write(line + "\n")
+
+
+def send_email(subject, body, to_addrs, cc_addrs=None):
+    if not SMTP_PASS:
+        raise SystemExit("Set ELPHEL_SMTP_PASS")
+    msg = EmailMessage()
+    msg["Subject"] = subject
+    msg["From"] = SMTP_USER
+    msg["To"] = ", ".join(to_addrs)
+    if cc_addrs:
+        msg["Cc"] = ", ".join(cc_addrs)
+    msg.set_content(body)
+
+    recipients = list(to_addrs) + (list(cc_addrs) if cc_addrs else [])
+    out_dir = Path(OUT_DIR)
+    date_dir = out_dir / email.utils.formatdate(localtime=True)[:10].replace(" ", "_")
+    date_dir.mkdir(parents=True, exist_ok=True)
+    safe_subject = _safe_subject(subject)
+    seq = f"{len(list(date_dir.glob('*.eml'))) + 1:04d}"
+    out_path = date_dir / f"{seq}_{safe_subject}.eml"
+    out_path.write_bytes(msg.as_bytes())
+    meta_path = date_dir / f"{seq}_meta.txt"
+    meta_path.write_text(
+        f"From: {SMTP_USER}\nTo: {', '.join(to_addrs)}\nCc: {', '.join(cc_addrs or [])}\nSubject: {subject}\n",
+        encoding="utf-8",
+    )
+    _write_index(out_dir / "index.csv", [email.utils.formatdate(localtime=True), SMTP_USER, ",".join(to_addrs), ",".join(cc_addrs or []), subject, str(out_path)])
+    with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) as s:
+        s.login(SMTP_USER, SMTP_PASS)
+        s.send_message(msg, to_addrs=recipients)
+
+
+def main():
+    subject = os.environ.get("EMAIL_SUBJECT", "MCP access test")
+    body = os.environ.get("EMAIL_BODY", "Hi,\n\nTest message.\n")
+    to_addrs = os.environ.get("EMAIL_TO", "").split(",")
+    cc_addrs = os.environ.get("EMAIL_CC", "").split(",") if os.environ.get("EMAIL_CC") else []
+    to_addrs = [a.strip() for a in to_addrs if a.strip()]
+    cc_addrs = [a.strip() for a in cc_addrs if a.strip()]
+    if not to_addrs:
+        raise SystemExit("Set EMAIL_TO=addr1,addr2")
+    send_email(subject, body, to_addrs, cc_addrs)
+
+
+if __name__ == "__main__":
+    main()