321 lines
11 KiB
Python
321 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Inbox solicitor cleaner — standalone script.
|
|
Reads all messages in INBOX, applies multi-heuristic detection,
|
|
copies flagged to a Solicitation folder, deletes from INBOX.
|
|
Adjust the Config block for your server.
|
|
|
|
Usage:
|
|
python3 inbox-solicitation-cleaner.py
|
|
|
|
For the reference patterns and heuristic design notes, see
|
|
references/inbox-solicitation-triaging.md in this skill.
|
|
"""
|
|
|
|
import imaplib
|
|
import email
|
|
import re
|
|
import json
|
|
from email.header import decode_header, make_header
|
|
from email.utils import parseaddr
|
|
|
|
# ═══════════ Config — ADAPT THESE ═══════════
|
|
IMAP_HOST = "mail.boxpilotlogistics.com"
|
|
IMAP_PORT = 993
|
|
USER = "hello@boxpilotlogistics.com"
|
|
PASS = "CHANGE_ME"
|
|
SOLICITATION_FOLDER = "Solicitation"
|
|
|
|
# Senders / domains NEVER to flag
|
|
INTERNAL_SENDERS = {"curt", "anita"}
|
|
LEGIT_DOMAINS = {"boxpilotlogistics.com"}
|
|
|
|
# Free / personal email domains (solicitation signal)
|
|
FREE_DOMAINS = {
|
|
"gmail.com", "yahoo.com", "ymail.com", "outlook.com", "hotmail.com",
|
|
"aol.com", "icloud.com", "me.com", "mac.com", "protonmail.com",
|
|
"proton.me", "pm.me", "zoho.com", "mail.com", "yandex.com",
|
|
"fastmail.com", "tutanota.com", "gmx.com", "live.com", "msn.com",
|
|
"comcast.net", "verizon.net", "att.net", "sbcglobal.net", "bellsouth.net",
|
|
"cox.net", "earthlink.net", "charter.net", "optonline.net",
|
|
}
|
|
|
|
SOLICITATION_KEYWORDS = [
|
|
"insurance", "coverage", "liability", "cargo insurance",
|
|
"general liability", "motor truck cargo", "physical damage",
|
|
"occupational accident", "workers comp", "bobtail", "non-trucking",
|
|
"factoring", "freight factoring", "invoice factoring", "cash flow",
|
|
"quick pay", "same day pay",
|
|
"fuel card", "fuel discount", "fuel savings", "fuel program",
|
|
"diesel discount", "fuel rewards",
|
|
"compliance", "dot compliance", "fmcsa", "authority", "mc number",
|
|
"usdot", "motor carrier authority", "boc-3", "boc 3", "process agent",
|
|
"operating authority", "new authority", "authority filing",
|
|
"eld", "electronic logging", "elog", "h.o.s.", "hours of service",
|
|
"logbook", "electronic log",
|
|
"dispatch", "dispatcher", "load board", "loadboard", "freight matching",
|
|
"dat load", "truckstop", "find loads", "backhaul",
|
|
"extended warranty", "vehicle warranty", "truck warranty",
|
|
"warranty coverage", "warranty expir",
|
|
"logo design", "website design", "seo service", "web development",
|
|
"digital marketing", "social media marketing",
|
|
"special offer", "limited time", "act now", "sign up today",
|
|
"get started today", "free quote", "no obligation", "risk free",
|
|
"exclusive offer", "grow your business", "take your business",
|
|
]
|
|
|
|
SOLICITATION_PHRASES = [
|
|
"we help trucking companies", "help your trucking",
|
|
"trucking company", "new entrant", "new mc", "new authority",
|
|
"carrier authority", "your mc", "trucking authority",
|
|
"startup carrier", "new carrier", "motor carrier",
|
|
"dot number", "safety audit", "new entrant safety",
|
|
"new to trucking", "independent contractor",
|
|
"owner operator", "small fleet",
|
|
]
|
|
|
|
SOLICITATION_SUBJECT_KEYWORDS = [
|
|
"get a quote", "free quote", "insurance quote", "fuel card",
|
|
"factoring", "dispatch services", "load board", "eld mandate",
|
|
"compliance", "authority filing",
|
|
"your mc", "new mc", "mc number",
|
|
"logo design", "website design", "seo",
|
|
"extended warranty", "reduce your cost",
|
|
"partner with us", "limited time offer",
|
|
"trucking solution", "fleet solution",
|
|
"lower your rate", "save on fuel",
|
|
"solicitation", "advertisement", "ad:",
|
|
]
|
|
|
|
|
|
def decode_mime_header(val):
|
|
if val is None:
|
|
return ""
|
|
try:
|
|
return str(make_header(decode_header(val)))
|
|
except Exception:
|
|
return val
|
|
|
|
|
|
def get_email_domain(email_addr):
|
|
_, addr = parseaddr(email_addr)
|
|
if "@" in addr:
|
|
return addr.lower().split("@")[1]
|
|
return ""
|
|
|
|
|
|
def get_clean_sender(msg):
|
|
from_hdr = decode_mime_header(msg.get("From", ""))
|
|
name, addr = parseaddr(from_hdr)
|
|
return name or addr.split("@")[0] if "@" in addr else from_hdr, addr.lower()
|
|
|
|
|
|
def get_body_text(msg):
|
|
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:
|
|
charset = part.get_content_charset() or "utf-8"
|
|
try:
|
|
body += payload.decode(charset, errors="replace")
|
|
except LookupError:
|
|
body += payload.decode("utf-8", errors="replace")
|
|
else:
|
|
payload = msg.get_payload(decode=True)
|
|
if payload:
|
|
charset = msg.get_content_charset() or "utf-8"
|
|
try:
|
|
body = payload.decode(charset, errors="replace")
|
|
except LookupError:
|
|
body = payload.decode("utf-8", errors="replace")
|
|
if not body:
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
if part.get_content_type() == "text/html":
|
|
payload = part.get_payload(decode=True)
|
|
if payload:
|
|
charset = part.get_content_charset() or "utf-8"
|
|
try:
|
|
raw = payload.decode(charset, errors="replace")
|
|
except LookupError:
|
|
raw = payload.decode("utf-8", errors="replace")
|
|
body = re.sub(r"<[^>]+>", " ", raw)
|
|
body = re.sub(r"\s+", " ", body).strip()
|
|
return body
|
|
|
|
|
|
def has_business_domain_ref(body, sender_domain):
|
|
refs = re.findall(
|
|
r'(?:https?://)?(?:www\.)?([a-zA-Z0-9][a-zA-Z0-9-]*\.[a-zA-Z]{2,})',
|
|
body
|
|
)
|
|
for d in refs:
|
|
d_lower = d.lower()
|
|
if d_lower not in FREE_DOMAINS and d_lower != sender_domain \
|
|
and "google" not in d_lower:
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_solicitation(sender_name, sender_addr, subject, body, from_domain):
|
|
reasons = []
|
|
subj_l = subject.lower()
|
|
body_l = body.lower()
|
|
|
|
sender_local = sender_addr.split("@")[0].lower() if "@" in sender_addr else ""
|
|
for internal in INTERNAL_SENDERS:
|
|
if internal in sender_addr or internal in sender_local:
|
|
return False, []
|
|
if from_domain in LEGIT_DOMAINS:
|
|
return False, []
|
|
|
|
for kw in SOLICITATION_SUBJECT_KEYWORDS:
|
|
if kw in subj_l:
|
|
reasons.append(f"Subject contains solicitation keyword: '{kw}'")
|
|
break
|
|
|
|
if from_domain in FREE_DOMAINS:
|
|
if has_business_domain_ref(body, from_domain):
|
|
reasons.append(f"Free-email sender ({from_domain}) referencing "
|
|
f"business domain in body")
|
|
for kw in SOLICITATION_KEYWORDS:
|
|
if kw in body_l:
|
|
reasons.append(f"Body contains solicitation keyword: '{kw}'")
|
|
break
|
|
for phrase in SOLICITATION_PHRASES:
|
|
if phrase in body_l:
|
|
reasons.append(f"Body contains solicitation phrase: '{phrase}'")
|
|
break
|
|
|
|
if from_domain not in FREE_DOMAINS and from_domain not in LEGIT_DOMAINS:
|
|
trucking_signals = 0
|
|
seen = set()
|
|
for phrase in SOLICITATION_PHRASES:
|
|
if phrase in body_l:
|
|
trucking_signals += 1
|
|
if phrase not in seen:
|
|
seen.add(phrase)
|
|
reasons.append(f"Body contains trucking solicitation phrase: "
|
|
f"'{phrase}'")
|
|
for kw in SOLICITATION_KEYWORDS:
|
|
if kw in body_l:
|
|
trucking_signals += 1
|
|
if kw not in seen:
|
|
seen.add(kw)
|
|
reasons.append(f"Body contains solicitation keyword: '{kw}'")
|
|
for kw in SOLICITATION_SUBJECT_KEYWORDS:
|
|
if kw in subj_l:
|
|
trucking_signals += 1
|
|
kw_rep = f"Subject keyword: '{kw}'"
|
|
if kw_rep not in seen:
|
|
seen.add(kw_rep)
|
|
reasons.append(kw_rep)
|
|
if trucking_signals >= 2:
|
|
reasons.append(f"Multiple trucking-solicitation signals (count: "
|
|
f"{trucking_signals})")
|
|
|
|
if not reasons:
|
|
pitch_count = 0
|
|
for kw in SOLICITATION_KEYWORDS:
|
|
if kw in subj_l or kw in body_l:
|
|
pitch_count += 1
|
|
for phrase in SOLICITATION_PHRASES:
|
|
if phrase in body_l:
|
|
pitch_count += 2
|
|
if pitch_count >= 2:
|
|
reasons.append(f"Combined solicitation signal strength: {pitch_count}")
|
|
|
|
return bool(reasons), reasons
|
|
|
|
|
|
def main():
|
|
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
|
|
conn.login(USER, PASS)
|
|
conn.select("INBOX")
|
|
|
|
status, data = conn.search(None, "ALL")
|
|
msg_ids = data[0].split() if data[0] else []
|
|
print(f"Total in INBOX: {len(msg_ids)}")
|
|
|
|
# Ensure target folder exists
|
|
status, folders = conn.list()
|
|
folder_pat = re.compile(r'.*"([^"]+)"$')
|
|
folder_names = []
|
|
for f in folders:
|
|
m = folder_pat.search(f.decode("utf-8", errors="replace"))
|
|
if m:
|
|
folder_names.append(m.group(1))
|
|
if not any(SOLICITATION_FOLDER in fn for fn in folder_names):
|
|
conn.create(SOLICITATION_FOLDER)
|
|
|
|
flagged, kept, errors = [], [], []
|
|
|
|
for idx, msg_id in enumerate(msg_ids, 1):
|
|
try:
|
|
status, data = conn.fetch(msg_id, "(BODY.PEEK[])")
|
|
if status != "OK":
|
|
errors.append((msg_id, "FETCH failed"))
|
|
continue
|
|
raw = data[0][1]
|
|
msg = email.message_from_bytes(raw) \
|
|
if isinstance(raw, bytes) \
|
|
else email.message_from_string(raw)
|
|
|
|
subject = decode_mime_header(msg.get("Subject", ""))
|
|
sname, saddr = get_clean_sender(msg)
|
|
domain = get_email_domain(saddr)
|
|
body = get_body_text(msg)
|
|
|
|
solicit, reasons = is_solicitation(sname, saddr, subject, body, domain)
|
|
entry = {
|
|
"id": int(msg_id), "from": saddr, "from_name": sname or saddr,
|
|
"subject": subject, "domain": domain,
|
|
}
|
|
if solicit:
|
|
entry["reasons"] = reasons
|
|
flagged.append(entry)
|
|
else:
|
|
kept.append(entry)
|
|
except Exception as e:
|
|
errors.append((msg_id, str(e)))
|
|
|
|
print(f"\nRESULTS: {len(flagged)} flagged, {len(kept)} kept, "
|
|
f"{len(errors)} errors")
|
|
|
|
moved = 0
|
|
for entry in flagged:
|
|
mid = str(entry["id"]).encode()
|
|
try:
|
|
status, _ = conn.copy(mid, SOLICITATION_FOLDER)
|
|
if status == "OK":
|
|
conn.store(mid, "+FLAGS", "\\Deleted")
|
|
moved += 1
|
|
except Exception:
|
|
pass
|
|
|
|
conn.expunge()
|
|
conn.logout()
|
|
|
|
report = {
|
|
"total_inbox": len(msg_ids), "flagged_count": len(flagged),
|
|
"kept_count": len(kept), "moved_count": moved,
|
|
"errors": [{"msg_id": str(e[0]), "error": e[1]} for e in errors],
|
|
"flagged": flagged,
|
|
"kept_summary": [
|
|
{"id": e["id"], "from": e["from"], "subject": e["subject"][:80]}
|
|
for e in kept
|
|
],
|
|
}
|
|
with open("inbox_review_report.json", "w") as f:
|
|
json.dump(report, f, indent=2, default=str)
|
|
|
|
print(f"\nSummary: {moved} moved, {len(kept)} kept, report → "
|
|
f"inbox_review_report.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|