first commit
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
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]
|
||||
Reference in New Issue
Block a user