open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2025-05-23 14:56:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

更新 Python 通知文件 (#832)

Browse Source 
已在 Python3.6.x（云函数平台），3.8.x，3.10.0 测试通过
特性如下
1. 支持 11 种推送方式，未来还会继续添加
2. 多线程通知
3. 支持一言（一句话）和控制台输出开关
4. 对于现有推送方式，补充了配置变量
This commit is contained in:
Oreo 2021-11-06 20:45:53 +08:00 committed by GitHub
parent 1becb26aaf
commit 6b74cccc34
1 changed files with 477 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

745
sample/notify.py
View File

@ -1,306 +1,515 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# _*_ coding:utf-8 _*_ # _*_ coding:utf-8 _*_
import sys
import os
cur_path = os.path.abspath(os.path.dirname(__file__))
root_path = os.path.split(cur_path)[0]
sys.path.append(root_path)
import requests
import json
import traceback
import time
import hmac
import hashlib
import base64 import base64
import urllib.parse import hashlib
from requests.adapters import HTTPAdapter import hmac
from urllib3.util import Retry import json
import os
import re import re
import threading
import time
import urllib.parse
import requests
# 原先的 print 函数和主线程的锁
_print = print
mutex = threading.Lock()
# 定义新的 print 函数
def print(text, *args, **kw):
"""
使输出有序进行，不出现多线程同一时间输出导致错乱的问题。
"""
with mutex:
_print(text, *args, **kw)
# 通知服务 # 通知服务
BARK = '' # bark服务,此参数如果以http或者https开头则判定为自建bark服务; secrets可填; # fmt: off
SCKEY = '' # Server酱的SCKEY; secrets可填 push_config = {
TG_BOT_TOKEN = '' # tg机器人的TG_BOT_TOKEN; secrets可填 'HITOKOTO': False, # 启用一言（随机句子）
TG_USER_ID = '' # tg机器人的TG_USER_ID; secrets可填
TG_PROXY_IP = '' # tg机器人的TG_PROXY_IP; secrets可填
TG_PROXY_PORT = '' # tg机器人的TG_PROXY_PORT; secrets可填
DD_BOT_TOKEN = '' # 钉钉机器人的DD_BOT_TOKEN; secrets可填
DD_BOT_SECRET = '' # 钉钉机器人的DD_BOT_SECRET; secrets可填
QYWX_AM = '' # 企业微信应用的QYWX_AM; secrets可填 参考http://note.youdao.com/s/HMiudGkb
QYWX_KEY = '' # 企业微信机器人的QYWX_KEY; secrets可填
notify_mode = [] 'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm/
'BARK_ARCHIVE': '', # bark 推送是否存档
'BARK_GROUP': '', # bark 推送分组
'BARK_SOUND': '', # bark 推送声音
# GitHub action运行需要填写对应的secrets 'CONSOLE': True, # 控制台输出
if "BARK" in os.environ and os.environ["BARK"]:
BARK = os.environ["BARK"] 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET
if "PUSH_KEY" in os.environ and os.environ["PUSH_KEY"]: 'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN
SCKEY = os.environ["PUSH_KEY"]
if "TG_BOT_TOKEN" in os.environ and os.environ["TG_BOT_TOKEN"] and "TG_USER_ID" in os.environ and os.environ["TG_USER_ID"]: 'FSKEY': '', # 飞书机器人的 FSKEY
TG_BOT_TOKEN = os.environ["TG_BOT_TOKEN"]
TG_USER_ID = os.environ["TG_USER_ID"] 'GOBOT_URL': '', # go-cqhttp
if "DD_BOT_TOKEN" in os.environ and os.environ["DD_BOT_TOKEN"] and "DD_BOT_SECRET" in os.environ and os.environ["DD_BOT_SECRET"]: # 推送到个人QQ：http://127.0.0.1/send_private_msg
DD_BOT_TOKEN = os.environ["DD_BOT_TOKEN"] # 群：http://127.0.0.1/send_group_msg
DD_BOT_SECRET = os.environ["DD_BOT_SECRET"] 'GOBOT_QQ': '', # go-cqhttp 的推送群或用户
if "QYWX_AM" in os.environ and os.environ["QYWX_AM"]: # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ
QYWX_AM = os.environ["QYWX_AM"] # /send_group_msg 时填入 group_id=QQ群
if "QYWX_KEY" in os.environ and os.environ["QYWX_KEY"]: 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token
QYWX_KEY = os.environ["QYWX_KEY"]
'IGOT_PUSH_KEY': '', # iGot 聚合推送的 IGOT_PUSH_KEY
'PUSH_KEY': '', # server 酱的 PUSH_KEY，兼容旧版与 Turbo 版
'PUSH_PLUS_TOKEN': '', # push+ 微信推送的用户令牌
'PUSH_PLUS_USER': '', # push+ 微信推送的群组编码
'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY
'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE
'QYWX_AM': '', # 企业微信应用
'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
}
notify_function = []
# fmt: on
# 首先读取 面板变量 或者 github action 运行变量
for k in push_config:
if os.getenv(k):
v = os.getenv(k)
push_config[k] = v
if BARK: def bark(title: str, content: str) -> None:
notify_mode.append('bark') """
print("BARK 推送打开") 使用 bark 推送消息。
if SCKEY: """
notify_mode.append('sc_key') if not push_config.get("BARK_PUSH"):
print("Server酱 推送打开") print("bark 服务的 BARK_PUSH 未设置!!\n取消推送")
if TG_BOT_TOKEN and TG_USER_ID:
notify_mode.append('telegram_bot')
print("Telegram 推送打开")
if DD_BOT_TOKEN and DD_BOT_SECRET:
notify_mode.append('dingding_bot')
print("钉钉机器人 推送打开")
if QYWX_AM:
notify_mode.append('qywxapp_bot')
print("企业微信应用 推送打开")
if QYWX_KEY:
notify_mode.append('qyw_bot')
print("企业微信机器人 推送打开")
def bark(title, content):
print("\n")
if not BARK:
print("bark服务的bark_token未设置!!\n取消推送")
return return
print("bark服务启动") print("bark 服务启动")
url = None
if BARK.startswith('http'): if push_config.get("BARK_PUSH").startswith("http"):
url = f"""{BARK}/{title}/{content}""" url = f'{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}'
else: else:
url = f"""https://api.day.app/{BARK}/{title}/{content}""" url = f'https://api.day.app/{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}'
bark_params = {
"BARK_ARCHIVE": "isArchive",
"BARK_GROUP": "group",
"BARK_SOUND": "sound",
}
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(),
):
params += f"{bark_params.get(pair[0])}={pair[1]}&"
if params:
url = url + "?" + params.rstrip("&")
response = requests.get(url).json() response = requests.get(url).json()
if response['code'] == 200:
print('推送成功！')
else:
print('推送失败！')
def serverJ(title, content): if response["code"] == 200:
print("\n") print("bark 推送成功！")
if not SCKEY: else:
print("server酱服务的SCKEY未设置!!\n取消推送") print("bark 推送失败！")
def console(title: str, content: str) -> None:
"""
使用 控制台 推送消息。
"""
print(f"{title}\n\n{content}")
def dingding_bot(title: str, content: str) -> None:
"""
使用 钉钉机器人 推送消息。
"""
if not push_config.get("DD_BOT_SECRET") or not push_config.get("DD_BOT_TOKEN"):
print("钉钉机器人 服务的 DD_BOT_SECRET 或者 DD_BOT_TOKEN 未设置!!\n取消推送")
return return
print("serverJ服务启动") print("钉钉机器人 服务启动")
timestamp = str(round(time.time() * 1000))
secret_enc = push_config.get("DD_BOT_SECRET").encode("utf-8")
string_to_sign = "{}\n{}".format(timestamp, push_config.get("DD_BOT_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))
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"}
data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}}
response = requests.post(
url=url, data=json.dumps(data), headers=headers, timeout=15
).json()
if not response["errcode"]:
print("钉钉机器人 推送成功！")
else:
print("钉钉机器人 推送失败！")
def feishu_bot(title: str, content: str) -> None:
"""
使用 飞书机器人 推送消息。
"""
if not push_config.get("FSKEY"):
print("飞书 服务的 FSKEY 未设置!!\n取消推送")
return
print("飞书 服务启动")
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:
print("飞书 推送成功！")
else:
print("飞书 推送失败！错误信息如下：\n", response)
def go_cqhttp(title: str, content: str) -> None:
"""
使用 go_cqhttp 推送消息。
"""
if not push_config.get("GOBOT_URL") or not push_config.get("GOBOT_QQ"):
print("go-cqhttp 服务的 GOBOT_URL 或 GOBOT_QQ 未设置!!\n取消推送")
return
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":
print("go-cqhttp 推送成功！")
else:
print("go-cqhttp 推送失败！")
def iGot(title: str, content: str) -> None:
"""
使用 iGot 推送消息。
"""
if not push_config.get("IGOT_PUSH_KEY"):
print("iGot 服务的 IGOT_PUSH_KEY 未设置!!\n取消推送")
return
print("iGot 服务启动")
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:
print("iGot 推送成功！")
else:
print(f'iGot 推送失败！{response["errMsg"]}')
def serverJ(title: str, content: str) -> None:
"""
通过 serverJ 推送消息。
"""
if not push_config.get("PUSH_KEY"):
print("serverJ 服务的 PUSH_KEY 未设置!!\n取消推送")
return
print("serverJ 服务启动")
data = {"text": title, "desp": content.replace("\n", "\n\n")}
if push_config.get("PUSH_KEY").index("SCT") != -1:
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:
print("serverJ 推送成功！")
else:
print(f'serverJ 推送失败！错误码：{response["message"]}')
def pushplus_bot(title: str, content: str) -> None:
"""
通过 push+ 推送消息。
"""
if not push_config.get("PUSH_PLUS_TOKEN"):
print("PUSHPLUS 服务的 PUSH_PLUS_TOKEN 未设置!!\n取消推送")
return
print("PUSHPLUS 服务启动")
url = "http://www.pushplus.plus/send"
data = { data = {
"text": title, "token": push_config.get("PUSH_PLUS_TOKEN"),
"desp": content.replace("\n", "\n\n") "title": title,
"content": content,
"topic": push_config.get("PUSH_PLUS_USER"),
} }
response = requests.post(f"https://sc.ftqq.com/{SCKEY}.send", data=data).json() body = json.dumps(data).encode(encoding="utf-8")
if response['errno'] == 0: headers = {"Content-Type": "application/json"}
print('推送成功！') response = requests.post(url=url, data=body, headers=headers).json()
else:
print('推送失败！')
def telegram_bot(title, content): if response["code"] == 200:
print("\n") print("PUSHPLUS 推送成功！")
bot_token = TG_BOT_TOKEN
user_id = TG_USER_ID else:
if not bot_token or not user_id:
print("tg服务的bot_token或者user_id未设置!!\n取消推送") url_old = "http://pushplus.hxtrip.com/send"
response = requests.post(url=url_old, data=body, headers=headers).json()
if response["code"] == 200:
print("PUSHPLUS(hxtrip) 推送成功！")
else:
print("PUSHPLUS 推送失败！")
def qmsg_bot(title: str, content: str) -> None:
"""
使用 qmsg 推送消息。
"""
if not push_config.get("QMSG_KEY") or not push_config.get("QMSG_TYPE"):
print("qmsg 的 QMSG_KEY 或者 QMSG_TYPE 未设置!!\n取消推送")
return return
print("tg服务启动") print("qmsg 服务启动")
url=f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage"
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
payload = {'chat_id': str(TG_USER_ID), 'text': f'{title}\n\n{content}', 'disable_web_page_preview': 'true'}
proxies = None
if TG_PROXY_IP and TG_PROXY_PORT:
proxyStr = "http://{}:{}".format(TG_PROXY_IP, TG_PROXY_PORT)
proxies = {"http": proxyStr, "https": proxyStr}
response = requests.post(url=url, headers=headers, params=payload, proxies=proxies).json()
if response['ok']:
print('推送成功！')
else:
print('推送失败！')
def dingding_bot(title, content): url = f'https://qmsg.zendee.cn/{push_config.get("QMSG_TYPE")}/{push_config.get("QMSG_KEY")}'
timestamp = str(round(time.time() * 1000)) # 时间戳 payload = {"msg": f'{title}\n\n{content.replace("----", "-")}'.encode("utf-8")}
secret_enc = DD_BOT_SECRET.encode('utf-8') response = requests.post(url=url, params=payload).json()
string_to_sign = '{}\n{}'.format(timestamp, DD_BOT_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)) # 签名
print('开始使用 钉钉机器人 推送消息...', end='')
url = f'https://oapi.dingtalk.com/robot/send?access_token={DD_BOT_TOKEN}&timestamp={timestamp}&sign={sign}'
headers = {'Content-Type': 'application/json;charset=utf-8'}
data = {
'msgtype': 'text',
'text': {'content': f'{title}\n\n{content}'}
}
response = requests.post(url=url, data=json.dumps(data), headers=headers, timeout=15).json()
if not response['errcode']:
print('推送成功！')
else:
print('推送失败！')
def qywxapp_bot(title, content): if response["code"] == 0:
print("\n") print("qmsg 推送成功！")
if not QYWX_AM: else:
print("企业微信应用的QYWX_AM未设置!!\n取消推送") print(f'qmsg 推送失败！{response["reason"]}')
def wecom_app(title: str, content: str) -> None:
"""
通过 企业微信 APP 推送消息。
"""
if not push_config.get("QYWX_AM"):
print("QYWX_AM 未设置!!\n取消推送")
return return
print("企业微信应用启动") QYWX_AM_AY = re.split(",", push_config.get("QYWX_AM"))
qywx_app_params = QYWX_AM.split(',') if 4 < len(QYWX_AM_AY) > 5:
url='https://qyapi.weixin.qq.com/cgi-bin/gettoken' print("QYWX_AM 设置错误!!\n取消推送")
headers= { return
'Content-Type': 'application/json', print("企业微信 APP 服务启动")
}
payload = {
'corpid': qywx_app_params[0],
'corpsecret': qywx_app_params[1],
}
response = requests.post(url=url, headers=headers, data=json.dumps(payload), timeout=15).json()
accesstoken = response["access_token"]
html = content.replace("\n", "<br/>")
options = None corpid = QYWX_AM_AY[0]
if not qywx_app_params[4]: corpsecret = QYWX_AM_AY[1]
options = { touser = QYWX_AM_AY[2]
'msgtype': 'text', agentid = QYWX_AM_AY[3]
'text': { try:
content: f'{title}\n\n${content}' media_id = QYWX_AM_AY[4]
} except IndexError:
} media_id = ""
elif qywx_app_params[4] == '0': wx = WeCom(corpid, corpsecret, agentid)
options = { # 如果没有配置 media_id 默认就以 text 方式发送
'msgtype': 'textcard', if not media_id:
'textcard': { message = title + "\n\n" + content
title: f'{title}', response = wx.send_text(message, touser)
description: f'{content}',
btntxt: '更多'
}
}
elif qywx_app_params[4] == '1':
options = {
'msgtype': 'text',
'text': {
content: f'{title}\n\n${content}'
}
}
else: else:
options = { response = wx.send_mpnews(title, content, media_id, touser)
'msgtype': 'mpnews',
'mpnews': { if response == "ok":
'articles': [ print("企业微信推送成功！")
else:
print("企业微信推送失败！错误信息如下：\n", response)
class WeCom:
def __init__(self, corpid, corpsecret, agentid):
self.CORPID = corpid
self.CORPSECRET = corpsecret
self.AGENTID = agentid
def get_access_token(self):
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
values = {
"corpid": self.CORPID,
"corpsecret": self.CORPSECRET,
}
req = requests.post(url, params=values)
data = json.loads(req.text)
return data["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()
)
send_values = {
"touser": touser,
"msgtype": "text",
"agentid": self.AGENTID,
"text": {"content": message},
"safe": "0",
}
send_msges = bytes(json.dumps(send_values), "utf-8")
respone = requests.post(send_url, send_msges)
respone = respone.json()
return respone["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()
)
send_values = {
"touser": touser,
"msgtype": "mpnews",
"agentid": self.AGENTID,
"mpnews": {
"articles": [
{ {
'title': f'{title}', "title": title,
'thumb_media_id': f'{qywx_app_params[4]}', "thumb_media_id": media_id,
'author': '智能助手', "author": "Author",
'content_source_url': '', "content_source_url": "",
'content': f'{html}', "content": message.replace("\n", "<br/>"),
'digest': f'{content}' "digest": message,
} }
] ]
},
} }
} send_msges = bytes(json.dumps(send_values), "utf-8")
respone = requests.post(send_url, send_msges)
url=f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={accesstoken}" respone = respone.json()
data = { return respone["errmsg"]
'touser': f'{change_user_id(content)}',
'agentid': f'{qywx_app_params[3]}',
'safe': '0'
}
data.update(options)
headers = {
'Content-Type': 'application/json',
}
response = requests.post(url=url, headers=headers, data=json.dumps(data)).json()
if response['errcode'] == 0:
print('推送成功！')
else:
print('推送失败！')
def qyw_bot(title, content): def wecom_bot(title: str, content: str) -> None:
print('开始使用 企业微信机器人 推送消息...', end='')
url = f'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={QYWX_KEY}'
headers = {'Content-Type': 'application/json;charset=utf-8'}
data = {
'msgtype': 'text',
'text': {'content': f'{title}\n\n{content}'}
}
response = requests.post(url=url, data=json.dumps(data), headers=headers, timeout=15).json()
if response['errcode'] == 0:
print('推送成功！')
else:
print('推送失败！')
def change_user_id(desp):
qywx_app_params = QYWX_AM.split(',')
if qywx_app_params[2]:
userIdTmp = qywx_app_params[2].split("|")
userId = ""
for i in range(len(userIdTmp)):
count1 = f"账号{i + 1}"
count2 = f"签到号{i + 1}"
if re.search(count1, desp) or re.search(count2, desp):
userId = userIdTmp[i]
if not userId:
userId = qywx_app_params[2]
return userId
else:
return "@all"
def send(title, content):
""" """
使用 bark, telegram bot, dingding bot, serverJ 发送手机推送 通过 企业微信机器人 推送消息。
:param title: """
:param content: if not push_config.get("QYWX_KEY"):
print("企业微信机器人 服务的 QYWX_KEY 未设置!!\n取消推送")
return
print("企业微信机器人服务启动")
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(
url=url, data=json.dumps(data), headers=headers, timeout=15
).json()
if response["errcode"] == 0:
print("企业微信机器人推送成功！")
else:
print("企业微信机器人推送失败！")
def telegram_bot(title: str, content: str) -> None:
"""
使用 telegram 机器人 推送消息。
"""
if not push_config.get("TG_BOT_TOKEN") or not push_config.get("TG_USER_ID"):
print("tg 服务的 bot_token 或者 user_id 未设置!!\n取消推送")
return
print("tg 服务启动")
if push_config.get("TG_API_HOST"):
url = f"https://{push_config.get('TG_API_HOST')}/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage"
else:
url = (
f"https://api.telegram.org/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage"
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
payload = {
"chat_id": str(push_config.get("TG_USER_ID")),
"text": f"{title}\n\n{content}",
"disable_web_page_preview": "true",
}
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"
):
push_config["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"]:
print("tg 推送成功！")
else:
print("tg 推送失败！")
def one() -> str:
"""
获取一条一言。
:return: :return:
""" """
for i in notify_mode: url = "https://v1.hitokoto.cn/"
if i == 'bark': res = requests.get(url).json()
if BARK: return res["hitokoto"] + " ----" + res["from"]
bark(title=title, content=content)
else:
print('未启用 bark') if push_config.get("BARK_PUSH"):
continue notify_function.append(bark)
if i == 'sc_key': if push_config.get("CONSOLE"):
if SCKEY: notify_function.append(console)
serverJ(title=title, content=content) if push_config.get("DD_BOT_TOKEN") and push_config.get("DD_BOT_SECRET"):
else: notify_function.append(dingding_bot)
print('未启用 Server酱') if push_config.get("FSKEY"):
continue notify_function.append(feishu_bot)
elif i == 'dingding_bot': if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"):
if DD_BOT_TOKEN and DD_BOT_SECRET: notify_function.append(go_cqhttp)
dingding_bot(title=title, content=content) if push_config.get("IGOT_PUSH_KEY"):
else: notify_function.append(iGot)
print('未启用 钉钉机器人') if push_config.get("PUSH_KEY"):
continue notify_function.append(serverJ)
elif i == 'telegram_bot': if push_config.get("PUSH_PLUS_TOKEN"):
if TG_BOT_TOKEN and TG_USER_ID: notify_function.append(pushplus_bot)
telegram_bot(title=title, content=content) if push_config.get("QMSG_KEY") and push_config.get("QMSG_TYPE"):
else: notify_function.append(qmsg_bot)
print('未启用 telegram机器人') if push_config.get("QYWX_AM"):
continue notify_function.append(wecom_app)
elif i == 'qywxapp_bot': if push_config.get("QYWX_KEY"):
if QYWX_AM: notify_function.append(wecom_bot)
qywxapp_bot(title=title, content=content) if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"):
else: notify_function.append(telegram_bot)
print('未启用 企业微信应用推送')
continue
elif i == 'qyw_bot': def send(title: str, content: str) -> None:
if QYWX_KEY: if not content:
qyw_bot(title=title, content=content) print(f"{title} 推送内容为空！")
else: return
print('未启用 企业微信机器人')
continue hitokoto = push_config.get("HITOKOTO")
else:
print('此类推送方式不存在') text = one() if hitokoto else ""
content += "\n\n" + text
ts = [
threading.Thread(target=mode, args=(title, content), name=mode.__name__)
for mode in notify_function
]
[t.start() for t in ts]
[t.join() for t in ts]
def main(): def main():
send('title', 'content') send("title", "content")
if __name__ == '__main__': if __name__ == "__main__":
main() main()