"""
Email Agent Database
Manages SQLite database for email logs, templates, and queue
"""
import sqlite3
from datetime import datetime
from typing import List, Optional, Dict, Any
from contextlib import contextmanager
class EmailDatabase:
"""Manages email agent database operations"""
def __init__(self, db_path: str = 'email_agent.db'):
"""
Initialize database connection
Args:
db_path: Path to SQLite database file
"""
self.db_path = db_path
self._init_schema()
@contextmanager
def get_connection(self):
"""Context manager for database connections"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def _init_schema(self):
"""Initialize database schema"""
with self.get_connection() as conn:
cursor = conn.cursor()
# Email log table
cursor.execute("""
CREATE TABLE IF NOT EXISTS email_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
from_email TEXT NOT NULL,
from_name TEXT,
subject TEXT,
body TEXT,
category TEXT,
priority TEXT,
sentiment TEXT,
action_taken TEXT,
response_sent TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
resolved BOOLEAN DEFAULT 0,
resolution_time INTEGER,
message_id TEXT UNIQUE
)
""")
# Email templates table
cursor.execute("""
CREATE TABLE IF NOT EXISTS email_templates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT UNIQUE,
name TEXT,
subject_template TEXT,
body_template TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
# Email queue table
cursor.execute("""
CREATE TABLE IF NOT EXISTS email_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
to_email TEXT NOT NULL,
to_name TEXT,
subject TEXT,
body TEXT,
action TEXT,
status TEXT DEFAULT 'pending',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
processed_at DATETIME,
error_message TEXT
)
""")
# Create indices for faster queries
cursor.execute("CREATE INDEX IF NOT EXISTS idx_email_log_status ON email_log(category, priority)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_queue_status ON email_queue(status)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_email_log_timestamp ON email_log(timestamp)")
conn.commit()
def log_email(self, email_data: Dict[str, Any]) -> int:
"""
Log an incoming email
Args:
email_data: Dictionary with email details
Returns:
ID of inserted row
"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO email_log (
from_email, from_name, subject, body,
category, priority, sentiment, action_taken,
message_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
email_data.get('from_email'),
email_data.get('from_name'),
email_data.get('subject'),
email_data.get('body'),
email_data.get('category'),
email_data.get('priority'),
email_data.get('sentiment'),
email_data.get('action_taken'),
email_data.get('message_id')
))
conn.commit()
return cursor.lastrowid
def update_email_log(self, log_id: int, updates: Dict[str, Any]) -> bool:
"""Update an email log entry"""
with self.get_connection() as conn:
cursor = conn.cursor()
set_clause = ", ".join([f"{k} = ?" for k in updates.keys()])
values = list(updates.values()) + [log_id]
cursor.execute(f"UPDATE email_log SET {set_clause} WHERE id = ?", values)
conn.commit()
return cursor.rowcount > 0
def get_email_log(self, log_id: int) -> Optional[Dict]:
"""Get email log entry by ID"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM email_log WHERE id = ?", (log_id,))
row = cursor.fetchone()
return dict(row) if row else None
def get_unresolved_emails(self, limit: int = 10) -> List[Dict]:
"""Get unresolved emails"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM email_log
WHERE resolved = 0
ORDER BY timestamp DESC
LIMIT ?
""", (limit,))
return [dict(row) for row in cursor.fetchall()]
def queue_email_response(self, response_data: Dict[str, Any]) -> int:
"""
Queue an email response to be sent
Args:
response_data: Dictionary with email response details
Returns:
ID of queued email
"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO email_queue (
to_email, to_name, subject, body, action, status
) VALUES (?, ?, ?, ?, ?, ?)
""", (
response_data.get('to_email'),
response_data.get('to_name'),
response_data.get('subject'),
response_data.get('body'),
response_data.get('action'),
'pending'
))
conn.commit()
return cursor.lastrowid
def get_pending_emails(self, limit: int = 10) -> List[Dict]:
"""Get pending emails from queue"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM email_queue
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT ?
""", (limit,))
return [dict(row) for row in cursor.fetchall()]
def mark_email_sent(self, queue_id: int) -> bool:
"""Mark queued email as sent"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
UPDATE email_queue
SET status = 'sent', processed_at = ?
WHERE id = ?
""", (datetime.now().isoformat(), queue_id))
conn.commit()
return cursor.rowcount > 0
def mark_email_failed(self, queue_id: int, error: str) -> bool:
"""Mark queued email as failed"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
UPDATE email_queue
SET status = 'failed', error_message = ?, processed_at = ?
WHERE id = ?
""", (error, datetime.now().isoformat(), queue_id))
conn.commit()
return cursor.rowcount > 0
def add_template(self, category: str, name: str,
subject_template: str, body_template: str) -> bool:
"""Add or update email template"""
with self.get_connection() as conn:
cursor = conn.cursor()
# Try update first
cursor.execute("""
UPDATE email_templates
SET name = ?, subject_template = ?, body_template = ?, updated_at = ?
WHERE category = ?
""", (name, subject_template, body_template, datetime.now().isoformat(), category))
# If no rows updated, insert
if cursor.rowcount == 0:
cursor.execute("""
INSERT INTO email_templates (category, name, subject_template, body_template)
VALUES (?, ?, ?, ?)
""", (category, name, subject_template, body_template))
conn.commit()
return True
def get_template(self, category: str) -> Optional[Dict]:
"""Get email template for category"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM email_templates WHERE category = ?", (category,))
row = cursor.fetchone()
return dict(row) if row else None
def get_statistics(self) -> Dict[str, Any]:
"""Get email agent statistics"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) as total FROM email_log")
total_emails = cursor.fetchone()['total']
cursor.execute("""
SELECT category, COUNT(*) as count
FROM email_log
GROUP BY category
""")
by_category = {row['category']: row['count'] for row in cursor.fetchall()}
cursor.execute("""
SELECT priority, COUNT(*) as count
FROM email_log
GROUP BY priority
""")
by_priority = {row['priority']: row['count'] for row in cursor.fetchall()}
cursor.execute("""
SELECT COUNT(*) as resolved
FROM email_log
WHERE resolved = 1
""")
resolved = cursor.fetchone()['resolved']
return {
'total_emails': total_emails,
'resolved': resolved,
'by_category': by_category,
'by_priority': by_priority,
'auto_response_rate': resolved / total_emails if total_emails > 0 else 0
}