Intermediate

Step 4: Smart Features

In this lesson, you will add intelligent features that transform your inbox from a wall of text into an organized task list. You will build email summarization for long threads, action item extraction that detects deadlines and assignments, follow-up reminders for unanswered emails, and a daily digest that gives you a morning briefing.

Smart Features Overview

Email Batch
    |
    +---> [Summarizer]       --> one-line + detailed summaries
    +---> [Action Extractor] --> tasks with deadlines
    +---> [Follow-Up Detector] --> reminder scheduling
    |
    v
[Daily Digest Generator]
    |
    v
Morning briefing email or dashboard card

Email Summarization

Long email threads are painful to read. The summarizer produces both a one-line summary (for inbox list view) and a detailed summary (for the email detail view):

# app/ai/smart.py
"""Smart features: summarization, action items, follow-ups, digests."""
import json
from datetime import datetime, timedelta
from openai import OpenAI

from app.config import config
from app.database import (
    SessionLocal, Email, Classification,
    ActionItem, FollowUp, Draft
)

client = OpenAI(api_key=config.openai_api_key)


# ─── Summarization ─────────────────────────────────────────────

SUMMARY_PROMPT = """Summarize this email thread. Return a JSON object with:

1. "one_line": A single sentence summary (max 80 characters)
2. "detailed": A 2-4 sentence summary covering key points, decisions, and open questions
3. "key_points": List of 3-5 bullet points (strings)

Return ONLY the JSON object.

--- EMAIL ---
From: {sender}
Subject: {subject}
Date: {date}

{body}
--- END EMAIL ---"""


def summarize_email(email: Email) -> dict:
    """
    Generate summaries for an email.

    Args:
        email: Email ORM object.

    Returns:
        Dict with one_line, detailed, and key_points.
    """
    body = (email.body_text or "")[:4000]

    prompt = SUMMARY_PROMPT.format(
        sender=f"{email.sender_name} <{email.sender}>",
        subject=email.subject or "(no subject)",
        date=email.date.strftime("%Y-%m-%d %H:%M") if email.date else "",
        body=body,
    )

    response = client.chat.completions.create(
        model=config.openai_model,
        messages=[
            {
                "role": "system",
                "content": "You are a precise email summarizer. Return valid JSON."
            },
            {"role": "user", "content": prompt},
        ],
        temperature=0.2,
        max_tokens=300,
        response_format={"type": "json_object"},
    )

    raw = response.choices[0].message.content.strip()
    try:
        result = json.loads(raw)
    except json.JSONDecodeError:
        result = {
            "one_line": email.snippet[:80] if email.snippet else "",
            "detailed": email.snippet or "",
            "key_points": [],
        }

    return result

Action Item Extraction

The action extractor identifies tasks, deadlines, and assignments from email content. It detects phrases like "please review by Friday", "can you send me", and "deadline is March 30":

# Add to app/ai/smart.py

ACTION_ITEM_PROMPT = """Analyze this email and extract ALL action items.
An action item is anything the recipient needs to do, decide, or follow up on.

Return a JSON object with:
{
  "action_items": [
    {
      "description": "What needs to be done (max 200 chars)",
      "assignee": "who should do it (e.g., 'me', 'sender', 'team')",
      "due_date": "ISO date if mentioned, null otherwise",
      "urgency": "high" | "normal" | "low"
    }
  ],
  "has_deadline": true/false,
  "requires_response": true/false
}

If there are no action items, return {"action_items": [], "has_deadline": false, "requires_response": false}.

Return ONLY the JSON object.

--- EMAIL ---
From: {sender}
Subject: {subject}
Date: {date}

{body}
--- END EMAIL ---"""


def extract_action_items(email: Email) -> dict:
    """
    Extract action items from an email.

    Args:
        email: Email ORM object.

    Returns:
        Dict with action_items list, has_deadline, requires_response.
    """
    body = (email.body_text or "")[:3000]

    prompt = ACTION_ITEM_PROMPT.format(
        sender=f"{email.sender_name} <{email.sender}>",
        subject=email.subject or "(no subject)",
        date=email.date.strftime("%Y-%m-%d %H:%M") if email.date else "",
        body=body,
    )

    response = client.chat.completions.create(
        model=config.openai_model,
        messages=[
            {
                "role": "system",
                "content": "You extract action items precisely. Return valid JSON."
            },
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
        max_tokens=400,
        response_format={"type": "json_object"},
    )

    raw = response.choices[0].message.content.strip()
    try:
        result = json.loads(raw)
    except json.JSONDecodeError:
        result = {
            "action_items": [],
            "has_deadline": False,
            "requires_response": False,
        }

    return result


def extract_and_store_action_items(email_id: int) -> list:
    """
    Extract action items from an email and store them in the database.

    Args:
        email_id: Database ID of the email.

    Returns:
        List of ActionItem objects created.
    """
    session = SessionLocal()
    items = []

    try:
        email = session.query(Email).get(email_id)
        if not email:
            raise ValueError(f"Email {email_id} not found")

        result = extract_action_items(email)

        for item_data in result.get("action_items", []):
            due_date = None
            if item_data.get("due_date"):
                try:
                    due_date = datetime.fromisoformat(
                        item_data["due_date"]
                    )
                except (ValueError, TypeError):
                    pass

            action_item = ActionItem(
                email_id=email.id,
                description=item_data.get("description", "")[:500],
                due_date=due_date,
                is_completed=False,
            )
            session.add(action_item)
            items.append(action_item)

        session.commit()
        print(f"Extracted {len(items)} action items from: {email.subject[:50]}")

    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

    return items

Follow-Up Reminders

Detect emails that need follow-up and schedule reminders. The system checks for unanswered emails, pending requests, and time-sensitive threads:

# Add to app/ai/smart.py

def detect_follow_ups(days_threshold: int = 2, limit: int = 50) -> list:
    """
    Find emails that may need follow-up and create reminders.

    Criteria:
    - Classified as 'task' or 'question' category
    - Priority is 'high' or 'urgent'
    - No draft has been sent yet
    - Older than days_threshold

    Args:
        days_threshold: Days since email was received.
        limit: Maximum reminders to create.

    Returns:
        List of FollowUp objects created.
    """
    session = SessionLocal()
    follow_ups = []
    cutoff = datetime.utcnow() - timedelta(days=days_threshold)

    try:
        # Find emails needing follow-up
        emails = (
            session.query(Email)
            .join(Classification)
            .outerjoin(Draft, (Draft.email_id == Email.id) &
                       (Draft.status == "sent"))
            .outerjoin(FollowUp)
            .filter(
                Classification.category.in_(["task", "question"]),
                Classification.priority.in_(["urgent", "high"]),
                Email.date < cutoff,
                Draft.id.is_(None),   # No sent reply
                FollowUp.id.is_(None),  # No existing reminder
            )
            .limit(limit)
            .all()
        )

        for email in emails:
            reminder_date = datetime.utcnow() + timedelta(hours=4)

            follow_up = FollowUp(
                email_id=email.id,
                reminder_date=reminder_date,
                reason=(
                    f"No reply sent to {email.sender_name or email.sender} "
                    f"regarding: {email.subject[:80]}"
                ),
                is_dismissed=False,
            )
            session.add(follow_up)
            follow_ups.append(follow_up)

        session.commit()
        print(f"Created {len(follow_ups)} follow-up reminders")

    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

    return follow_ups


def get_pending_follow_ups() -> list:
    """Get all follow-ups that are due and not dismissed."""
    session = SessionLocal()
    try:
        now = datetime.utcnow()
        follow_ups = (
            session.query(FollowUp, Email)
            .join(Email, FollowUp.email_id == Email.id)
            .filter(
                FollowUp.reminder_date <= now,
                FollowUp.is_dismissed == False,
            )
            .order_by(FollowUp.reminder_date.asc())
            .all()
        )
        return follow_ups
    finally:
        session.close()

Daily Digest Generator

The digest summarizes your inbox activity into a concise morning briefing:

# Add to app/ai/smart.py

DIGEST_PROMPT = """Generate a concise daily email digest from these inbox stats
and emails. Format it as a briefing someone can scan in 30 seconds.

Stats:
- Total new emails: {total_new}
- Urgent: {urgent_count}
- High priority: {high_count}
- Tasks requiring action: {task_count}
- Pending follow-ups: {followup_count}

Top priority emails:
{priority_emails}

Pending action items:
{action_items}

Write a brief, scannable digest with sections:
1. Quick Summary (2-3 sentences)
2. Needs Immediate Attention (urgent items)
3. Action Items Due Today
4. FYI (notable non-urgent items)

Keep it concise and actionable."""


def generate_daily_digest() -> str:
    """
    Generate a daily digest summarizing inbox activity.

    Returns:
        Formatted digest text.
    """
    session = SessionLocal()

    try:
        today = datetime.utcnow().replace(hour=0, minute=0, second=0)
        yesterday = today - timedelta(days=1)

        # Count new emails by priority
        from sqlalchemy import func

        total_new = (
            session.query(Email)
            .filter(Email.fetched_at >= yesterday)
            .count()
        )

        priority_counts = (
            session.query(
                Classification.priority,
                func.count(Classification.id)
            )
            .join(Email)
            .filter(Email.fetched_at >= yesterday)
            .group_by(Classification.priority)
            .all()
        )
        counts = dict(priority_counts)

        # Get top priority emails
        top_emails = (
            session.query(Email, Classification)
            .join(Classification)
            .filter(
                Email.fetched_at >= yesterday,
                Classification.priority.in_(["urgent", "high"]),
            )
            .order_by(Email.date.desc())
            .limit(10)
            .all()
        )

        priority_email_text = "\n".join(
            f"- [{c.priority.upper()}] {e.subject} (from {e.sender_name or e.sender})"
            for e, c in top_emails
        ) or "No urgent or high-priority emails."

        # Get pending action items
        pending_items = (
            session.query(ActionItem, Email)
            .join(Email)
            .filter(ActionItem.is_completed == False)
            .order_by(ActionItem.due_date.asc().nullslast())
            .limit(10)
            .all()
        )

        action_item_text = "\n".join(
            f"- {ai.description} (from: {e.subject[:40]})"
            for ai, e in pending_items
        ) or "No pending action items."

        # Get follow-up count
        followup_count = (
            session.query(FollowUp)
            .filter(FollowUp.is_dismissed == False)
            .count()
        )

        # Generate digest with LLM
        prompt = DIGEST_PROMPT.format(
            total_new=total_new,
            urgent_count=counts.get("urgent", 0),
            high_count=counts.get("high", 0),
            task_count=len(pending_items),
            followup_count=followup_count,
            priority_emails=priority_email_text,
            action_items=action_item_text,
        )

        response = client.chat.completions.create(
            model=config.openai_model,
            messages=[
                {
                    "role": "system",
                    "content": "You write concise, scannable email digests."
                },
                {"role": "user", "content": prompt},
            ],
            temperature=0.3,
            max_tokens=600,
        )

        digest = response.choices[0].message.content.strip()
        return digest

    finally:
        session.close()

Background Scheduler

Tie all smart features together with APScheduler for automatic periodic execution:

# app/scheduler.py
"""Background job scheduler for periodic tasks."""
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger

from app.config import config


def create_scheduler() -> BackgroundScheduler:
    """
    Create and configure the background scheduler.

    Returns:
        Configured BackgroundScheduler (not started).
    """
    scheduler = BackgroundScheduler()

    # Poll Gmail for new emails
    scheduler.add_job(
        func=_poll_emails,
        trigger=IntervalTrigger(minutes=config.poll_interval_minutes),
        id="poll_emails",
        name="Poll Gmail for new emails",
        replace_existing=True,
    )

    # Classify new emails (runs 1 minute after polling)
    scheduler.add_job(
        func=_classify_emails,
        trigger=IntervalTrigger(minutes=config.poll_interval_minutes + 1),
        id="classify_emails",
        name="Classify unprocessed emails",
        replace_existing=True,
    )

    # Extract action items (runs 2 minutes after polling)
    scheduler.add_job(
        func=_extract_actions,
        trigger=IntervalTrigger(minutes=config.poll_interval_minutes + 2),
        id="extract_actions",
        name="Extract action items",
        replace_existing=True,
    )

    # Check for follow-ups (every 4 hours)
    scheduler.add_job(
        func=_check_follow_ups,
        trigger=IntervalTrigger(hours=4),
        id="check_follow_ups",
        name="Check for follow-up reminders",
        replace_existing=True,
    )

    # Daily digest (8:00 AM)
    scheduler.add_job(
        func=_generate_digest,
        trigger=CronTrigger(hour=8, minute=0),
        id="daily_digest",
        name="Generate daily digest",
        replace_existing=True,
    )

    return scheduler


def _poll_emails():
    """Job: Fetch new emails from Gmail."""
    try:
        from app.gmail import fetch_and_store_emails
        fetch_and_store_emails()
    except Exception as e:
        print(f"[Scheduler] Email poll error: {e}")


def _classify_emails():
    """Job: Classify unprocessed emails."""
    try:
        from app.ai.classifier import classify_unprocessed_emails
        classify_unprocessed_emails()
    except Exception as e:
        print(f"[Scheduler] Classification error: {e}")


def _extract_actions():
    """Job: Extract action items from recent emails."""
    try:
        from app.database import SessionLocal, Email, ActionItem
        from app.ai.smart import extract_and_store_action_items

        session = SessionLocal()
        # Find emails without action items extracted
        emails = (
            session.query(Email)
            .outerjoin(ActionItem)
            .filter(ActionItem.id.is_(None))
            .order_by(Email.date.desc())
            .limit(20)
            .all()
        )
        session.close()

        for email in emails:
            try:
                extract_and_store_action_items(email.id)
            except Exception as e:
                print(f"[Scheduler] Action extraction error: {e}")
    except Exception as e:
        print(f"[Scheduler] Action extraction error: {e}")


def _check_follow_ups():
    """Job: Create follow-up reminders."""
    try:
        from app.ai.smart import detect_follow_ups
        detect_follow_ups()
    except Exception as e:
        print(f"[Scheduler] Follow-up check error: {e}")


def _generate_digest():
    """Job: Generate and print daily digest."""
    try:
        from app.ai.smart import generate_daily_digest
        digest = generate_daily_digest()
        print(f"\n{'='*60}")
        print("DAILY DIGEST")
        print(f"{'='*60}")
        print(digest)
        print(f"{'='*60}\n")
    except Exception as e:
        print(f"[Scheduler] Digest error: {e}")

Testing Smart Features

# Test all smart features
python -c "
from app.database import SessionLocal, Email
from app.ai.smart import (
    summarize_email,
    extract_action_items,
    generate_daily_digest
)

session = SessionLocal()
email = session.query(Email).order_by(Email.date.desc()).first()
session.close()

if email:
    print('=== Summary ===')
    summary = summarize_email(email)
    print(f'One-line: {summary[\"one_line\"]}')
    print(f'Detailed: {summary[\"detailed\"]}')
    for point in summary.get('key_points', []):
        print(f'  - {point}')
    print()

    print('=== Action Items ===')
    actions = extract_action_items(email)
    for item in actions.get('action_items', []):
        due = item.get('due_date', 'No deadline')
        print(f'  [{item[\"urgency\"]}] {item[\"description\"]} (due: {due})')
    if not actions.get('action_items'):
        print('  No action items found')
    print()

    print('=== Daily Digest ===')
    digest = generate_daily_digest()
    print(digest)
else:
    print('No emails found. Run email integration first.')
"

Key Takeaways

• Summarization uses a slightly higher token limit (4000 chars) to capture more context from long threads.
• Action item extraction detects deadlines by looking for date patterns and urgency phrases in the email body.
• Follow-up reminders use a combination of classification data (priority, category) and time elapsed to identify emails needing attention.
• The daily digest aggregates all inbox intelligence into a scannable morning briefing.
• APScheduler runs all features on configurable intervals without blocking the web server.

What Is Next

In the next lesson, you will build the web interface — a Flask-powered dashboard with priority inbox view, draft review/edit panel, action item tracker, and one-click send workflow.