Add rich email notification with template file
All checks were successful
Build And Test / publish (push) Successful in 47s
All checks were successful
Build And Test / publish (push) Successful in 47s
- Created src/templates/notif.txt for customizable notification format - Include sender, recipients, subject, date, body (truncated), attachments, unsubscribe info - Fetch full email content for notifications 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,9 +1,35 @@
|
||||
"""Pydantic models for webhook notification payloads."""
|
||||
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
from pydantic import BaseModel
|
||||
|
||||
# Load notification template
|
||||
_TEMPLATE_PATH = os.path.join(os.path.dirname(__file__), "..", "templates", "notif.txt")
|
||||
_DEFAULT_TEMPLATE = """📧 New Email Received
|
||||
|
||||
From: {sender_name} <{sender_email}>
|
||||
To: {recipients}
|
||||
Subject: {subject}
|
||||
Date: {date}
|
||||
|
||||
{body}
|
||||
|
||||
---
|
||||
{attachments_line}
|
||||
{unsubscribe_line}
|
||||
"""
|
||||
|
||||
|
||||
def _load_template() -> str:
|
||||
"""Load notification template from file."""
|
||||
try:
|
||||
with open(_TEMPLATE_PATH, "r") as f:
|
||||
return f.read()
|
||||
except FileNotFoundError:
|
||||
return _DEFAULT_TEMPLATE
|
||||
|
||||
|
||||
class EmailNotificationPayload(BaseModel):
|
||||
"""Webhook payload for new email notifications to Poke."""
|
||||
@@ -27,23 +53,67 @@ class EmailNotificationPayload(BaseModel):
|
||||
# Content
|
||||
subject: str
|
||||
snippet: Optional[str] = None
|
||||
body: Optional[str] = None
|
||||
|
||||
# Metadata
|
||||
date: datetime
|
||||
has_attachments: bool
|
||||
is_flagged: bool
|
||||
attachment_count: int = 0
|
||||
|
||||
# Unsubscribe info
|
||||
can_unsubscribe: bool = False
|
||||
unsubscribe_url: Optional[str] = None
|
||||
|
||||
# Threading
|
||||
in_reply_to: Optional[str] = None
|
||||
|
||||
def to_webhook_format(self) -> dict:
|
||||
"""Convert to the format expected by Poke webhook."""
|
||||
# Format sender
|
||||
sender = self.from_name if self.from_name else self.from_email
|
||||
template = _load_template()
|
||||
|
||||
# Build message
|
||||
message = f"New email from {sender}\nSubject: {self.subject}"
|
||||
if self.snippet:
|
||||
message += f"\n\n{self.snippet}"
|
||||
# Format sender
|
||||
sender_name = self.from_name or self.from_email.split("@")[0]
|
||||
|
||||
# Format recipients
|
||||
recipients = ", ".join(self.to_emails[:3])
|
||||
if len(self.to_emails) > 3:
|
||||
recipients += f" (+{len(self.to_emails) - 3} more)"
|
||||
|
||||
# Format date
|
||||
date_str = self.date.strftime("%b %d, %Y at %H:%M")
|
||||
|
||||
# Format body (truncate if too long)
|
||||
body = self.body or self.snippet or "(No content)"
|
||||
max_body_len = 500
|
||||
if len(body) > max_body_len:
|
||||
body = body[:max_body_len].rsplit(" ", 1)[0] + "..."
|
||||
|
||||
# Format attachments line
|
||||
if self.has_attachments:
|
||||
attachments_line = f"📎 {self.attachment_count} attachment(s)"
|
||||
else:
|
||||
attachments_line = ""
|
||||
|
||||
# Format unsubscribe line
|
||||
if self.can_unsubscribe:
|
||||
unsubscribe_line = "🔕 This email can be unsubscribed"
|
||||
else:
|
||||
unsubscribe_line = ""
|
||||
|
||||
# Build message from template
|
||||
message = template.format(
|
||||
sender_name=sender_name,
|
||||
sender_email=self.from_email,
|
||||
recipients=recipients,
|
||||
subject=self.subject,
|
||||
date=date_str,
|
||||
body=body,
|
||||
attachments_line=attachments_line,
|
||||
unsubscribe_line=unsubscribe_line,
|
||||
)
|
||||
|
||||
# Clean up empty lines at the end
|
||||
message = message.strip()
|
||||
|
||||
return {"message": message}
|
||||
|
||||
@@ -278,6 +278,22 @@ class EmailMonitor:
|
||||
self, email: EmailSummary, mailbox: str
|
||||
) -> tuple[bool, Optional[str]]:
|
||||
"""Send webhook notification for a new email."""
|
||||
# Fetch full email for body and unsubscribe info
|
||||
full_email = self.email_service.read_email(mailbox, email.id, format="text")
|
||||
|
||||
# Get body and unsubscribe details from full email if available
|
||||
body = None
|
||||
can_unsubscribe = False
|
||||
unsubscribe_url = None
|
||||
attachment_count = 0
|
||||
|
||||
if full_email:
|
||||
body = full_email.body_text
|
||||
attachment_count = len(full_email.attachments)
|
||||
if full_email.unsubscribe:
|
||||
can_unsubscribe = full_email.unsubscribe.available
|
||||
unsubscribe_url = full_email.unsubscribe.http_url
|
||||
|
||||
payload = EmailNotificationPayload(
|
||||
timestamp=datetime.now(timezone.utc),
|
||||
email_id=email.id,
|
||||
@@ -287,9 +303,13 @@ class EmailMonitor:
|
||||
to_emails=[addr.email for addr in email.to_addresses],
|
||||
subject=email.subject,
|
||||
snippet=email.snippet,
|
||||
body=body,
|
||||
date=email.date,
|
||||
has_attachments=email.has_attachments,
|
||||
is_flagged=email.is_flagged,
|
||||
attachment_count=attachment_count,
|
||||
can_unsubscribe=can_unsubscribe,
|
||||
unsubscribe_url=unsubscribe_url,
|
||||
)
|
||||
|
||||
return await self.webhook_service.send_new_email_notification(payload)
|
||||
|
||||
12
src/templates/notif.txt
Normal file
12
src/templates/notif.txt
Normal file
@@ -0,0 +1,12 @@
|
||||
📧 New Email Received
|
||||
|
||||
From: {sender_name} <{sender_email}>
|
||||
To: {recipients}
|
||||
Subject: {subject}
|
||||
Date: {date}
|
||||
|
||||
{body}
|
||||
|
||||
---
|
||||
{attachments_line}
|
||||
{unsubscribe_line}
|
||||
Reference in New Issue
Block a user