"""
Alert System Module
Proactive renewal alert system with state management
"""
from src.mysql_db import execute_query
from src.gmail_tools import send_renewal_alert
from datetime import datetime, timedelta
def check_renewals_3_days_ahead(user_email):
"""
Check for renewals 3 days ahead and send alerts
Args:
user_email: Email address to send alerts to
"""
try:
query = """
SELECT s.id, s.service, s.amount, s.renewal_date
FROM subscriptions s
WHERE s.renewal_date = DATE_ADD(CURDATE(), INTERVAL 3 DAY)
AND s.status = 'active'
AND NOT EXISTS (
SELECT 1 FROM renewal_alerts ra
WHERE ra.subscription_id = s.id
AND ra.alert_type = '3_day_alert'
AND ra.alert_sent = TRUE
AND ra.scheduled_date = s.renewal_date
)
"""
renewals = execute_query(query, fetch_all=True)
alerts_sent = []
errors = []
for renewal in renewals:
alert_result = send_renewal_alert(
user_email,
renewal['service'],
renewal['amount'],
renewal['renewal_date']
)
if alert_result.get('success'):
mark_alert_sent(renewal['id'], '3_day_alert', renewal['renewal_date'])
alerts_sent.append({
'subscription_id': renewal['id'],
'service': renewal['service'],
'amount': renewal['amount'],
'renewal_date': renewal['renewal_date']
})
else:
errors.append({
'subscription_id': renewal['id'],
'service': renewal['service'],
'error': alert_result.get('error')
})
return {
'success': True,
'alerts_sent': len(alerts_sent),
'details': alerts_sent,
'errors': errors
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def mark_alert_sent(subscription_id, alert_type, scheduled_date):
"""
Mark alert as sent in database
Args:
subscription_id: Subscription ID
alert_type: Type of alert (e.g., '3_day_alert')
scheduled_date: Date the renewal is scheduled
"""
try:
query = """
INSERT INTO renewal_alerts
(subscription_id, alert_type, alert_sent, alert_sent_at, scheduled_date)
VALUES (%s, %s, TRUE, NOW(), %s)
"""
execute_query(query, (subscription_id, alert_type, scheduled_date))
return {
'success': True,
'message': f"Alert marked as sent for subscription {subscription_id}"
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def get_alert_history(subscription_id=None, limit=50):
"""
Get history of sent alerts
Args:
subscription_id: Optional subscription ID to filter
limit: Maximum number of alerts to return
"""
try:
if subscription_id:
query = """
SELECT ra.*, s.service, s.amount
FROM renewal_alerts ra
JOIN subscriptions s ON ra.subscription_id = s.id
WHERE ra.subscription_id = %s
ORDER BY ra.alert_sent_at DESC
LIMIT %s
"""
params = (subscription_id, limit)
else:
query = """
SELECT ra.*, s.service, s.amount
FROM renewal_alerts ra
JOIN subscriptions s ON ra.subscription_id = s.id
ORDER BY ra.alert_sent_at DESC
LIMIT %s
"""
params = (limit,)
results = execute_query(query, params, fetch_all=True)
return {
'success': True,
'alerts': results,
'count': len(results)
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def reset_alert_status(subscription_id, alert_type):
"""
Reset alert status to allow resending
Args:
subscription_id: Subscription ID
alert_type: Type of alert to reset
"""
try:
query = """
DELETE FROM renewal_alerts
WHERE subscription_id = %s AND alert_type = %s
"""
result = execute_query(query, (subscription_id, alert_type))
return {
'success': True,
'message': f"Alert status reset for subscription {subscription_id}",
'affected_rows': result['affected_rows']
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
