accounts.fcm_service

  1"""
  2FCM (Firebase Cloud Messaging) service for sending notifications to users.
  3
  4Features:
  5- Initializes Firebase app using service account credentials.
  6- Send notification to a single user via FCM token.
  7- Send notification to multiple users via FCM tokens.
  8- Memo notification helpers for memo creation and tagging.
  9
 10Functions:
 11- send_fcm_to_user(token, title, body, data): Send notification to a single user.
 12- send_fcm_to_users(tokens, title, body, data): Send notification to multiple users.
 13- memo_created_notification(info, memoId): Notify approvers and responders when a memo is created.
 14- memo_tagged_notification(info, tagged_role_name, tagged_role_id, memoId): Notify responders when a memo is tagged to their role.
 15"""
 16
 17import firebase_admin
 18from firebase_admin import credentials, messaging
 19import os
 20from .models import User
 21from django.db.models import Q
 22
 23BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 24FIREBASE_CRED_PATH = os.path.join(BASE_DIR, 'memo-track-firebase-adminsdk-fbsvc-060fb4e2e6.json')
 25
 26if not firebase_admin._apps:
 27    cred = credentials.Certificate(FIREBASE_CRED_PATH)
 28    firebase_admin.initialize_app(cred)
 29
 30def send_fcm_to_user(token: str, title: str, body: str, data: dict = None):
 31    """
 32    Send a notification to a single user via FCM token.
 33
 34    Args:
 35        token (str): FCM token of the user.
 36        title (str): Notification title.
 37        body (str): Notification body.
 38        data (dict): Additional data payload.
 39
 40    Returns:
 41        dict: Success or error response.
 42    """
 43    if not token:
 44        return {"error": "Missing FCM token"}
 45
 46    message = messaging.Message(
 47        notification=messaging.Notification(title=title, body=body),
 48        token=token,
 49        data=data or {}
 50    )
 51
 52    try:
 53        response = messaging.send(message)
 54        return {"success": True, "response": response}
 55    except Exception as e:
 56        return {"error": str(e)}
 57
 58def send_fcm_to_users(tokens: list[str], title: str, body: str, data: dict = None):
 59    """
 60    Send a notification to multiple users via FCM tokens.
 61
 62    Args:
 63        tokens (list[str]): List of FCM tokens.
 64        title (str): Notification title.
 65        body (str): Notification body.
 66        data (dict): Additional data payload.
 67
 68    Returns:
 69        dict: Success/failure counts and failed tokens.
 70    """
 71    if not tokens:
 72        return {"error": "No FCM tokens provided"}
 73
 74    message = messaging.MulticastMessage(
 75        notification=messaging.Notification(title=title, body=body),
 76        tokens=tokens,
 77        data=data or {},
 78    )
 79
 80    try:
 81        response = messaging.send_each_for_multicast(message)
 82        failed_tokens = []
 83
 84        for idx, resp in enumerate(response.responses):
 85            if not resp.success:
 86                failed_tokens.append(tokens[idx])
 87
 88        print({
 89            "success": response.success_count,
 90            "failure": response.failure_count,
 91            "failed_tokens": failed_tokens,
 92        })
 93        return {
 94            "success": response.success_count,
 95            "failure": response.failure_count,
 96            "failed_tokens": failed_tokens,
 97        }
 98
 99    except Exception as e:
100        print(f"Error sending FCM messages: {str(e)}")
101        return {"error": str(e)}
102    
103def memo_created_notification(info, memoId):
104    """
105    Send notification to approvers and responders when a memo is created.
106
107    Args:
108        info (dict): Memo info containing block, ward, and role details.
109        memoId (str): Memo ID.
110
111    Returns:
112        dict: FCM send result.
113    """
114    title = "New Memo Created"
115    body = f"Memo has been created at Block:{info['current_block_name']},floor {info['current_floor']}, ward {info['current_ward_name']}"
116    data = {
117        "type": "memo_created",
118        "memo_id": memoId,
119    }
120    approver_tokens = list(
121        User.objects.filter(
122            is_logged_in=True,
123            current_block=info["current_block_id"],
124            role__is_approver=True
125        ).exclude(Q(fcm_token__isnull=True) | Q(fcm_token=''))
126        .values_list('fcm_token', flat=True)
127    )
128
129    responder_tokens = list(
130        User.objects.filter(
131            is_logged_in=True,
132            role__id=info["tagged_role_id"]
133        ).exclude(Q(fcm_token__isnull=True) | Q(fcm_token=''))
134        .values_list('fcm_token', flat=True)
135    )
136    tokens = approver_tokens + responder_tokens
137    print(f"Tokens for notification: {tokens}")
138    if len(tokens) > 0:
139        return send_fcm_to_users(tokens, title, body, data)
140    
141def memo_tagged_notification(info, tagged_role_name, tagged_role_id, memoId):
142    """
143    Send notification to responders when a memo is tagged to their role.
144
145    Args:
146        info (dict): Memo info containing block, ward, and role details.
147        tagged_role_name (str): Name of the tagged role.
148        tagged_role_id (int): ID of the tagged role.
149        memoId (str): Memo ID.
150
151    Returns:
152        dict: FCM send result.
153    """
154    if not tagged_role_id or not tagged_role_name:
155        return 
156    title = f"New Memo Tagged to your role: {tagged_role_name}"
157    body = f"Memo has been tagged to you at Block: {info['current_block_name']}, Floor: {info['current_floor']}, Ward: {info['current_ward_name']}"
158    data = {
159        "type": "memo_created",
160        "memo_id": memoId,
161    }
162    responder_tokens = list(
163        User.objects.filter(
164            is_logged_in=True,
165            role__id=tagged_role_id
166        ).exclude(Q(fcm_token__isnull=True) | Q(fcm_token=''))
167        .values_list('fcm_token', flat=True)
168    )
169    print(f"Tokens for notification: {responder_tokens}")
170    if len(responder_tokens) > 0:
171        return send_fcm_to_users(responder_tokens, title, body, data)