hermes-recovery/scripts/boxpilot-triage.py

#!/usr/bin/env python3
"""BoxPilot Logistics email triage — detect and move solicitations.

Heuristic: if sender is from a free/personal email domain (gmail, yahoo, etc.)
AND the email body or subject references a different business domain,
it's likely solicitation. Move to INBOX.Solicitation.
"""
import imaplib, email, re, os, json, hashlib
from email.utils import parsedate_to_datetime
from datetime import datetime, timedelta, timezone

HOST = "mail.boxpilotlogistics.com"
PORT = 993
USER = "hello@boxpilotlogistics.com"
PW = "1New.opportunity"
SOLICIT_FOLDER = "Solicitation"
CHECK_HOURS = 48

# Free/personal email domains — if sender is from these AND claims another company, it's sus
FREE_DOMAINS = {
    "gmail.com", "yahoo.com", "yahoo.co.uk", "ymail.com", "outlook.com",
    "hotmail.com", "live.com", "msn.com", "aol.com", "icloud.com",
    "me.com", "mac.com", "protonmail.com", "proton.me", "pm.me",
    "mail.com", "inbox.com", "zoho.com", "yandex.com", "gmx.com",
    "fastmail.com", "tutanota.com", "rediffmail.com", "libero.it",
    "web.de", "gmx.de", "t-online.de", "online.de",
}

def safe_str(val):
    if val is None: return ""
    if isinstance(val, str): return val
    try: return str(val)
    except: return ""

def get_body_text(msg):
    """Extract plain text body."""
    body = ""
    if msg.is_multipart():
        for part in msg.walk():
            ct = part.get_content_type()
            if ct == "text/plain":
                payload = part.get_payload(decode=True)
                if payload:
                    body += payload.decode("utf-8", errors="replace")
            elif ct == "text/html" and not body:
                payload = part.get_payload(decode=True)
                if payload:
                    body += payload.decode("utf-8", errors="replace")
    else:
        payload = msg.get_payload(decode=True)
        if payload:
            body = payload.decode("utf-8", errors="replace")
    return body

def extract_domains(text):
    """Extract all domain names from text (from URLs and email addresses)."""
    domains = set()
    # URLs
    for m in re.finditer(r'https?://(?:www\.)?([a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.[a-z]{2,}(?:\.[a-z]{2,})?)', text, re.IGNORECASE):
        d = m.group(1).lower()
        # Skip common non-business domains
        if d not in ('google.com', 'youtube.com', 'facebook.com', 'linkedin.com', 'twitter.com', 'x.com', 'instagram.com'):
            domains.add(d)
    # Email addresses in body
    for m in re.finditer(r'[a-z0-9._%+-]+@([a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.[a-z]{2,}(?:\.[a-z]{2,})?)', text, re.IGNORECASE):
        d = m.group(1).lower()
        if d not in FREE_DOMAINS:
            domains.add(d)
    return domains

def is_solicitation(msg):
    """Determine if a message is likely solicitation."""
    sender = safe_str(msg.get("From", "")).lower()
    subject = safe_str(msg.get("Subject", "")).lower()
    body = get_body_text(msg).lower()
    full_text = subject + " " + body

    # Extract sender's domain
    sender_domain = ""
    m = re.search(r'@([a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.[a-z]{2,}(?:\.[a-z]{2,})?)', sender)
    if m:
        sender_domain = m.group(1).lower()

    # If sender is from a free domain, check if they claim another company
    if sender_domain in FREE_DOMAINS:
        # Extract business domains mentioned in the email
        biz_domains = extract_domains(full_text)
        # If they reference a business domain different from their sender domain
        # AND that business domain is a real company domain (not another free domain)
        legit_biz = [d for d in biz_domains if d not in FREE_DOMAINS]
        if legit_biz:
            return True, f"Free domain ({sender_domain}) referencing business domain(s): {', '.join(list(legit_biz)[:3])}"

    # Also flag common solicitation subject patterns from free domains
    if sender_domain in FREE_DOMAINS:
        spam_kw = ["quote", "insurance", "factoring", "fuel", "dispatch", "load board",
                    "freight", "logistics", "carrier", "trucking", "broker", "authority",
                    "mc authority", "dot authority", "compliance", "ifta", "irp",
                    "elog", "eld", "safety", "audit", "lease", "warranty", "extended warranty"]
        found = [kw for kw in spam_kw if kw in subject]
        if found:
            return True, f"Free domain ({sender_domain}) with solicitation subject keywords: {', '.join(found)}"

    return False, ""

def main():
    try:
        conn = imaplib.IMAP4_SSL(HOST, PORT)
        conn.login(USER, PW)
    except Exception as e:
        print(f"ERROR: Connection failed: {e}")
        return

    # Select inbox
    conn.select("INBOX")

    # Search for recent unread messages
    since = (datetime.now(timezone.utc) - timedelta(hours=CHECK_HOURS)).strftime("%d-%b-%Y")
    status, data = conn.search(None, f'(UNSEEN SINCE {since})')
    if status != "OK" or not data[0]:
        conn.logout()
        return  # silent

    moved = 0
    for num in data[0].split():
        status, msg_data = conn.fetch(num, "(RFC822)")
        if status != "OK":
            continue
        raw = email.message_from_bytes(msg_data[0][1])
        result, reason = is_solicitation(raw)
        if result:
            # Copy to Solicitation folder, then delete from inbox
            copy_status = conn.copy(num, SOLICIT_FOLDER)
            if copy_status[0] == "OK":
                conn.store(num, "+FLAGS", "\\Deleted")
                moved += 1
                subj = safe_str(raw.get("Subject", "(no subject)"))
                sender = safe_str(raw.get("From", "(unknown)"))

    # Expunge deleted messages
    if moved > 0:
        conn.expunge()

    conn.logout()

    if moved > 0:
        print(f"📬 BoxPilot — Moved {moved} solicitation(s) to Solicitation folder")
    # silent if nothing moved

if __name__ == "__main__":
    main()