PyBot/SendCore/FeishuSendBot.py

# -*- 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)