import os
from app.common.services.email_service import send_email
from app.common.twillio_helper import get_twillio_client
from app.repositories.base_repo import BaseRepo
from app.common.models import Advisor, SystemFailureLogs
from datetime import datetime, timedelta, timezone
from app.twilio_service.services.twilio_service import send_sms
from app.utils.helpers.common import get_app_url
from app.common.constants import SeverityLevel
class SystemFailureRepo(BaseRepo):
def __init__(self, db):
super().__init__(db)
def create_unique_log(self, component: str, severity: int, message: str = None, twilio_call_id: str = None):
log = self.db.query(SystemFailureLogs).filter(
SystemFailureLogs.component == component,
SystemFailureLogs.twilio_call_id == twilio_call_id,
SystemFailureLogs.twilio_call_id.isnot(None)
).first()
if log:
return log
else:
return self.create_log(component, severity, message, twilio_call_id)
def create_log(self, component: str, severity: int, message: str = None, twilio_call_id: str = None):
log = self.create(SystemFailureLogs, data={"component": component, "severity": severity, "message": message, "twilio_call_id": twilio_call_id})
self.check_for_alert(severity, component, message)
return log
def check_for_alert(self, severity: int, component: str, message: str = None):
ten_minutes_ago = datetime.now(timezone.utc) - timedelta(minutes=10)
recent_logs = self.db.query(SystemFailureLogs).filter(
SystemFailureLogs.created_at >= ten_minutes_ago,
SystemFailureLogs.severity == severity,
SystemFailureLogs.component == component
).count()
print(f"Recent logs: {recent_logs}")
if recent_logs >= 3:
if component == "transfer_call":
self.switch_to_ultravox()
self.send_ultravox_switch_alert()
self.send_system_failure_alert(component, severity, message)
def switch_to_ultravox(self):
client = get_twillio_client()
twilio_number = os.getenv("TWILIO_NUMBER")
trunkId = os.getenv("TRUNK_ID")
number = client.incoming_phone_numbers \
.list(phone_number=twilio_number)[0]
number_sid = number.sid
voice_url = f'{get_app_url()}/advisor/api/v1/twilio/incoming-webhook'
phone_number = client.incoming_phone_numbers(number_sid).update(
voice_url=voice_url,
voice_method='POST',
status_callback=f'{get_app_url()}/advisor/api/v1/twilio/call-status-callback',
status_callback_method='POST'
)
client.trunking.v1.trunks(trunkId).phone_numbers(number_sid).delete()
def send_system_failure_alert(self,component: str, severity: int, message: str):
self.send_system_failure_email(component, severity, message)
self.send_system_failure_sms(component, severity, message)
def send_system_failure_email(self, component: str, severity: int, message: str):
"""
Send email notification for system failures
"""
# Get recipient emails from environment variable or use default
manager_emails = []
manager_emails.append("shahzad@fluten.ai")
manager_emails.append("smalik@fluten.ai")
manager_emails.append("mimj1800@gmail.com")
recipient_emails = ",".join(manager_emails)
subject = "DealerPulse System Failure Alert"
email_body = f"""Dear Team,
A system failure has been detected in DealerPulse that needs your urgent attention:
Component: {component}
Severity: {SeverityLevel(severity).name}
Message: {message}
Please investigate this issue as soon as possible and take necessary actions to resolve it.
Regards,
DealerPulse System"""
html_body = f"""
Dear Team,
A system failure has been detected in DealerPulse that needs your urgent attention:
Component: {component}
Severity: {SeverityLevel(severity).name}
Message: {message if message is not None else "N/A"}
Please investigate this issue as soon as possible and take necessary actions to resolve it.
Regards,
DealerPulse System
"""
send_email(
recipient=recipient_emails,
subject=subject,
body=email_body,
html_body=html_body
)
def send_system_failure_sms(self, component: str, severity: int, message: str):
to_numbers = ['+923017357180', '+923334046969']
for number in to_numbers:
send_sms(
to_number=number,
message=f"System failure alert: {component} {SeverityLevel(severity).name} {message}",
from_number=os.getenv('TWILIO_FROM_NUMBER')
)
def send_ultravox_switch_alert(self):
self.send_ultravox_switch_email()
self.send_ultravox_switch_sms()
def send_ultravox_switch_email(self):
"""
Send email notification for ultravox switch
"""
# Get recipient emails from environment variable or use default
manager_emails = []
manager_emails.append("shahzad@fluten.ai")
manager_emails.append("smalik@fluten.ai")
manager_emails.append("mimj1800@gmail.com")
recipient_emails = ",".join(manager_emails)
subject = "DealerPulse System Failure Alert"
email_body = f"""Dear Team,
Due to multiple system failures, system has been automatically switched to ultravox.
Regards,
DealerPulse System"""
html_body = f"""
Dear Team,
Due to multiple system failures, system has been automatically switched to ultravox.
Regards,
DealerPulse System
"""
send_email(
recipient=recipient_emails,
subject=subject,
body=email_body,
html_body=html_body
)
def send_ultravox_switch_sms(self):
to_numbers = ['+923017357180', '+923334046969']
for number in to_numbers:
send_sms(
to_number=number,
message=f"System has been automatically switched to ultravox due to multiple system failures.",
from_number=os.getenv('TWILIO_FROM_NUMBER')
)