← back to Qwen38 Viewer
mail-bridge.py
198 lines
#!/usr/bin/env python3
"""
qwen mail bridge — lets ALLOWLISTED senders email qwen@agentabrams.com and get
an inline reply from the local uncensored qwen3.8-27b-heretic (Ollama).
ARMED per Steve's explicit authorization (2026-08-19).
Hard safety controls (this fronts an UNCENSORED model + sends outbound mail):
* ALLOWLIST-ONLY: replies exclusively to the 4 addresses below. Any other
sender -> marked read, logged, NEVER answered (not an open relay).
* LOOP GUARD: never replies to itself, mailer-daemon, no-reply, or any message
carrying Auto-Submitted / Precedence:bulk / List-* headers.
* RATE CAP: at most MAX_PER_RUN replies per poll, MAX_PER_SENDER_DAY per sender.
* Body + token caps.
Runs once per invocation (launchd StartInterval). Idempotent via IMAP \\Seen.
"""
import os, sys, json, ssl, time, email, imaplib, smtplib, re, pathlib, datetime
from email.message import EmailMessage
from email.utils import parseaddr, formataddr, make_msgid
from urllib import request as urlreq
MAILBOX = "qwen@agentabrams.com"
IMAP_HOST = "imap.purelymail.com"
SMTP_HOST = "smtp.purelymail.com"
SMTP_PORT = 587
OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("MODEL", "qwen3.8-27b-heretic")
ALLOWLIST = {a.lower() for a in [
"info@designerwallcoverings.com",
"steve@designerwallcoverings.com",
"browntwn@gmail.com",
"theagentabrams@gmail.com",
]}
MAX_PER_RUN = 6
MAX_PER_SENDER_DAY = 40
MAX_BODY_CHARS = 6000
NUM_PREDICT = 1200
STATE = pathlib.Path.home() / ".qwen-mail-bridge-state.json"
LOG = pathlib.Path.home() / ".qwen-mail-bridge.log"
def log(m):
line = f"{datetime.datetime.now().isoformat(timespec='seconds')} {m}"
print(line)
try:
with LOG.open("a") as f: f.write(line + "\n")
except OSError: pass
def get_pw():
# read the mailbox password from the master secrets .env (never logged)
p = os.environ.get("QWEN_MAIL_PASS")
if p: return p
envf = pathlib.Path.home() / "Projects/secrets-manager/.env"
for ln in envf.read_text().splitlines():
if ln.startswith("QWEN_MAIL_PASS="):
return ln.split("=", 1)[1].strip().strip('"').strip("'")
raise SystemExit("QWEN_MAIL_PASS not found")
def load_state():
try: return json.loads(STATE.read_text())
except Exception: return {}
def save_state(s):
try: STATE.write_text(json.dumps(s))
except OSError: pass
def is_bulk(msg):
if (msg.get("Auto-Submitted") or "").lower() not in ("", "no"): return True
if (msg.get("Precedence") or "").lower() in ("bulk", "list", "junk"): return True
if msg.get("List-Id") or msg.get("List-Unsubscribe"): return True
return False
def _decode(part):
try: return part.get_content()
except Exception:
pl = part.get_payload(decode=True)
return pl.decode("utf-8", "ignore") if pl else ""
def _strip_html(h):
h = re.sub(r"(?is)<(script|style).*?</\1>", " ", h)
h = re.sub(r"(?s)<[^>]+>", " ", h)
return re.sub(r"[ \t]*\n", "\n", re.sub(r"[ \t]+", " ", h)).strip()
def plain_body(msg):
text, html = "", ""
if msg.is_multipart():
for part in msg.walk():
ct = part.get_content_type()
if "attachment" in str(part.get("Content-Disposition")): continue
if ct == "text/plain" and not text: text = _decode(part)
elif ct == "text/html" and not html: html = _decode(part)
else:
if msg.get_content_type() == "text/html": html = _decode(msg)
else: text = _decode(msg)
return text.strip() or _strip_html(html)
def strip_quotes(body):
out = []
for ln in body.splitlines():
if ln.strip().startswith(">"): continue
if re.match(r"^On .* wrote:$", ln.strip()): break
if ln.strip() in ("--", "-- "): break
out.append(ln)
return "\n".join(out).strip()
def ask_qwen(prompt):
data = json.dumps({"model": MODEL, "messages": [{"role": "user", "content": prompt}],
"stream": False, "keep_alive": "30m",
"options": {"temperature": 0.8, "num_predict": NUM_PREDICT}}).encode()
req = urlreq.Request(f"{OLLAMA}/api/chat", data=data, headers={"Content-Type": "application/json"})
with urlreq.urlopen(req, timeout=300) as r:
j = json.loads(r.read().decode())
return (j.get("message", {}) or {}).get("content", "").strip() or "(no response)"
def main():
pw = get_pw()
state = load_state()
today = datetime.date.today().isoformat()
counts = state.get(today, {})
replied = 0
ctx = ssl.create_default_context()
M = imaplib.IMAP4_SSL(IMAP_HOST, 993, ssl_context=ctx)
M.login(MAILBOX, pw)
M.select("INBOX")
typ, data = M.search(None, "UNSEEN")
ids = data[0].split() if data and data[0] else []
log(f"poll: {len(ids)} unseen")
smtp = None
for num in ids:
if replied >= MAX_PER_RUN:
log("rate: MAX_PER_RUN hit, stopping"); break
typ, md = M.fetch(num, "(RFC822)")
msg = email.message_from_bytes(md[0][1])
frm = parseaddr(msg.get("From", ""))[1].lower()
base = re.sub(r"\+[^@]*@", "@", frm) # strip +tag so plus-addresses can't dodge the self-guard
subj = msg.get("Subject", "(no subject)")
# --- safety gates ---
if base == MAILBOX or "mailer-daemon" in frm or frm.startswith(("no-reply", "noreply")):
log(f"skip loop-guard from={frm}"); M.store(num, "+FLAGS", "\\Seen"); continue
if is_bulk(msg):
log(f"skip bulk/auto from={frm}"); M.store(num, "+FLAGS", "\\Seen"); continue
if frm not in ALLOWLIST:
log(f"IGNORE non-allowlisted from={frm} subj={subj!r}"); M.store(num, "+FLAGS", "\\Seen"); continue
if counts.get(frm, 0) >= MAX_PER_SENDER_DAY:
log(f"skip per-sender cap from={frm}"); M.store(num, "+FLAGS", "\\Seen"); continue
body = strip_quotes(plain_body(msg))[:MAX_BODY_CHARS].strip()
if not body:
log(f"skip empty body from={frm}"); M.store(num, "+FLAGS", "\\Seen"); continue
log(f"ANSWER from={frm} subj={subj!r} chars={len(body)}")
try:
answer = ask_qwen(body)
except Exception as e:
log(f"qwen error: {e}"); continue
reply = EmailMessage()
reply["From"] = formataddr(("Qwen3.8 Heretic", MAILBOX))
reply["To"] = frm
reply["Subject"] = subj if subj.lower().startswith("re:") else f"Re: {subj}"
mid = msg.get("Message-ID")
if mid:
reply["In-Reply-To"] = mid
reply["References"] = ((msg.get("References", "") + " " + mid)).strip()
reply["Message-ID"] = make_msgid(domain="agentabrams.com")
reply["Auto-Submitted"] = "auto-replied" # be a good citizen; prevents remote loops
reply.set_content(answer + "\n\n— qwen3.8-27b-heretic (local, uncensored) via agentabrams.com")
try:
if smtp is None: # connect once, reuse across replies
smtp = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=60)
smtp.starttls(context=ctx); smtp.login(MAILBOX, pw)
smtp.send_message(reply)
M.store(num, "+FLAGS", "\\Seen")
counts[frm] = counts.get(frm, 0) + 1
replied += 1
log(f"replied to {frm} ({len(answer)} chars)")
except Exception as e:
log(f"smtp error to {frm}: {e}")
M.store(num, "+FLAGS", "\\Seen") # mark seen -> no retry storm / dup replies
try: smtp.quit()
except Exception: pass
smtp = None # force fresh connection next time
if smtp is not None:
try: smtp.quit()
except Exception: pass
M.logout()
state = {today: counts} # keep only today's counts
save_state(state)
log(f"done: {replied} replied")
if __name__ == "__main__":
main()