飞书自定义机器人 Webhook 接入实战

public class FeishuWebhook {
    private static final Logger logger = LoggerFactory.getLogger(FeishuWebhook.class);
    public static final String DEFAULT_DATETIME_FORMATTER_STR = "yyyy-MM-dd HH:mm:ss";
    public static final DateTimeFormatter DEFAULT_DATETIME_FORMATTER = DateTimeFormatter.ofPattern(DEFAULT_DATETIME_FORMATTER_STR);

    public static void send(String url, String secret, AlertDO alertDO) {
        logger.info("FeishuWebhook.send, url:{}, alertDO={}", url, alertDO);
        JSONObject requestBody = createRequestBody(secret, alertDO);
        logger.info("requestBody:{}", requestBody);
        // 假设 HttpUtils.postForJsonObject 已封装好 POST 请求
        JSONObject result = HttpUtils.postForJsonObject(url, null, null, requestBody);
        logger.info("result:{}", result);
    }

    private static String genSign(String secret, long timestamp) throws NoSuchAlgorithmException, InvalidKeyException {
        // 签名字符串：timestamp + "\n" + secret
        String stringToSign = timestamp + "\n" + secret;
        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(stringToSign.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
        return new String(Base64.encodeBase64(signData));
    }

    private static JSONObject createRequestBody(String secret, AlertDO alertDO) {
        JSONObject requestBody = new JSONObject();
        long seconds = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().getEpochSecond();
        logger.info("seconds: {}", seconds);

        try {
            String sign = genSign(secret, seconds);
            requestBody.put("timestamp", seconds);
            requestBody.put("sign", sign);
            requestBody.put("msg_type", "post");

            String title = "通知";
            String message = alertDO.getMessage();
            String detail = alertDO.getDetail();
            String startTime = DEFAULT_DATETIME_FORMATTER.format(LocalDateTime.ofInstant(alertDO.getStartTime().toInstant(), ZoneId.systemDefault()));
            String endTime = DEFAULT_DATETIME_FORMATTER.format(LocalDateTime.ofInstant(alertDO.getEndTime().toInstant(), ZoneId.systemDefault()));

            requestBody.put("content", createOuterContent(title, message, detail, startTime, endTime));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return requestBody;
    }

    private static JSONObject createOuterContent(String title, String message, String detail, String startTime, String endTime) {
        JSONObject result = new JSONObject();
        result.put("post", createPostJsonObject(title, message, detail, startTime, endTime));
        return result;
    }

    private static JSONObject createPostJsonObject(String title, String message, String detail, String startTime, String endTime) {
        JSONObject result = new JSONObject();
        result.put("zh_cn", createZhCNJsonObject(title, message, detail, startTime, endTime));
        return result;
    }

    private static JSONObject createZhCNJsonObject(String title, String message, String detail, String startTime, String endTime) {
        JSONObject result = new JSONObject();
        result.put("title", title);
        result.put("content", createContentList(message, detail, startTime, endTime));
        return result;
    }

    private static JSONArray createContentList(String message, String detail, String startTime, String endTime) {
        JSONArray result = new JSONArray();
        result.add(createItem("message", message));
        result.add(createItem("detail", detail));
        result.add(createItem("startTime", startTime));
        result.add(createItem("endTime", endTime));
        return result;
    }

    private static JSONArray createItem(String tag, String text) {
        JSONArray item = new JSONArray();
        item.add(createInnerHeadContent(tag));
        item.add(createInnerTextContent(text));
        return item;
    }

    private static JSONObject createInnerHeadContent(String tag) {
        JSONObject result = new JSONObject();
        result.put("tag", "text");
        result.put("text", tag + ": ");
        return result;
    }

    private static JSONObject createInnerTextContent(String text) {
        JSONObject result = new JSONObject();
        result.put("tag", "text");
        result.put("text", text);
        return result;
    }
}

4. Python 实现示例

Python 版本更简洁，利用 requests 库即可快速发起请求。这里提供带签名和不带签名两种场景的代码参考。

带签名验证

import base64
import hashlib
import hmac
from datetime import datetime
import requests

WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/xx"
WEBHOOK_SECRET = "ssssssss"

class LarkBot:
    def __init__(self, secret: str):
        if not secret:
            raise ValueError("invalid secret key")
        self.secret = secret

    def gen_sign(self, timestamp: int) -> str:
        string_to_sign = '{}\n{}'.format(timestamp, self.secret)
        hmac_code = hmac.new(
            string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(hmac_code).decode('utf-8')
        return sign

    def send(self):
        timestamp = int(datetime.now().timestamp())
        sign = self.gen_sign(timestamp)
        params = {
            "timestamp": timestamp,
            "sign": sign,
            "msg_type": "post",
            "content": {
                "post": {
                    "zh_cn": {
                        "title": "项目更新通知",
                        "content": [
                            [{"tag": "text", "text": "项目有更新: "}, {"tag": "a", "text": "请查看", "href": "http://www.example.com/"}]
                        ]
                    }
                }
            }
        }
        resp = requests.post(url=WEBHOOK_URL, json=params)
        resp.raise_for_status()
        result = resp.json()
        if result.get("code") and result["code"] != 0:
            print(result["msg"])
            return
        print("消息发送成功")

def main():
    bot = LarkBot(secret=WEBHOOK_SECRET)
    bot.send()

if __name__ == '__main__':
    main()

无签名验证（仅测试用）

如果机器人未开启安全设置，可直接发送文本消息。

from datetime import datetime
import requests

WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/sss"

class LarkBot:
    def send(self, content: str):
        timestamp = int(datetime.now().timestamp())
        params = {
            "timestamp": timestamp,
            "msg_type": "text",
            "content": {"text": content},
        }
        resp = requests.post(url=WEBHOOK_URL, json=params)
        resp.raise_for_status()
        result = resp.json()
        if result.get("code") and result["code"] != 0:
            print(result["msg"])
            return
        print("消息发送成功")

def main():
    bot = LarkBot()
    bot.send(content="我是一只高级鸽子！")

if __name__ == '__main__':
    main()

5. 注意事项

• 密钥保护：Webhook 地址和密钥相当于密码，切勿提交至公共代码仓库。
• 时区处理：时间戳计算务必考虑服务器时区，建议统一使用 UTC 或明确指定系统默认时区。
• 错误处理：生产环境应捕获 HTTP 异常并记录日志，避免因网络波动导致静默失败。