added more robust personal email rejection handling

Author: antonidebicki

src/agent/assets/all_email_provider_domains.txt

@@ -1,4 +1,6 @@
 import re
+from functools import lru_cache
+from pathlib import Path
 from typing import Dict, Optional
 
 from openai import AsyncOpenAI
@@ -27,14 +29,54 @@
     "protonmail.com",
 }
 
+EMAIL_PROVIDER_BLACKLIST_PATH = Path(__file__).parent / "assets" / "all_email_provider_domains.txt"
+
 
 EMAIL_REGEX = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.IGNORECASE)
 
 
+def _normalize_domain(domain: Optional[str]) -> str:
+    raw = (domain or "").strip().lower()
+    if not raw:
+        return ""
+
+    if "@" in raw:
+        raw = raw.split("@")[-1]
+    if raw.startswith("www."):
+        raw = raw[4:]
+    return raw.strip(". ")
+
+
+@lru_cache(maxsize=1)
+def load_personal_email_blacklist() -> set[str]:
+    domains = {d.lower() for d in COMMON_PERSONAL_DOMAINS}
+
+    try:
+        for line in EMAIL_PROVIDER_BLACKLIST_PATH.read_text(encoding="utf-8", errors="ignore").splitlines():
+            candidate = _normalize_domain(line)
+            if not candidate:
+                continue
+            if candidate.startswith("#") or candidate.startswith("/*"):
+                continue
+            if re.fullmatch(r"[a-z0-9.-]+\.[a-z]{2,}", candidate):
+                domains.add(candidate)
+    except Exception as exc:
+        logger.warning("Failed to load email provider blacklist from %s: %s", EMAIL_PROVIDER_BLACKLIST_PATH, exc)
+
+    return domains
+
+
+def is_blacklisted_email_domain(domain: Optional[str]) -> bool:
+    normalized = _normalize_domain(domain)
+    if not normalized:
+        return False
+    return normalized in load_personal_email_blacklist()
+
+
 def _extract_email_domain(email: Optional[str]) -> str:
     if not email or "@" not in email:
         return ""
-    return email.split("@")[-1].strip().lower()
+    return _normalize_domain(email)
 
 
 def _extract_form_email(email_subject: str, email_body: str) -> str:
@@ -121,14 +163,14 @@ async def run_company_verification(
             "status": "company verification skipped: missing form email domain",
         }
 
-    if form_domain in COMMON_PERSONAL_DOMAINS:
+    if is_blacklisted_email_domain(form_domain):
         fallback = CompanyVerificationResult(
             is_corporate_email=False,
             is_legit_company=False,
             company_type="unknown",
             company_name=company_name or None,
             sender_domain=form_domain,
-            reason="form email domain is a known personal email provider",
+            reason="form email domain is in personal/provider blacklist",
         )
         return {
             "company_verification": fallback.model_dump(),

src/agent/company_verification_layer.py

@@ -10,6 +10,7 @@
 from openai import AsyncOpenAI
 
 from src.agent.config import Config
+from src.agent.company_verification_layer import is_blacklisted_email_domain
 from src.agent.graph_schemas import (
     EmailClassificationOutput,
     EmailClassificationRequest,
@@ -341,6 +342,31 @@ async def classify_email(state: EmailClassificationState):
         }
 
 
+async def enforce_personal_email_blacklist(state: EmailClassificationState):
+    classification = dict(state.classification or {})
+    contact_email = str(classification.get("email") or "").strip().lower()
+    email_domain = contact_email.split("@")[-1] if "@" in contact_email else ""
+
+    if not email_domain:
+        return {
+            "status": "email blacklist check skipped: missing contact email domain",
+        }
+
+    if not is_blacklisted_email_domain(email_domain):
+        return {
+            "status": "email blacklist check passed",
+        }
+
+    classification["action"] = "disqualify"
+    classification["salesperson"] = "none"
+    classification["blacklist_reason"] = f"personal/provider email domain blocked: {email_domain}"
+
+    return {
+        "classification": classification,
+        "status": "email disqualified by personal/provider blacklist",
+    }
+
+
 email_classification_graph_builder = StateGraph(
     EmailClassificationState,
     input_schema=EmailClassificationRequest,
@@ -350,10 +376,12 @@ async def classify_email(state: EmailClassificationState):
 email_classification_graph_builder.add_node("build_query_text", build_query_text)
 email_classification_graph_builder.add_node("retrieve_context", retrieve_context)
 email_classification_graph_builder.add_node("classify_email", classify_email)
+email_classification_graph_builder.add_node("enforce_personal_email_blacklist", enforce_personal_email_blacklist)
 
 email_classification_graph_builder.add_edge(START, "build_query_text")
 email_classification_graph_builder.add_edge("build_query_text", "retrieve_context")
 email_classification_graph_builder.add_edge("retrieve_context", "classify_email")
-email_classification_graph_builder.add_edge("classify_email", END)
+email_classification_graph_builder.add_edge("classify_email", "enforce_personal_email_blacklist")
+email_classification_graph_builder.add_edge("enforce_personal_email_blacklist", END)
 
 email_classification_graph = email_classification_graph_builder.compile()

src/agent/email_classification_graph.py

@@ -13,6 +13,7 @@
 from langgraph.graph import END, START, StateGraph
 
 from src.agent.config import Config
+from src.agent.company_verification_layer import is_blacklisted_email_domain
 from src.agent.graph_schemas import EmailClassificationResult, EmailIngestionOutput, EmailIngestionRequest, EmailIngestionState
 from src.agent.logger import get_logger
 from src.agent.excel_tracker import EmailClassificationExcelTracker
@@ -537,6 +538,31 @@ async def classify_email(state: EmailIngestionState):
         }
 
 
+async def enforce_personal_email_blacklist(state: EmailIngestionState):
+    classification = dict(state.classification or {})
+    contact_email = str(classification.get("email") or "").strip().lower()
+    email_domain = contact_email.split("@")[-1] if "@" in contact_email else ""
+
+    if not email_domain:
+        return {
+            "status": "email blacklist check skipped: missing contact email domain",
+        }
+
+    if not is_blacklisted_email_domain(email_domain):
+        return {
+            "status": "email blacklist check passed",
+        }
+
+    classification["action"] = "disqualify"
+    classification["salesperson"] = "none"
+    classification["blacklist_reason"] = f"personal/provider email domain blocked: {email_domain}"
+
+    return {
+        "classification": classification,
+        "status": "email disqualified by personal/provider blacklist",
+    }
+
+
 def _append_to_excel_sync(
     thread_id: str,
     created_at: str,
@@ -729,13 +755,15 @@ async def forward_to_salesperson(state: EmailIngestionState):
 email_ingestion_graph_builder.add_node("download_attachments", download_attachments)
 email_ingestion_graph_builder.add_node("extract_attachment_text", extract_attachment_text)
 email_ingestion_graph_builder.add_node("classify_email", classify_email)
+email_ingestion_graph_builder.add_node("enforce_personal_email_blacklist", enforce_personal_email_blacklist)
 email_ingestion_graph_builder.add_node("forward_to_salesperson", forward_to_salesperson)
 
 email_ingestion_graph_builder.add_edge(START, "get_email_messages")
 email_ingestion_graph_builder.add_edge("get_email_messages", "download_attachments")
 email_ingestion_graph_builder.add_edge("download_attachments", "extract_attachment_text")
 email_ingestion_graph_builder.add_edge("extract_attachment_text", "classify_email")
-email_ingestion_graph_builder.add_edge("classify_email", "forward_to_salesperson")
+email_ingestion_graph_builder.add_edge("classify_email", "enforce_personal_email_blacklist")
+email_ingestion_graph_builder.add_edge("enforce_personal_email_blacklist", "forward_to_salesperson")
 email_ingestion_graph_builder.add_edge("forward_to_salesperson", END)
 
 email_ingestion_graph = email_ingestion_graph_builder.compile()

src/agent/email_ingestion_graph.py

@@ -1,7 +1,10 @@
 import os
-from dotenv import load_dotenv
-from msal import ConfidentialClientApplication
+import json
 from pathlib import Path
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+from dotenv import load_dotenv
 
 # A standalone script if you need an access token for postman or other testing purposes. 
 
@@ -18,25 +21,30 @@ def get_access_token(self):
         if not all([client_id, client_secret, tenant_id]):
             raise ValueError(f"Missing credentials: CLIENT_ID={client_id}, CLIENT_SECRET={bool(client_secret)}, TENANT_ID={tenant_id}")
 
-        msal_app = ConfidentialClientApplication(
-            client_id=client_id,
-            client_credential=client_secret,
-            authority=f"https://login.microsoftonline.com/{tenant_id}",
-        )
- 
-        result = msal_app.acquire_token_silent(
-            scopes=["https://graph.microsoft.com/.default"],
-            account=None,
+        token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
+        payload = {
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "scope": "https://graph.microsoft.com/.default",
+            "grant_type": "client_credentials",
+        }
+
+        body = urlencode(payload).encode("utf-8")
+        request = Request(
+            token_url,
+            data=body,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            method="POST",
         )
-        if not result:
-            result = msal_app.acquire_token_for_client(
-                scopes=["https://graph.microsoft.com/.default"]
-            )
- 
-        if not result:
-            return None
 
-        return result.get("access_token")
+        try:
+            with urlopen(request, timeout=30) as response:
+                response_data = response.read().decode("utf-8", errors="ignore")
+                result = json.loads(response_data or "{}")
+                return result.get("access_token")
+        except Exception as exc:
+            print(f"Token request failed: {exc}")
+            return None
 
 token = GraphClient().get_access_token()
 print(f"{token} ..." if token else "NO TOKEN")