PyBot/SendBot.py

# -*- coding: utf-8 -*-

import requests
import json
import hashlib
import base64
import hmac
import time
import yaml


# # 官方提供用以验证签名的函数
# def gen_sign(timestamp, secret):
#     # 拼接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 sign

def gen_sign():
    with open('./config.yaml', 'r', encoding="utf-8") as file:
        config = yaml.safe_load(file)
    secret = f"{config['secret']}"
    # print(secret)
    webhook_url = f"https://open.feishu.cn/open-apis/bot/v2/hook/{config['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)





# 测试用消息体
test_msg = {
    "timestamp": f"{timestamp}",
    "sign": f"{sign}",
    "msg_type": "interactive",
    "card": {
        "elements": 
        [{
            "tag": "markdown",
            "content": "请单件文档查看昨天讨论的方案相关飞书文档，注意作者为 <font color=red> **张三** <font> 版本为 \n*002* ,版本 ~~001~~ 已经删除。文件地址是 [https://www.feishu.cn](https://www.feishu.cn),打开次数:1次"
        },
        {
            "actions": 
            [{
                "tag": "button",
                "text": {
                        "content": "点此访问原文",
                        "tag": "lark_md"
                },
                "url": "https://www.masonliu.com",
                "type": "default",
                "value": {}
            }],
            "tag": "action"
        }],
        "header": {
                "title": {
                        "content": "网安资讯传递",
                        "tag": "plain_text"
                }
        }
    }
}


if __name__ == '__main__':
    SendToFeishu(test_msg, "测试消息")