1. 项目缘起为什么需要自动化发送图文消息在企业的日常运营中信息同步是刚需。无论是项目进度日报、系统监控告警、市场活动数据快报还是团队内部的优秀案例分享都需要一个高效、直观的渠道来触达相关人员。过去我们可能依赖人工在群里复制粘贴文字和图片或者手动上传文件不仅效率低下而且容易出错、遗漏尤其是在需要定时或触发生成的场景下。飞书作为一款集成了IM、日历、文档、云盘的协同办公平台其群组功能是企业内部沟通的核心。而飞书开放平台提供的“机器人”能力则为我们打通了自动化流程与即时沟通的“最后一公里”。想象一下你的数据平台在凌晨3点完成了当日报表的生成一个机器人悄无声息地将一份图文并茂的分析总结推送到项目群你的CI/CD流水线在每次构建成功后自动将更新日志和构建状态截图发到研发群你的客服系统将当日的用户反馈高频词云图在下班前推送给运营团队……这一切都可以通过一个简单的Python脚本实现。这个项目的核心价值就是将结构化的数据或程序运行结果转化为富媒体消息并精准、自动地投递到指定的飞书群聊中实现信息流转的无人化与智能化。它不仅仅是“发送一条消息”而是构建了一个可编程的、可靠的企业信息中枢接口。2. 核心概念解析飞书机器人、消息类型与权限在动手写代码之前我们必须厘清几个关键概念这决定了我们能否正确、安全地使用这项功能。2.1 什么是飞书机器人飞书机器人本质上是一个特殊的飞书账号它由程序控制可以主动向群聊或用户发送消息也可以被动接收并处理它的消息。与我们熟悉的微信公众号后台机器人或钉钉机器人类似它是我们程序与飞书IM系统交互的“代理”。每个机器人都拥有一个唯一的webhook地址我们的程序通过向这个地址发送HTTP POST请求就能驱动机器人在对应的群或会话中发言。这里有一个至关重要的安全理念机器人的权限是与其所在的群或会话绑定的。一个机器人被创建后它本身并不能随意给任何群发消息。你必须将它添加到目标群聊中它才获得了向该群发言的“门票”。同样如果你希望它给单人发消息你需要单独与这个机器人发起聊天。因此自动化流程的第一步往往是“将机器人拉入群聊”。2.2 富文本与图文消息不止于纯文本飞书机器人支持多种消息类型远不止简单的文字。对于我们的“图文信息”场景主要涉及以下两种并且常常组合使用富文本post消息这是功能最强大的消息类型之一。它允许你构建一个结构化的“文章”包含标题、正文、字体加粗/变色、引用、分割线以及关键——内嵌图片。这里的“内嵌图片”并非上传一个图片文件而是需要你先将图片上传到飞书服务器获取一个唯一的image_key然后在post消息体中引用这个key。post消息在群聊中会以一张精美的“卡片”形式呈现信息密度高阅读体验好。图片image消息这种类型用于发送纯粹的图片。与post内嵌图片类似也需要先上传获取image_key。发送后在聊天窗口中会直接显示该图片。交互卡片interactive消息功能更复杂可以包含按钮、选择器、表单等交互元素适用于需要用户点击反馈的场景。对于简单的图文推送post通常已足够。我们的目标“图文信息”在大多数业务场景下指的就是一个包含文字描述和配图的post消息。例如一篇运营日报的标题是文字核心数据图表是图片。2.3 权限与安全webhook与token机器人有两种主要的调用方式对应不同的权限级别webhook这是最简单、最常用的方式。在群聊中添加机器人后可以在机器人设置里找到一个以https://open.feishu.cn/open-apis/bot/v2/hook/开头的URL。这个URL包含了该机器人在这个特定群聊中的发送权限。任何人拿到这个URL都可以向该群发送消息。因此webhook的安全性完全依赖于这个URL的保密性。切记不要将它提交到公开的代码仓库如GitHub务必通过环境变量或配置服务器来管理。tenant_access_token这是更高级、更安全的方式。你需要创建一个企业自建应用并为其开通机器人能力。通过应用的App ID和App Secret可以获取一个具有时效性的tenant_access_token。用这个token调用消息发送接口可以指定发送到任意已添加该机器人的群或用户。这种方式权限更大适合需要跨多个群发送消息或集成到更复杂后端系统的场景。同时它支持主动获取与被动响应消息。对于入门和大多数单群推送场景使用webhook足矣。本文将主要基于webhook进行讲解因为它步骤更少更容易理解。3. 环境准备与飞书后台配置“工欲善其事必先利其器”。在编写Python脚本前我们需要完成两件事配置飞书机器人并搭建Python开发环境。3.1 在飞书群中创建并配置机器人打开目标飞书群进入你希望接收自动化消息的飞书群。点击群设置在群聊天窗口右上角点击“...”或群名称进入“群设置”。添加机器人找到“群机器人”或“设置机器人”选项。点击“添加机器人”。选择“自定义机器人”在机器人列表里选择“自定义机器人”。配置机器人信息机器人名字给你的机器人起个名字比如“数据日报机器人”、“系统监控官”。机器人描述可选填写机器人的主要用途。安全设置强烈建议配置自定义关键词只有包含指定关键词的消息才会被发送。例如设置为“日报”那么你的消息内容中必须含有“日报”二字。这是防止webhookURL泄露后被滥用的第一道防线。IP白名单填写你部署脚本的服务器公网IP地址。只有来自这些IP的请求才会被处理。这是最推荐的安全措施。注意不要勾选“消息卡片回调”或“Outgoing Webhook”除非你需要处理用户的回复。点击“添加”完成后你就创建了一个属于该群的机器人。获取webhook地址添加成功后在机器人列表中找到你刚创建的机器人点击“设置”或“复制地址”即可获得那个至关重要的webhook URL。请立即妥善保存例如存入本地的密码管理器或服务器的环境变量中。注意一个机器人可以被添加到多个群每个群都会生成一个独立的webhook URL。向哪个URL发消息机器人就在哪个群出现。3.2 Python环境搭建与库安装本项目对Python环境要求非常宽松Python 3.6及以上版本均可。核心是安装用于发送HTTP请求的库。requests库是Python社区的事实标准简单易用。# 使用pip安装requests库 pip install requests如果你使用虚拟环境如venv或conda请确保在正确的环境中安装。对于生产环境建议将依赖写入requirements.txt文件requests2.25.1然后通过pip install -r requirements.txt安装。4. 实战核心发送图文post消息的完整代码拆解这是整个项目的核心环节。我们将把过程分解为两步首先上传图片获取image_key然后组装并发送post消息。4.1 第一步上传图片并获取image_key飞书不允许直接通过webhook上传图片二进制流。你必须先调用另一个“上传图片”接口将图片传到飞书服务器它会返回一个image_key。这个key在后续发送消息时就代表了这张图片。关键点上传图片接口需要使用tenant_access_token进行认证而不是webhook。这意味着即使你最终用webhook发消息在上传图片这一步也需要一个具有相应权限的token。对于自建应用机器人你可以用App ID和App Secret换token。但对于我们刚刚创建的、仅通过webhook使用的群自定义机器人它没有App Secret因此无法直接调用上传图片接口。这似乎成了一个死循环解决方案是方案A推荐创建一个企业自建应用并开通机器人能力。用这个应用的凭证上传图片用这个应用的webhook或基于token的发送接口发消息。这是最规范、权限最完整的方式。方案B变通如果你只有群自定义机器人的webhook又想发图可以将图片先上传到其他公共图床如公司内网的文件服务器、或支持外链的云存储然后在post消息中使用img标签引用图片的公开URL。但这种方式依赖图床的稳定性且可能不符合企业安全规范。为了教程的完整性我们先讲解规范方案A中的图片上传步骤。假设你已经创建了自建应用并获得了app_id和app_secret。import requests import json def get_tenant_access_token(app_id, app_secret): 获取 tenant_access_token url https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal headers {Content-Type: application/json; charsetutf-8} payload { app_id: app_id, app_secret: app_secret } response requests.post(url, headersheaders, datajson.dumps(payload)) result response.json() if result.get(code) 0: return result[tenant_access_token] else: raise Exception(fFailed to get tenant_access_token: {result}) def upload_image(image_path, tenant_access_token): 上传图片到飞书获取 image_key :param image_path: 本地图片文件路径 :param tenant_access_token: 租户访问令牌 :return: 图片的 image_key url https://open.feishu.cn/open-apis/im/v1/images headers { Authorization: fBearer {tenant_access_token}, # 注意这里的认证方式 } # 请求体格式为 multipart/form-data files { image: open(image_path, rb) } data { image_type: message # 图片类型用于消息 } response requests.post(url, headersheaders, filesfiles, datadata) result response.json() if result.get(code) 0: image_key result[data][image_key] print(f图片上传成功image_key: {image_key}) return image_key else: raise Exception(fFailed to upload image: {result}) # 使用示例 if __name__ __main__: APP_ID your_app_id # 替换为你的 App ID APP_SECRET your_app_secret # 替换为你的 App Secret IMAGE_PATH ./daily_report_chart.png # 替换为你的图片路径 try: token get_tenant_access_token(APP_ID, APP_SECRET) image_key upload_image(IMAGE_PATH, token) # 保存这个 image_key用于后续发送消息 # 例如存入变量或临时文件 except Exception as e: print(f操作失败: {e})代码解读与注意事项get_tenant_access_token函数用于获取有效期为2小时的tenant_access_token。在生产环境中你应该缓存这个token并在过期前刷新而不是每次上传都重新获取。upload_image函数中headers的Authorization字段格式为Bearer {token}。files参数用于上传二进制文件requests库会自动处理multipart/form-data格式。image_type固定为message表示此图片用于消息。获取到的image_key有效期很长可以重复使用。但如果图片需要更新你需要上传新图片获取新的key。4.2 第二步构造并发送post消息拿到image_key后我们就可以构造最终的图文消息了。这里我们使用webhook方式发送因为它最直观。post消息体的结构相对复杂它遵循飞书定义的“消息卡片”格式。一个简单的图文post结构如下def send_post_message(webhook_url, title, content_text, image_key): 通过 webhook 发送 post 图文消息 :param webhook_url: 群机器人的 webhook 地址 :param title: 消息标题 :param content_text: 消息正文文本 :param image_key: 通过上传接口获取的图片 key headers { Content-Type: application/json } # 构建消息体 payload { msg_type: post, # 消息类型为 post content: { post: { zh_cn: { # 中文语言版本 title: title, content: [ # 第一个段落文本 [ { tag: text, text: content_text } ], # 第二个段落图片 [ { tag: img, image_key: image_key, width: 600, # 图片显示宽度单位像素 height: 400 # 图片显示高度单位像素 } ] ] } } } } response requests.post(webhook_url, headersheaders, datajson.dumps(payload)) result response.json() if result.get(code) 0 and result.get(msg) success: print(图文消息发送成功) else: print(f消息发送失败: {result}) # 常见错误code 9499msg “消息内容安全校验不通过”可能是触发了飞书的内容安全规则。 # 使用示例 if __name__ __main__: WEBHOOK_URL https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx # 替换为你的 webhook TITLE 【数据日报】2023-10-27 业务核心指标 CONTENT 各位同事好以下是昨日2023-10-26的核心业务数据概览\n- 活跃用户数125,432环比2.3%\n- 订单成交额¥1,234,567环比5.6%\n- 用户平均停留时长15分32秒保持稳定。\n详情请查阅附件图表。 IMAGE_KEY img_xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx # 替换为上传得到的 image_key send_post_message(WEBHOOK_URL, TITLE, CONTENT, IMAGE_KEY)消息体结构深度解析msg_type: 固定为post。content.post.zh_cn: 指定语言为简体中文。你也可以定义其他语言版本如en_us。content: 这是一个列表的列表。外层列表的每个元素代表一个“行”或“段落”。内层列表的每个元素代表该段落内的一个“元素”。元素类型{tag: text, text: ...}: 文本元素。支持简单的Markdown语法如**加粗**、*斜体*、行内代码。{tag: img, image_key: ..., width: 600, height: 400}: 图片元素。width和height用于控制显示尺寸飞书客户端会按比例缩放。你可以构建更复杂的内容例如在文本后加一个链接按钮content: [ [ {tag: text, text: 查看详细报告}, { tag: a, text: 点击这里, href: https://your-bi-system.com/report/123 } ], [ {tag: img, image_key: image_key, width: 600, height: 400} ] ]5. 生产级实践错误处理、异步与部署将脚本运行在本地测试成功只是第一步。要让它成为可靠的生产力工具还需要考虑更多。5.1 健壮的错误处理与重试机制网络请求可能失败飞书接口可能暂时不可用图片可能上传失败。我们的脚本必须能妥善处理这些异常。import time from tenacity import retry, stop_after_attempt, wait_exponential # 安装 tenacity: pip install tenacity retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def send_message_with_retry(webhook_url, payload): 带重试机制的消息发送函数 try: response requests.post(webhook_url, jsonpayload, timeout10) # 设置超时 response.raise_for_status() # 如果HTTP状态码不是200抛出HTTPError result response.json() if result.get(code) 0: return True else: # 飞书业务逻辑错误如内容安全拦截重试可能无效 print(f飞书接口返回业务错误: {result}) if result.get(code) 9499: # 内容安全校验失败 # 记录日志可能需要人工审核内容 log_error_to_file(Content safety check failed, payload) return False except requests.exceptions.Timeout: print(请求超时正在重试...) raise # 触发 tenacity 重试 except requests.exceptions.ConnectionError: print(网络连接错误正在重试...) raise except requests.exceptions.RequestException as e: print(f请求发生异常: {e}) # 对于其他请求异常可以选择重试或直接失败 raise # 在主函数中调用 def main(): # ... 准备 payload ... success send_message_with_retry(WEBHOOK_URL, payload) if not success: # 重试后仍失败发送告警例如发邮件、发另一个高优先级的机器人消息 send_alert_to_admin(飞书机器人消息发送失败请检查)要点使用tenacity库实现优雅的重试逻辑避免因临时网络抖动导致失败。区分错误类型网络超时、连接错误可以重试但像9499内容安全这类业务错误重试是没用的需要记录日志并人工介入。设置超时避免脚本因网络问题无限期挂起。最终告警如果所有重试都失败应有兜底机制通知管理员。5.2 异步发送提升性能如果你的脚本需要频繁发送消息或者是在一个Web服务器中响应请求并发送飞书通知同步的requests.post可能会阻塞主线程。使用异步库aiohttp可以提升性能。import aiohttp import asyncio async def async_send_message(webhook_url, payload): 异步发送消息 timeout aiohttp.ClientTimeout(total10) async with aiohttp.ClientSession(timeouttimeout) as session: try: async with session.post(webhook_url, jsonpayload) as response: result await response.json() if result.get(code) 0: return True else: print(f发送失败: {result}) return False except asyncio.TimeoutError: print(异步请求超时) return False except Exception as e: print(f异步请求异常: {e}) return False # 在异步上下文中调用 async def main_async(): tasks [] for msg_data in list_of_messages: # 假设有多个消息要发 task asyncio.create_task(async_send_message(WEBHOOK_URL, msg_data)) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理 results...5.3 部署与调度让脚本自动运行脚本写好了总不能每天手动运行。常见的部署方式有Linux服务器 Crontab最经典的方式。将脚本放在服务器上使用crontab -e设置定时任务。# 每天上午9点执行 0 9 * * * /usr/bin/python3 /path/to/your/feishu_bot.py /path/to/log.log 21云函数/Serverless推荐更现代、无需管理服务器的方式。例如阿里云函数计算、腾讯云SCF、AWS Lambda。将代码打包部署并配置定时触发器。云函数通常有免费的额度非常适合这种轻量级定时任务。CI/CD平台如果你的消息生成与代码构建、测试相关可以在GitLab CI、Jenkins、GitHub Actions的流水线最后一步添加发送飞书通知的步骤。容器化部署如果脚本环境复杂可以打包成Docker镜像在K8s或通过docker run配合系统定时任务执行。部署注意事项敏感信息管理绝对不要将webhook_url、app_secret等硬编码在脚本里。必须使用环境变量、云平台的密钥管理服务如KMS或配置文件且配置文件本身不被提交到仓库。import os WEBHOOK_URL os.environ.get(FEISHU_WEBHOOK_URL) if not WEBHOOK_URL: raise ValueError(请设置 FEISHU_WEBHOOK_URL 环境变量)日志记录脚本应输出详细的运行日志便于排查问题。可以输出到文件或集成到云平台的日志服务中。监控告警为脚本的执行状态设置监控。例如云函数失败时发送告警或者脚本自身在最终失败时尝试通过其他渠道如短信、另一个高可用机器人发送告警。6. 避坑指南与高级技巧在实际开发和运维中我踩过不少坑也总结了一些让机器人更好用的技巧。6.1 常见问题排查错误码9499消息内容安全校验不通过原因这是最常见的问题。飞书后台有内容安全过滤机制可能触发了某些关键词如涉及敏感信息、广告、暴力等。排查检查消息标题和正文是否包含可能被误判的词汇。尝试发送极简内容如“测试”看是否成功。如果是图片检查图片内容或文件名。前往飞书开放平台后台查看该机器人的“安全设置”中是否启用了“自定义关键词”。如果你设置了关键词“日报”那么消息中必须包含“日报”二字。解决调整文案避免敏感词。或检查并满足自定义关键词要求。错误码99991663无效的 tenant_access_token原因tenant_access_token已过期有效期2小时或app_id/app_secret错误。解决实现token的自动刷新和缓存机制。不要每次调用都重新获取而是在内存或Redis中缓存token并在接近过期时刷新。机器人“发了消息但群里没看到”原因1机器人被移出群聊或者webhook URL对应的机器人已不存在。原因2消息被群管理员设置的消息折叠规则过滤了某些企业版功能。原因3网络问题导致发送请求成功但飞书服务端处理失败。虽然code为0但实际未投递。解决检查机器人是否仍在群内用最简单的文本消息测试查看飞书开放平台后台的“事件日志”看是否有发送记录和详细状态。图片上传成功但发送消息时提示image_key无效原因image_key与使用的token或机器人不匹配。例如用应用A的token上传的图片尝试用应用B的webhook发送这是不允许的。图片资源是绑定到上传它的应用上的。解决确保上传图片和发送消息使用的是同一个飞书应用或同一个机器人的凭证体系。6.2 让消息更美观高级post格式基础的图文已经不错但我们可以做得更专业。混合丰富内容在同一行内混合文本、链接、人。[ { tag: text, text: 本报告由 }, { tag: at, user_id: ou_xxxxxx // 特定成员 }, { tag: text, text: 生成更多细节请 }, { tag: a, text: 查看文档, href: https://your-wiki.com/doc/123 } ]使用分割线和备注让结构更清晰。content: [ [{tag: text, text: ## 核心指标, style: {bold: true}}], [{tag: text, text: UV: 10k}], [{tag: hr}], // 分割线 [{tag: text, text: ## 详细数据}], [{tag: img, image_key: img_123}], [{tag: note, elements: [{tag: text, text: 数据截止时间2023-10-27 00:00}]}] // 备注小字 ]6.3 一个完整的、可复用的生产脚本框架最后我将分享一个我常用的脚本框架它集成了配置管理、错误处理、日志和主逻辑。# feishu_reporter.py import os import json import logging from datetime import datetime import requests from tenacity import retry, stop_after_attempt, wait_exponential # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(feishu_bot.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) class FeishuBot: def __init__(self, webhook_urlNone, app_idNone, app_secretNone): self.webhook_url webhook_url or os.getenv(FEISHU_WEBHOOK_URL) self.app_id app_id or os.getenv(FEISHU_APP_ID) self.app_secret app_secret or os.getenv(FEISHU_APP_SECRET) self._tenant_access_token None self._token_expiry None self._validate_config() def _validate_config(self): if not self.webhook_url: raise ValueError(未配置飞书机器人 Webhook URL) # 如果需要上传图片则检查 app_id 和 app_secret # if not self.app_id or not self.app_secret: # raise ValueError(未配置飞书应用 App ID 和 Secret) def _get_token(self): 获取并缓存 tenant_access_token if self._tenant_access_token and self._token_expiry and datetime.now() self._token_expiry: return self._tenant_access_token url https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal payload {app_id: self.app_id, app_secret: self.app_secret} try: resp requests.post(url, jsonpayload, timeout5) resp.raise_for_status() data resp.json() if data[code] 0: self._tenant_access_token data[tenant_access_token] # 令牌有效期通常为2小时这里设置1小时50分钟后过期预留缓冲 self._token_expiry datetime.now().timestamp() (2 * 3600 - 600) logger.info(Tenant access token 获取成功) return self._tenant_access_token else: raise Exception(f获取 token 失败: {data}) except Exception as e: logger.error(f获取 tenant_access_token 异常: {e}) raise def upload_image(self, image_path): 上传图片需要自建应用权限 token self._get_token() url https://open.feishu.cn/open-apis/im/v1/images headers {Authorization: fBearer {token}} files {image: open(image_path, rb)} data {image_type: message} try: resp requests.post(url, headersheaders, filesfiles, datadata, timeout30) resp.raise_for_status() result resp.json() if result[code] 0: image_key result[data][image_key] logger.info(f图片上传成功: {image_path} - key: {image_key}) return image_key else: logger.error(f图片上传接口返回错误: {result}) return None except Exception as e: logger.error(f上传图片 {image_path} 异常: {e}) return None retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def send_post(self, title, content_blocks, image_keyNone): 发送 post 消息 :param title: 标题 :param content_blocks: 内容块列表每个块是一个元素列表例如 [[{元素1}, {元素2}], [{图片元素}]] :param image_key: 可选的图片 key如果提供会自动添加到内容末尾 content content_blocks.copy() if image_key: content.append([{tag: img, image_key: image_key, width: 600, height: 400}]) payload { msg_type: post, content: { post: { zh_cn: { title: title, content: content } } } } logger.debug(f发送消息负载: {json.dumps(payload, indent2, ensure_asciiFalse)}) try: resp requests.post(self.webhook_url, jsonpayload, timeout10) resp.raise_for_status() result resp.json() if result.get(code) 0: logger.info(f消息发送成功: {title}) return True else: logger.error(f消息发送失败业务错误: {result}) # 特殊处理内容安全错误 if result.get(code) 9499: logger.critical(消息被内容安全策略拦截请检查文案和图片。) return False except requests.exceptions.RequestException as e: logger.error(f消息发送请求异常: {e}) raise # 触发重试 def generate_daily_report(): 模拟生成日报内容和图表返回标题、内容块、图片路径 # 这里是你的业务逻辑例如从数据库查询数据用matplotlib生成图表 title f【业务日报】{datetime.now().strftime(%Y-%m-%d)} content [ [{tag: text, text: 昨日核心数据概览, style: {bold: True}}], [{tag: text, text: - 新增用户: 1,234}], [{tag: text, text: - 订单总数: 567}], [{tag: text, text: - 平均响应时间: 125ms}], [{tag: hr}], [{tag: text, text: 详细趋势如下图所示}], ] # 假设生成了图表保存为文件 chart_path /tmp/daily_chart.png # ... 你的图表生成代码 ... return title, content, chart_path if __name__ __main__: # 初始化机器人 bot FeishuBot() # 从环境变量读取配置 # 1. 生成报告内容 title, content_blocks, chart_path generate_daily_report() # 2. 上传图表图片如果需要且配置了应用凭证 image_key None if chart_path and os.path.exists(chart_path) and bot.app_id: image_key bot.upload_image(chart_path) # 可选发送后删除临时图片文件 # os.remove(chart_path) # 3. 发送图文消息 success bot.send_post(title, content_blocks, image_key) if not success: logger.error(日报发送失败请检查) # 这里可以加入告警逻辑例如发送邮件 else: logger.info(日报发送任务完成。)这个框架提供了良好的起点你可以根据实际业务需求填充generate_daily_report函数并配置好环境变量一个可靠的企业级飞书图文消息机器人就搭建完成了。记住自动化是为了释放人力但监控和日志同样重要确保这个“沉默”的助手一直在可靠地工作。
