47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
|
import re
|
||
|
|
from typing import Dict, List, Any, Tuple
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class EmailIntent:
|
||
|
|
is_actionable: bool
|
||
|
|
confidence: float
|
||
|
|
intent_type: str
|
||
|
|
reason: str
|
||
|
|
|
||
|
|
class EmailTriage:
|
||
|
|
def __init__(self):
|
||
|
|
self.non_actionable_patterns = [
|
||
|
|
r'no-reply@', r'noreply@', r'newsletter', r'promotion',
|
||
|
|
r'unsubscribe', r'confirm your email', r'password reset'
|
||
|
|
]
|
||
|
|
self.actionable_patterns = [
|
||
|
|
r'\?', r'can you', r'could you', r'please', r'help',
|
||
|
|
r'urgent', r'asap', r'follow up', r'status', r'update'
|
||
|
|
]
|
||
|
|
self.non_actionable_regex = [re.compile(p, re.IGNORECASE) for p in self.non_actionable_patterns]
|
||
|
|
self.actionable_regex = [re.compile(p, re.IGNORECASE) for p in self.actionable_patterns]
|
||
|
|
|
||
|
|
def analyze_email(self, email: Dict[str, Any]) -> EmailIntent:
|
||
|
|
from_addr = email.get('from', '').lower()
|
||
|
|
subject = email.get('subject', '').lower()
|
||
|
|
snippet = email.get('snippet', '').lower()
|
||
|
|
text = f"{from_addr} {subject} {snippet}"
|
||
|
|
|
||
|
|
# Check non-actionable first
|
||
|
|
for pattern in self.non_actionable_regex:
|
||
|
|
if pattern.search(text):
|
||
|
|
return EmailIntent(False, 0.9, 'automated', 'Automated email detected')
|
||
|
|
|
||
|
|
# Calculate actionable score
|
||
|
|
score = sum(len(p.findall(text)) * 0.2 for p in self.actionable_regex)
|
||
|
|
score += text.count('?') * 0.3
|
||
|
|
|
||
|
|
if score > 0.3:
|
||
|
|
return EmailIntent(True, min(score, 0.9), 'actionable', f'Score: {score:.2f}')
|
||
|
|
|
||
|
|
return EmailIntent(False, 0.5, 'unclear', 'No clear indicators')
|
||
|
|
|
||
|
|
def filter_actionable_emails(self, emails: List[Dict[str, Any]]) -> List[Tuple[Dict[str, Any], EmailIntent]]:
|
||
|
|
return [(email, self.analyze_email(email)) for email in emails
|
||
|
|
if self.analyze_email(email).is_actionable]
|