From 524eba6455f252d12a9223d479393399c3b9b3c6 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 16 Dec 2023 17:01:00 -0500 Subject: [PATCH] Added WeCom Bot Support (#1016) --- KEYWORDS | 1 + README.md | 1 + apprise/plugins/NotifyWeComBot.py | 258 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 2 +- test/test_plugin_wecombot.py | 90 ++++++++++ 5 files changed, 351 insertions(+), 1 deletion(-) create mode 100644 apprise/plugins/NotifyWeComBot.py create mode 100644 test/test_plugin_wecombot.py diff --git a/KEYWORDS b/KEYWORDS index c8afd826..f9c80b49 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -97,6 +97,7 @@ Twitter Voipms Vonage Webex +WeCom Bot WhatsApp Windows XBMC diff --git a/README.md b/README.md index 1a6bef51..82a42aa9 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,7 @@ The table below identifies the services this tool supports and some example serv | [Twist](https://github.com/caronc/apprise/wiki/Notify_twist) | twist:// | (TCP) 443 | twist://pasword:login
twist://password:login/#channel
twist://password:login/#team:channel
twist://password:login/#team:channel1/channel2/#team3:channel | [XBMC](https://github.com/caronc/apprise/wiki/Notify_xbmc) | xbmc:// or xbmcs:// | (TCP) 8080 or 443 | xbmc://hostname
xbmc://user@hostname
xbmc://user:password@hostname:port | [Webex Teams (Cisco)](https://github.com/caronc/apprise/wiki/Notify_wxteams) | wxteams:// | (TCP) 443 | wxteams://Token +| [WeCom Bot](https://github.com/caronc/apprise/wiki/Notify_wecombot) | wecombot:// | (TCP) 443 | wecombot://BotKey | [WhatsApp](https://github.com/caronc/apprise/wiki/Notify_whatsapp) | whatsapp:// | (TCP) 443 | whatsapp://AccessToken@FromPhoneID/ToPhoneNo
whatsapp://Template:AccessToken@FromPhoneID/ToPhoneNo | [Zulip Chat](https://github.com/caronc/apprise/wiki/Notify_zulip) | zulip:// | (TCP) 443 | zulip://botname@Organization/Token
zulip://botname@Organization/Token/Stream
zulip://botname@Organization/Token/Email diff --git a/apprise/plugins/NotifyWeComBot.py b/apprise/plugins/NotifyWeComBot.py new file mode 100644 index 00000000..1c3b25d7 --- /dev/null +++ b/apprise/plugins/NotifyWeComBot.py @@ -0,0 +1,258 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +# WeCom for PC +# 1. On WeCom for PC, find the target WeCom group for receiving alarm +# notifications. +# 2. Right-click the WeCom group. In the window that appears, click +# "Add Group Bot". +# 3. In the window that appears, click Create a Bot. +# 4. In the window that appears, enter a custom bot name and click Add. +# 5. You will be provided a Webhook URL that looks like: +# https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=abcd +# +# WeCom for Web +# 1. On WebCom for Web, open the target WeCom group for receiving alarm +# notifications. +# 2. Click the group settings icon in the upper-right corner. +# 3. On the group settings page, choose Group Bots > Add a Bot. +# 4. On the management page for adding bots, enter a custom name for the new +# bot. +# 5. Click Add, copy the webhook address, and configure the API callback by +# following Step 2. + +# the URL will look something like this: +# https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=abcd +# ^ +# | +# webhook key +# +# This plugin also supports taking the URL (as identified above) directly +# as well. + +import re +import requests +from json import dumps + +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..utils import validate_regex +from ..AppriseLocale import gettext_lazy as _ + + +class NotifyWeComBot(NotifyBase): + """ + A wrapper for WeCom Bot Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'WeCom Bot' + + # The services URL + service_url = 'https://weixin.qq.com/' + + # The default secure protocol + secure_protocol = 'wecombot' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_wecombot' + + # Plain Text Notification URL + notify_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={key}' + + # Define object templates + templates = ( + '{schema}://{key}', + ) + + # The title is not used + title_maxlen = 0 + + # Define our template arguments + template_tokens = dict(NotifyBase.template_tokens, **{ + # The Bot Key can be found at the end of the webhook provided (?key=) + 'key': { + 'name': _('Bot Webhook Key'), + 'type': 'string', + 'required': True, + 'private': True, + 'regex': (r'^[a-z0-9_-]+$', 'i'), + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + # You can optionally pass IRC colors into + 'key': { + 'alias_of': 'key', + }, + }) + + def __init__(self, key, **kwargs): + """ + Initialize WeCom Bot Object + """ + super().__init__(**kwargs) + + # Assign our bot webhook + self.key = validate_regex( + key, *self.template_tokens['key']['regex']) + if not self.key: + msg = 'An invalid WeCom Bot Webhook Key ' \ + '({}) was specified.'.format(key) + self.logger.warning(msg) + raise TypeError(msg) + + # Prepare our notification URL now: + self.api_url = self.notify_url.format( + key=self.key, + ) + return + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Prepare our parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + return '{schema}://{key}/?{params}'.format( + schema=self.secure_protocol, + key=self.pprint(self.key, privacy, safe=''), + params=NotifyWeComBot.urlencode(params), + ) + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + wrapper to _send since we can alert more then one channel + """ + + # prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json; charset=utf-8', + } + + # Prepare our payload + payload = { + 'msgtype': "text", + 'text': { + 'content': body, + } + } + + self.logger.debug('WeCom Bot GET URL: %s (cert_verify=%r)' % ( + self.api_url, self.verify_certificate)) + self.logger.debug('WeCom Bot Payload: %s' % str(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + self.api_url, + data=dumps(payload).encode('utf-8'), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyWeComBot.http_response_code_lookup(r.status_code) + + self.logger.warning( + 'Failed to send WeCom Bot notification: ' + '{}{}error={}.'.format( + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug('Response Details:\r\n{}'.format(r.content)) + + # Return; we're done + return False + + else: + self.logger.info('Sent WeCom Bot notification.') + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending WeCom Bot ' + 'notification.') + self.logger.debug('Socket Exception: %s' % str(e)) + + # Return; we're done + return False + + return True + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + + """ + + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't load the results + return results + + # The first token is stored in the hostname + results['key'] = NotifyWeComBot.unquote(results['host']) + + # The 'key' makes it easier to use yaml configuration + if 'key' in results['qsd'] and len(results['qsd']['key']): + results['key'] = \ + NotifyWeComBot.unquote(results['qsd']['key']) + + return results + + @staticmethod + def parse_native_url(url): + """ + Support https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=BOTKEY + """ + + result = re.match( + r'^https?://qyapi\.weixin\.qq\.com/cgi-bin/webhook/send/?\?key=' + r'(?P[A-Z0-9_-]+)/?' + r'&?(?P.+)?$', url, re.I) + + if result: + return NotifyWeComBot.parse_url( + '{schema}://{key}{params}'.format( + schema=NotifyWeComBot.secure_protocol, + key=result.group('key'), + params='' if not result.group('params') + else '?' + result.group('params'))) + + return None diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index ffa04798..7f05a9f9 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -51,7 +51,7 @@ Pushover, PushSafer, Pushy, PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema Gateway, Twilio, Twitter, Twist, XBMC, -Voipms, Vonage, WhatsApp, Webex Teams} +Voipms, Vonage, WeCom Bot, WhatsApp, Webex Teams} Name: python-%{pypi_name} Version: 1.6.0 diff --git a/test/test_plugin_wecombot.py b/test/test_plugin_wecombot.py new file mode 100644 index 00000000..0f90ae8c --- /dev/null +++ b/test/test_plugin_wecombot.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +import requests + +from apprise.plugins.NotifyWeComBot import NotifyWeComBot +from helpers import AppriseURLTester + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ('wecombot://', { + 'instance': TypeError, + }), + ('wecombot://:@/', { + 'instance': TypeError, + }), + ('wecombot://botkey', { + # Minimum requirements met + 'instance': NotifyWeComBot, + }), + ('wecombot://?key=botkey', { + # Test ?key= + 'instance': NotifyWeComBot, + }), + # Support Native URLs + ('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=BOTKEY', { + 'instance': NotifyWeComBot, + }), + ('https://qyapi.weixin.qq.com/cgi-bin/webhook/send/?key=BOTKEY&data=123', { + # another variation (more parameters don't obstruct our key) + 'instance': NotifyWeComBot, + }), + ('wecombot://botkey', { + 'instance': NotifyWeComBot, + # force a failure + 'response': False, + 'requests_response_code': requests.codes.internal_server_error, + }), + ('wecombot://botkey', { + 'instance': NotifyWeComBot, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('wecombot://botkey', { + 'instance': NotifyWeComBot, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + }), +) + + +def test_plugin_wecombot_urls(): + """ + NotifyWeComBot() Apprise URLs + + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all()