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

修复 python Server酱推送

Browse Source
This commit is contained in:
whyour 2023-03-19 22:54:50 +08:00
parent f1f009da3b
commit a5ecb204b2
1 changed files with 28 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
sample/notify.py
View File

@ -164,7 +164,8 @@ def dingding_bot(title: str, content: str) -> None:
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 = "{}\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
@ -220,7 +221,7 @@ def go_cqhttp(title: str, content: str) -> None:
print("go-cqhttp 推送失败！")
def gotify(title:str,content:str) -> None:
def gotify(title: str, content: str) -> None:
"""
使用 gotify 推送消息。
"""
@ -230,8 +231,9 @@ def gotify(title:str,content:str) -> None:
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()
data = {"title": title, "message": content,
"priority": push_config.get("GOTIFY_PRIORITY")}
response = requests.post(url, data=data).json()
if response.get("id"):
print("gotify 推送成功！")
@ -269,10 +271,10 @@ def serverJ(title: str, content: str) -> None:
print("serverJ 服务启动")
data = {"text": title, "desp": content.replace("\n", "\n\n")}
if push_config.get("PUSH_KEY").index("SCT") != -1:
if push_config.get("PUSH_KEY").find("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'
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:
@ -289,7 +291,8 @@ def pushdeer(title: str, content: str) -> None:
print("PushDeer 服务的 DEER_KEY 未设置!!\n取消推送")
return
print("PushDeer 服务启动")
data = {"text": title, "desp": content, "type": "markdown", "pushkey": push_config.get("DEER_KEY")}
data = {"text": title, "desp": content, "type": "markdown",
"pushkey": push_config.get("DEER_KEY")}
url = 'https://api2.pushdeer.com/message/push'
if push_config.get("DEER_URL"):
url = push_config.get("DEER_URL")
@ -320,7 +323,6 @@ def chat(title: str, content: str) -> None:
print("Chat 推送失败！错误信息：", response)
def pushplus_bot(title: str, content: str) -> None:
"""
通过 push+ 推送消息。
@ -348,7 +350,8 @@ def pushplus_bot(title: str, content: str) -> None:
url_old = "http://pushplus.hxtrip.com/send"
headers["Accept"] = "application/json"
response = requests.post(url=url_old, data=body, headers=headers).json()
response = requests.post(
url=url_old, data=body, headers=headers).json()
if response["code"] == 200:
print("PUSHPLUS(hxtrip) 推送成功！")
@ -367,7 +370,8 @@ def qmsg_bot(title: str, content: str) -> None:
print("qmsg 服务启动")
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:
@ -553,14 +557,14 @@ def aibotk(title: str, content: str) -> None:
data = {
"apiKey": push_config.get("AIBOTK_KEY"),
"roomName": push_config.get("AIBOTK_NAME"),
"message": {"type": 1, "content": f'【青龙快讯】\n\n${title}\n${content}' }
"message": {"type": 1, "content": f'【青龙快讯】\n\n{title}\n{content}'}
}
else:
url = "https://api-bot.aibotk.com/openapi/v1/chat/contact"
data = {
"apiKey": push_config.get("AIBOTK_KEY"),
"name": push_config.get("AIBOTK_NAME"),
"message": {"type": 1, "content": f'【青龙快讯】\n\n${title}\n${content}' }
"message": {"type": 1, "content": f'【青龙快讯】\n\n{title}\n{content}'}
}
body = json.dumps(data).encode(encoding="utf-8")
headers = {"Content-Type": "application/json"}
@ -582,14 +586,19 @@ def smtp(title: str, content: str) -> None:
print("SMTP 邮件 服务启动")
message = MIMEText(content, 'plain', 'utf-8')
message['From'] = formataddr((Header(push_config.get("SMTP_NAME"), 'utf-8').encode(), push_config.get("SMTP_EMAIL")))
message['To'] = formataddr((Header(push_config.get("SMTP_NAME"), 'utf-8').encode(), push_config.get("SMTP_EMAIL")))
message['From'] = formataddr((Header(push_config.get(
"SMTP_NAME"), 'utf-8').encode(), push_config.get("SMTP_EMAIL")))
message['To'] = formataddr((Header(push_config.get(
"SMTP_NAME"), 'utf-8').encode(), push_config.get("SMTP_EMAIL")))
message['Subject'] = Header(title, 'utf-8')
try:
smtp_server = smtplib.SMTP_SSL(push_config.get("SMTP_SERVER")) if push_config.get("SMTP_SSL") == 'true' else smtplib.SMTP(push_config.get("SMTP_SERVER"))
smtp_server.login(push_config.get("SMTP_EMAIL"), push_config.get("SMTP_PASSWORD"))
smtp_server.sendmail(push_config.get("SMTP_EMAIL"), push_config.get("SMTP_EMAIL"), message.as_bytes())
smtp_server = smtplib.SMTP_SSL(push_config.get("SMTP_SERVER")) if push_config.get(
"SMTP_SSL") == 'true' else smtplib.SMTP(push_config.get("SMTP_SERVER"))
smtp_server.login(push_config.get("SMTP_EMAIL"),
push_config.get("SMTP_PASSWORD"))
smtp_server.sendmail(push_config.get("SMTP_EMAIL"),
push_config.get("SMTP_EMAIL"), message.as_bytes())
smtp_server.close()
print("SMTP 邮件 推送成功！")
except Exception as e:
@ -660,7 +669,8 @@ def send(title: str, content: str) -> None:
content += "\n\n" + text
ts = [
threading.Thread(target=mode, args=(title, content), name=mode.__name__)
threading.Thread(target=mode, args=(
title, content), name=mode.__name__)
for mode in notify_function
]
[t.start() for t in ts]