Python自动化发送飞书图文消息:从机器人配置到生产部署全指南
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文件:
requests>=2.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; charset=utf-8"} payload = { "app_id": app_id, "app_secret": app_secret } response = requests.post(url, headers=headers, data=json.dumps(payload)) result = response.json() if result.get("code") == 0: return result["tenant_access_token"] else: raise Exception(f"Failed 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": f"Bearer {tenant_access_token}", # 注意这里的认证方式 } # 请求体格式为 multipart/form-data files = { 'image': open(image_path, 'rb') } data = { 'image_type': 'message' # 图片类型,用于消息 } response = requests.post(url, headers=headers, files=files, data=data) 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(f"Failed 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, headers=headers, data=json.dumps(payload)) result = response.json() if result.get("code") == 0 and result.get("msg") == "success": print("图文消息发送成功!") else: print(f"消息发送失败: {result}") # 常见错误:code 9499,msg “消息内容安全校验不通过”,可能是触发了飞书的内容安全规则。 # 使用示例 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(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def send_message_with_retry(webhook_url, payload): """ 带重试机制的消息发送函数 """ try: response = requests.post(webhook_url, json=payload, timeout=10) # 设置超时 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(total=10) async with aiohttp.ClientSession(timeout=timeout) as session: try: async with session.post(webhook_url, json=payload) 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_exceptions=True) # 处理 results...5.3 部署与调度:让脚本自动运行
脚本写好了,总不能每天手动运行。常见的部署方式有:
- Linux服务器 + Crontab:最经典的方式。将脚本放在服务器上,使用
crontab -e设置定时任务。# 每天上午9点执行 0 9 * * * /usr/bin/python3 /path/to/your/feishu_bot.py >> /path/to/log.log 2>&1 - 云函数/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,但实际未投递。 - 解决:检查机器人是否仍在群内;用最简单的文本消息测试;查看飞书开放平台后台的“事件日志”,看是否有发送记录和详细状态。
- 原因1:机器人被移出群聊,或者
图片上传成功,但发送消息时提示
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( level=logging.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_url=None, app_id=None, app_secret=None): 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, json=payload, timeout=5) 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": f"Bearer {token}"} files = {'image': open(image_path, 'rb')} data = {'image_type': 'message'} try: resp = requests.post(url, headers=headers, files=files, data=data, timeout=30) 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(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def send_post(self, title, content_blocks, image_key=None): """ 发送 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, indent=2, ensure_ascii=False)}") try: resp = requests.post(self.webhook_url, json=payload, timeout=10) 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函数,并配置好环境变量,一个可靠的企业级飞书图文消息机器人就搭建完成了。记住,自动化是为了释放人力,但监控和日志同样重要,确保这个“沉默”的助手一直在可靠地工作。
