配置文件
# 飞书机器人key
bot_app = {
"app_id": "************************************",
"app_secret": "************************************"
}
}
# 飞书群机器人
bot = {
"bot_test": "************************************", # 测试机器人
}
url_api = {
"url_msg": "https://open.feishu.cn/open-apis/bot/v2/hook/", # 发送普通消息时用
"url_card": "https://open.feishu.cn/open-apis/message/v4/send/", # 发送卡片消息时用
"url_img": "https://open.feishu.cn/open-apis/image/v4/put/", # 发送图片消息时用
"url_token": "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/" # 获取机器人的key
}
主程序
import json
import requests
from .config import url_api, bot_app, bot1_app
def get_token(app):
"""获取应用token"""
url = url_api['url_token']
headers = {"Content-Type": "text/plain"}
r = requests.post(url, headers=headers, json=app)
return json.loads(r.text)
def upload_image(image_path):
"""上传图片"""
with open(image_path, 'rb') as f:
image = f.read()
resp = requests.post(
url=url_api['url_img'],
headers={'Authorization': "Bearer " + get_token(bot_app)['tenant_access_token']},
files={
"image": image
},
data={
"image_type": "message"
},
stream=True)
resp.raise_for_status()
content = resp.json()
if content.get("code") == 0:
return content['data']['image_key']
else:
return Exception("Call Api Error, errorCode is %s" % content["code"])
def send_text(Text, bot):
"""发送普通消息"""
url = url_api['url_msg'] + bot
headers = {"Content-Type": "text/plain"}
data = {
"msg_type": "text",
"content": {
"text": Text
}
}
r = requests.post(url, headers=headers, json=data)
return r.text
def send_img(path, bot):
"""发送图片消息"""
url = url_api['url_msg'] + bot
headers = {"Content-Type": "text/plain"}
data = {
"msg_type": "image",
"content": {
"image_key": upload_image(path)
}
}
r = requests.post(url, headers=headers, json=data)
return r.text
def send_markdown(text, bot):
"""发送富文本消息"""
url = url_api['url_msg'] + bot
headers = {"Content-Type": "text/plain"}
data = {
"msg_type": "interactive",
"card": {
"config": {
"wide_screen_mode": True
},
"header": {
"title": {
"tag": "plain_text",
"content": "提醒"
},
"template": "red"
},
"elements": [{"tag": "div",
"text": {"content": text,
"tag": "lark_md"}}]}
}
r = requests.post(url, headers=headers, json=data)
return r.text
def send_card(Text, bot):
"""发送卡片消息"""
url = url_api['url_msg'] + bot
headers = {"Content-Type": "text/plain"}
data = {
"msg_type": "interactive",
"card": Text
}
r = requests.post(url, headers=headers, json=data)
return r.text
def send_file(localfile, bot):
"""发送文件，根据send_markdown改的，主要是将七牛的地址放进来"""
url = url = url_api['url_msg'] + bot
headers = {"Content-Type": "text/plain"}
key = localfile.split('/')[-1]
data = {
"msg_type": "post",
"content": {
"post": {
"zh_cn": {
"title": key.split('.')[0],
"content": [[{"tag": "text",
"text": "数据链接: "},
{"tag": "a",
"text": "点击下载",
"href": qn_url(localfile)}]]
}}}}
r = requests.post(url, headers=headers, json=data)
return r.text
def send_fscard(mail, data):
"""发送飞书卡片消息"""
header = {
"Authorization": "Bearer " + get_token(bot1_app)['tenant_access_token'],
"Content-Type": "application/json; charset=utf-8"
}
body = {
"msg_type": "interactive",
"email": mail,
"card": data
}
r = requests.post(url_api['url_card'], json=body, headers=header)
return r.text
