影刀RPA实操指南:钉钉与飞书双平台消息推送——一套代码适配两个平台的通知方案
之前写过企业微信的群机器人消息推送。但在实际工作中,很多公司用的是钉钉或飞书。
这篇文章覆盖钉钉和飞书两个平台的消息推送方案,并且告诉你如何写一套"适配层",让同一个流程能无缝切换平台。
三个平台消息推送的对比
| 特性 | 企业微信 | 钉钉 | 飞书 |
|---|---|---|---|
| 接入方式 | Webhook URL | Webhook URL | Webhook URL |
| 安全验证 | 无(URL即权限) | 加签(可选) | 签名校验(可选) |
| Markdown支持 | 部分支持 | 支持 | 完全支持 |
| @人功能 | 通过手机号/UserID | 通过手机号 | 通过open_id |
| 富文本卡片 | 不支持 | 支持 | 支持(消息卡片) |
| 文件发送 | 不支持直接发 | 不支持直接发 | 不支持直接发 |
三个平台的Webhook接入方式基本一致:先在IM里创建一个群机器人,得到一个Webhook URL,然后向这个URL发送HTTP POST请求,消息就出现在群里了。
钉钉群机器人消息推送
获取Webhook地址
打开钉钉PC端 → 进入目标群聊
群设置 → 智能群助手 → 添加机器人
选择"自定义机器人"
安全设置选择"自定义关键词"(填"RPA通知")
复制Webhook地址
发送文本消息
importrequestsimportjsondefsend_dingtalk_text(webhook_url,content,at_mobiles=None):""" 发送钉钉文本消息 webhook_url: 钉钉机器人的Webhook地址 content: 消息内容 at_mobiles: 要@的人的手机号列表,如 ['13800138000'] """payload={"msgtype":"text","text":{"content":content}}ifat_mobiles:payload["at"]={"atMobiles":at_mobiles,"isAtAll":False}response=requests.post(webhook_url,json=payload)result=response.json()ifresult.get("errcode")==0:print("✅ 钉钉消息发送成功")else:print(f"❌ 钉钉消息发送失败:{result.get('errmsg')}")returnresult# 使用send_dingtalk_text("https://oapi.dingtalk.com/robot/send?access_token=xxx","【RPA通知】今日竞品价格采集已完成,共采集850条商品数据。",at_mobiles=["13800138000"])发送Markdown消息
拼多多店群自动化上架方案
钉钉的Markdown消息更丰富:
defsend_dingtalk_markdown(webhook_url,title,text):payload={"msgtype":"markdown","markdown":{"title":title,"text":text}}response=requests.post(webhook_url,json=payload)returnresponse.json()# 发送日报格式markdown_content=""" ## 📊 竞品价格日报  **日期**:2026-06-11 **数据来源**:拼多多、淘宝 | 平台 | 采集商品数 | 均价 | 较昨日变化 | |------|-----------|------|-----------| | 拼多多 | 450 | ¥32.5 | -2.1% | | 淘宝 | 400 | ¥38.2 | +0.8% | > 异常提醒:拼多多"XX商品"降价超过15%,请关注 """send_dingtalk_markdown(webhook_url,"竞品价格日报",markdown_content)钉钉的安全设置
钉钉机器人有三种安全策略,创建时按需选择:
- 自定义关键词:消息内容必须包含指定关键词(如"RPA通知")才会发送
- 加签:需要计算签名,更安全但代码复杂
- IP白名单:限制只有指定IP的请求才能触发
如果选了"加签"模式,需要额外的签名计算:
importtimeimporthmacimporthashlibimportbase64importurllib.parsedefsign_dingtalk(secret):"""钉钉加签模式下的签名计算"""timestamp=str(round(time.time()*1000))secret_enc=secret.encode('utf-8')string_to_sign=f'{timestamp}\n{secret}'string_to_sign_enc=string_to_sign.encode('utf-8')hmac_code=hmac.new(secret_enc,string_to_sign_enc,digestmod=hashlib.sha256).digest()sign=urllib.parse.quote_plus(base64.b64encode(hmac_code))returntimestamp,sign# 在Webhook URL后拼接参数timestamp,sign=sign_dingtalk("你的加签密钥")signed_url=f"{webhook_url}×tamp={timestamp}&sign={sign}"飞书群机器人消息推送
获取Webhook地址
- 打开飞书PC端 → 进入目标群聊
- 群设置 → 群机器人 → 添加机器人
- 选择"自定义机器人"
- 飞书的Webhook地址自带安全校验Token,复制即可
发送文本消息
defsend_feishu_text(webhook_url,content):"""发送飞书文本消息"""payload={"msg_type":"text","content":{"text":content}}response=requests.post(webhook_url,json=payload)result=response.json()ifresult.get("code")==0:print("✅ 飞书消息发送成功")else:print(f"❌ 飞书消息发送失败:{result.get('msg')}")returnresult# 使用send_feishu_text("https://open.feishu.cn/open-apis/bot/v2/hook/xxx","【RPA通知】流程运行完毕")发送富文本消息(飞书的超能力)
飞书支持"消息卡片",比纯文本好看很多:
defsend_feishu_card(webhook_url,title,content_items):""" 发送飞书卡片消息 content_items: [{"tag": "text", "text": "内容"}, ...] """payload={"msg_type":"interactive","card":{"header":{"title":{"tag":"plain_text","content":title},"template":"blue"# 蓝色标题栏},"elements":[{"tag":"div","text":{"tag":"lark_md","content":"\n".join(content_items)}},{"tag":"hr"# 分割线},{"tag":"note","elements":[{"tag":"plain_text","content":f"发送时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"}]}]}}response=requests.post(webhook_url,json=payload)returnresponse.json()# 发一张漂亮的日报卡片send_feishu_card(feishu_webhook,"📊 今日数据汇总",["**采集平台**:拼多多、淘宝、1688","**采集商品数**:1,250","**均价**:¥35.80","","⚠️ **异常提醒**:3个商品降价超过15%",])封装一个统一的消息推送层
如果你的团队不同组用不同的IM,写一个统一接口:
classMessagePusher:"""统一的消息推送器,适配企业微信/钉钉/飞书"""def__init__(self,platform,webhook_url):self.platform=platform self.webhook_url=webhook_urldefsend_text(self,content):ifself.platform=='wecom':returnself._send_wecom_text(content)elifself.platform=='dingtalk':returnself._send_dingtalk_text(content)elifself.platform=='feishu':returnself._send_feishu_text(content)else:print(f"❌ 不支持的平台:{self.platform}")defsend_report(self,title,items):"""发送格式化报告"""ifself.platform=='dingtalk':md=f"##{title}\n\n"+"\n".join(f"-{item}"foriteminitems)returnself._send_dingtalk_markdown(title,md)elifself.platform=='feishu':returnself._send_feishu_card(title,items)else:text=f"{title}\n"+"\n".join(f"•{item}"foriteminitems)returnself.send_text(text)# ... 各平台的私有方法# 使用时只需要改一行配置pusher=MessagePusher('feishu','https://open.feishu.cn/open-apis/bot/v2/hook/xxx')pusher.send_report("竞品价格日报",["拼多多:450条,均价¥32.5","淘宝:400条,均价¥38.2","异常商品:3个",])消息推送的最佳实践
1. 不要刷屏
每小时最多推一条。如果采集频率高,把结果汇总后定时推送,而不是采集一条发一条。
2. 区分严重级别
🔴 错误:流程运行失败 → 立即推送 + @人 🟡 警告:某步骤异常但流程继续 → 汇总到日报 🟢 信息:流程正常运行 → 可选推送 [video(video-KV65Kuc4-1781341567728)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营?)]3. 消息要有"可操作性"
坏消息:"采集失败了"
好消息:"拼多多价格采集失败(超时)。点击这里查看日志:[链接]。请检查代理IP是否正常。"
好的消息告诉接收人:发生了什么、为什么、现在要做什么。
4. 限制Webhook请求频率
三个平台都有频率限制(通常每分钟20条左右)。如果你的流程可能高频发送,加上缓冲:
importtimefromcollectionsimportdequeclassRateLimiter:def__init__(self,max_per_minute=15):self.requests=deque()self.max_per_minute=max_per_minutedefwait_if_needed(self):now=time.time()# 清理1分钟前的记录whileself.requestsandself.requests[0]<now-60:self.requests.popleft()iflen(self.requests)>=self.max_per_minute:wait_time=60-(now-self.requests[0])print(f"⏳ 频率限制,等待{wait_time:.1f}秒...")time.sleep(wait_time)self.requests.append(time.time())写在最后
IM消息推送是RPA自动化的"最后一公里"——流程跑完了,如果没人知道结果,等于没跑。
最省事的方案是:选公司主用的IM平台,打通Webhook,然后在所有流程的结束时统一调一次"发送通知"。后续如果换平台,只需要改MessagePusher的配置,不需要动任何流程代码。
内容标签:影刀RPA · 钉钉 · 飞书 · 消息推送 · Webhook · 群机器人 · 通知自动化
作者:林焱
系列说明:本文是「影刀RPA新手到高手」系列教程之一,面向需要多IM平台消息推送的开发者。