PyBot/SendCore/FeishuSendBot.py
2024-12-15 02:14:39 +08:00

79 lines
2.4 KiB
Python

# -*- coding: utf-8 -*-
import requests
import json
import hashlib
import base64
import hmac
import time
import yaml
def gen_sign():
with open('./config.yaml', 'r', encoding="utf-8") as file:
config = yaml.safe_load(file)
secret = f"{config['fs_secret']}"
# print(secret)
webhook_url = f"https://open.feishu.cn/open-apis/bot/v2/hook/{config['fs_key']}"
# print(webhook_url)
timestamp = int(time.time())
#拼接timestamp和secret
string_to_sign = '{}\n{}'.format(timestamp, secret)
hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
# 对结果进行base64处理
sign = base64.b64encode(hmac_code).decode('utf-8')
return webhook_url, timestamp, sign
webhook_url, timestamp, sign = gen_sign()
# 主函数
def SendToFeishu(body, header, webhook_url, timestamp, sign):
msg = {
"timestamp": f"{timestamp}",
"sign": f"{sign}",
"msg_type": "interactive",
"card": {
"elements":
[{
"tag": "markdown",
"content": f"{body}"
}],
"header": {
"title": {
"content": f"{header}",
"tag": "plain_text"
}
}
}
}
# WebHook请求头
headers = {
"Content-Type": "application/json",
}
msg_encode=json.dumps(msg,ensure_ascii=True).encode("utf-8")
response=requests.post(url=webhook_url,data=msg_encode,headers=headers)
try:
# 解析 JSON 字符串
response_data = json.loads(response.content)
# 检查 msg 字段
if response_data.get('msg') == 'success':
return f"飞书发送 {header} 成功"
elif response_data.get('msg') == 'sign match fail or timestamp is not within one hour from current time':
return f"{header} 发送失败: 签名验证错误,请检查签名密钥是否正确!"
else:
return f"{header} 发送失败: 其他错误,请检查请求参数是否正确!\n原因:{response_data.get('msg')}"
except json.JSONDecodeError as e:
return f"{header} JSON解析错误: {e}"
# print(sign)
# print(response.content)
msg = "测试消息"
if __name__ == '__main__':
result = SendToFeishu(msg, "测试消息")
print(result)