279 lines
11 KiB
Python
279 lines
11 KiB
Python
from flask import request
|
|
from models import Notif, NotifType, db, EmailTemplate, Mail, KeyValueSettings
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime, timedelta
|
|
from flask_login import current_user
|
|
from sqlalchemy import desc
|
|
import logging
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
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 using the configured SMTP settings.
|
|
|
|
Args:
|
|
mail: The Mail object containing the email details
|
|
|
|
Returns:
|
|
bool: True if email was sent successfully, False otherwise
|
|
"""
|
|
smtp_settings = get_smtp_settings()
|
|
if not smtp_settings:
|
|
logger.error("Cannot send email: SMTP settings not configured")
|
|
return False
|
|
|
|
try:
|
|
# Create message
|
|
msg = MIMEMultipart()
|
|
msg['From'] = f"{smtp_settings['smtp_from_name']} <{smtp_settings['smtp_from_email']}>"
|
|
msg['To'] = mail.recipient
|
|
msg['Subject'] = mail.subject
|
|
|
|
# Attach HTML body
|
|
msg.attach(MIMEText(mail.body, 'html'))
|
|
|
|
# Create SMTP connection
|
|
if smtp_settings['smtp_security'] == 'ssl':
|
|
server = smtplib.SMTP_SSL(smtp_settings['smtp_host'], int(smtp_settings['smtp_port']))
|
|
else:
|
|
server = smtplib.SMTP(smtp_settings['smtp_host'], int(smtp_settings['smtp_port']))
|
|
if smtp_settings['smtp_security'] == 'tls':
|
|
server.starttls()
|
|
|
|
# Login if credentials are provided
|
|
if smtp_settings['smtp_username'] and smtp_settings['smtp_password']:
|
|
server.login(smtp_settings['smtp_username'], smtp_settings['smtp_password'])
|
|
|
|
# Send email
|
|
server.send_message(msg)
|
|
server.quit()
|
|
|
|
# Update mail status
|
|
mail.status = 'sent'
|
|
mail.sent_at = datetime.utcnow()
|
|
db.session.commit()
|
|
|
|
logger.info(f"Email sent successfully to {mail.recipient}")
|
|
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 details
|
|
if notif.details:
|
|
for key, value in notif.details.items():
|
|
# Handle nested keys (e.g., room.name -> room_name)
|
|
if '.' in key:
|
|
parts = key.split('.')
|
|
if len(parts) == 2:
|
|
obj_name, attr = parts
|
|
if obj_name in notif.details and isinstance(notif.details[obj_name], dict):
|
|
if attr in notif.details[obj_name]:
|
|
filled_body = filled_body.replace(f'{{{{ {key} }}}}', str(notif.details[obj_name][attr]))
|
|
else:
|
|
filled_body = filled_body.replace(f'{{{{ {key} }}}}', str(value))
|
|
|
|
# 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)
|
|
|
|
# Add timestamps
|
|
filled_body = filled_body.replace('{{ created_at }}', notif.timestamp.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'))
|
|
|
|
# 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 |