165 lines
5.7 KiB
Python
165 lines
5.7 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import Optional, Dict, Any
|
|
import json
|
|
from app.models.notification import Notification, NotificationType, NotificationSettings
|
|
from app.models.user import User
|
|
|
|
|
|
class NotificationService:
|
|
@staticmethod
|
|
def create_notification(
|
|
db: Session,
|
|
recipient_id: int,
|
|
notification_type: NotificationType,
|
|
title: str,
|
|
message: str,
|
|
sender_id: Optional[int] = None,
|
|
data: Optional[Dict[Any, Any]] = None,
|
|
) -> Notification:
|
|
"""Create a new notification"""
|
|
|
|
# Check if user wants this type of notification
|
|
settings = (
|
|
db.query(NotificationSettings)
|
|
.filter(NotificationSettings.user_id == recipient_id)
|
|
.first()
|
|
)
|
|
|
|
if settings and not NotificationService._should_send_notification(
|
|
settings, notification_type
|
|
):
|
|
return None
|
|
|
|
notification = Notification(
|
|
recipient_id=recipient_id,
|
|
sender_id=sender_id,
|
|
type=notification_type,
|
|
title=title,
|
|
message=message,
|
|
data=json.dumps(data) if data else None,
|
|
)
|
|
|
|
db.add(notification)
|
|
db.commit()
|
|
db.refresh(notification)
|
|
return notification
|
|
|
|
@staticmethod
|
|
def _should_send_notification(
|
|
settings: NotificationSettings, notification_type: NotificationType
|
|
) -> bool:
|
|
"""Check if user wants to receive this type of notification"""
|
|
type_mapping = {
|
|
NotificationType.CONNECTION_REQUEST: settings.connection_requests,
|
|
NotificationType.CONNECTION_ACCEPTED: settings.connection_requests,
|
|
NotificationType.POST_LIKE: settings.post_interactions,
|
|
NotificationType.POST_COMMENT: settings.post_interactions,
|
|
NotificationType.EVENT_INVITATION: settings.event_reminders,
|
|
NotificationType.EVENT_REMINDER: settings.event_reminders,
|
|
NotificationType.PRAYER_REQUEST: settings.prayer_requests,
|
|
NotificationType.MINISTRY_ASSIGNMENT: settings.ministry_updates,
|
|
NotificationType.DONATION_REMINDER: settings.donation_reminders,
|
|
NotificationType.BIRTHDAY_REMINDER: settings.birthday_reminders,
|
|
NotificationType.ANNOUNCEMENT: settings.announcements,
|
|
}
|
|
return type_mapping.get(notification_type, True)
|
|
|
|
@staticmethod
|
|
def mark_as_read(db: Session, notification_id: int, user_id: int) -> bool:
|
|
"""Mark notification as read"""
|
|
notification = (
|
|
db.query(Notification)
|
|
.filter(
|
|
Notification.id == notification_id, Notification.recipient_id == user_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if notification:
|
|
notification.is_read = True
|
|
from datetime import datetime
|
|
|
|
notification.read_at = datetime.utcnow()
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def mark_all_as_read(db: Session, user_id: int) -> int:
|
|
"""Mark all notifications as read for a user"""
|
|
from datetime import datetime
|
|
|
|
count = (
|
|
db.query(Notification)
|
|
.filter(Notification.recipient_id == user_id, Notification.is_read is False)
|
|
.update({"is_read": True, "read_at": datetime.utcnow()})
|
|
)
|
|
db.commit()
|
|
return count
|
|
|
|
@staticmethod
|
|
def get_unread_count(db: Session, user_id: int) -> int:
|
|
"""Get count of unread notifications"""
|
|
return (
|
|
db.query(Notification)
|
|
.filter(Notification.recipient_id == user_id, Notification.is_read is False)
|
|
.count()
|
|
)
|
|
|
|
@staticmethod
|
|
def create_connection_request_notification(
|
|
db: Session, sender: User, recipient_id: int
|
|
):
|
|
"""Create notification for connection request"""
|
|
return NotificationService.create_notification(
|
|
db=db,
|
|
recipient_id=recipient_id,
|
|
sender_id=sender.id,
|
|
notification_type=NotificationType.CONNECTION_REQUEST,
|
|
title="New Connection Request",
|
|
message=f"{sender.first_name} {sender.last_name} wants to connect with you",
|
|
data={"sender_name": f"{sender.first_name} {sender.last_name}"},
|
|
)
|
|
|
|
@staticmethod
|
|
def create_post_like_notification(
|
|
db: Session, liker: User, post_author_id: int, post_id: int
|
|
):
|
|
"""Create notification for post like"""
|
|
if liker.id == post_author_id: # Don't notify self
|
|
return None
|
|
|
|
return NotificationService.create_notification(
|
|
db=db,
|
|
recipient_id=post_author_id,
|
|
sender_id=liker.id,
|
|
notification_type=NotificationType.POST_LIKE,
|
|
title="Post Liked",
|
|
message=f"{liker.first_name} {liker.last_name} liked your post",
|
|
data={
|
|
"post_id": post_id,
|
|
"liker_name": f"{liker.first_name} {liker.last_name}",
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def create_post_comment_notification(
|
|
db: Session, commenter: User, post_author_id: int, post_id: int
|
|
):
|
|
"""Create notification for post comment"""
|
|
if commenter.id == post_author_id: # Don't notify self
|
|
return None
|
|
|
|
return NotificationService.create_notification(
|
|
db=db,
|
|
recipient_id=post_author_id,
|
|
sender_id=commenter.id,
|
|
notification_type=NotificationType.POST_COMMENT,
|
|
title="New Comment",
|
|
message=f"{commenter.first_name} {commenter.last_name} commented on your post",
|
|
data={
|
|
"post_id": post_id,
|
|
"commenter_name": f"{commenter.first_name} {commenter.last_name}",
|
|
},
|
|
)
|