This repository has been archived by the owner on Jan 16, 2023. It is now read-only.
forked from valetzx/uptimekumaonreplit
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
105 lines (93 loc) · 3.92 KB
/
main.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
#这个文件需要新建一个Python仓库并将这个文件放入Python仓库运行!
#在Python仓库中点绿色run之前 在shell中执行 `pip install python-multipart`
import base64
import json
import requests
import uvicorn
from fastapi import FastAPI, Form
#如何注册企业微信请看:https://github.com/riba2534/wecomchan
wecom_cid = "引号中填企业微信id"
wecom_aid = "引号中填企业微信应用id"
wecom_secret = "引号中填企业微信应用秘钥secret"
app = FastAPI()
def send_to_wecom(text, wecom_touid='@all'):
global wecom_cid, wecom_aid, wecom_secret
get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
response = requests.get(get_token_url).content
access_token = json.loads(response).get('access_token')
if access_token and len(access_token) > 0:
send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
data = {
"touser": wecom_touid,
"agentid": wecom_aid,
"msgtype": "text",
"text": {
"content": text
},
"duplicate_check_interval": 600
}
response = requests.post(send_msg_url, data=json.dumps(data)).content
return response
else:
return False
def send_to_wecom_image(base64_content, wecom_touid='@all'):
global wecom_cid, wecom_aid, wecom_secret
get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
response = requests.get(get_token_url).content
access_token = json.loads(response).get('access_token')
if access_token and len(access_token) > 0:
upload_url = f'https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=image'
upload_response = requests.post(upload_url, files={
"picture": base64.b64decode(base64_content)
}).json()
if "media_id" in upload_response:
media_id = upload_response['media_id']
else:
return False
send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
data = {
"touser": wecom_touid,
"agentid": wecom_aid,
"msgtype": "image",
"image": {
"media_id": media_id
},
"duplicate_check_interval": 600
}
response = requests.post(send_msg_url, data=json.dumps(data)).content
return response
else:
return False
def send_to_wecom_markdown(text, wecom_touid='@all'):
global wecom_cid, wecom_aid, wecom_secret
get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
response = requests.get(get_token_url).content
access_token = json.loads(response).get('access_token')
if access_token and len(access_token) > 0:
send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
data = {
"touser": wecom_touid,
"agentid": wecom_aid,
"msgtype": "markdown",
"markdown": {
"content": text
},
"duplicate_check_interval": 600
}
response = requests.post(send_msg_url, data=json.dumps(data)).content
return response
else:
return False
@app.post("/")
def main(type: str = Form(...), title: str = Form(...), body: str = Form(...), wecom_touid='@all'):
if type == 'note':
data = send_to_wecom(title + '\n' + body, wecom_touid)
elif type == 'image':
data = send_to_wecom_image(body, wecom_touid)
elif type == 'markdown':
data = send_to_wecom_markdown(title + '\n' + body, wecom_touid)
else:
data = send_to_wecom(title + '\n' + body, wecom_touid)
return data
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=3000, log_level="info")