157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
"""
|
|
Notification services for emails, alerts, and reminders.
|
|
"""
|
|
from typing import Dict, Any
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from django.core.mail import send_mail
|
|
from django.template.loader import render_to_string
|
|
from django.conf import settings
|
|
from backend.core.repositories.services.services import ServiceRepository
|
|
from backend.core.repositories.invoices.invoices import InvoiceRepository
|
|
|
|
|
|
class NotificationService:
|
|
"""
|
|
Service for sending notifications and alerts.
|
|
"""
|
|
|
|
@staticmethod
|
|
def send_email(
|
|
recipient_email: str,
|
|
subject: str,
|
|
template_name: str,
|
|
context: Dict[str, Any]
|
|
) -> bool:
|
|
"""
|
|
Send an email using a template.
|
|
|
|
Args:
|
|
recipient_email: Recipient email address
|
|
subject: Email subject
|
|
template_name: Template name
|
|
context: Template context
|
|
|
|
Returns:
|
|
True if email was sent successfully, False otherwise
|
|
"""
|
|
try:
|
|
html_content = render_to_string(f"emails/{template_name}.html", context)
|
|
text_content = render_to_string(f"emails/{template_name}.txt", context)
|
|
|
|
send_mail(
|
|
subject=subject,
|
|
message=text_content,
|
|
from_email=settings.DEFAULT_FROM_EMAIL,
|
|
recipient_list=[recipient_email],
|
|
html_message=html_content,
|
|
fail_silently=False
|
|
)
|
|
|
|
return True
|
|
except Exception as e:
|
|
# Log the error
|
|
print(f"Error sending email: {str(e)}")
|
|
return False
|
|
|
|
@staticmethod
|
|
def send_service_reminders() -> int:
|
|
"""
|
|
Send service reminders for tomorrow's services.
|
|
|
|
Returns:
|
|
Number of reminders sent
|
|
"""
|
|
tomorrow = timezone.now().date() + timedelta(days=1)
|
|
|
|
# Get services scheduled for tomorrow
|
|
services = ServiceRepository.get_all(date=tomorrow)
|
|
|
|
reminder_count = 0
|
|
|
|
for service in services:
|
|
# Get the account
|
|
account = service.account
|
|
|
|
# Skip if no account (shouldn't happen)
|
|
if not account:
|
|
continue
|
|
|
|
# Get team members
|
|
team_members = service.team_members.all()
|
|
|
|
# Send reminder to each team member
|
|
for member in team_members:
|
|
# Skip if no email
|
|
if not member.email:
|
|
continue
|
|
|
|
# Prepare context
|
|
context = {
|
|
'service': service,
|
|
'account': account,
|
|
'team_member': member,
|
|
'service_time': f"{service.deadline_start.strftime('%I:%M %p')} - {service.deadline_end.strftime('%I:%M %p')}" if service.deadline_start and service.deadline_end else "All day"
|
|
}
|
|
|
|
# Send email
|
|
success = NotificationService.send_email(
|
|
recipient_email=member.email,
|
|
subject=f"Service Reminder: {account.name} - {tomorrow.strftime('%m/%d/%Y')}",
|
|
template_name="service_reminder",
|
|
context=context
|
|
)
|
|
|
|
if success:
|
|
reminder_count += 1
|
|
|
|
return reminder_count
|
|
|
|
@staticmethod
|
|
def send_overdue_invoice_reminders() -> int:
|
|
"""
|
|
Send reminders for overdue invoices.
|
|
|
|
Returns:
|
|
Number of reminders sent
|
|
"""
|
|
# Get overdue invoices
|
|
overdue_invoices = InvoiceRepository.get_overdue()
|
|
|
|
reminder_count = 0
|
|
|
|
for invoice in overdue_invoices:
|
|
# Get the customer
|
|
customer = invoice.customer
|
|
|
|
# Skip if no customer (shouldn't happen)
|
|
if not customer:
|
|
continue
|
|
|
|
# Skip if no billing email
|
|
if not customer.billing_email:
|
|
continue
|
|
|
|
# Calculate days overdue
|
|
days_overdue = (timezone.now().date() - invoice.date).days - 30
|
|
|
|
# Prepare context
|
|
context = {
|
|
'invoice': invoice,
|
|
'customer': customer,
|
|
'days_overdue': days_overdue,
|
|
'total_amount': invoice.total_amount
|
|
}
|
|
|
|
# Send email
|
|
success = NotificationService.send_email(
|
|
recipient_email=customer.billing_email,
|
|
subject=f"Overdue Invoice Reminder: {invoice.id}",
|
|
template_name="overdue_invoice_reminder",
|
|
context=context
|
|
)
|
|
|
|
if success:
|
|
reminder_count += 1
|
|
|
|
return reminder_count |