From cfb6f1be1013af5e2a309cae6ffbfeabb40fe300 Mon Sep 17 00:00:00 2001 From: zhonghuaifang Date: Tue, 7 Feb 2023 10:32:15 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dcqhttp=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- function/bika/sendNotify.py | 34 ++++++++++++++++++++++----------- function/bilibili/sendNotify.py | 34 ++++++++++++++++++++++----------- function/sendNotify.js | 2 +- function/sendNotify.py | 34 ++++++++++++++++++++++----------- function/wps/sendNotify.py | 34 ++++++++++++++++++++++----------- 5 files changed, 93 insertions(+), 45 deletions(-) diff --git a/function/bika/sendNotify.py b/function/bika/sendNotify.py index 34b1e41..bf6704f 100644 --- a/function/bika/sendNotify.py +++ b/function/bika/sendNotify.py @@ -136,7 +136,8 @@ def serverNotify(self, text, desp): 'text': text, 'desp': desp } - response = json.dumps(requests.post(url, data).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data).json(), ensure_ascii=False) datas = json.loads(response) # print(datas) if datas['code'] == 0: @@ -154,7 +155,8 @@ def BarkNotify(self, text, desp): url = sendNotify.BARK_PUSH + '/' + urllib.parse.quote(text) + '/' + urllib.parse.quote( desp) + '?sound=' + sendNotify.BARK_SOUND headers = {'Content-type': "application/x-www-form-urlencoded"} - response = json.dumps(requests.get(url, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.get( + url, headers=headers).json(), ensure_ascii=False) data = json.loads(response) # print(data) if data['code'] == 400: @@ -175,7 +177,8 @@ def tgBotNotify(self, text, desp): headers = {'Content-type': "application/x-www-form-urlencoded"} body = 'chat_id=' + sendNotify.TG_USER_ID + '&text=' + urllib.parse.quote( text) + '\n\n' + urllib.parse.quote(desp) + '&disable_web_page_preview=true' - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) data = json.loads(response) if data['ok']: @@ -209,11 +212,14 @@ def dingNotify(self, text, desp): 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() + hmac_code = hmac.new( + secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) - url = 'https://oapi.dingtalk.com/robot/send?access_token=' + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign + url = 'https://oapi.dingtalk.com/robot/send?access_token=' + \ + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign - response = requests.post(url=url, data=json.dumps(data), headers=headers).text + response = requests.post( + url=url, data=json.dumps(data), headers=headers).text if json.loads(response)['errcode'] == 0: print('\n钉钉发送通知消息成功\n') else: @@ -258,7 +264,8 @@ def pushNotify(self, text, desp): } body = json.dumps(data).encode(encoding='utf-8') headers = {'Content-Type': 'application/json'} - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) datas = json.loads(response) if datas['code'] == 200: print('\npush+发送通知消息成功\n') @@ -275,14 +282,16 @@ def sendWechat(self, desp): if sendNotify.QYWX_AM != '': # 获得access_token url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' - token_param = '?corpid=' + sendNotify.QYWX_AM.split(',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] + token_param = '?corpid=' + \ + sendNotify.QYWX_AM.split( + ',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] token_data = requests.get(url + token_param) token_data.encoding = 'utf-8' token_data = token_data.json() access_token = token_data['access_token'] - #发送内容 + # 发送内容 content = desp - #创建要发送的消息 + # 创建要发送的消息 data = { "touser": sendNotify.QYWX_AM.split(',')[2], "msgtype": "text", @@ -308,7 +317,10 @@ def go_cqhttp(self, title, content): return print("go-cqhttp 服务启动") - url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&{sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + if '/send_group_msg' in sendNotify.GOBOT_URL: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&group_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + else: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&user_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' try: response = requests.get(url, timeout=15) diff --git a/function/bilibili/sendNotify.py b/function/bilibili/sendNotify.py index 34b1e41..bf6704f 100644 --- a/function/bilibili/sendNotify.py +++ b/function/bilibili/sendNotify.py @@ -136,7 +136,8 @@ def serverNotify(self, text, desp): 'text': text, 'desp': desp } - response = json.dumps(requests.post(url, data).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data).json(), ensure_ascii=False) datas = json.loads(response) # print(datas) if datas['code'] == 0: @@ -154,7 +155,8 @@ def BarkNotify(self, text, desp): url = sendNotify.BARK_PUSH + '/' + urllib.parse.quote(text) + '/' + urllib.parse.quote( desp) + '?sound=' + sendNotify.BARK_SOUND headers = {'Content-type': "application/x-www-form-urlencoded"} - response = json.dumps(requests.get(url, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.get( + url, headers=headers).json(), ensure_ascii=False) data = json.loads(response) # print(data) if data['code'] == 400: @@ -175,7 +177,8 @@ def tgBotNotify(self, text, desp): headers = {'Content-type': "application/x-www-form-urlencoded"} body = 'chat_id=' + sendNotify.TG_USER_ID + '&text=' + urllib.parse.quote( text) + '\n\n' + urllib.parse.quote(desp) + '&disable_web_page_preview=true' - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) data = json.loads(response) if data['ok']: @@ -209,11 +212,14 @@ def dingNotify(self, text, desp): 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() + hmac_code = hmac.new( + secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) - url = 'https://oapi.dingtalk.com/robot/send?access_token=' + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign + url = 'https://oapi.dingtalk.com/robot/send?access_token=' + \ + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign - response = requests.post(url=url, data=json.dumps(data), headers=headers).text + response = requests.post( + url=url, data=json.dumps(data), headers=headers).text if json.loads(response)['errcode'] == 0: print('\n钉钉发送通知消息成功\n') else: @@ -258,7 +264,8 @@ def pushNotify(self, text, desp): } body = json.dumps(data).encode(encoding='utf-8') headers = {'Content-Type': 'application/json'} - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) datas = json.loads(response) if datas['code'] == 200: print('\npush+发送通知消息成功\n') @@ -275,14 +282,16 @@ def sendWechat(self, desp): if sendNotify.QYWX_AM != '': # 获得access_token url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' - token_param = '?corpid=' + sendNotify.QYWX_AM.split(',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] + token_param = '?corpid=' + \ + sendNotify.QYWX_AM.split( + ',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] token_data = requests.get(url + token_param) token_data.encoding = 'utf-8' token_data = token_data.json() access_token = token_data['access_token'] - #发送内容 + # 发送内容 content = desp - #创建要发送的消息 + # 创建要发送的消息 data = { "touser": sendNotify.QYWX_AM.split(',')[2], "msgtype": "text", @@ -308,7 +317,10 @@ def go_cqhttp(self, title, content): return print("go-cqhttp 服务启动") - url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&{sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + if '/send_group_msg' in sendNotify.GOBOT_URL: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&group_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + else: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&user_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' try: response = requests.get(url, timeout=15) diff --git a/function/sendNotify.js b/function/sendNotify.js index 9b05332..605e646 100644 --- a/function/sendNotify.js +++ b/function/sendNotify.js @@ -181,7 +181,7 @@ function gobotNotify(text, desp, time = 2100) { return new Promise((resolve) => { if (GOBOT_URL) { const options = { - url: `${GOBOT_URL}?access_token=${GOBOT_TOKEN}&${GOBOT_QQ}`, + url: `${GOBOT_URL}?access_token=${GOBOT_TOKEN}&${GOBOT_URL.indexOf("/send_group_msg") !== -1?'group_id=':'user_id='}${GOBOT_QQ}`, json: {message:`${text}\n${desp}`}, headers: { 'Content-Type': 'application/json', diff --git a/function/sendNotify.py b/function/sendNotify.py index 34b1e41..bf6704f 100644 --- a/function/sendNotify.py +++ b/function/sendNotify.py @@ -136,7 +136,8 @@ def serverNotify(self, text, desp): 'text': text, 'desp': desp } - response = json.dumps(requests.post(url, data).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data).json(), ensure_ascii=False) datas = json.loads(response) # print(datas) if datas['code'] == 0: @@ -154,7 +155,8 @@ def BarkNotify(self, text, desp): url = sendNotify.BARK_PUSH + '/' + urllib.parse.quote(text) + '/' + urllib.parse.quote( desp) + '?sound=' + sendNotify.BARK_SOUND headers = {'Content-type': "application/x-www-form-urlencoded"} - response = json.dumps(requests.get(url, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.get( + url, headers=headers).json(), ensure_ascii=False) data = json.loads(response) # print(data) if data['code'] == 400: @@ -175,7 +177,8 @@ def tgBotNotify(self, text, desp): headers = {'Content-type': "application/x-www-form-urlencoded"} body = 'chat_id=' + sendNotify.TG_USER_ID + '&text=' + urllib.parse.quote( text) + '\n\n' + urllib.parse.quote(desp) + '&disable_web_page_preview=true' - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) data = json.loads(response) if data['ok']: @@ -209,11 +212,14 @@ def dingNotify(self, text, desp): 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() + hmac_code = hmac.new( + secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) - url = 'https://oapi.dingtalk.com/robot/send?access_token=' + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign + url = 'https://oapi.dingtalk.com/robot/send?access_token=' + \ + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign - response = requests.post(url=url, data=json.dumps(data), headers=headers).text + response = requests.post( + url=url, data=json.dumps(data), headers=headers).text if json.loads(response)['errcode'] == 0: print('\n钉钉发送通知消息成功\n') else: @@ -258,7 +264,8 @@ def pushNotify(self, text, desp): } body = json.dumps(data).encode(encoding='utf-8') headers = {'Content-Type': 'application/json'} - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) datas = json.loads(response) if datas['code'] == 200: print('\npush+发送通知消息成功\n') @@ -275,14 +282,16 @@ def sendWechat(self, desp): if sendNotify.QYWX_AM != '': # 获得access_token url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' - token_param = '?corpid=' + sendNotify.QYWX_AM.split(',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] + token_param = '?corpid=' + \ + sendNotify.QYWX_AM.split( + ',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] token_data = requests.get(url + token_param) token_data.encoding = 'utf-8' token_data = token_data.json() access_token = token_data['access_token'] - #发送内容 + # 发送内容 content = desp - #创建要发送的消息 + # 创建要发送的消息 data = { "touser": sendNotify.QYWX_AM.split(',')[2], "msgtype": "text", @@ -308,7 +317,10 @@ def go_cqhttp(self, title, content): return print("go-cqhttp 服务启动") - url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&{sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + if '/send_group_msg' in sendNotify.GOBOT_URL: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&group_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + else: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&user_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' try: response = requests.get(url, timeout=15) diff --git a/function/wps/sendNotify.py b/function/wps/sendNotify.py index 34b1e41..bf6704f 100644 --- a/function/wps/sendNotify.py +++ b/function/wps/sendNotify.py @@ -136,7 +136,8 @@ def serverNotify(self, text, desp): 'text': text, 'desp': desp } - response = json.dumps(requests.post(url, data).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data).json(), ensure_ascii=False) datas = json.loads(response) # print(datas) if datas['code'] == 0: @@ -154,7 +155,8 @@ def BarkNotify(self, text, desp): url = sendNotify.BARK_PUSH + '/' + urllib.parse.quote(text) + '/' + urllib.parse.quote( desp) + '?sound=' + sendNotify.BARK_SOUND headers = {'Content-type': "application/x-www-form-urlencoded"} - response = json.dumps(requests.get(url, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.get( + url, headers=headers).json(), ensure_ascii=False) data = json.loads(response) # print(data) if data['code'] == 400: @@ -175,7 +177,8 @@ def tgBotNotify(self, text, desp): headers = {'Content-type': "application/x-www-form-urlencoded"} body = 'chat_id=' + sendNotify.TG_USER_ID + '&text=' + urllib.parse.quote( text) + '\n\n' + urllib.parse.quote(desp) + '&disable_web_page_preview=true' - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) data = json.loads(response) if data['ok']: @@ -209,11 +212,14 @@ def dingNotify(self, text, desp): 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() + hmac_code = hmac.new( + secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) - url = 'https://oapi.dingtalk.com/robot/send?access_token=' + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign + url = 'https://oapi.dingtalk.com/robot/send?access_token=' + \ + sendNotify.DD_BOT_TOKEN + '×tamp=' + timestamp + '&sign=' + sign - response = requests.post(url=url, data=json.dumps(data), headers=headers).text + response = requests.post( + url=url, data=json.dumps(data), headers=headers).text if json.loads(response)['errcode'] == 0: print('\n钉钉发送通知消息成功\n') else: @@ -258,7 +264,8 @@ def pushNotify(self, text, desp): } body = json.dumps(data).encode(encoding='utf-8') headers = {'Content-Type': 'application/json'} - response = json.dumps(requests.post(url, data=body, headers=headers).json(), ensure_ascii=False) + response = json.dumps(requests.post( + url, data=body, headers=headers).json(), ensure_ascii=False) datas = json.loads(response) if datas['code'] == 200: print('\npush+发送通知消息成功\n') @@ -275,14 +282,16 @@ def sendWechat(self, desp): if sendNotify.QYWX_AM != '': # 获得access_token url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' - token_param = '?corpid=' + sendNotify.QYWX_AM.split(',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] + token_param = '?corpid=' + \ + sendNotify.QYWX_AM.split( + ',')[0] + '&corpsecret=' + sendNotify.QYWX_AM.split(',')[1] token_data = requests.get(url + token_param) token_data.encoding = 'utf-8' token_data = token_data.json() access_token = token_data['access_token'] - #发送内容 + # 发送内容 content = desp - #创建要发送的消息 + # 创建要发送的消息 data = { "touser": sendNotify.QYWX_AM.split(',')[2], "msgtype": "text", @@ -308,7 +317,10 @@ def go_cqhttp(self, title, content): return print("go-cqhttp 服务启动") - url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&{sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + if '/send_group_msg' in sendNotify.GOBOT_URL: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&group_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' + else: + url = f'{sendNotify.GOBOT_URL}?access_token={sendNotify.GOBOT_TOKEN}&user_id={sendNotify.GOBOT_QQ}&message=标题:{title}\n内容:{content}' try: response = requests.get(url, timeout=15)