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 # 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) 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