- Update type annotations to modern syntax (dict, list, X | Y) - Remove unnecessary elif after return - Minor style improvements Note: Some linting warnings remain (unused content param, inline conditions) but these are minor style issues and do not affect functionality. All tests pass with these changes.
220 lines
6.2 KiB
Python
220 lines
6.2 KiB
Python
"""Notification email compressor for terminal-friendly display."""
|
|
|
|
from typing import Any
|
|
|
|
from .notification_detector import (
|
|
NotificationType,
|
|
classify_notification,
|
|
extract_notification_summary,
|
|
is_notification_email,
|
|
)
|
|
|
|
|
|
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, NotificationType | None]:
|
|
"""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: NotificationType | None,
|
|
) -> 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: NotificationType | None,
|
|
) -> 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)
|
|
return NotificationCompressor(mode=mode)
|