ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python+飞书机器人:自动化消息推送实战指南

Python+飞书机器人:自动化消息推送实战指南 “发个通知而已能有多麻烦”说实话我以前也这么想。直到有一天我负责的一个数据同步任务在凌晨2点挂了群里安安静静第二天早上才被人发现。那一刻我才意识到手动发通知这事看着只是复制粘贴实际上是一个巨大的隐形时间黑洞还是一个不可靠的风险点。后来我花了一下午把团队里的通知全部切到飞书机器人用Python一套脚本搞定所有消息推送。之后不管是服务报警、每日报表还是任务完成提醒全都自动跑再也不用惦记“今天谁发通知”这种事了。这套方案有多简单简单到不夸张地说5分钟就够了。你不需要会写复杂的代码不需要搭建服务器更不需要看一堆冗长的官方文档。只要一个Webhook地址一个Python脚本就能把任意消息推到飞书群里。这篇文章我尽量把每一步拆开讲清楚从创建机器人到代码封装再到常见坑位排查全程贴着实际应用来你可以直接照着抄。1. 为什么“发个通知”这件事值得专门写一套工具很多人觉得发通知嘛群里复制粘贴一下不就完了但如果你真的负责过这类事会发现这里面有几个很扎心的痛点尤其是在团队协作场景里。第一个痛点是重复且无聊。“今日订单量XXX异常订单XX笔”“本周上线版本稳定无重大事故”……这种日报、周报、播报内容其实就几个数字在变但你必须每天手动编辑、手动发送。一个月30天你至少搭进去30次机械劳动。一次两次还行时间久了纯粹是消耗。第二个痛点是容易漏。人不是机器总有忙忘的时候。我见过因为忘了发上线通知导致测试团队在旧环境白白测了一上午的情况。也见过某个报警通知没盯着服务挂了40分钟才被业务方发现。这些“漏”带来的连带成本往往远超发一条消息本身。第三个痛点是无法沉淀。手动发消息发完就完了消息格式、发送记录、接收对象都不可控。但如果用代码去发通知的模板可以固定下来触发条件可以写进程序里历史记录可以通过日志回溯。这种“可复现、可追溯、可变更”的能力才是自动化消息推送真正的价值所在。再往深一层说飞书机器人推送解决的其实是程序世界和信息接收者之间的桥梁问题。你的Python脚本跑在服务器上可能是凌晨跑批可能是异常时的救火队。它怎么在正确的时间把正确的话说给正确的人听答案就是让代码自己去调飞书机器人的接口把一条结构化的消息投递到群里。人只需要在群里看着结果不需要参与过程。所以这套内容的适用对象很明确日常有通知、报表、报警需求但还在手动操作的运营和研发刚学Python、想找一个能立刻见效的小项目的初学者以及团队协作里负责“上下传达”的各种角色。它不要求你有高深的编程能力只要会跑Python脚本就能上手。2. 飞书自定义机器人先花3分钟把接收端准备好在用Python写代码之前第一步是在飞书里把机器人建好。这一步网上教程不少但很多都只讲了操作路径没说清楚背后的机制容易让人后面调接口时一头雾水。2.1 创建机器人的完整操作路径找一个你有权限的飞书群按照下面路径操作进入群聊点击右上角的“设置”图标在设置面板里找到“群机器人”点击进入点击“添加机器人”在弹出的列表里选择“自定义机器人”给机器人起个名字比如“运维报警助手”“数据播报机器人”建议名字里带上用途关键字按需要勾选“签名校验”这一步建议勾上后面我会专门讲原因点击“完成”后飞书会给你生成一个Webhook地址形如https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx创建完成后建议先把Webhook地址复制到自己的笔记工具里存好。飞书页面上只展示这一次后面想找就比较麻烦。2.2 Webhook地址到底意味着什么很多教程就直接告诉你“复制这个链接”但没说这到底是个什么东西。我来打个比方。Webhook地址本质上就是你给机器人设置的一个“收件箱”。别人往这个地址发一封特定格式的信机器人就会读取信的内容然后说给群里的人听。这里的“信”就是HTTP POST请求特定格式就是JSON。所以你其实根本不需要去研究飞书开放平台那套复杂的SDK也不需要注册应用、不需要app_id和app_secret默认情况下。你只需要一个能发HTTP请求的工具往这个URL上POST一串JSON数据消息就出去了。而Python里的requests库就是最顺手的“邮递员”。这也是为什么说5分钟能搞定——因为最核心的接口调用本质上就是一次简单的POST请求没有任何魔法。2.3 安全开关签名校验的底层逻辑创建机器人时有个“签名校验”选项很多人会选择跳过觉得多一事不如少一事。我的建议是除非你只是在自己电脑上临时测试否则一定要开。不开签名可能发生什么如果你的Webhook地址不小心泄露了群截图外传、代码提交到公开仓库、离职同事手机里还有聊天记录任何拿到这个链接的人都能往你群里发消息。轻则发广告骚扰重则冒充官方账号发诈骗内容后果挺严重的。开了签名校验后相当于在这个“收件箱”上加了一把密码锁。请求方必须同时提供正确的密钥签名飞书服务器才会受理。校验原理是这样的调用方生成当前时间的Unix时间戳秒级把时间戳和密钥拼接成一个字符串{时间戳}\n{密钥}用HMAC-SHA256算法对这个字符串做加密把加密结果做Base64编码得到签名请求体里带上时间戳和签名飞书服务器用同样的逻辑重新计算一遍比对是否一致同时校验时间戳与服务器时间偏差是否在1小时内这套机制的好处是就算Webhook地址泄露了对方没有密钥很难伪造合法的签名。我后面第5章会给完整的Python实现。3. Python推送第一课最小可用的文本消息demo接收端准备好了现在开始写代码。先跑通一个最小demo让你直观看到“代码往Webhook丢JSON群里出消息”的完整链路。3.1 环境准备不折腾的安装方式代码基于Python 3.6目前主流的3.8、3.9、3.10、3.11我都实测过没问题。你只需要额外装一个requests库它是Python世界里最常用的HTTP请求库没有之一。pip install requests如果你用的是虚拟环境建议在项目里用venv先激活虚拟环境再装。如果你只在服务器上跑脚本直接全局装也行Python的包管理没那么娇气。3.2 发送第一条消息的完整代码代码如下非常简短import requests WEBHOOK_URL https://open.feishu.cn/open-apis/bot/v2/hook/你的Webhook地址 def send_text(text): payload { msg_type: text, content: { text: text } } resp requests.post(WEBHOOK_URL, jsonpayload, timeout10) result resp.json() print(fHTTP状态码: {resp.status_code}) print(f业务返回码: {result.get(code)}) print(f返回信息: {result.get(msg)}) if __name__ __main__: send_text(大家好这是一条来自Python机器人的消息)这里有两个容易踩的细节值得说一下。第一个细节是requests.post的json参数。之所以用jsonpayload而不是datapayload是因为json会自动帮你做两件事把Python字典序列化成JSON字符串并且把Content-Type请求头设置为application/json。很多初学者用datapayload发中文消息结果群里全是乱码或者直接报参数错误多半就是栽在这里。用json参数最省心。第二个细节是timeout要设置。Webhook请求看着简单但如果飞书接口偶发超时没有timeout的请求可能会让你的程序一直卡在那里看起来像死循环。设置10秒超时是很有必要的额外保险生产环境里我会再配合重试机制。3.3 运行结果怎么看运行脚本后如果一切正常你会看到类似输出HTTP状态码: 200 业务返回码: 0 返回信息: success飞书的返回结构是{code: 0, msg: success}code为0代表成功。在这个层面HTTP状态码反而不是最关键的判断依据即使HTTP返回200业务code也可能非0所以调试时两个都要看。如果群里没收到消息也别慌先看返回的code对应什么错误。最常见的几个19001参数错误多半是JSON结构不对或字段拼写错误19002签名校验失败你开了签名但没带对sign19003时间戳不合法通常是时间偏差超过1小时9499内部错误多为触发频率限制或机器人被移出群这一步跑通了你的“自动化消息推送”地基就完成了。接下来要思考的是怎么把消息发得更好看、更有用。4. 从纯文本到富文本再到卡片消息怎么发才不招人烦消息能发出去只是第一步。真正用过一段时间你会发现一条干巴巴的纯文本和一条结构清晰、有重点的富文本/卡片消息接收者的阅读体验和响应速度完全是两回事。4.1 text和post什么时候用哪个飞书自定义机器人支持多种消息类型其中日常最常用的是text纯文本和post富文本和interactive消息卡片。我用下来的一般准则消息类型适用场景优点注意点text临时提醒、简单通知结构简单容易拼装没有排版信息一长容易淹没post日报、周报、结构化信息支持标题、加粗、超链接JSON嵌套稍复杂interactive告警、重大通知、需要强提醒视觉冲击强支持颜色模板和交互卡片结构较复杂需要花时间学我的建议是日常碎事用text日报和周报用post报警用interactive。三个都掌握基本能覆盖90%的推送场景。4.2 富文本post消息的组装逻辑post消息长这样JSON结构稍微多了一层嵌套但理清楚就好def send_post(webhook_url, title, lines): payload { msg_type: post, content: { post: { zh_cn: { title: title, content: lines } } } } resp requests.post(webhook_url, jsonpayload, timeout10) print(resp.json())其中content是一个二维数组每一行是一组元素一行里可以放多个不同格式的字段。举个例子我要发一条带超链接的日报lines [ [ {tag: text, text: 今日订单量}, {tag: text, text: 12345, style: bold}, {tag: text, text: | 成交额}, {tag: text, text: 98.6万, style: bold} ], [ {tag: text, text: 异常订单}, {tag: text, text: 5笔, style: red}, {tag: a, text: 点击查看详情, href: https://example.com/orders/abnormal} ] ]style字段支持bold加粗、red红字等链接用tag: a配合href。这样发出来的消息段落分明群里扫一眼就能抓重点不用从大段文字里自己找信息。4.3 消息卡片把告警做得让人“不得不看”如果说post是“有排版的消息”那interactive就是“长着一张脸的消息”。它的视觉元素更丰富有头部标题、有颜色模板、有按钮交互。一条最简告警卡片长这样payload { msg_type: interactive, card: { header: { title: { tag: plain_text, content: 【线上告警】订单服务异常 }, template: red }, elements: [ { tag: div, text: { tag: lark_md, content: **服务名**order-service\n**异常时间**2024-06-01 02:15:33\n**错误信息**连接数据库超时 } }, { tag: action, actions: [ { tag: button, text: {tag: plain_text, content: 去处理}, url: https://example.com/alert/123, type: danger } ] } ] } }template字段可以设置卡片的整体颜色red表示红色告警还有orange、green、blue、grey等。elements里可以放文本div、图片img、按钮action等。按钮能带跳转链接这一点用来做“收到告警点按钮直接跳转处理页面”体验非常好。我实际用下来的心得是卡片消息别啥都往上面堆。红色的卡片发多了大家就麻木了真正出事时反而没人看。平时能用post解决的问题不需要动用interactive。4.4 想发图片和文件怎么办自定义机器人的接口实际上只支持text、post、interactive这几种消息载体。如果你想把本地图片、Excel文件直接推进群里官方是不支持的。这里有两个替代方案方案一图片走URL。在消息卡片的img元素里放一个公网可访问的图片URL飞书会按链接渲染图片。适合发图表、截图。缺点是你得有地方存放这张图片并生成URL。方案二换用应用机器人。如果你确实要发本地文件就得升级到飞书开放平台的“企业自建应用机器人”走im/v1/files接口先上传文件然后用im/v1/messages接口发送。这也意味着你需要申请app_id和app_secret流程比自定义机器人重一些但能力边界扩展了很多。所以我的建议很直接初期先用自定义机器人把消息推送跑起来等确实有发文件、建表格这类刚需时再升级成应用机器人。不要一开始就想着上大杀器够用就好。5. 带签名校验的项目级封装团队消息推送的干净写法如果你的机器人只是自己调试用不签名也没太大所谓。但一旦要给团队用或者要长期挂在服务器上我会强烈建议把签名校验加上并且把发消息的能力封装成一个简洁的模块。这样团队成员想推送时就一行代码调用不需要每个人都知道飞书接口细节。5.1 签名校验的完整实现首先是生成签名。import base64 import hashlib import hmac import time def gen_sign(timestamp: str, secret: str) - str: 飞书签名生成规则 把 timestamp \n secret 拼成字符串 用 HMAC-SHA256 加密后做 Base64 编码 string_to_sign f{timestamp}\n{secret} hmac_code hmac.new( string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() sign base64.b64encode(hmac_code).decode(utf-8) return sign然后是发送消息的完整函数把之前学的三种消息类型都封装进去import requests import time class FeishuBot: def __init__(self, webhook_url: str, secret: str None): self.webhook_url webhook_url self.secret secret def _build_payload(self, msg_type: str, content: dict) - dict: payload {msg_type: msg_type, content: content} if self.secret: timestamp str(int(time.time())) payload[timestamp] timestamp payload[sign] gen_sign(timestamp, self.secret) return payload def _send(self, payload: dict) - dict: try: resp requests.post( self.webhook_url, jsonpayload, timeout10 ) return resp.json() except requests.exceptions.RequestException as e: return {code: -1, msg: f请求异常: {e}} def send_text(self, text: str) - dict: payload self._build_payload(text, {text: text}) return self._send(payload) def send_post(self, title: str, lines: list) - dict: content {post: {zh_cn: {title: title, content: lines}}} payload self._build_payload(post, content) return self._send(payload) def send_card(self, header_title: str, template: str, elements: list) - dict: content { card: { header: { title: {tag: plain_text, content: header_title}, template: template }, elements: elements } } payload self._build_payload(interactive, content) return self._send(payload)用起来就是这样的感觉bot FeishuBot( webhook_urlhttps://open.feishu.cn/open-apis/bot/v2/hook/xxx, secret你的密钥 ) bot.send_text(构建完成可以开始测试了) abnormal_lines [ [{tag: text, text: 今日异常订单}, {tag: text, text: 5笔, style: red}] ] bot.send_post(每日订单异常报告, abnormal_lines) bot.send_card( header_title【系统告警】数据库连接异常, templatered, elements[ {tag: div, text: {tag: lark_md, content: **实例**db-01\n**状态**不可达}} ] )这个封装类的好处在于业务代码和飞书接口细节解耦了。团队成员关心的是“我要发一段文字、一张卡片”而不是“我今天又要构造什么样的JSON”。5.2 密钥和配置的存放建议团队协作时Webhook地址和密钥千万别直接硬编码在脚本里尤其别提交到公开的代码仓库。我之前在GitHub上搜到一个公司的Webhook泄露那种群被轰炸的场面看一次就忘不了。推荐的存放方式有两种环境变量在服务器上通过export设置Python里用os.getenv(FEISHU_WEBHOOK_URL)读取配置文件建议放在.env文件里再加上.gitignore把该文件排除掉避免传到仓库import os from dotenv import load_dotenv load_dotenv() bot FeishuBot( webhook_urlos.getenv(FEISHU_WEBHOOK_URL), secretos.getenv(FEISHU_SECRET) )5.3 多机器人管理一个团队可能同时有“业务播报”“报警通知”“发布通知”好几个机器人。建议配置里按用途拆分比如BOTS { alert: FeishuBot(webhook_urlos.getenv(FEISHU_ALERT_URL), secretos.getenv(FEISHU_ALERT_SECRET)), report: FeishuBot(webhook_urlos.getenv(FEISHU_REPORT_URL), secretos.getenv(FEISHU_REPORT_SECRET)), deploy: FeishuBot(webhook_urlos.getenv(FEISHU_DEPLOY_URL), secretos.getenv(FEISHU_DEPLOY_SECRET)) }这样做的收益是群与群职责分离接收者不会被无关消息打扰。而且某个群的机器人密钥泄露了只需要重新生成那一个其他不用动。6. 实测最容易翻车的几个点与完整排查过程代码写好了但真实跑起来总会遇到幺蛾子。下面这几个坑是我自己和身边的同事都踩过的每一个都附带完整的排查思路。6.1 案例一开了签名校验后一直报19002这是一个非常典型的场景。你按官方文档写了签名逻辑但发送时飞书一直返回code: 19002msg一般会说是签名错误。这时候怎么查第一步先复现。检查你的请求体里是否真的带上了timestamp和sign字段。有些人把签名逻辑封装好了但漏了把这两个字段加进最终发送的payload里。打印出来看一眼没有就直接定位了。第二步验证签名算法。飞书提供一个在线的签名校验工具你把密钥、时间戳贴进去它可以生成一个标准sign。和你Python里生成的sign比对一下。如果不一样大概率是算法细节有问题。比如有些人会把密钥和时间戳的顺序搞反或者用成了SHA1而不是SHA256。第三步检查时间戳是不是实时生成的。有些人在调试时图方便把timestamp写死成常量或者缓存了之前的timestamp。飞书要求时间戳与服务器时间偏差不超过1小时一旦偏差过大返回的就是19003而不是19002但如果你在代码里把异常吞掉了看到的就会是表面上的“发送失败”这时候逐层打印日志的重要性就体现出来了。第四步确认密钥没被“污染”。从飞书后台复制密钥时有时候会多复制一个换行符或空格。在代码里打印repr(secret)一眼就能看出有没有藏着不可见字符。我自己就遇到过密钥末尾多了一个\n排查了快20分钟才抓到。提示排查这类问题最有效的习惯是“把每个环节的输入输出都打印出来”。发送前打印payload发送后打印响应。不要凭感觉猜直接看数据。6.2 案例二消息发出去中文乱码或显示异常这个问题在requests里其实很少见只要用了json参数。但如果你看到乱码通常是下面两种情况之一一是你在某些系统上手动把整个JSON拼成了字符串然后用data发送且字符串编码不对。这种情况下飞书拿到的body可能是GBK编码的字节流解析出来自然乱码。解决办法就是别手动拼JSON规规矩矩用json参数。二是消息内容里带了非法字符比如特殊控制字符、未转义的引号。解决方法是把内容写成干净的纯文本或者用json.dumps确保转义正确。如果内容来自外部文件或数据库建议先做一下清洗。6.3 案例三Webhook地址不小心泄露了这个坑属于安全类事故但真的会发生。有一次我一个同事为了测试直接把Webhook地址贴在了公开的GitHub issue里结果没过多久群里就进来了大量垃圾消息。那次的处理步骤是这样的立刻在飞书后台把该机器人的Webhook作废重新生成一个重新开启签名校验排查GitHub历史记录确认没有其他敏感信息泄露这里也说一句Webhook本质上就是一个“能往群里说话”的凭据它的安全等级应该和账号密码一样对待。凡是可能被外部看到的场景务必加签名校验。6.4 错误码速查表日常开发中我把遇到的错误码整理成了一个表贴在项目文档里供团队快速对照错误码含义最可能的触发原因处理建议0成功无不用处理19001参数错误JSON格式不对、字段缺失检查payload结构对照官方文档19002签名错误签名算法不对、密钥不对用官方工具比对签名19003时间戳错误服务器时间偏差过大校时确保timestamp是当前时间19004请求频率超限短时间发送过多加延时分批发9499内部错误机器人被移出群、群不存在检查群状态这张表不用背但建议收藏遇到问题先对号入座别一上来就改代码。7. 从“能发消息”到“会干活”三个可以立刻落地的自动化场景代码会写了坑也排了接下来最重要的问题是这个能力怎么用到实际工作里。下面三个场景你直接套用就能落地。7.1 场景一定时任务推送每日播报最常见的需求是每天早上固定时间推送一条摘要比如“今日待办”“昨日报表”“服务器状态”。实现方式不复杂就是“定时触发 Python脚本 飞书机器人”。在Linux服务器上用crontab做定时0 9 * * * /usr/bin/python3 /opt/scripts/daily_report.py /var/log/daily_report.log 21这条命令的意思是每天9点整跑一次daily_report.py脚本并把标准输出和错误都追加到日志文件里。脚本内部完成数据读取、格式化、推送。写定时任务时有几个细节很关键用绝对路径。crontab环境下的PATH和你在命令行里不一样直接在脚本里写相对路径或依赖~很可能跑不通日志必须留。定时任务出问题时最难查日志是唯一线索脚本开头加#!/usr/bin/env python3然后赋予执行权限这样可以直接执行一个简单的日报脚本长这样#!/usr/bin/env python3 # -*- coding: utf-8 -*- import datetime import requests webhook_url https://open.feishu.cn/open-apis/bot/v2/hook/xxx def get_today_data(): # 这里换成你自己的数据查询逻辑比如读数据库、读API order_count 12345 abnormal_count 5 return order_count, abnormal_count def main(): today datetime.date.today().isoformat() order_count, abnormal_count get_today_data() lines [ [{tag: text, text: f日期: {today}}], [{tag: text, text: 订单量: }, {tag: text, text: str(order_count), style: bold}], [{tag: text, text: 异常订单: }, {tag: text, text: str(abnormal_count), style: red if abnormal_count else green}] ] payload { msg_type: post, content: {post: {zh_cn: {title: 每日业务播报, content: lines}}} } resp requests.post(webhook_url, jsonpayload, timeout10) print(resp.json()) if __name__ __main__: main()Windows上的话可以用“任务计划程序”替代crontab配置思路完全一样。7.2 场景二程序异常自动报警这个是所有场景里性价比最高的一个。你的Python程序、爬虫、数据处理任务出了异常第一时间通知到人而不是等第二天看日志。核心思路就是在异常处理块里调机器人import traceback def run_job(): try: # 你的业务逻辑 do_something() except Exception as e: error_detail traceback.format_exc() bot.send_card( header_title【任务失败】数据同步任务, templatered, elements[ {tag: div, text: {tag: lark_md, content: f**错误信息**{e}\n**堆栈**\n\n{error_detail[-500:]}\n}} ] ) raise注意这里有两个小技巧用traceback.format_exc()拿到完整堆栈而不是只发一个str(e)。不然收到“出现异常”却看不到哪里异常等于没报堆栈信息别全发截取最后500个字符就好。飞书消息内容有大小限制而真实堆栈可能非常长全发出去容易踩到上限我自己的经验是报警消息一定要带上任务名、时间、主机名、错误摘要这四个要素。不然凌晨3点收到一条“程序报错了”的消息你还得爬起来查是哪个程序在哪个机器上跑。7.3 场景三数据报表自动汇总推送如果你经常要做“把Excel统计一下、把结果发群里”这类工作Python的pandas库加上飞书机器人能把整个流程压缩成一条命令。思路是用pandas读取数据源Excel、CSV、数据库做统计汇总然后把汇总结果拼成post或卡片消息推送出去。import pandas as pd df pd.read_excel(/data/orders.xlsx) summary df.groupby(status).size().reset_index(namecount) lines [] for _, row in summary.iterrows(): lines.append([ {tag: text, text: f{row[status]}: }, {tag: text, text: str(row[count]), style: bold} ]) bot.send_post(订单状态汇总, lines)这个场景的进阶方向是把数据查询逻辑直接接进数据库每天定时跑或者每次同步完成后自动跑。投入产出比极高因为“数据统计发通知”这类工作手动做既慢又容易算错用代码做就是几秒钟的事。我个人建议刚开始时别做太复杂先从“文本通知”和“异常报警”这两个切入。等跑顺了再慢慢加富文本、卡片、定时任务。这套能力最大的价值不是让你学会某个具体接口而是让你建立一种思维凡是“程序知道、人需要知道”的信息都应该让代码自动去说而不是等人去转述。我自己的群里现在每天固定收到5条自动消息没有一条需要我手动去编这种“机器管机器、人只看结果”的状态才是自动化的意义。
返回列表