"""Email campaign delivery tracking — outbound email campaigns and delivery metrics."""
from datetime import datetime, timezone
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from app.models import db, EmailCampaign, EmailDeliveryLog
from app.utils.csrf import require_csrf
email_campaigns_bp = Blueprint('email_campaigns', __name__, url_prefix='/api/email-campaigns')
def _require_super_admin():
"""Check that current user is super_admin. Returns error tuple or None."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
if current_user.role != 'super_admin':
return jsonify({'error': 'Admin access required'}), 403
return None
# =============================================================================
# Campaign CRUD
# =============================================================================
@email_campaigns_bp.route('/', methods=['GET'])
@login_required
def list_campaigns():
"""List all email campaigns."""
err = _require_super_admin()
if err:
return err
campaigns = EmailCampaign.query.order_by(EmailCampaign.created_at.desc()).all()
return jsonify({
'items': [c.to_dict() for c in campaigns],
'total': len(campaigns),
})
@email_campaigns_bp.route('/', methods=['POST'])
@login_required
@require_csrf
def create_campaign():
"""Create a new email campaign."""
err = _require_super_admin()
if err:
return err
data = request.get_json()
if not data or not data.get('name'):
return jsonify({'error': 'Name is required'}), 400
campaign = EmailCampaign(
name=data['name'],
status=data.get('status', 'draft'),
subject=data.get('subject', ''),
body=data.get('body', ''),
created_by=current_user.id,
)
db.session.add(campaign)
# Create delivery logs for recipients if provided
recipients = data.get('recipients', [])
if recipients:
for recipient in recipients:
log = EmailDeliveryLog(
campaign_id=campaign.id,
recipient=recipient.get('email') if isinstance(recipient, dict) else recipient,
status='pending',
)
db.session.add(log)
db.session.commit()
return jsonify({'campaign': campaign.to_dict(), 'message': 'Campaign created'}), 201
@email_campaigns_bp.route('/<campaign_id>', methods=['GET'])
@login_required
def get_campaign(campaign_id):
"""Get campaign detail with delivery logs."""
err = _require_super_admin()
if err:
return err
campaign = db.session.get(EmailCampaign, campaign_id)
if not campaign:
return jsonify({'error': 'Campaign not found'}), 404
logs = EmailDeliveryLog.query.filter_by(campaign_id=campaign_id).order_by(
EmailDeliveryLog.created_at.desc()
).all()
return jsonify({
'campaign': campaign.to_dict(),
'logs': [l.to_dict() for l in logs],
})
@email_campaigns_bp.route('/<campaign_id>', methods=['PUT'])
@login_required
@require_csrf
def update_campaign(campaign_id):
"""Update campaign status or details."""
err = _require_super_admin()
if err:
return err
campaign = db.session.get(EmailCampaign, campaign_id)
if not campaign:
return jsonify({'error': 'Campaign not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request body'}), 400
if 'status' in data:
# Trigger actual email sending when status changes to 'sending'
if data['status'] == 'sending' and campaign.status == 'draft':
from app.utils.mail import send_campaign_email
# Get pending recipients
logs = EmailDeliveryLog.query.filter_by(
campaign_id=campaign_id,
status='pending'
).all()
recipients = [log.recipient for log in logs]
# Send emails
results = send_campaign_email(
campaign_id=campaign_id,
recipients=recipients,
subject=campaign.subject,
html_body=campaign.body,
)
# Update campaign counters
campaign.total_sent = (campaign.total_sent or 0) + results['sent']
campaign.failed = (campaign.failed or 0) + results['failed']
campaign.status = data['status']
if 'subject' in data:
campaign.subject = data['subject']
if 'body' in data:
campaign.body = data['body']
# Update counters if provided
for field in ('total_sent', 'delivered', 'opened', 'bounced', 'failed'):
if field in data:
setattr(campaign, field, data[field])
# Set sent_at when status changes to sending/sent
if data.get('status') in ('sending', 'sent') and not campaign.sent_at:
campaign.sent_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({'campaign': campaign.to_dict(), 'message': 'Campaign updated'})
@email_campaigns_bp.route('/<campaign_id>/send', methods=['POST'])
@login_required
@require_csrf
def send_campaign(campaign_id):
"""Send emails for a campaign."""
err = _require_super_admin()
if err:
return err
campaign = db.session.get(EmailCampaign, campaign_id)
if not campaign:
return jsonify({'error': 'Campaign not found'}), 404
if campaign.status != 'draft':
return jsonify({'error': 'Campaign must be in draft status to send'}), 400
# Get pending recipients
logs = EmailDeliveryLog.query.filter_by(
campaign_id=campaign_id,
status='pending'
).all()
recipients = [log.recipient for log in logs]
if not recipients:
return jsonify({'error': 'No pending recipients'}), 400
from app.utils.mail import send_campaign_email
# Send emails
results = send_campaign_email(
campaign_id=campaign_id,
recipients=recipients,
subject=campaign.subject,
html_body=campaign.body,
)
# Update campaign
campaign.status = 'sending'
campaign.sent_at = datetime.now(timezone.utc)
campaign.total_sent = (campaign.total_sent or 0) + results['sent']
campaign.failed = (campaign.failed or 0) + results['failed']
if results['sent'] > 0 and results['failed'] == 0:
campaign.status = 'sent'
db.session.commit()
return jsonify({
'campaign': campaign.to_dict(),
'results': results,
'message': f"Sent {results['sent']}, failed {results['failed']}"
})
@email_campaigns_bp.route('/<campaign_id>', methods=['DELETE'])
@login_required
@require_csrf
def delete_campaign(campaign_id):
"""Delete a campaign and its logs."""
err = _require_super_admin()
if err:
return err
campaign = db.session.get(EmailCampaign, campaign_id)
if not campaign:
return jsonify({'error': 'Campaign not found'}), 404
db.session.delete(campaign)
db.session.commit()
return jsonify({'message': 'Campaign deleted'})
# =============================================================================
# Delivery Log management
# =============================================================================
@email_campaigns_bp.route('/<campaign_id>/logs', methods=['GET'])
@login_required
def list_logs(campaign_id):
"""List delivery logs for a campaign."""
err = _require_super_admin()
if err:
return err
campaign = db.session.get(EmailCampaign, campaign_id)
if not campaign:
return jsonify({'error': 'Campaign not found'}), 404
logs = EmailDeliveryLog.query.filter_by(campaign_id=campaign_id).order_by(
EmailDeliveryLog.created_at.desc()
).all()
return jsonify({
'items': [l.to_dict() for l in logs],
'total': len(logs),
})
@email_campaigns_bp.route('/<campaign_id>/logs/<log_id>', methods=['PUT'])
@login_required
@require_csrf
def update_log(campaign_id, log_id):
"""Update delivery log status (e.g. mark as delivered/opened)."""
err = _require_super_admin()
if err:
return err
log = EmailDeliveryLog.query.filter_by(id=log_id, campaign_id=campaign_id).first()
if not log:
return jsonify({'error': 'Log entry not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request body'}), 400
if 'status' in data:
log.status = data['status']
if 'error_message' in data:
log.error_message = data['error_message']
now = datetime.now(timezone.utc)
if data.get('status') == 'sent' and not log.sent_at:
log.sent_at = now
if data.get('status') in ('delivered', 'opened') and not log.delivered_at:
log.delivered_at = now
if data.get('status') == 'opened' and not log.opened_at:
log.opened_at = now
db.session.commit()
return jsonify({'log': log.to_dict(), 'message': 'Log updated'})