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

增加符合OneBot标准的通知方式

Browse Source
This commit is contained in:
Qian23333 2025-03-24 21:38:53 +08:00 committed by GitHub
parent 2370923a75
commit e94475dcb2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 56 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
sample/notify.py
View File

@ -58,6 +58,11 @@ push_config = {
# /send_group_msg 时填入 group_id=QQ群 # /send_group_msg 时填入 group_id=QQ群
'GOBOT_TOKEN': '', # go-cqhttp 的 access_token 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token
'ONEBOT_URL': '', # OneBot 的推送地址，以send_msg结尾
'ONEBOT_USER': '', # OneBot 的推送对象，QQ号
'ONEBOT_GROUP': '', # OneBot 的推送对象，群号
'ONEBOT_TOKEN': '', # OneBot 的 access_token，可选
'GOTIFY_URL': '', # gotify地址,如https://push.example.de:8080 'GOTIFY_URL': '', # gotify地址,如https://push.example.de:8080
'GOTIFY_TOKEN': '', # gotify的消息应用token 'GOTIFY_TOKEN': '', # gotify的消息应用token
'GOTIFY_PRIORITY': 0, # 推送消息优先级,默认为0 'GOTIFY_PRIORITY': 0, # 推送消息优先级,默认为0
@ -254,6 +259,53 @@ def go_cqhttp(title: str, content: str) -> None:
print("go-cqhttp 推送失败！") print("go-cqhttp 推送失败！")
def one_bot(title: str, content: str) -> None:
"""
使用 one_bot 推送消息。
"""
if not push_config.get("ONEBOT_USER") and not push_config.get("ONEBOT_GROUP"):
return
if not push_config.get("ONEBOT_URL"):
return
print("OneBot 服务启动")
url = push_config.get("ONEBOT_URL").rstrip("/")
user_id = push_config.get("ONEBOT_USER")
group_id = push_config.get("ONEBOT_GROUP")
token = push_config.get("ONEBOT_TOKEN")
if not url.endswith("/send_msg"):
url += "/send_msg"
headers = {'Content-Type': "application/json"}
message = [{"type": "text", "data": {"text": f"{title}\n\n{content}"}}]
data_private = {"message": message}
data_group = {"message": message}
if token is not None:
headers["Authorization"] = f"Bearer {token}"
if user_id is not None:
data_private["message_type"] = "private"
data_private["user_id"] = user_id
response_private = requests.post(url, data=json.dumps(data_private), headers=headers).json()
if response_private["status"] == "ok":
print("OneBot 私聊推送成功！")
else:
print("OneBot 私聊推送失败！")
if group_id is not None:
data_group["message_type"] = "group"
data_group["group_id"] = group_id
response_group = requests.post(url, data=json.dumps(data_group), headers=headers).json()
if response_group["status"] == "ok":
print("OneBot 群聊推送成功！")
else:
print("OneBot 群聊推送失败！")
def gotify(title: str, content: str) -> None: def gotify(title: str, content: str) -> None:
""" """
使用 gotify 推送消息。 使用 gotify 推送消息。
@ -985,6 +1037,8 @@ def add_notify_function():
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("ONEBOT_URL"):
notify_function.append(one_bot)
if push_config.get("GOTIFY_URL") and push_config.get("GOTIFY_TOKEN"): if push_config.get("GOTIFY_URL") and push_config.get("GOTIFY_TOKEN"):
notify_function.append(gotify) notify_function.append(gotify)
if push_config.get("IGOT_PUSH_KEY"): if push_config.get("IGOT_PUSH_KEY"):