From 710c6321aea5d045ec99b43720378ae406abb999 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BD=99=E5=BC=A6?= Date: Sun, 10 Apr 2022 10:22:01 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=E6=B6=88=E6=81=AF=E6=8E=A8?= =?UTF-8?q?=E9=80=81APP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- checksendNotify.py | 299 +++++++++++++++++++++++++-------------------- 1 file changed, 167 insertions(+), 132 deletions(-) diff --git a/checksendNotify.py b/checksendNotify.py index 68482f0..13799e1 100644 --- a/checksendNotify.py +++ b/checksendNotify.py @@ -8,9 +8,11 @@ import os import re import threading import time +import traceback import urllib.parse import requests +import tomli # 原先的 print 函数和主线程的锁 _print = print @@ -29,63 +31,75 @@ def print(text, *args, **kw): # 通知服务 # fmt: off push_config = { - 'HITOKOTO': True, # 启用一言（随机句子） + 'HITOKOTO': False, # 启用一言（随机句子） - 'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm/ - 'BARK_ARCHIVE': '', # bark 推送是否存档 - 'BARK_GROUP': '', # bark 推送分组 - 'BARK_SOUND': '', # bark 推送声音 + 'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm + 'BARK_ARCHIVE': '', # bark 推送是否存档 + 'BARK_GROUP': '', # bark 推送分组 + 'BARK_SOUND': '', # bark 推送声音 - 'CONSOLE': False, # 控制台输出 + 'CONSOLE': True, # 控制台输出 - 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET - 'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN + 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET + 'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN - 'FSKEY': '', # 飞书机器人的 FSKEY + 'FSKEY': '', # 飞书机器人的 FSKEY - 'GOBOT_URL': '', # go-cqhttp - # 推送到个人QQ：http://127.0.0.1/send_private_msg - # 群：http://127.0.0.1/send_group_msg - 'GOBOT_QQ': '', # go-cqhttp 的推送群或用户 - # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ - # /send_group_msg 时填入 group_id=QQ群 - 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token + 'GOBOT_URL': '', # go-cqhttp + # 推送到个人QQ：http://127.0.0.1/send_private_msg + # 群：http://127.0.0.1/send_group_msg + 'GOBOT_QQ': '', # go-cqhttp 的推送群或用户 + # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ + # /send_group_msg 时填入 group_id=QQ群 + 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token - 'GOTIFY_URL': '', # gotify地址,如https://push.example.de:8080 - 'GOTIFY_TOKEN': '', # gotify的消息应用token - 'GOTIFY_PRIORITY': 0, # 推送消息优先级,默认为0 + 'IGOT_PUSH_KEY': '', # iGot 聚合推送的 IGOT_PUSH_KEY - 'IGOT_PUSH_KEY': '', # iGot 聚合推送的 IGOT_PUSH_KEY + 'PUSH_KEY': '', # server 酱的 PUSH_KEY，兼容旧版与 Turbo 版 - 'PUSH_KEY': '', # server 酱的 PUSH_KEY，兼容旧版与 Turbo 版 + 'PUSH_PLUS_TOKEN': '', # push+ 微信推送的用户令牌 + 'PUSH_PLUS_USER': '', # push+ 微信推送的群组编码 - 'PUSH_PLUS_TOKEN': '', # push+ 微信推送的用户令牌 - 'PUSH_PLUS_USER': '', # push+ 微信推送的群组编码 + 'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY + 'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE - 'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY - 'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE + 'QYWX_AM': '', # 企业微信应用 - 'QYWX_AM': '', # 企业微信应用 + 'QYWX_KEY': '', # 企业微信机器人 - 'QYWX_KEY': '', # 企业微信机器人 - - 'TG_BOT_TOKEN': '', # tg 机器人的 TG_BOT_TOKEN，例：1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ - 'TG_USER_ID': '', # tg 机器人的 TG_USER_ID，例：1434078534 - 'TG_API_HOST': '', # tg 代理 api - 'TG_PROXY_AUTH': '', # tg 代理认证参数 - 'TG_PROXY_HOST': '', # tg 机器人的 TG_PROXY_HOST - 'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT + 'TG_BOT_TOKEN': '', # tg 机器人的 TG_BOT_TOKEN，例：1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ + 'TG_USER_ID': '', # tg 机器人的 TG_USER_ID，例：1434078534 + 'TG_API_HOST': '', # tg 代理 api + 'TG_PROXY_AUTH': '', # tg 代理认证参数 + 'TG_PROXY_HOST': '', # tg 机器人的 TG_PROXY_HOST + 'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT + 'MI_PUSH_ALIAS': '', } notify_function = [] # fmt: on # 首先读取 面板变量 或者 github action 运行变量 for k in push_config: - if os.getenv(k): - v = os.getenv(k) + if v := os.getenv(k): push_config[k] = v +# 读取配置文件中的变量 (会覆盖环境变量) +# CONFIG_PATH = os.getenv("NOTIFY_CONFIG_PATH") +# if os.path.exists(CONFIG_PATH): +# print(f"通知配置文件存在：{CONFIG_PATH}。") +# try: +# for k, v in dict(tomli.load(open(CONFIG_PATH, "rb"))).items(): +# if k in push_config: +# push_config[k] = v +# except tomli.TOMLDecodeError: +# print( +# f"错误：配置文件 {CONFIG_PATH} 格式不对，请学习 https://toml.io/cn/v1.0.0\n错误信息：\n{traceback.format_exc()}" +# ) +# elif CONFIG_PATH: +# print(f"{CONFIG_PATH} 配置的通知文件不存在，请检查文件位置或删除对应环境变量！") + + def bark(title: str, content: str) -> None: """ 使用 bark 推送消息。 @@ -96,7 +110,7 @@ def bark(title: str, content: str) -> None: print("bark 服务启动") if push_config.get("BARK_PUSH").startswith("http"): - url = f'{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' + url = f'{push_config.get("BARK_PUSH").rstrip("/")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' else: url = f'https://api.day.app/{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' @@ -107,21 +121,23 @@ def bark(title: str, content: str) -> None: } params = "" for pair in filter( - lambda pairs: pairs[0].startswith("BARK_") - and pairs[0] != "BARK_PUSH" - and pairs[1] - and bark_params.get(pairs[0]), - push_config.items(), + lambda pairs: pairs[0].startswith("BARK_") + and pairs[0] != "BARK_PUSH" + and pairs[1] + and bark_params.get(pairs[0]), + push_config.items(), ): params += f"{bark_params.get(pair[0])}={pair[1]}&" if params: url = url + "?" + params.rstrip("&") - response = requests.get(url).json() - if response["code"] == 200: + datas = requests.get(url, timeout=15).json() + if datas.get("code") == 200: print("bark 推送成功！") + elif datas.get("code") == 400: + print("bark 推送失败！找不到 Key 对应的 DeviceToken。") else: - print("bark 推送失败！") + print(f"bark 推送失败！响应数据：{datas}") def console(title: str, content: str) -> None: @@ -151,14 +167,14 @@ def dingding_bot(title: str, content: str) -> None: url = f'https://oapi.dingtalk.com/robot/send?access_token={push_config.get("DD_BOT_TOKEN")}×tamp={timestamp}&sign={sign}' headers = {"Content-Type": "application/json;charset=utf-8"} data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} - response = requests.post( + + datas = requests.post( url=url, data=json.dumps(data), headers=headers, timeout=15 ).json() - - if not response["errcode"]: + if datas.get("errcode") == 0: print("钉钉机器人 推送成功！") else: - print("钉钉机器人 推送失败！") + print(f"钉钉机器人 推送失败！响应数据：{datas}") def feishu_bot(title: str, content: str) -> None: @@ -172,12 +188,13 @@ def feishu_bot(title: str, content: str) -> None: url = f'https://open.feishu.cn/open-apis/bot/v2/hook/{push_config.get("FSKEY")}' data = {"msg_type": "text", "content": {"text": f"{title}\n\n{content}"}} - response = requests.post(url, data=json.dumps(data)).json() - if response.get("StatusCode") == 0: + datas = requests.post(url, data=json.dumps(data), timeout=15) + datas = datas.json + if datas.get("StatusCode") == 0: print("飞书 推送成功！") else: - print("飞书 推送失败！错误信息如下：\n", response) + print(f"飞书 推送失败！响应数据：{datas}") def go_cqhttp(title: str, content: str) -> None: @@ -190,31 +207,12 @@ def go_cqhttp(title: str, content: str) -> None: print("go-cqhttp 服务启动") url = f'{push_config.get("GOBOT_URL")}?access_token={push_config.get("GOBOT_TOKEN")}&{push_config.get("GOBOT_QQ")}&message=标题:{title}\n内容:{content}' - response = requests.get(url).json() - if response["status"] == "ok": + datas = requests.get(url, timeout=15).json() + if datas.get("status") == "ok": print("go-cqhttp 推送成功！") else: - print("go-cqhttp 推送失败！") - - -def gotify(title:str,content:str) -> None: - """ - 使用 gotify 推送消息。 - """ - if not push_config.get("GOTIFY_URL") or not push_config.get("GOTIFY_TOKEN"): - print("gotify 服务的 GOTIFY_URL 或 GOTIFY_TOKEN 未设置!!\n取消推送") - return - print("gotify 服务启动") - - url = f'{push_config.get("GOTIFY_URL")}/message?token={push_config.get("GOTIFY_TOKEN")}' - data = {"title": title,"message": content,"priority": push_config.get("GOTIFY_PRIORITY")} - response = requests.post(url,data=data).json() - - if response.get("id"): - print("gotify 推送成功！") - else: - print("gotify 推送失败！") + print(f"go-cqhttp 推送失败！响应数据：{datas}") def iGot(title: str, content: str) -> None: @@ -229,12 +227,12 @@ def iGot(title: str, content: str) -> None: url = f'https://push.hellyw.com/{push_config.get("IGOT_PUSH_KEY")}' data = {"title": title, "content": content} headers = {"Content-Type": "application/x-www-form-urlencoded"} - response = requests.post(url, data=data, headers=headers).json() - if response["ret"] == 0: + datas = requests.post(url, data=data, headers=headers, timeout=15).json() + if datas.get("ret") == 0: print("iGot 推送成功！") else: - print(f'iGot 推送失败！{response["errMsg"]}') + print(f'iGot 推送失败！错误信息：{datas.get("errMsg")}') def serverJ(title: str, content: str) -> None: @@ -251,12 +249,14 @@ def serverJ(title: str, content: str) -> None: url = f'https://sctapi.ftqq.com/{push_config.get("PUSH_KEY")}.send' else: url = f'https://sc.ftqq.com/${push_config.get("PUSH_KEY")}.send' - response = requests.post(url, data=data).json() - if response.get("errno") == 0 or response.get("code") == 0: + datas = requests.post(url, data=data, timeout=15).json() + if datas.get("errno") == 0 or datas.get("code") == 0: print("serverJ 推送成功！") + elif datas.get("code") == 40001: + print("serverJ 推送失败！PUSH_KEY 错误。") else: - print(f'serverJ 推送失败！错误码：{response["message"]}') + print(f'serverJ 推送失败！错误码：{datas.get("message")}') def pushplus_bot(title: str, content: str) -> None: @@ -277,22 +277,22 @@ def pushplus_bot(title: str, content: str) -> None: } body = json.dumps(data).encode(encoding="utf-8") headers = {"Content-Type": "application/json"} - response = requests.post(url=url, data=body, headers=headers).json() - if response["code"] == 200: + datas = requests.post(url=url, data=body, headers=headers, timeout=15).json() + if datas.get("code") == 200: print("PUSHPLUS 推送成功！") - - else: - - url_old = "http://pushplus.hxtrip.com/send" + elif datas.get("code") == 600: + url2 = "http://pushplus.hxtrip.com/send" headers["Accept"] = "application/json" - response = requests.post(url=url_old, data=body, headers=headers).json() - - if response["code"] == 200: + datas2 = requests.post(url=url2, data=body, headers=headers, timeout=15).json() + if datas2.get("code") == 200: print("PUSHPLUS(hxtrip) 推送成功！") - + elif datas2.get("code") == 600: + print("PUSHPLUS 推送失败！PUSH_PLUS_TOKEN 错误。") else: - print("PUSHPLUS 推送失败！") + print(f"PUSHPLUS(hxtrip) 推送失败！响应数据：{datas2}") + else: + print(f"PUSHPLUS 推送失败！响应数据：{datas}") def qmsg_bot(title: str, content: str) -> None: @@ -306,12 +306,25 @@ def qmsg_bot(title: str, content: str) -> None: url = f'https://qmsg.zendee.cn/{push_config.get("QMSG_TYPE")}/{push_config.get("QMSG_KEY")}' payload = {"msg": f'{title}\n\n{content.replace("----", "-")}'.encode("utf-8")} - response = requests.post(url=url, params=payload).json() - if response["code"] == 0: + datas = requests.post(url=url, params=payload, timeout=15).json() + if datas.get("code") == 0: print("qmsg 推送成功！") else: - print(f'qmsg 推送失败！{response["reason"]}') + print(f'qmsg 推送失败！错误信息：{datas.get("reason")}') + + +def mi_push(title: str, content: str) -> None: + """ + 使用来自消息接收APP的推送服务 + """ + if not push_config.get("MI_PUSH_ALIAS"): + print("MI_PUSH_ALIAS 未设置!!\n取消推送") + url = "http://tdtt.top" + data = {"title": title, "content": content, "alias": push_config.get("MI_PUSH_ALIAS")} + datas = requests.post(url=url, data=data) + print(f"失败 {datas}") if not datas.status_code == requests.codes.ok else print('MI_PUSH推送成功') + return def wecom_app(title: str, content: str) -> None: @@ -339,14 +352,13 @@ def wecom_app(title: str, content: str) -> None: # 如果没有配置 media_id 默认就以 text 方式发送 if not media_id: message = title + "\n\n" + content - response = wx.send_text(message, touser) + datas = wx.send_text(message, touser) else: - response = wx.send_mpnews(title, content, media_id, touser) - - if response == "ok": + datas = wx.send_mpnews(title, content, media_id, touser) + if datas == "ok": print("企业微信推送成功！") else: - print("企业微信推送失败！错误信息如下：\n", response) + print(f"企业微信推送失败！错误信息：{datas}") class WeCom: @@ -361,14 +373,14 @@ class WeCom: "corpid": self.CORPID, "corpsecret": self.CORPSECRET, } - req = requests.post(url, params=values) - data = json.loads(req.text) - return data["access_token"] + req = requests.post(url, params=values, timeout=15) + datas = json.loads(req.text) + return datas.get("access_token") def send_text(self, message, touser="@all"): send_url = ( - "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" - + self.get_access_token() + "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + + self.get_access_token() ) send_values = { "touser": touser, @@ -378,14 +390,13 @@ class WeCom: "safe": "0", } send_msges = bytes(json.dumps(send_values), "utf-8") - respone = requests.post(send_url, send_msges) - respone = respone.json() - return respone["errmsg"] + datas = requests.post(send_url, send_msges, timeout=15).json() + return datas.get("errmsg") def send_mpnews(self, title, message, media_id, touser="@all"): send_url = ( - "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" - + self.get_access_token() + "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + + self.get_access_token() ) send_values = { "touser": touser, @@ -405,9 +416,8 @@ class WeCom: }, } send_msges = bytes(json.dumps(send_values), "utf-8") - respone = requests.post(send_url, send_msges) - respone = respone.json() - return respone["errmsg"] + datas = requests.post(send_url, send_msges, timeout=15).json() + return datas.get("errmsg") def wecom_bot(title: str, content: str) -> None: @@ -422,14 +432,14 @@ def wecom_bot(title: str, content: str) -> None: url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={push_config.get('QYWX_KEY')}" headers = {"Content-Type": "application/json;charset=utf-8"} data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} - response = requests.post( + + datas = requests.post( url=url, data=json.dumps(data), headers=headers, timeout=15 ).json() - - if response["errcode"] == 0: - print("企业微信机器人推送成功！") + if datas.get("errcode") == 0: + print("企业微信机器人 推送成功！") else: - print("企业微信机器人推送失败！") + print(f"企业微信机器人 推送失败！响应数据：{datas}") def telegram_bot(title: str, content: str) -> None: @@ -456,25 +466,29 @@ def telegram_bot(title: str, content: str) -> None: proxies = None if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"): if push_config.get("TG_PROXY_AUTH") is not None and "@" not in push_config.get( - "TG_PROXY_HOST" + "TG_PROXY_HOST" ): push_config["TG_PROXY_HOST"] = ( - push_config.get("TG_PROXY_AUTH") - + "@" - + push_config.get("TG_PROXY_HOST") + push_config.get("TG_PROXY_AUTH") + + "@" + + push_config.get("TG_PROXY_HOST") ) proxyStr = "http://{}:{}".format( push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT") ) proxies = {"http": proxyStr, "https": proxyStr} - response = requests.post( - url=url, headers=headers, params=payload, proxies=proxies - ).json() - if response["ok"]: + datas = requests.post( + url=url, headers=headers, params=payload, proxies=proxies, timeout=15 + ).json() + if datas.get("ok") == True: print("tg 推送成功！") + elif datas.get("error_code") == 400: + print("tg 推送失败！请主动给 bot 发送一条消息并检查接收用户 TG_USER_ID 是否正确。") + elif datas.get("error_code") == 401: + print("tg 推送失败！TG_BOT_TOKEN 填写错误。") else: - print("tg 推送失败！") + print(f"tg 推送失败！响应数据：{datas}") def one() -> str: @@ -482,9 +496,12 @@ def one() -> str: 获取一条一言。 :return: """ - url = "https://v1.hitokoto.cn/" - res = requests.get(url).json() - return res["hitokoto"] + " ----" + res["from"] + try: + url = "https://v1.hitokoto.cn/" + res = requests.get(url).json() + return res["hitokoto"] + " ----" + res["from"] + except requests.exceptions.ConnectionError: + return "" if push_config.get("BARK_PUSH"): @@ -497,8 +514,6 @@ if push_config.get("FSKEY"): notify_function.append(feishu_bot) if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"): notify_function.append(go_cqhttp) -if push_config.get("GOTIFY_URL") and push_config.get("GOTIFY_TOKEN"): - notify_function.append(gotify) if push_config.get("IGOT_PUSH_KEY"): notify_function.append(iGot) if push_config.get("PUSH_KEY"): @@ -513,6 +528,26 @@ if push_config.get("QYWX_KEY"): notify_function.append(wecom_bot) if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"): notify_function.append(telegram_bot) +if push_config.get("MI_PUSH_ALIAS"): + notify_function.append(mi_push) + + +def excepthook(args, /): + if issubclass(args.exc_type, requests.exceptions.RequestException): + print( + f"网络异常，请检查你的网络连接、推送服务器和代理配置，该错误和账号配置无关。信息：{str(args.exc_type)}, {args.thread.name}" + ) + elif issubclass(args.exc_type, json.JSONDecodeError): + print( + f"推送返回值非 json 格式，请检查网址和账号是否填写正确。信息：{str(args.exc_type)}, {args.thread.name}" + ) + else: + global default_hook + default_hook(args) + + +default_hook = threading.excepthook +threading.excepthook = excepthook def send(title: str, content: str) -> None: @@ -523,7 +558,7 @@ def send(title: str, content: str) -> None: hitokoto = push_config.get("HITOKOTO") text = one() if hitokoto else "" - content += "\n" + text + content += "\n\n" + text ts = [ threading.Thread(target=mode, args=(title, content), name=mode.__name__)