飞书机器人实战：用Python脚本实现CI/CD打包成功自动通知（附完整代码）
在DevOps实践中，持续集成与持续部署（CI/CD）已成为现代软件开发的标准流程。每当代码提交后，自动化构建和测试流程随即启动，但构建结果的通知环节却常常被忽视——开发团队要么频繁刷新构建页面，要么被动等待邮件通知。这种低效的沟通方式完全可以通过飞书机器人API实现即时、可视化的改进。
本文将手把手带您实现一个典型的自动化场景：当Jenkins或GitLab CI完成代码打包构建后，自动通过飞书机器人向相关群组发送富文本通知，包含构建状态、耗时、版本号等关键信息。不同于简单的API调用教程，我们将聚焦三个核心价值点：如何设计有实用价值的通知内容、如何处理飞书API的权限陷阱、如何将脚本无缝集成到现有CI流程中。以下是完整的实现路径：
1. 飞书机器人前期配置
1.1 创建自定义机器人应用
首先登录飞书开放平台，进入"开发者后台"创建新应用。关键配置项包括：
- 应用名称：建议包含"CI/CD"等标识，如"订单系统构建通知"
- 应用描述：明确说明用途"用于构建结果通知"
- 权限范围：选择"企业自建应用"
创建完成后，进入"凭证与基础信息"页面，记录以下关键参数：
App ID: cli_xxxxxx App Secret: xxxxxx1.2 配置必要的API权限
飞书机器人的消息发送能力依赖于精细的权限控制，需要手动添加以下权限：
| 权限名称 | 权限描述 | 是否必需 |
|---|---|---|
| 获取群组信息 | 读取机器人所在群聊的基本信息 | 是 |
| 发送消息 | 向单个用户或群组发送消息 | 是 |
| 批量发送消息 | 同时向多个用户发送相同消息 | 可选 |
| 获取用户user_id | 通过手机号/邮箱查询用户信息 | 可选 |
注意：权限申请后需要管理员审核，通常需要1-2小时生效。建议提前准备以免阻塞后续开发。
1.3 将机器人添加到目标群聊
在飞书客户端中，右键点击目标群组选择"设置"，进入"群机器人"标签页：
- 点击"添加机器人"
- 搜索并选择刚创建的应用
- 确认添加后，记录群聊的
chat_id（可通过右键群名片获取）
2. Python脚本开发实战
2.1 获取访问凭证(access_token)
飞书API的所有请求都需要携带有效的access_token。以下是Python实现示例：
import requests def get_feishu_token(app_id, app_secret): url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" headers = {"Content-Type": "application/json"} payload = { "app_id": app_id, "app_secret": app_secret } response = requests.post(url, headers=headers, json=payload) if response.status_code == 200 and response.json().get("code") == 0: return response.json()["tenant_access_token"] else: raise Exception(f"获取token失败: {response.text}")关键注意事项：
- token有效期为2小时，需要缓存并定期刷新
- 每个企业每天有1000次的token获取限制
- 生产环境建议使用Redis等缓存机制
2.2 构建富文本消息体
飞书支持多种消息格式，对于构建通知推荐使用"富文本"格式：
def build_ci_message(build_status, duration, version, commit_msg): status_emoji = "✅" if build_status == "success" else "❌" return { "msg_type": "interactive", "card": { "header": { "title": { "content": f"构建通知 {status_emoji}", "tag": "plain_text" }, "template": "green" if build_status == "success" else "red" }, "elements": [ { "tag": "div", "text": { "content": f"**项目**：订单系统后端\n**状态**：{build_status}\n**耗时**：{duration}s\n**版本**：{version}", "tag": "lark_md" } }, { "tag": "div", "text": { "content": f"**最近提交**：{commit_msg}", "tag": "lark_md" } }, { "tag": "action", "actions": [ { "tag": "button", "text": { "content": "查看构建详情", "tag": "plain_text" }, "type": "primary", "url": "https://jenkins.example.com/job/order-system" } ] } ] } }2.3 完整消息发送实现
结合前两步，完整的消息发送函数如下：
def send_feishu_message(chat_id, message, access_token): url = f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}" } payload = { "receive_id": chat_id, "content": json.dumps(message), "msg_type": message["msg_type"] } response = requests.post(url, headers=headers, json=payload) result = response.json() if response.status_code == 200 and result.get("code") == 0: print("消息发送成功") return True elif result.get("code") == 99991401: raise Exception("token过期或无效") else: raise Exception(f"发送失败: {result.get('msg')}")3. 集成到CI/CD流程
3.1 Jenkins集成方案
在Jenkins项目的"构建后操作"中添加Execute shell步骤：
#!/bin/bash if [ $BUILD_RESULT == "SUCCESS" ]; then STATUS="success" else STATUS="failure" fi python3 /opt/scripts/feishu_notify.py \ --status $STATUS \ --duration $BUILD_DURATION \ --version $BUILD_TAG \ --commit "$GIT_COMMIT_MSG"对应的Python脚本参数处理部分：
import argparse parser = argparse.ArgumentParser() parser.add_argument("--status", required=True) parser.add_argument("--duration", type=int, required=True) parser.add_argument("--version", required=True) parser.add_argument("--commit", required=True) args = parser.parse_args() message = build_ci_message( build_status=args.status, duration=args.duration, version=args.version, commit_msg=args.commit )3.2 GitLab CI集成方案
在.gitlab-ci.yml中添加after_script阶段：
after_script: - | if [ "$CI_JOB_STATUS" == "success" ]; then STATUS="success" else STATUS="failed" fi python3 feishu_notify.py \ --status $STATUS \ --duration $CI_JOB_DURATION \ --version $CI_COMMIT_SHORT_SHA \ --commit "$CI_COMMIT_MESSAGE"4. 高级功能扩展
4.1 消息@特定负责人
当构建失败时，可以自动@代码提交者或模块负责人：
def add_mention(message, user_id): message["card"]["elements"].insert(1, { "tag": "div", "text": { "content": f"<at id={user_id}></at> 请及时处理", "tag": "lark_md" } }) return message获取user_id的两种方式：
- 通过飞书API根据邮箱查询
- 建立git账号与飞书账号的映射表
4.2 构建历史趋势统计
在消息卡片中添加最近5次构建的成功率统计：
def add_build_stats(message, stats): success_rate = (stats["success"] / stats["total"]) * 100 message["card"]["elements"].append({ "tag": "div", "text": { "content": f"**近期构建**：成功 {stats['success']}/{stats['total']} 次 ({success_rate:.1f}%)", "tag": "lark_md" } })4.3 安全加固方案
对于敏感信息处理建议：
- 将App Secret存储在Jenkins Credentials或Vault中
- 对消息接口启用IP白名单限制
- 设置消息发送频率限制（如每分钟不超过5条）
# 从环境变量读取敏感配置 import os app_id = os.getenv("FEISHU_APP_ID") app_secret = os.getenv("FEISHU_APP_SECRET")
