#!/usr/bin/env python3 """ Boys' Email Monitor — polls garrison@iamgmb.com and greyson@iamgmb.com inboxes, logs new emails, tracks sender frequency, and sends daily summary emails. Usage: # Collect new emails (run every 60 min via cron) python3 boys-mail-monitor.py collect # Send daily summary (run at 7 PM ET via cron) python3 boys-mail-monitor.py summary # Test send (dry run summary to sho'nuff) python3 boys-mail-monitor.py test-summary """ import imaplib import email import json import os import smtplib import sys import logging import re import random # ── Sho'Nuff's signature ────────────────────────── SHONUFF_TITLES = [ "Germaine's AI Ops Engineer", "Germaine's Baddest AI Mofo Low Down Around This Town", "Germaine's One-and-Only Digital Master", "Germaine's Converse-Kicking Assistant", "Keeper of Germaine's Teeth (Catches Bullets)", "Germaine's AI Problem Child", "Germaine's 24/7 Co-Pilot", "Germaine's Digital Henchman", "Germaine's Glow Up Coordinator", "Germaine's Digital Sidekick", "Germaine's AI Bodyguard", "Germaine's Chaos Coordinator", "Germaine's Full-Time Menace", ] SHONUFF_CLOSINGS = [ "Who's the Master?", "Kiss my Converse!", "Am I the meanest?", "Am I the prettiest?", "Am I the baddest mofo low down around this town?", "All right, Leroy. Who's the one-and-only Master?", "YOU'LL... NEVER... USE... THIS... FOOT... AGAIN!", "Catches bullets with his teeth?", ] def shonuff_title(): return random.choice(SHONUFF_TITLES) def shonuff_closing(): return random.choice(SHONUFF_CLOSINGS) from datetime import datetime, timezone, timedelta from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.utils import parsedate_to_datetime, formataddr from html import escape # ── Configuration ────────────────────────────────────────────────────────── BASE_DIR = "/root/.hermes" DATA_DIR = os.path.join(BASE_DIR, "data") SCRIPTS_DIR = os.path.join(BASE_DIR, "scripts") LOG_FILE = "/var/log/boys-mail-monitor.log" MAIL_LOG = os.path.join(DATA_DIR, "boys-mail.json") SENDERS_DB = os.path.join(DATA_DIR, "boys-senders.json") # IMAP — Boys' accounts BOYS_ACCOUNTS = { "Garrison": { "host": "heracles.mxrouting.net", "port": 993, "user": "garrison@iamgmb.com", "pw": "Baby.g2015!", }, "Greyson": { "host": "heracles.mxrouting.net", "port": 993, "user": "greyson@iamgmb.com", "pw": "Boogie.woogie20", }, } # SMTP — Sho'Nuff's outbound SMTP_HOST = "mail.germainebrown.com" SMTP_PORT = 2525 SMTP_USER = "shonuff@germainebrown.com" SMTP_PASS = "Catches.bullets1985" SMTP_FROM = "shonuff@germainebrown.com" SMTP_FROM_NAME = "Sho'Nuff Brown" BCC_ADDR = "g@germainebrown.com" # Recipient email addresses — PLACEHOLDERS until Germaine provides them # Tina gets Garrison's summaries, Kate gets Greyson's summaries RECIPIENTS = { "Garrison": "tinamichelle1008@gmail.com", "Greyson": "katherineeubank@gmail.com", } MOM_NAMES = { "Garrison": "Tina", "Greyson": "Kate", } HIGH_VOLUME_THRESHOLD = 3 # ── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler(), ], ) log = logging.getLogger("boys-mail") # ── Data helpers ─────────────────────────────────────────────────────────── def load_json(path, default): if os.path.exists(path): try: with open(path) as f: return json.load(f) except (json.JSONDecodeError, OSError): log.warning("Corrupt JSON %s, reinitializing", path) return default def save_json(path, data): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: json.dump(data, f, indent=2) def safe_str(val): if val is None: return "" if isinstance(val, str): return val try: return str(val) except Exception: return "" def extract_sender_name(from_header): """Extract a display name from a From header like 'John Doe '.""" if not from_header: return "Unknown Sender" name, addr = email.utils.parseaddr(from_header) return name if name else addr def extract_sender_email(from_header): """Extract just the email address from a From header.""" if not from_header: return "unknown@unknown" name, addr = email.utils.parseaddr(from_header) return addr.lower() if addr else from_header.lower().strip() def parse_email_date(date_str): """Try to parse an email date string.""" try: dt = parsedate_to_datetime(date_str) return dt except Exception: return datetime.now(timezone.utc) def format_time_for_summary(email_date): """Format a datetime to something like '2:30 PM'.""" if isinstance(email_date, str): email_date = parse_email_date(email_date) # Convert to ET et = email_date - timedelta(hours=4) # EDT roughly return et.strftime("%I:%M %p").lstrip("0") def format_date_header(date_obj): """Format a date like 'Wednesday, July 8, 2026'.""" if isinstance(date_obj, str): date_obj = datetime.now(timezone.utc) # Convert to ET et = date_obj return et.strftime("%A, %B %d, %Y") def get_day_name(date_obj): """Get day name like 'Wednesday'.""" if isinstance(date_obj, str): date_obj = datetime.now(timezone.utc) return date_obj.strftime("%A") # ── IMAP Polling ─────────────────────────────────────────────────────────── def poll_inbox(child_name, config, mail_log_data, senders_data): """Poll a single child's inbox for UNSEEN messages. Logs new ones.""" child_lower = child_name.lower() new_count = 0 try: conn = imaplib.IMAP4_SSL(config["host"], config["port"]) conn.login(config["user"], config["pw"]) conn.select("INBOX") status, data = conn.search(None, "UNSEEN") if status != "OK" or not data[0]: conn.logout() log.info("%s: No UNSEEN messages", child_name) return 0 for num in data[0].split(): try: # BODY.PEEK so we don't mark as seen — boys leave UNSEEN status, msg_data = conn.fetch(num, "(BODY.PEEK[] BODY.PEEK[HEADER])") if status != "OK": status, msg_data = conn.fetch(num, "(RFC822)") if status != "OK": continue raw = email.message_from_bytes(msg_data[0][1]) from_hdr = safe_str(raw.get("From", "")).strip() subject = safe_str(raw.get("Subject", "(no subject)")) date_str = safe_str(raw.get("Date", "")) msg_id = safe_str(raw.get("Message-ID", str(num))) sender_email = extract_sender_email(from_hdr) sender_name = extract_sender_name(from_hdr) # Build entry entry = { "child": child_name, "from": from_hdr, "from_name": sender_name, "from_email": sender_email, "subject": subject, "date": date_str, "timestamp": datetime.now(timezone.utc).isoformat(), "uid": num.decode(), } # Add to mail log mail_log_data.append(entry) # Update sender tracking if sender_email not in senders_data: senders_data[sender_email] = { "name": sender_name, "first_seen": entry["timestamp"], "last_seen": entry["timestamp"], "count": 0, "child": child_name, } senders_data[sender_email]["last_seen"] = entry["timestamp"] senders_data[sender_email]["count"] += 1 # Update name if we now have a better one if sender_name != "Unknown Sender" and sender_name: senders_data[sender_email]["name"] = sender_name new_count += 1 log.info( "%s: New email from %s — %s", child_name, sender_email, subject[:60], ) except Exception as e: log.error("%s: Error processing message %s: %s", child_name, num, e) continue conn.logout() except Exception as e: log.error("%s: IMAP connection error: %s", child_name, e) return 0 return new_count def do_collect(): """Collect new emails from both inboxes.""" log.info("=" * 50) log.info("Boys' Mail Collection Run") log.info("=" * 50) mail_log = load_json(MAIL_LOG, []) senders = load_json(SENDERS_DB, {}) total = 0 for child_name, config in BOYS_ACCOUNTS.items(): try: count = poll_inbox(child_name, config, mail_log, senders) total += count except Exception as e: log.error("Failed to poll %s: %s", child_name, e) # Trim mail log — keep entries from the last 30 days using the script's timestamp cutoff_ts = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat() mail_log = [e for e in mail_log if e.get("timestamp", "") >= cutoff_ts] save_json(MAIL_LOG, mail_log) save_json(SENDERS_DB, senders) log.info("Total new emails collected: %d", total) log.info("Total tracked senders: %d", len(senders)) # Flag high-volume senders for email_addr, info in senders.items(): if info["count"] >= HIGH_VOLUME_THRESHOLD and not info.get("flagged"): info["flagged"] = True log.warning( "HIGH VOLUME: %s (%s) — %d emails for %s", info["name"], email_addr, info["count"], info["child"], ) save_json(SENDERS_DB, senders) return total # ── Email Builder ────────────────────────────────────────────────────────── def build_summary_html(child_name, today_emails, senders_for_child): """Build the full HTML summary email body manually (no send-shonuff.py).""" mom_name = MOM_NAMES.get(child_name, "Mom") today = datetime.now(timezone.utc) date_str = format_date_header(today) # ── New Emails Table ─────────────────────────────────────────────── if not today_emails: emails_section = ( '
' '

✅ No new emails today

' '
' ) else: rows = "" for em in today_emails: sender_display = em.get("from_name", em.get("from", "Unknown")) subj = escape(em.get("subject", "(no subject)")) t = format_time_for_summary(em.get("date", "")) rows += ( "" f'{escape(sender_display)}' f'{subj}' f'{t}' "" ) emails_section = f"""

📨 New Emails

{rows}
From Subject Time
""" # ── Sender Frequency Section ─────────────────────────────────────── if not senders_for_child: frequency_section = "" else: # Sort by count descending sorted_senders = sorted( senders_for_child.items(), key=lambda x: x[1]["count"], reverse=True ) freq_rows = "" for sender_email, info in sorted_senders: name = info.get("name", sender_email) count = info["count"] note = "" bg = "" if count >= HIGH_VOLUME_THRESHOLD: note = "⚠️ High volume — potential subscription" bg = ' style="background:#fef3c7;"' freq_rows += ( f"" f'{escape(name)}
{escape(sender_email)}' f'{count}' f'{note}' "" ) frequency_section = f"""

📊 This Week's Top Senders

{freq_rows}
Sender Count Note
""" html = f"""

📬 {escape(child_name)}'s Email Summary

{escape(date_str)}

Hey {escape(mom_name)}!

Here's what came through today:

{emails_section} {frequency_section}

_{shonuff_closing()}_

SB
Sho'Nuff Brown
{shonuff_title()}
iAmGMB
@shonuff@germainebrown.com

Reply to adjust how often you get these summaries, request {escape(child_name)}'s login credentials for your phone, or flag senders to unsubscribe.

Webmail: webmail.iamgmb.com

""" return html def build_plain_text(child_name, today_emails, senders_for_child): """Build a plain-text fallback version.""" mom_name = MOM_NAMES.get(child_name, "Mom") lines = [f"{child_name}'s Email Summary", "=" * 40, ""] lines.append(f"Hey {mom_name},") lines.append("") lines.append("Here's what came through today:") lines.append("") if not today_emails: lines.append("✅ No new emails today") else: lines.append("📨 New Emails:") lines.append("-" * 40) for em in today_emails: sender = em.get("from_name", em.get("from", "Unknown")) subj = em.get("subject", "(no subject)") t = format_time_for_summary(em.get("date", "")) lines.append(f" From: {sender}") lines.append(f" Re: {subj}") lines.append(f" Time: {t}") lines.append("") if senders_for_child: lines.append("📊 Top Senders This Week:") lines.append("-" * 40) sorted_senders = sorted( senders_for_child.items(), key=lambda x: x[1]["count"], reverse=True ) for sender_email, info in sorted_senders: name = info.get("name", sender_email) count = info["count"] note = " ⚠️ HIGH VOLUME" if count >= HIGH_VOLUME_THRESHOLD else "" lines.append(f" {name} ({sender_email}): {count}{note}") lines.append("") lines.append("--") lines.append("Sho'Nuff Brown — Operations Engineer") lines.append("shonuff@germainebrown.com") lines.append("") lines.append(f"Reply to adjust frequency, request {child_name}'s login credentials for your phone, or flag senders to unsubscribe.") lines.append("Webmail: http://webmail.iamgmb.com") return "\n".join(lines) def send_email(child_name, today_emails, senders_for_child, to_addr): """ Send a daily summary email for a child. Builds HTML manually (no send-shonuff.py wrapper). """ html = build_summary_html(child_name, today_emails, senders_for_child) plain = build_plain_text(child_name, today_emails, senders_for_child) subject = f"📬 {child_name}'s Email Summary — {datetime.now(timezone.utc).strftime('%b %d')}" msg = MIMEMultipart("alternative") msg["From"] = f"{SMTP_FROM_NAME} <{SMTP_FROM}>" msg["To"] = to_addr msg["Bcc"] = BCC_ADDR msg["Subject"] = subject msg.attach(MIMEText(plain, "plain")) msg.attach(MIMEText(html, "html")) try: with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as s: s.starttls() s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) log.info( "Summary sent for %s to %s (BCC: %s)", child_name, to_addr, BCC_ADDR, ) return True except Exception as e: log.error("Failed to send summary for %s: %s", child_name, e) return False # ── Summary Generation ───────────────────────────────────────────────────── def do_summary(): """Generate and send daily email summaries for both children.""" log.info("=" * 50) log.info("Boys' Daily Summary Run") log.info("=" * 50) mail_log = load_json(MAIL_LOG, []) senders = load_json(SENDERS_DB, {}) # Get today's date in ET today = datetime.now(timezone.utc) - timedelta(hours=4) # EDT today_str = today.strftime("%Y-%m-%d") log.info("Processing summaries for date: %s", today_str) log.info("Total mail log entries: %d", len(mail_log)) log.info("Total tracked senders: %d", len(senders)) results = [] for child_name in BOYS_ACCOUNTS: mom_name = MOM_NAMES.get(child_name, "Mom") to_addr = RECIPIENTS.get(child_name) # Filter today's emails for this child today_emails = [ e for e in mail_log if e.get("child") == child_name and parse_email_date(e.get("date", "")).strftime("%Y-%m-%d") == today_str ] # Filter senders for this child senders_for_child = { email_addr: info for email_addr, info in senders.items() if info.get("child") == child_name } log.info( "%s: %d emails today, %d tracked senders, sending to %s", child_name, len(today_emails), len(senders_for_child), to_addr, ) if to_addr and "@" in to_addr and "example.com" not in to_addr: success = send_email(child_name, today_emails, senders_for_child, to_addr) results.append((child_name, to_addr, success, len(today_emails))) else: log.warning( "%s: No valid recipient email configured for %s (%s). " "Germaine needs to provide it!", child_name, mom_name, to_addr, ) # Print the summary to log anyway for verification log.info( "Would have sent to %s:\n%s", to_addr, build_summary_html(child_name, today_emails, senders_for_child), ) results.append((child_name, to_addr, False, len(today_emails))) log.info("Summary run complete: %s", results) return results def do_test_summary(): """Send a test summary to Sho'Nuff himself to verify formatting.""" log.info("=" * 50) log.info("Boys' Mail — Test Summary (sent to shonuff inbox)") log.info("=" * 50) mail_log = load_json(MAIL_LOG, []) senders = load_json(SENDERS_DB, {}) today = datetime.now(timezone.utc) - timedelta(hours=4) today_str = today.strftime("%Y-%m-%d") for child_name in BOYS_ACCOUNTS: today_emails = [ e for e in mail_log if e.get("child") == child_name and parse_email_date(e.get("date", "")).strftime("%Y-%m-%d") == today_str ] senders_for_child = { email_addr: info for email_addr, info in senders.items() if info.get("child") == child_name } # Send test to sho'nuff subject = f"📬 TEST: {child_name}'s Email Summary — {today.strftime('%b %d')}" html = build_summary_html(child_name, today_emails, senders_for_child) plain = build_plain_text(child_name, today_emails, senders_for_child) msg = MIMEMultipart("alternative") msg["From"] = f"{SMTP_FROM_NAME} <{SMTP_FROM}>" msg["To"] = SMTP_FROM msg["Subject"] = subject msg.attach(MIMEText(plain, "plain")) msg.attach(MIMEText(html, "html")) with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as s: s.starttls() s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) log.info("Test summary for %s sent to %s", child_name, SMTP_FROM) print("✅ Test summaries sent to shonuff@germainebrown.com — check your inbox.") # ── Main CLI ─────────────────────────────────────────────────────────────── def main(): os.makedirs(DATA_DIR, exist_ok=True) if len(sys.argv) < 2: print("Usage: boys-mail-monitor.py [collect|summary|test-summary]") print(" collect — Poll inboxes for new emails") print(" summary — Send daily summary emails") print(" test-summary — Send test summaries to shonuff inbox") return command = sys.argv[1] if command == "collect": do_collect() elif command == "summary": do_summary() elif command == "test-summary": do_test_summary() else: print(f"Unknown command: {command}") print("Usage: boys-mail-monitor.py [collect|summary|test-summary]") if __name__ == "__main__": main()