146 lines
5.1 KiB
Python
146 lines
5.1 KiB
Python
from flask import request
|
|
from models import Notif, NotifType, db, EmailTemplate, Mail
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime, timedelta
|
|
from flask_login import current_user
|
|
from sqlalchemy import desc
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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
|
|
if notif.user:
|
|
filled_body = filled_body.replace('{{ user.username }}', notif.user.username)
|
|
if notif.details:
|
|
for key, value in notif.details.items():
|
|
filled_body = filled_body.replace(f'{{{{ {key} }}}}', str(value))
|
|
|
|
# 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)
|
|
logger.debug(f"Created mail record: {mail}")
|
|
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 |