Muryor
/
check
mirror of https://github.com/yuxian158/check.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

接入消息推送APP

pull/15/head
余弦 3 years ago
parent
baf903c300
commit
710c6321ae
1 changed files with 167 additions and 132 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 299
      checksendNotify.py

299
checksendNotify.py
Unescape View File

@ -8,9 +8,11 @@ import os
import re import re
import threading import threading
import time import time
import traceback
import urllib.parse import urllib.parse
import requests import requests
import tomli
# 原先的 print 函数和主线程的锁 # 原先的 print 函数和主线程的锁
_print = print _print = print
@ -29,63 +31,75 @@ def print(text, *args, **kw):
# 通知服务 # 通知服务
# fmt: off # fmt: off
push_config = { push_config = {
'HITOKOTO': True, # 启用一言（随机句子） 'HITOKOTO': False, # 启用一言（随机句子）
'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm/ 'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm
'BARK_ARCHIVE': '', # bark 推送是否存档 'BARK_ARCHIVE': '', # bark 推送是否存档
'BARK_GROUP': '', # bark 推送分组 'BARK_GROUP': '', # bark 推送分组
'BARK_SOUND': '', # bark 推送声音 'BARK_SOUND': '', # bark 推送声音
'CONSOLE': False, # 控制台输出 'CONSOLE': True, # 控制台输出
'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET
'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN 'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN
'FSKEY': '', # 飞书机器人的 FSKEY 'FSKEY': '', # 飞书机器人的 FSKEY
'GOBOT_URL': '', # go-cqhttp 'GOBOT_URL': '', # go-cqhttp
# 推送到个人QQ：http://127.0.0.1/send_private_msg # 推送到个人QQ：http://127.0.0.1/send_private_msg
# 群：http://127.0.0.1/send_group_msg # 群：http://127.0.0.1/send_group_msg
'GOBOT_QQ': '', # go-cqhttp 的推送群或用户 'GOBOT_QQ': '', # go-cqhttp 的推送群或用户
# GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ
# /send_group_msg 时填入 group_id=QQ群 # /send_group_msg 时填入 group_id=QQ群
'GOBOT_TOKEN': '', # go-cqhttp 的 access_token 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token
'GOTIFY_URL': '', # gotify地址,如https://push.example.de:8080 'IGOT_PUSH_KEY': '', # iGot 聚合推送的 IGOT_PUSH_KEY
'GOTIFY_TOKEN': '', # gotify的消息应用token
'GOTIFY_PRIORITY': 0, # 推送消息优先级,默认为0
'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+ 微信推送的用户令牌 'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY
'PUSH_PLUS_USER': '', # push+ 微信推送的群组编码 'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE
'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY 'QYWX_AM': '', # 企业微信应用
'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE
'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_BOT_TOKEN': '', # tg 机器人的 TG_BOT_TOKEN，例：1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ 'TG_API_HOST': '', # tg 代理 api
'TG_USER_ID': '', # tg 机器人的 TG_USER_ID，例：1434078534 'TG_PROXY_AUTH': '', # tg 代理认证参数
'TG_API_HOST': '', # tg 代理 api 'TG_PROXY_HOST': '', # tg 机器人的 TG_PROXY_HOST
'TG_PROXY_AUTH': '', # tg 代理认证参数 'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT
'TG_PROXY_HOST': '', # tg 机器人的 TG_PROXY_HOST 'MI_PUSH_ALIAS': '',
'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT
} }
notify_function = [] notify_function = []
# fmt: on # fmt: on
# 首先读取 面板变量 或者 github action 运行变量 # 首先读取 面板变量 或者 github action 运行变量
for k in push_config: for k in push_config:
if os.getenv(k): if v := os.getenv(k):
v = os.getenv(k)
push_config[k] = v 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: def bark(title: str, content: str) -> None:
""" """
使用 bark 推送消息。 使用 bark 推送消息。
@ -96,7 +110,7 @@ def bark(title: str, content: str) -> None:
print("bark 服务启动") print("bark 服务启动")
if push_config.get("BARK_PUSH").startswith("http"): 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: else:
url = f'https://api.day.app/{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' 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 = "" params = ""
for pair in filter( for pair in filter(
lambda pairs: pairs[0].startswith("BARK_") lambda pairs: pairs[0].startswith("BARK_")
and pairs[0] != "BARK_PUSH" and pairs[0] != "BARK_PUSH"
and pairs[1] and pairs[1]
and bark_params.get(pairs[0]), and bark_params.get(pairs[0]),
push_config.items(), push_config.items(),
): ):
params += f"{bark_params.get(pair[0])}={pair[1]}&" params += f"{bark_params.get(pair[0])}={pair[1]}&"
if params: if params:
url = url + "?" + params.rstrip("&") 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 推送成功！") print("bark 推送成功！")
elif datas.get("code") == 400:
print("bark 推送失败！找不到 Key 对应的 DeviceToken。")
else: else:
print("bark 推送失败！") print(f"bark 推送失败！响应数据：{datas}")
def console(title: str, content: str) -> None: 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")}&timestamp={timestamp}&sign={sign}' url = f'https://oapi.dingtalk.com/robot/send?access_token={push_config.get("DD_BOT_TOKEN")}&timestamp={timestamp}&sign={sign}'
headers = {"Content-Type": "application/json;charset=utf-8"} headers = {"Content-Type": "application/json;charset=utf-8"}
data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} 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 url=url, data=json.dumps(data), headers=headers, timeout=15
).json() ).json()
if datas.get("errcode") == 0:
if not response["errcode"]:
print("钉钉机器人 推送成功！") print("钉钉机器人 推送成功！")
else: else:
print("钉钉机器人 推送失败！") print(f"钉钉机器人 推送失败！响应数据：{datas}")
def feishu_bot(title: str, content: str) -> None: 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")}' 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}"}} 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("飞书 推送成功！") print("飞书 推送成功！")
else: else:
print("飞书 推送失败！错误信息如下：\n", response) print(f"飞书 推送失败！响应数据：{datas}")
def go_cqhttp(title: str, content: str) -> None: def go_cqhttp(title: str, content: str) -> None:
@ -190,31 +207,12 @@ def go_cqhttp(title: str, content: str) -> None:
print("go-cqhttp 服务启动") 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}' 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 推送成功！") print("go-cqhttp 推送成功！")
else: else:
print("go-cqhttp 推送失败！") print(f"go-cqhttp 推送失败！响应数据：{datas}")
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 推送失败！")
def iGot(title: str, content: str) -> None: 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")}' url = f'https://push.hellyw.com/{push_config.get("IGOT_PUSH_KEY")}'
data = {"title": title, "content": content} data = {"title": title, "content": content}
headers = {"Content-Type": "application/x-www-form-urlencoded"} 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 推送成功！") print("iGot 推送成功！")
else: else:
print(f'iGot 推送失败！{response["errMsg"]}') print(f'iGot 推送失败！错误信息：{datas.get("errMsg")}')
def serverJ(title: str, content: str) -> None: 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' url = f'https://sctapi.ftqq.com/{push_config.get("PUSH_KEY")}.send'
else: else:
url = f'https://sc.ftqq.com/${push_config.get("PUSH_KEY")}.send' 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 推送成功！") print("serverJ 推送成功！")
elif datas.get("code") == 40001:
print("serverJ 推送失败！PUSH_KEY 错误。")
else: else:
print(f'serverJ 推送失败！错误码：{response["message"]}') print(f'serverJ 推送失败！错误码：{datas.get("message")}')
def pushplus_bot(title: str, content: str) -> None: 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") body = json.dumps(data).encode(encoding="utf-8")
headers = {"Content-Type": "application/json"} 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 推送成功！") print("PUSHPLUS 推送成功！")
elif datas.get("code") == 600:
else: url2 = "http://pushplus.hxtrip.com/send"
url_old = "http://pushplus.hxtrip.com/send"
headers["Accept"] = "application/json" headers["Accept"] = "application/json"
response = requests.post(url=url_old, data=body, headers=headers).json() datas2 = requests.post(url=url2, data=body, headers=headers, timeout=15).json()
if datas2.get("code") == 200:
if response["code"] == 200:
print("PUSHPLUS(hxtrip) 推送成功！") print("PUSHPLUS(hxtrip) 推送成功！")
elif datas2.get("code") == 600:
print("PUSHPLUS 推送失败！PUSH_PLUS_TOKEN 错误。")
else: else:
print("PUSHPLUS 推送失败！") print(f"PUSHPLUS(hxtrip) 推送失败！响应数据：{datas2}")
else:
print(f"PUSHPLUS 推送失败！响应数据：{datas}")
def qmsg_bot(title: str, content: str) -> None: 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")}' 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")} 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 推送成功！") print("qmsg 推送成功！")
else: 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: def wecom_app(title: str, content: str) -> None:
@ -339,14 +352,13 @@ def wecom_app(title: str, content: str) -> None:
# 如果没有配置 media_id 默认就以 text 方式发送 # 如果没有配置 media_id 默认就以 text 方式发送
if not media_id: if not media_id:
message = title + "\n\n" + content message = title + "\n\n" + content
response = wx.send_text(message, touser) datas = wx.send_text(message, touser)
else: else:
response = wx.send_mpnews(title, content, media_id, touser) datas = wx.send_mpnews(title, content, media_id, touser)
if datas == "ok":
if response == "ok":
print("企业微信推送成功！") print("企业微信推送成功！")
else: else:
print("企业微信推送失败！错误信息如下：\n", response) print(f"企业微信推送失败！错误信息：{datas}")
class WeCom: class WeCom:
@ -361,14 +373,14 @@ class WeCom:
"corpid": self.CORPID, "corpid": self.CORPID,
"corpsecret": self.CORPSECRET, "corpsecret": self.CORPSECRET,
} }
req = requests.post(url, params=values) req = requests.post(url, params=values, timeout=15)
data = json.loads(req.text) datas = json.loads(req.text)
return data["access_token"] return datas.get("access_token")
def send_text(self, message, touser="@all"): def send_text(self, message, touser="@all"):
send_url = ( send_url = (
"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="
+ self.get_access_token() + self.get_access_token()
) )
send_values = { send_values = {
"touser": touser, "touser": touser,
@ -378,14 +390,13 @@ class WeCom:
"safe": "0", "safe": "0",
} }
send_msges = bytes(json.dumps(send_values), "utf-8") send_msges = bytes(json.dumps(send_values), "utf-8")
respone = requests.post(send_url, send_msges) datas = requests.post(send_url, send_msges, timeout=15).json()
respone = respone.json() return datas.get("errmsg")
return respone["errmsg"]
def send_mpnews(self, title, message, media_id, touser="@all"): def send_mpnews(self, title, message, media_id, touser="@all"):
send_url = ( send_url = (
"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="
+ self.get_access_token() + self.get_access_token()
) )
send_values = { send_values = {
"touser": touser, "touser": touser,
@ -405,9 +416,8 @@ class WeCom:
}, },
} }
send_msges = bytes(json.dumps(send_values), "utf-8") send_msges = bytes(json.dumps(send_values), "utf-8")
respone = requests.post(send_url, send_msges) datas = requests.post(send_url, send_msges, timeout=15).json()
respone = respone.json() return datas.get("errmsg")
return respone["errmsg"]
def wecom_bot(title: str, content: str) -> None: 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')}" 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"} headers = {"Content-Type": "application/json;charset=utf-8"}
data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} 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 url=url, data=json.dumps(data), headers=headers, timeout=15
).json() ).json()
if datas.get("errcode") == 0:
if response["errcode"] == 0: print("企业微信机器人 推送成功！")
print("企业微信机器人推送成功！")
else: else:
print("企业微信机器人推送失败！") print(f"企业微信机器人 推送失败！响应数据：{datas}")
def telegram_bot(title: str, content: str) -> None: def telegram_bot(title: str, content: str) -> None:
@ -456,25 +466,29 @@ def telegram_bot(title: str, content: str) -> None:
proxies = None proxies = None
if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"): 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( 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["TG_PROXY_HOST"] = (
push_config.get("TG_PROXY_AUTH") push_config.get("TG_PROXY_AUTH")
+ "@" + "@"
+ push_config.get("TG_PROXY_HOST") + push_config.get("TG_PROXY_HOST")
) )
proxyStr = "http://{}:{}".format( proxyStr = "http://{}:{}".format(
push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT") push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT")
) )
proxies = {"http": proxyStr, "https": proxyStr} 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 推送成功！") 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: else:
print("tg 推送失败！") print(f"tg 推送失败！响应数据：{datas}")
def one() -> str: def one() -> str:
@ -482,9 +496,12 @@ def one() -> str:
获取一条一言。 获取一条一言。
:return: :return:
""" """
url = "https://v1.hitokoto.cn/" try:
res = requests.get(url).json() url = "https://v1.hitokoto.cn/"
return res["hitokoto"] + " ----" + res["from"] res = requests.get(url).json()
return res["hitokoto"] + " ----" + res["from"]
except requests.exceptions.ConnectionError:
return ""
if push_config.get("BARK_PUSH"): if push_config.get("BARK_PUSH"):
@ -497,8 +514,6 @@ if push_config.get("FSKEY"):
notify_function.append(feishu_bot) notify_function.append(feishu_bot)
if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"): if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"):
notify_function.append(go_cqhttp) 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"): if push_config.get("IGOT_PUSH_KEY"):
notify_function.append(iGot) notify_function.append(iGot)
if push_config.get("PUSH_KEY"): if push_config.get("PUSH_KEY"):
@ -513,6 +528,26 @@ if push_config.get("QYWX_KEY"):
notify_function.append(wecom_bot) notify_function.append(wecom_bot)
if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"): if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"):
notify_function.append(telegram_bot) 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: def send(title: str, content: str) -> None:
@ -523,7 +558,7 @@ def send(title: str, content: str) -> None:
hitokoto = push_config.get("HITOKOTO") hitokoto = push_config.get("HITOKOTO")
text = one() if hitokoto else "" text = one() if hitokoto else ""
content += "\n" + text content += "\n\n" + text
ts = [ ts = [
threading.Thread(target=mode, args=(title, content), name=mode.__name__) threading.Thread(target=mode, args=(title, content), name=mode.__name__)

Write Preview
Loading…
Cancel
Save