Spaces:
Runtime error
Runtime error
File size: 1,933 Bytes
a398c85 | 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 | from firebase_admin import messaging
from app import db
from google.cloud.firestore_v1.base_query import FieldFilter
# This function cannot be automation test because the requirement of another device to receive notification
def sendMessage(artifactId: str, message: str = None):
token = []
artifact = db.collection("artifacts").document(artifactId).get()
if not artifact.exists:
return
artifact = artifact.to_dict()
user_ref = db.collection("user").where(
filter=FieldFilter("artifacts", "array_contains", "artifacts/" + artifactId)
).stream()
for user in user_ref:
token.append(user.to_dict()["deviceId"])
if message is not None:
msg = messaging.MulticastMessage(
data={
"notification": {
"title": message,
"body": "Video "
+ artifact["name"]
+ " has done inference. Click here to see the video",
},
},
android=messaging.AndroidConfig(
notification=messaging.AndroidNotification(
icon="stock_ticker_update", color="#f45342"
),
),
tokens=token
)
else:
msg = messaging.MulticastMessage(
data={
"notification": {
"title": "Video " + artifact['name'] + " has done inference.",
"body": "Video "
+ artifact["name"]
+ " has done inference. Click here to see the video",
},
},
android=messaging.AndroidConfig(
notification=messaging.AndroidNotification(
icon="stock_ticker_update", color="#f45342"
),
),
tokens=token
)
response = messaging.send_multicast(msg)
return response.success_count
|