283 lines
12 KiB
Python
283 lines
12 KiB
Python
from flask import request
|
|
from models import Notif, NotifType, db, EmailTemplate, Mail, KeyValueSettings, User
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime, timedelta
|
|
from flask_login import current_user
|
|
from sqlalchemy import desc, and_
|
|
import logging
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.utils import formatdate
|
|
import json
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_smtp_settings() -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get SMTP settings from the database.
|
|
|
|
Returns:
|
|
Dictionary containing SMTP settings or None if not configured
|
|
"""
|
|
try:
|
|
smtp_settings = KeyValueSettings.get_value('smtp_settings')
|
|
if not smtp_settings:
|
|
logger.warning("No SMTP settings found in database")
|
|
return None
|
|
return smtp_settings
|
|
except Exception as e:
|
|
logger.error(f"Error retrieving SMTP settings: {str(e)}")
|
|
return None
|
|
|
|
def send_email_via_smtp(mail: Mail) -> bool:
|
|
"""Send an email synchronously"""
|
|
try:
|
|
# Get SMTP settings
|
|
smtp_settings = get_smtp_settings()
|
|
if not smtp_settings:
|
|
logger.error("SMTP settings not found")
|
|
mail.status = 'failed'
|
|
mail.error_message = "SMTP settings not found"
|
|
db.session.commit()
|
|
return False
|
|
|
|
# Create message
|
|
msg = MIMEMultipart()
|
|
msg['From'] = smtp_settings.sender_email
|
|
msg['To'] = mail.recipient_email
|
|
msg['Subject'] = mail.subject
|
|
msg['Date'] = formatdate(localtime=True)
|
|
|
|
# Add HTML content
|
|
msg.attach(MIMEText(mail.content, 'html'))
|
|
|
|
# Send email
|
|
with smtplib.SMTP(smtp_settings.smtp_server, smtp_settings.smtp_port) as server:
|
|
if smtp_settings.use_tls:
|
|
server.starttls()
|
|
if smtp_settings.username and smtp_settings.password:
|
|
server.login(smtp_settings.username, smtp_settings.password)
|
|
server.send_message(msg)
|
|
|
|
# Update mail status
|
|
mail.status = 'sent'
|
|
mail.sent_at = datetime.utcnow()
|
|
db.session.commit()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error sending email: {str(e)}")
|
|
mail.status = 'failed'
|
|
mail.error_message = str(e)
|
|
db.session.commit()
|
|
return False
|
|
|
|
def generate_mail_from_notification(notif: Notif) -> Optional[Mail]:
|
|
"""
|
|
Generate a mail record from a notification using the appropriate email template.
|
|
|
|
Args:
|
|
notif: The notification object to generate mail from
|
|
|
|
Returns:
|
|
The created Mail object or None if no template was found
|
|
"""
|
|
logger.debug(f"Generating mail for notification: {notif}")
|
|
|
|
# Convert notification type to template name format (e.g., 'account_created' -> 'Account Created')
|
|
template_name = ' '.join(word.capitalize() for word in notif.notif_type.split('_'))
|
|
|
|
# Find the corresponding email template based on notif_type
|
|
template = EmailTemplate.query.filter_by(name=template_name).first()
|
|
if not template:
|
|
logger.warning(f"No email template found for notification type: {notif.notif_type} (template name: {template_name})")
|
|
return None
|
|
|
|
try:
|
|
# Fill in the template with notification details
|
|
filled_body = template.body
|
|
|
|
# Add user information
|
|
if notif.user:
|
|
filled_body = filled_body.replace('{{ user.username }}', notif.user.username + ' ' + notif.user.last_name)
|
|
filled_body = filled_body.replace('{{ user.email }}', notif.user.email)
|
|
if hasattr(notif.user, 'company'):
|
|
filled_body = filled_body.replace('{{ user.company }}', notif.user.company or '')
|
|
if hasattr(notif.user, 'position'):
|
|
filled_body = filled_body.replace('{{ user.position }}', notif.user.position or '')
|
|
|
|
# Add sender information if available
|
|
if notif.sender:
|
|
filled_body = filled_body.replace('{{ sender.username }}', notif.sender.username + ' ' + notif.sender.last_name)
|
|
filled_body = filled_body.replace('{{ sender.email }}', notif.sender.email)
|
|
|
|
# Add site information
|
|
from models import SiteSettings
|
|
site_settings = SiteSettings.query.first()
|
|
if site_settings:
|
|
filled_body = filled_body.replace('{{ site.company_name }}', site_settings.company_name or '')
|
|
filled_body = filled_body.replace('{{ site.company_website }}', site_settings.company_website or '')
|
|
|
|
# Add notification-specific variables
|
|
if notif.details:
|
|
if 'setup_link' in filled_body and 'setup_link' in notif.details:
|
|
filled_body = filled_body.replace('{{ setup_link }}', notif.details['setup_link'])
|
|
if 'reset_link' in filled_body and 'reset_link' in notif.details:
|
|
filled_body = filled_body.replace('{{ reset_link }}', notif.details['reset_link'])
|
|
if 'expiry_time' in filled_body and 'expiry_time' in notif.details:
|
|
filled_body = filled_body.replace('{{ expiry_time }}', notif.details['expiry_time'])
|
|
if 'created_by' in filled_body and 'created_by' in notif.details:
|
|
filled_body = filled_body.replace('{{ created_by }}', notif.details['created_by'])
|
|
if 'deleted_by' in filled_body and 'deleted_by' in notif.details:
|
|
filled_body = filled_body.replace('{{ deleted_by }}', notif.details['deleted_by'])
|
|
if 'updated_by' in filled_body and 'updated_by' in notif.details:
|
|
filled_body = filled_body.replace('{{ updated_by }}', notif.details['updated_by'])
|
|
if 'remover.username' in filled_body and 'remover' in notif.details:
|
|
filled_body = filled_body.replace('{{ remover.username }}', notif.details['remover'])
|
|
if 'sender.username' in filled_body and 'sender' in notif.details:
|
|
filled_body = filled_body.replace('{{ sender.username }}', notif.details['sender'])
|
|
if 'conversation.name' in filled_body and 'conversation' in notif.details:
|
|
filled_body = filled_body.replace('{{ conversation.name }}', notif.details['conversation'])
|
|
if 'conversation.description' in filled_body and 'conversation_description' in notif.details:
|
|
filled_body = filled_body.replace('{{ conversation.description }}', notif.details['conversation_description'])
|
|
if 'message.content' in filled_body and 'message' in notif.details:
|
|
filled_body = filled_body.replace('{{ message.content }}', notif.details['message'])
|
|
if 'message.created_at' in filled_body and 'message_created_at' in notif.details:
|
|
filled_body = filled_body.replace('{{ message.created_at }}', notif.details['message_created_at'])
|
|
if 'message_link' in filled_body and 'message_link' in notif.details:
|
|
filled_body = filled_body.replace('{{ message_link }}', notif.details['message_link'])
|
|
if 'updated_fields' in filled_body and 'updated_fields' in notif.details:
|
|
filled_body = filled_body.replace('{{ updated_fields }}', notif.details['updated_fields'])
|
|
if 'created_at' in filled_body:
|
|
filled_body = filled_body.replace('{{ created_at }}', datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S'))
|
|
if 'updated_at' in filled_body:
|
|
filled_body = filled_body.replace('{{ updated_at }}', datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S'))
|
|
if 'deleted_at' in filled_body:
|
|
filled_body = filled_body.replace('{{ deleted_at }}', datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S'))
|
|
if 'removed_at' in filled_body:
|
|
filled_body = filled_body.replace('{{ removed_at }}', datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S'))
|
|
|
|
# Handle special URL variables
|
|
if 'room_link' in filled_body and 'room_id' in notif.details:
|
|
from flask import url_for
|
|
room_link = url_for('rooms.room', room_id=notif.details['room_id'], _external=True)
|
|
filled_body = filled_body.replace('{{ room_link }}', room_link)
|
|
|
|
if 'conversation_link' in filled_body and 'conversation_id' in notif.details:
|
|
from flask import url_for
|
|
conversation_link = url_for('conversations.conversation', conversation_id=notif.details['conversation_id'], _external=True)
|
|
filled_body = filled_body.replace('{{ conversation_link }}', conversation_link)
|
|
|
|
# Create a new Mail record
|
|
mail = Mail(
|
|
recipient=notif.user.email,
|
|
subject=template.subject,
|
|
body=filled_body,
|
|
status='pending',
|
|
template_id=template.id,
|
|
notif_id=notif.id
|
|
)
|
|
db.session.add(mail)
|
|
|
|
# Try to send the email immediately
|
|
if send_email_via_smtp(mail):
|
|
logger.debug(f"Email sent successfully for notification: {notif}")
|
|
else:
|
|
logger.warning(f"Failed to send email for notification: {notif}")
|
|
|
|
return mail
|
|
except Exception as e:
|
|
logger.error(f"Error generating mail from notification: {str(e)}")
|
|
return None
|
|
|
|
def create_notification(
|
|
notif_type: str,
|
|
user_id: int,
|
|
sender_id: Optional[int] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
generate_mail: bool = True
|
|
) -> Notif:
|
|
"""
|
|
Create a notification in the database and optionally generate a mail record.
|
|
|
|
Args:
|
|
notif_type: The type of notification (must match NotifType enum)
|
|
user_id: The ID of the user to notify
|
|
sender_id: Optional ID of the user who triggered the notification
|
|
details: Optional dictionary containing notification details
|
|
generate_mail: Whether to generate a mail record for this notification
|
|
|
|
Returns:
|
|
The created Notif object
|
|
"""
|
|
logger.debug(f"Creating notification of type: {notif_type}")
|
|
logger.debug(f"Notification details: {details}")
|
|
|
|
try:
|
|
notif = Notif(
|
|
notif_type=notif_type,
|
|
user_id=user_id,
|
|
sender_id=sender_id,
|
|
timestamp=datetime.utcnow(),
|
|
details=details or {},
|
|
read=False
|
|
)
|
|
|
|
logger.debug(f"Created notification object: {notif}")
|
|
db.session.add(notif)
|
|
|
|
# Generate mail if requested
|
|
if generate_mail:
|
|
mail = generate_mail_from_notification(notif)
|
|
if mail:
|
|
logger.debug(f"Generated mail record for notification: {mail}")
|
|
|
|
# Don't commit here - let the caller handle the transaction
|
|
logger.debug("Notification object added to session")
|
|
return notif
|
|
except Exception as e:
|
|
logger.error(f"Error creating notification: {str(e)}")
|
|
raise
|
|
|
|
def get_user_notifications(user_id: int, limit: int = 50, unread_only: bool = False) -> List[Notif]:
|
|
"""Get recent notifications for a specific user"""
|
|
query = Notif.query.filter_by(user_id=user_id)
|
|
if unread_only:
|
|
query = query.filter_by(read=False)
|
|
return query.order_by(desc(Notif.timestamp)).limit(limit).all()
|
|
|
|
def mark_notification_read(notif_id: int) -> bool:
|
|
"""Mark a notification as read"""
|
|
notif = Notif.query.get(notif_id)
|
|
if notif:
|
|
notif.read = True
|
|
db.session.commit()
|
|
return True
|
|
return False
|
|
|
|
def mark_all_notifications_read(user_id: int) -> int:
|
|
"""Mark all notifications as read for a user"""
|
|
result = Notif.query.filter_by(user_id=user_id, read=False).update({'read': True})
|
|
db.session.commit()
|
|
return result
|
|
|
|
def get_unread_count(user_id: int) -> int:
|
|
"""Get count of unread notifications for a user"""
|
|
return Notif.query.filter_by(user_id=user_id, read=False).count()
|
|
|
|
def delete_notification(notif_id: int) -> bool:
|
|
"""Delete a notification"""
|
|
notif = Notif.query.get(notif_id)
|
|
if notif:
|
|
db.session.delete(notif)
|
|
db.session.commit()
|
|
return True
|
|
return False
|
|
|
|
def delete_old_notifications(days: int = 30) -> int:
|
|
"""Delete notifications older than specified days"""
|
|
cutoff_date = datetime.utcnow() - timedelta(days=days)
|
|
result = Notif.query.filter(Notif.timestamp < cutoff_date).delete()
|
|
db.session.commit()
|
|
return result |