feat: Add notification email compression feature
Add intelligent notification email detection and compression: - Detect notification emails from GitLab, GitHub, Jira, Confluence, Datadog, Renovate - Extract structured summaries from notification emails - Compress notifications into terminal-friendly markdown format - Add configuration options for notification compression mode Features: - Rule-based detection using sender domains and subject patterns - Type-specific extractors for each notification platform - Configurable compression modes (summary, detailed, off) - Integrated with ContentContainer for seamless display Files added: - src/mail/notification_detector.py: Notification type detection - src/mail/notification_compressor.py: Content compression Modified: - src/mail/config.py: Add notification_compression_mode config - src/mail/widgets/ContentContainer.py: Integrate compressor - src/mail/app.py: Pass envelope data to display_content - PROJECT_PLAN.md: Document new feature
This commit is contained in:
221
src/mail/notification_compressor.py
Normal file
221
src/mail/notification_compressor.py
Normal file
@@ -0,0 +1,221 @@
|
||||
"""Notification email compressor for terminal-friendly display."""
|
||||
|
||||
from typing import Dict, Any, Optional
|
||||
from textual.widgets import Markdown
|
||||
|
||||
from .notification_detector import (
|
||||
is_notification_email,
|
||||
classify_notification,
|
||||
extract_notification_summary,
|
||||
NotificationType,
|
||||
)
|
||||
|
||||
|
||||
class NotificationCompressor:
|
||||
"""Compress notification emails into terminal-friendly summaries."""
|
||||
|
||||
def __init__(self, mode: str = "summary"):
|
||||
"""Initialize compressor.
|
||||
|
||||
Args:
|
||||
mode: Compression mode - "summary", "detailed", or "off"
|
||||
"""
|
||||
self.mode = mode
|
||||
|
||||
def should_compress(self, envelope: Dict[str, Any]) -> bool:
|
||||
"""Check if email should be compressed.
|
||||
|
||||
Args:
|
||||
envelope: Email envelope metadata
|
||||
|
||||
Returns:
|
||||
True if email should be compressed
|
||||
"""
|
||||
if self.mode == "off":
|
||||
return False
|
||||
|
||||
return is_notification_email(envelope)
|
||||
|
||||
def compress(
|
||||
self, content: str, envelope: Dict[str, Any]
|
||||
) -> tuple[str, Optional[NotificationType]]:
|
||||
"""Compress notification email content.
|
||||
|
||||
Args:
|
||||
content: Raw email content
|
||||
envelope: Email envelope metadata
|
||||
|
||||
Returns:
|
||||
Tuple of (compressed content, notification_type)
|
||||
"""
|
||||
|
||||
if not self.should_compress(envelope):
|
||||
return content, None
|
||||
|
||||
# Classify notification type
|
||||
notif_type = classify_notification(envelope, content)
|
||||
|
||||
# Extract summary
|
||||
summary = extract_notification_summary(content, notif_type)
|
||||
|
||||
# Format as markdown
|
||||
compressed = self._format_as_markdown(summary, envelope, notif_type)
|
||||
|
||||
return compressed, notif_type
|
||||
|
||||
def _format_as_markdown(
|
||||
self,
|
||||
summary: Dict[str, Any],
|
||||
envelope: Dict[str, Any],
|
||||
notif_type: Optional[NotificationType],
|
||||
) -> str:
|
||||
"""Format summary as markdown for terminal display.
|
||||
|
||||
Args:
|
||||
summary: Extracted summary data
|
||||
envelope: Email envelope metadata
|
||||
notif_type: Classified notification type
|
||||
|
||||
Returns:
|
||||
Markdown-formatted compressed email
|
||||
"""
|
||||
|
||||
from_addr = envelope.get("from", {}).get("name") or envelope.get(
|
||||
"from", {}
|
||||
).get("addr", "")
|
||||
subject = envelope.get("subject", "")
|
||||
|
||||
# Get icon
|
||||
icon = notif_type.icon if notif_type else "\uf0f3"
|
||||
|
||||
# Build markdown
|
||||
lines = []
|
||||
|
||||
# Header with icon
|
||||
if notif_type:
|
||||
lines.append(f"## {icon} {notif_type.name.title()} Notification")
|
||||
else:
|
||||
lines.append(f"## {icon} Notification")
|
||||
|
||||
lines.append("")
|
||||
|
||||
# Title/subject
|
||||
if summary.get("title"):
|
||||
lines.append(f"**{summary['title']}**")
|
||||
else:
|
||||
lines.append(f"**{subject}**")
|
||||
lines.append("")
|
||||
|
||||
# Metadata section
|
||||
if summary.get("metadata"):
|
||||
lines.append("### Details")
|
||||
for key, value in summary["metadata"].items():
|
||||
# Format key nicely
|
||||
key_formatted = key.replace("_", " ").title()
|
||||
lines.append(f"- **{key_formatted}**: {value}")
|
||||
lines.append("")
|
||||
|
||||
# Action items
|
||||
if summary.get("action_items"):
|
||||
lines.append("### Actions")
|
||||
for i, action in enumerate(summary["action_items"], 1):
|
||||
lines.append(f"{i}. {action}")
|
||||
lines.append("")
|
||||
|
||||
# Add footer
|
||||
lines.append("---")
|
||||
lines.append("")
|
||||
lines.append(f"*From: {from_addr}*")
|
||||
lines.append(
|
||||
"*This is a compressed notification view. Press `m` to toggle full view.*"
|
||||
)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class DetailedCompressor(NotificationCompressor):
|
||||
"""Compressor that includes more detail in summaries."""
|
||||
|
||||
def _format_as_markdown(
|
||||
self,
|
||||
summary: Dict[str, Any],
|
||||
envelope: Dict[str, Any],
|
||||
notif_type: Optional[NotificationType],
|
||||
) -> str:
|
||||
"""Format summary with more detail."""
|
||||
|
||||
from_addr = envelope.get("from", {}).get("name") or envelope.get(
|
||||
"from", {}
|
||||
).get("addr", "")
|
||||
subject = envelope.get("subject", "")
|
||||
date = envelope.get("date", "")
|
||||
|
||||
icon = notif_type.icon if notif_type else "\uf0f3"
|
||||
|
||||
lines = []
|
||||
|
||||
# Header
|
||||
lines.append(
|
||||
f"## {icon} {notif_type.name.title()} Notification"
|
||||
if notif_type
|
||||
else f"## {icon} Notification"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
# Subject and from
|
||||
lines.append(f"**Subject:** {subject}")
|
||||
lines.append(f"**From:** {from_addr}")
|
||||
lines.append(f"**Date:** {date}")
|
||||
lines.append("")
|
||||
|
||||
# Summary title
|
||||
if summary.get("title"):
|
||||
lines.append(f"### {summary['title']}")
|
||||
lines.append("")
|
||||
|
||||
# Metadata table
|
||||
if summary.get("metadata"):
|
||||
lines.append("| Property | Value |")
|
||||
lines.append("|----------|-------|")
|
||||
for key, value in summary["metadata"].items():
|
||||
key_formatted = key.replace("_", " ").title()
|
||||
lines.append(f"| {key_formatted} | {value} |")
|
||||
lines.append("")
|
||||
|
||||
# Action items
|
||||
if summary.get("action_items"):
|
||||
lines.append("### Action Items")
|
||||
for i, action in enumerate(summary["action_items"], 1):
|
||||
lines.append(f"- [ ] {action}")
|
||||
lines.append("")
|
||||
|
||||
# Key links
|
||||
if summary.get("key_links"):
|
||||
lines.append("### Important Links")
|
||||
for link in summary["key_links"]:
|
||||
lines.append(f"- [{link.get('text', 'Link')}]({link.get('url', '#')})")
|
||||
lines.append("")
|
||||
|
||||
# Footer
|
||||
lines.append("---")
|
||||
lines.append(
|
||||
"*This is a compressed notification view. Press `m` to toggle full view.*"
|
||||
)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def create_compressor(mode: str) -> NotificationCompressor:
|
||||
"""Factory function to create appropriate compressor.
|
||||
|
||||
Args:
|
||||
mode: Compression mode - "summary", "detailed", or "off"
|
||||
|
||||
Returns:
|
||||
NotificationCompressor instance
|
||||
"""
|
||||
|
||||
if mode == "detailed":
|
||||
return DetailedCompressor(mode=mode)
|
||||
else:
|
||||
return NotificationCompressor(mode=mode)
|
||||
Reference in New Issue
Block a user