diff --git a/frappe/push_notification.py b/frappe/push_notification.py index fd890b63b7..952d0f76e7 100644 --- a/frappe/push_notification.py +++ b/frappe/push_notification.py @@ -1,10 +1,12 @@ import json +from urllib.parse import urlparse import frappe from frappe.utils.response import Response -from urllib.parse import urlparse + from .frappeclient import FrappeClient + class PushNotification: project_name = "" @@ -26,10 +28,9 @@ class PushNotification: :param token: (str) The token to be added. :return: tuple[bool, str] First element is the success status, second element is the message. """ - data = self._send_post_request("notification_relay.api.token.add", { - "user_id": user_id, - "fcm_token": token - }) + data = self._send_post_request( + "notification_relay.api.token.add", {"user_id": user_id, "fcm_token": token} + ) return data["success"], data["message"] def remove_token(self, user_id: str, token: str) -> tuple[bool, str]: @@ -40,10 +41,9 @@ class PushNotification: :param token: (str) The token to be removed. :return: tuple[bool, str] First element is the success status, second element is the message. """ - data = self._send_post_request("notification_relay.api.token.remove", { - "user_id": user_id, - "fcm_token": token - }) + data = self._send_post_request( + "notification_relay.api.token.remove", {"user_id": user_id, "fcm_token": token} + ) return data["success"], data["message"] def add_topic(self, topic_name: str) -> bool: @@ -53,9 +53,7 @@ class PushNotification: :param topic_name: (str) The name of the topic. :return: bool True if successful, False otherwise. """ - data = self._send_post_request("notification_relay.api.topic.add", { - "topic_name": topic_name - }) + data = self._send_post_request("notification_relay.api.topic.add", {"topic_name": topic_name}) return data["success"] # Remove Topic @@ -66,9 +64,7 @@ class PushNotification: :param topic_name: (str) The name of the topic. :return: bool True if successful, False otherwise. """ - data = self._send_post_request("notification_relay.api.topic.remove", { - "topic_name": topic_name - }) + data = self._send_post_request("notification_relay.api.topic.remove", {"topic_name": topic_name}) return data["success"] def subscribe_topic(self, user_id: str, topic_name: str) -> bool: @@ -79,10 +75,9 @@ class PushNotification: :param topic_name: (str) The name of the topic. This topic should be already created. :return: """ - data = self._send_post_request("notification_relay.api.topic.subscribe", { - "user_id": user_id, - "topic_name": topic_name - }) + data = self._send_post_request( + "notification_relay.api.topic.subscribe", {"user_id": user_id, "topic_name": topic_name} + ) return data["success"] # Unsubscribe Topic (User) @@ -94,13 +89,20 @@ class PushNotification: :param topic_name: (str) The name of the topic. This topic should be already created. :return: bool True if successful, False otherwise. """ - data = self._send_post_request("notification_relay.api.topic.unsubscribe", { - "user_id": user_id, - "topic_name": topic_name - }) + data = self._send_post_request( + "notification_relay.api.topic.unsubscribe", {"user_id": user_id, "topic_name": topic_name} + ) return data["success"] - def send_notification_to_user(self, user_id: str, title: str, content: str, link:str=None, data=None, truncate_content:bool=False) -> bool: + def send_notification_to_user( + self, + user_id: str, + title: str, + content: str, + link: str = None, + data=None, + truncate_content: bool = False, + ) -> bool: """ Send notification to a user. @@ -121,15 +123,21 @@ class PushNotification: content = content[:1000] else: raise Exception("Content should be at max 1000 characters") - response_data = self._send_post_request("notification_relay.api.send_notification.user", { - "user_id": user_id, - "title": title, - "content": content, - "data": json.dumps(data) - }) + response_data = self._send_post_request( + "notification_relay.api.send_notification.user", + {"user_id": user_id, "title": title, "content": content, "data": json.dumps(data)}, + ) return response_data["success"] - def send_notification_to_topic(self, topic_name: str, title: str, content: str, link:str=None, data=None, truncate_content:bool=False) -> bool: + def send_notification_to_topic( + self, + topic_name: str, + title: str, + content: str, + link: str = None, + data=None, + truncate_content: bool = False, + ) -> bool: """ Send notification to a notification topic. @@ -150,12 +158,10 @@ class PushNotification: content = content[:1000] else: raise Exception("Content should be at max 1000 characters") - response_data = self._send_post_request("notification_relay.api.send_notification.topic", { - "topic_name": topic_name, - "title": title, - "content": content, - "data": json.dumps(data) - }) + response_data = self._send_post_request( + "notification_relay.api.send_notification.topic", + {"topic_name": topic_name, "title": title, "content": content, "data": json.dumps(data)}, + ) return response_data["success"] def is_enabled(self) -> bool: @@ -177,8 +183,12 @@ class PushNotification: :return: tuple[str, str] The API key and secret. """ notification_settings = frappe.get_doc("Push Notification Settings") - if notification_settings.api_key != "" and notification_settings.api_secret != ""\ - and notification_settings.api_key is not None and notification_settings.api_secret is not None: + if ( + notification_settings.api_key != "" + and notification_settings.api_secret != "" + and notification_settings.api_key is not None + and notification_settings.api_secret is not None + ): return notification_settings.api_key, notification_settings.api_secret else: # Generate new credentials @@ -190,7 +200,7 @@ class PushNotification: "protocol": self._get_site_protocol, "port": self._get_site_port, "token": token, - "webhook_route": "/api/method/frappe.push_notification.auth_webhook" + "webhook_route": "/api/method/frappe.push_notification.auth_webhook", } response = self._send_post_request("notification_relay.api.auth.get_credential", body, False) success = response["success"] @@ -202,7 +212,7 @@ class PushNotification: frappe.db.commit() return notification_settings.api_key, notification_settings.api_secret - def _send_post_request(self, method: str, params: dict, use_authentication: bool=True): + def _send_post_request(self, method: str, params: dict, use_authentication: bool = True): """ Send a POST request to the central relay server. @@ -244,8 +254,9 @@ class PushNotification: else: return "" + # Webhook which will be called by the central relay server for authentication -@frappe.whitelist(allow_guest=True, methods=['GET']) +@frappe.whitelist(allow_guest=True, methods=["GET"]) def auth_webhook(): token = frappe.cache().get(f"{frappe.local.site}:push_relay_registration_token") response = Response() @@ -260,19 +271,15 @@ def auth_webhook(): response.status_code = 200 return response + # Subscribe and Unsubscribe API @frappe.whitelist(methods=["GET"]) def subscribe(fcm_token: str): success, message = PushNotification().add_token(frappe.session.user, fcm_token) - return { - "success": success, - "message": message - } + return {"success": success, "message": message} + @frappe.whitelist(methods=["GET"]) def unsubscribe(fcm_token: str): success, message = PushNotification().remove_token(frappe.session.user, fcm_token) - return { - "success": success, - "message": message - } + return {"success": success, "message": message}