File size: 3,136 Bytes
bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 e8a6b78 bf41ce7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | import os
import sys
import json
import traceback
from typing import Any
import firebase_admin
from firebase_admin import messaging, firestore
class CloudMessage:
def __init__(self, firebase_app: firebase_admin.App):
self.firebase_app = firebase_app
self.db = firestore.client(app=firebase_app)
def get_tokens(self):
users_ref = self.db.collection("users")
docs = users_ref.stream()
registeration_tokens = []
for doc in docs:
registeration_tokens.append(doc.to_dict()["token"])
return registeration_tokens
def exception_detail(self, e):
error_class = e.__class__.__name__
detail = e.args[0]
cl, exc, tb = sys.exc_info()
lastCallStack = traceback.extract_tb(tb)[-1]
fileName = lastCallStack[0]
lineNum = lastCallStack[1]
funcName = lastCallStack[2]
errMsg = 'File "{}", line {}, in {}: [{}] {}'.format(
fileName, lineNum, funcName, error_class, detail
)
return errMsg
def send_message(self, notification, token_list):
if token_list == []:
return False, "token_list empty"
if notification.get("title") not in [None, ""]:
notify = messaging.Notification(
title=notification.get("title"),
body=notification.get("content", ""),
)
android_notify = messaging.AndroidNotification(
title=notification.get("title"),
body=notification.get("content", ""),
default_sound=True,
)
else:
notify = messaging.Notification(body=notification.get("content", ""))
android_notify = messaging.AndroidNotification(
body=notification.get("content", ""), default_sound=True
)
multi_msg = messaging.MulticastMessage(
notification=notify,
tokens=token_list,
data={}
if "route" not in notification
else {"direct": notification["route"]},
android=messaging.AndroidConfig(
notification=android_notify, priority="high"
),
apns=messaging.APNSConfig(
payload=messaging.APNSPayload(
messaging.Aps(sound=messaging.CriticalSound("default", volume=1.0))
)
),
)
response = messaging.send_multicast(
multi_msg,
app=self.firebase_app,
)
failed_tokens = []
if response.failure_count > 0:
responses = response.responses
for idx, resp in enumerate(responses):
if not resp.success:
# The order of responses corresponds to the order of the registration tokens.
failed_tokens.append(token_list[idx])
print("List of tokens that caused failures: {0}".format(failed_tokens))
return True, "send to {} devices, with {} successed, with {} failed.".format(
len(token_list), response.success_count, response.failure_count
)
|