Add methods to send messages directly

knockknock/__init__.py

@@ -1,12 +1,12 @@
-from knockknock.chime_sender import chime_sender
-from knockknock.discord_sender import discord_sender
-from knockknock.email_sender import email_sender
-from knockknock.slack_sender import slack_sender
-from knockknock.sms_sender import sms_sender
-from knockknock.telegram_sender import telegram_sender
-from knockknock.teams_sender import teams_sender
-from knockknock.desktop_sender import desktop_sender
-from knockknock.matrix_sender import matrix_sender
-from knockknock.dingtalk_sender import dingtalk_sender
-from knockknock.wechat_sender import wechat_sender
-from knockknock.rocketchat_sender import rocketchat_sender
+from knockknock.chime_sender import chime_sender, send_on_chime
+from knockknock.discord_sender import discord_sender, send_on_discord
+from knockknock.email_sender import email_sender, send_on_email
+from knockknock.slack_sender import slack_sender, send_on_slack
+from knockknock.sms_sender import sms_sender, send_on_sms
+from knockknock.telegram_sender import telegram_sender, send_on_telegram
+from knockknock.teams_sender import teams_sender, send_on_teams
+from knockknock.desktop_sender import desktop_sender, send_on_desktop
+from knockknock.matrix_sender import matrix_sender, send_on_matrix
+from knockknock.dingtalk_sender import dingtalk_sender, send_on_dingtalk
+from knockknock.wechat_sender import wechat_sender, send_on_wechat
+from knockknock.rocketchat_sender import rocketchat_sender, send_on_rocketchat

knockknock/chime_sender.py

@@ -23,8 +23,6 @@ def chime_sender(webhook_url: str, user_mentions: List[str] = []):
         Optional users alias or full email address to notify.
     """
 
-    dump = {}
-
     def decorator_sender(func):
         @functools.wraps(func)
         def wrapper_sender(*args, **kwargs):
@@ -49,11 +47,13 @@ def wrapper_sender(*args, **kwargs):
                     'Your training has started 🎬',
                     'Machine name: %s' % host_name,
                     'Main call: %s' % func_name,
-                    'Starting date: %s' % start_time.strftime(DATE_FORMAT),
-                    ' '.join(user_mentions)
+                    'Starting date: %s' % start_time.strftime(DATE_FORMAT)
                 ]
-                dump['Content'] = '\n'.join(contents)
-                requests.post(url=webhook_url, json=dump)
+                send_on_chime(
+                    "\n".join(contents),
+                    webhook_url,
+                    user_mentions
+                )
 
             try:
                 value = func(*args, **kwargs)
@@ -77,9 +77,11 @@ def wrapper_sender(*args, **kwargs):
                         contents.append('Main call returned value: %s' %
                                         "ERROR - Couldn't str the returned value.")
 
-                    contents.append(' '.join(user_mentions))
-                    dump['Content'] = '\n'.join(contents)
-                    requests.post(url=webhook_url, json=dump)
+                    send_on_chime(
+                        "\n".join(contents),
+                        webhook_url,
+                        user_mentions
+                    )
 
                 return value
 
@@ -94,13 +96,29 @@ def wrapper_sender(*args, **kwargs):
                     'Crash date: %s' % end_time.strftime(DATE_FORMAT),
                     'Crashed training duration: %s\n\n' % str(elapsed_time),
                     "Here's the error:", '%s\n\n' % ex,
-                    "Traceback:", '%s' % traceback.format_exc(),
-                    ' '.join(user_mentions)
+                    "Traceback:", '%s' % traceback.format_exc()
                 ]
-                dump['Content'] = '\n'.join(contents)
-                requests.post(url=webhook_url, json=dump)
+                send_on_chime(
+                    "\n".join(contents),
+                    webhook_url,
+                    user_mentions
+                )
                 raise ex
 
         return wrapper_sender
 
     return decorator_sender
+
+def send_on_chime(
+    contents: str,
+    webhook_url: str,
+    user_mentions: List[str] = []):
+
+    contents = "{}\n{}".format(
+        contents,
+        contents.append(' '.join(user_mentions))
+    )
+
+    requests.post(url=webhook_url, json={
+        "Content": contents
+    })

knockknock/desktop_sender.py

@@ -9,26 +9,6 @@
 DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
 
 def desktop_sender(title: str = "knockknock"):
-
-    def show_notification(text: str, title: str):
-        # Check the OS
-        if platform.system() == "Darwin":     
-            subprocess.run(["sh", "-c", "osascript -e 'display notification \"%s\" with title \"%s\"'" % (text, title)])
-
-        elif platform.system() == "Linux":
-            subprocess.run(["notify-send", title, text])
-
-        elif platform.system() == "Windows":
-            try:
-                from win10toast import ToastNotifier
-            except ImportError as err:
-                print('Error: to use Windows Desktop Notifications, you need to install `win10toast` first. Please run `pip install win10toast==0.9`.')
-
-            toaster = ToastNotifier()
-            toaster.show_toast(title,
-                               text,
-                               icon_path=None,
-                               duration=5)
 
     def decorator_sender(func):
         @functools.wraps(func)
@@ -54,8 +34,7 @@ def wrapper_sender(*args, **kwargs):
                             'Machine name: %s' % host_name,
                             'Main call: %s' % func_name,
                             'Starting date: %s' % start_time.strftime(DATE_FORMAT)]
-                text = '\n'.join(contents)
-                show_notification(text, title)
+                send_on_desktop('\n'.join(contents), title)
 
             try:
                 value = func(*args, **kwargs)
@@ -76,8 +55,7 @@ def wrapper_sender(*args, **kwargs):
                     except:
                         contents.append('Main call returned value: %s'% "ERROR - Couldn't str the returned value.")
 
-                    text = '\n'.join(contents)
-                    show_notification(text, title)
+                    send_on_desktop('\n'.join(contents), title)
 
                 return value
 
@@ -94,10 +72,34 @@ def wrapper_sender(*args, **kwargs):
                             '%s\n\n' % ex,
                             "Traceback:",
                             '%s' % traceback.format_exc()]
-                text = '\n'.join(contents)
-                show_notification(text, title)
+                send_on_desktop('\n'.join(contents), title)
                 raise ex
 
         return wrapper_sender
 
     return decorator_sender
+
+
+def send_on_desktop(
+    contents: str,
+    title: str = "knockknock"):
+    # Check the OS
+    if platform.system() == "Darwin":
+        subprocess.run(["sh", "-c", "osascript -e 'display notification \"%s\" with title \"%s\"'" % (contents, title)])
+
+    elif platform.system() == "Linux":
+        subprocess.run(["notify-send", title, contents])
+
+    elif platform.system() == "Windows":
+        try:
+            from win10toast import ToastNotifier
+        except ImportError as err:
+            print('Error: to use Windows Desktop Notifications, you need to install `win10toast` first. Please run `pip install win10toast==0.9`.')
+
+        toaster = ToastNotifier()
+        toaster.show_toast(
+            title,
+            contents,
+            icon_path=None,
+            duration=5
+        )

knockknock/dingtalk_sender.py

@@ -40,30 +40,6 @@ def dingtalk_sender(webhook_url: str,
         see `secret`
 
     """
-    msg_template = {
-        "msgtype": "text", 
-        "text": {
-            "content": ""
-        }, 
-        "at": {
-            "atMobiles": user_mentions,
-            "isAtAll": False
-        }
-    }
-
-    def _construct_encrypted_url():
-        '''
-        Visit https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq for details
-        '''
-        timestamp = round(datetime.datetime.now().timestamp() * 1000)
-        secret_enc = secret.encode('utf-8')
-        string_to_sign = '{}\n{}'.format(timestamp, secret)
-        string_to_sign_enc = string_to_sign.encode('utf-8')
-        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
-        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
-        encrypted_url = webhook_url + '&timestamp={}'.format(timestamp) \
-                        + '&sign={}'.format(sign) 
-        return encrypted_url
 
     def decorator_sender(func):
         @functools.wraps(func)
@@ -89,15 +65,13 @@ def wrapper_sender(*args, **kwargs):
                             'Machine name: %s' % host_name,
                             'Main call: %s' % func_name,
                             'Starting date: %s' % start_time.strftime(DATE_FORMAT)]
-                contents.extend(['@{}'.format(i) for i in user_mentions])
-                contents.extend(keywords)
-
-                msg_template['text']['content'] = '\n'.join(contents)
-                if secret:
-                    postto = _construct_encrypted_url()
-                    requests.post(postto, json=msg_template)
-                else:
-                    requests.post(webhook_url, json=msg_template)
+                send_on_dingtalk(
+                    '\n'.join(contents),
+                    webhook_url,
+                    user_mentions,
+                    secret,
+                    keywords
+                )
 
             try:
                 value = func(*args, **kwargs)
@@ -118,16 +92,14 @@ def wrapper_sender(*args, **kwargs):
                     except:
                         contents.append('Main call returned value: %s'% "ERROR - Couldn't str the returned value.")
 
-                    contents.extend(['@{}'.format(i) for i in user_mentions])
-                    contents.extend(keywords)
+                    send_on_dingtalk(
+                        '\n'.join(contents),
+                        webhook_url,
+                        user_mentions,
+                        secret,
+                        keywords
+                    )
 
-                    msg_template['text']['content'] = '\n'.join(contents)
-                    if secret:
-                        postto = _construct_encrypted_url()
-                        requests.post(postto, json=msg_template)
-                    else:
-                        requests.post(webhook_url, json=msg_template)
-                        print(msg_template)
 
                 return value
 
@@ -144,19 +116,62 @@ def wrapper_sender(*args, **kwargs):
                             '%s\n\n' % ex,
                             "Traceback:",
                             '%s' % traceback.format_exc()]
-                contents.extend(['@{}'.format(i) for i in user_mentions])
-                contents.extend(keywords)
-
-                msg_template['text']['content'] = '\n'.join(contents)
-                if secret:
-                    postto = _construct_encrypted_url()
-                    requests.post(postto, json=msg_template)
-                else:
-                    requests.post(webhook_url, json=msg_template)
-                    print(msg_template)
+                send_on_dingtalk(
+                    '\n'.join(contents),
+                    webhook_url,
+                    user_mentions,
+                    secret,
+                    keywords
+                )
 
                 raise ex
 
         return wrapper_sender
 
     return decorator_sender
+
+
+def send_on_dingtalk(
+        contents: str,
+        webhook_url: str,
+        user_mentions: List[str] = [],
+        secret: str = '',
+        keywords: List[str] = []):
+
+    def _construct_encrypted_url():
+        '''
+        Visit https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq for details
+        '''
+        timestamp = round(datetime.datetime.now().timestamp() * 1000)
+        secret_enc = secret.encode('utf-8')
+        string_to_sign = '{}\n{}'.format(timestamp, secret)
+        string_to_sign_enc = string_to_sign.encode('utf-8')
+        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+        encrypted_url = webhook_url + '&timestamp={}'.format(timestamp) \
+                        + '&sign={}'.format(sign)
+        return encrypted_url
+
+    contents = "{}\n{}\n{}".format(
+        contents,
+        "\n".join(['@{}'.format(i) for i in user_mentions]),
+        "\n".join(keywords)
+    )
+
+    msg_template = {
+        "msgtype": "text",
+        "text": {
+            "content": contents
+        },
+        "at": {
+            "atMobiles": user_mentions,
+            "isAtAll": False
+        }
+    }
+
+    if secret:
+        postto = _construct_encrypted_url()
+        requests.post(postto, json=msg_template)
+    else:
+        requests.post(webhook_url, json=msg_template)
+        print(msg_template)

knockknock/discord_sender.py

@@ -21,11 +21,7 @@ def discord_sender(webhook_url: str):
         set up your webhook and get your URL.
     """
     def decorator_sender(func):
-        def send_message(text: str):
-            headers = {'Content-Type': 'application/json'}
-            payload = json.dumps({'content': text})
-            r = requests.post(url=webhook_url, data=payload, headers=headers)
-
+
         @functools.wraps(func)
         def wrapper_sender(*args, **kwargs):
 
@@ -49,8 +45,7 @@ def wrapper_sender(*args, **kwargs):
                             'Machine name: %s' % host_name,
                             'Main call: %s' % func_name,
                             'Starting date: %s' % start_time.strftime(DATE_FORMAT)]
-                text = '\n'.join(contents)
-                send_message(text=text)
+                send_on_discord('\n'.join(contents), webhook_url)
 
             try:
                 value = func(*args, **kwargs)
@@ -71,8 +66,7 @@ def wrapper_sender(*args, **kwargs):
                     except:
                         contents.append('Main call returned value: %s'% "ERROR - Couldn't str the returned value.")
 
-                    text = '\n'.join(contents)
-                    send_message(text=text)
+                    send_on_discord('\n'.join(contents), webhook_url)
 
                 return value
 
@@ -89,10 +83,19 @@ def wrapper_sender(*args, **kwargs):
                             '%s\n\n' % ex,
                             "Traceback:",
                             '%s' % traceback.format_exc()]
-                text = '\n'.join(contents)
-                send_message(text=text)
+                send_on_discord('\n'.join(contents), webhook_url)
                 raise ex
 
         return wrapper_sender
 
     return decorator_sender
+
+
+def send_on_discord(
+        contents: str,
+        webhook_url: str):
+
+    headers = {'Content-Type': 'application/json'}
+    payload = json.dumps({'content': contents})
+
+    requests.post(url=webhook_url, data=payload, headers=headers)