-
Notifications
You must be signed in to change notification settings - Fork 20
/
dingding.py
138 lines (112 loc) · 3.98 KB
/
dingding.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import json
import requests
from ..config import config_path, load_api_keys
api_keys = load_api_keys(config_path)
dingding_url = api_keys.get("dingding_url")
wake_word = ""
# 发送消息到钉钉的函数
def send_dingding(content, msgtype="markdown"):
global dingding_url, wake_word
webhook_url = dingding_url
headers = {"Content-Type": "application/json; charset=utf-8"}
if msgtype == "text":
data = {
"msgtype": msgtype,
"text": {
"content": wake_word + "\n" + content if wake_word != "" else content,
},
}
elif msgtype == "markdown":
data = {
"msgtype": msgtype,
"markdown": {"text": content, "title": wake_word},
}
# 发送POST请求到钉钉服务器
response = requests.post(webhook_url, headers=headers, data=json.dumps(data))
# 返回响应状态码
return response.status_code
class Dingding_tool:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"is_enable": ("BOOLEAN", {"default": True}),
"msgtype": (["text", "markdown"], {"default": "markdown"}),
"key_word": ("STRING", {"default": ""}),
},
"optional": {
"url": ("STRING", {}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("tool",)
FUNCTION = "web"
# OUTPUT_NODE = False
CATEGORY = "大模型派对(llm_party)/工具(tools)"
def web(self, is_enable=True, url=None, msgtype="markdown", key_word=""):
if is_enable == False:
return (None,)
global dingding_url, wake_word
if url is not None and url != "":
dingding_url = url
else:
dingding_url = api_keys.get("dingding_url")
if key_word is not None and key_word != "":
wake_word = key_word
output = [
{
"type": "function",
"function": {
"name": "send_dingding",
"description": "向钉钉发送消息",
"parameters": {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "需要发送的消息,支持Markdown格式",
},
"msgtype": {
"type": "string",
"description": "消息类型,支持text、markdown",
"default": msgtype,
},
},
"required": ["content", "msgtype"],
},
},
}
]
out = json.dumps(output, ensure_ascii=False)
return (out,)
class Dingding:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"content": ("STRING", {"default": "hello world"}),
"is_enable": ("BOOLEAN", {"default": True}),
"msgtype": (["text", "markdown"], {"default": "markdown"}),
"key_word": ("STRING", {"default": ""}),
},
"optional": {
"url": ("STRING", {}),
},
}
RETURN_TYPES = ()
RETURN_NAMES = ()
FUNCTION = "web"
OUTPUT_NODE = True
CATEGORY = "大模型派对(llm_party)/函数(function)"
def web(self, is_enable=True, url=None, content="hello world", msgtype="markdown", key_word=""):
if is_enable == False:
return (None,)
global dingding_url, wake_word
if url is not None and url != "":
dingding_url = url
else:
dingding_url = api_keys.get("dingding_url")
if key_word is not None and key_word != "":
wake_word = key_word
send_dingding(content, msgtype)
return ()