# 使用Python编写飞书消息发送程序 下面我将详细介绍如何使用Python编写发送飞书消息的程序，涵盖两种主要实现方式：使用Webhook机器人和使用飞书开放平台API。 ## 1. 准备工作 在开始编写代码前，需要先完成以下配置： ### 1.1 Webhook机器人方式配置 - 在飞书群聊中添加自定义机器人 - 获取Webhook URL - 配置安全设置（可选）：自定义关键词、IP白名单、签名校验[ref_2] ### 1.2 开放平台API方式配置 - 在飞书开放平台创建应用 - 获取应用凭证：app_id、app_secret - 配置应用权限：获取消息发送权限[ref_4] ## 2. 使用Webhook机器人发送消息 ### 2.1 基础文本消息发送 ```python import requests import json import time class FeishuWebhookBot: """飞书Webhook机器人封装类""" def __init__(self, webhook_url): self.webhook_url = webhook_url self.headers = { 'Content-Type': 'application/json' } def send_text(self, content): """ 发送文本消息 :param content: 消息内容 :return: 发送结果 """ data = { "msg_type": "text", "content": { "text": content } } try: response = requests.post( self.webhook_url, headers=self.headers, data=json.dumps(data) ) return response.json() except Exception as e: return {"code": -1, "msg": f"发送失败: {str(e)}"} # 使用示例 if __name__ == "__main__": # 替换为你的实际webhook URL webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" bot = FeishuWebhookBot(webhook_url) # 发送文本消息 result = bot.send_text("这是一条测试消息，来自Python程序！") print("发送结果:", result) ``` ### 2.2 发送富文本消息 ```python def send_rich_text(self, title, content_elements): """ 发送富文本消息 :param title: 消息标题 :param content_elements: 内容元素列表，每个元素为字典格式 :return: 发送结果 """ # 构建富文本内容结构 elements = [] for element in content_elements: if element.get("tag") == "text": elements.append({ "tag": "text", "text": element.get("text", "") }) elif element.get("tag") == "a": elements.append({ "tag": "a", "text": element.get("text", ""), "href": element.get("href", "") }) elif element.get("tag") == "at": elements.append({ "tag": "at", "user_id": element.get("user_id", "") }) data = { "msg_type": "post", "content": { "post": { "zh_cn": { "title": title, "content": [elements] } } } } try: response = requests.post( self.webhook_url, headers=self.headers, data=json.dumps(data) ) return response.json() except Exception as e: return {"code": -1, "msg": f"发送失败: {str(e)}"} # 富文本消息使用示例 elements = [ {"tag": "text", "text": "这是一条富文本消息\n"}, {"tag": "a", "text": "点击查看详情", "href": "https://www.feishu.cn"}, {"tag": "text", "text": "\n"}, {"tag": "at", "user_id": "user_id"} # 替换为实际用户ID ] result = bot.send_rich_text("重要通知", elements) print("富文本发送结果:", result) ``` ## 3. 使用飞书开放平台API发送消息 ### 3.1 获取访问令牌 ```python class FeishuAPIClient: """飞书API客户端""" def __init__(self, app_id, app_secret): self.app_id = app_id self.app_secret = app_secret self.base_url = "https://open.feishu.cn/open-apis" self.access_token = None self.token_expire_time = 0 def get_access_token(self): """ 获取访问令牌 :return: access_token """ # 检查token是否过期（通常2小时有效） if self.access_token and time.time() < self.token_expire_time: return self.access_token url = f"{self.base_url}/auth/v3/tenant_access_token/internal" data = { "app_id": self.app_id, "app_secret": self.app_secret } try: response = requests.post(url, json=data) result = response.json() if result.get("code") == 0: self.access_token = result["tenant_access_token"] self.token_expire_time = time.time() + result["expire"] - 300 # 提前5分钟刷新 return self.access_token else: raise Exception(f"获取token失败: {result}") except Exception as e: raise Exception(f"获取访问令牌异常: {str(e)}") ``` ### 3.2 发送消息到指定群组 ```python def send_message_to_chat(self, chat_id, msg_type, content): """ 发送消息到指定群组 :param chat_id: 群组ID :param msg_type: 消息类型 :param content: 消息内容 :return: 发送结果 """ access_token = self.get_access_token() url = f"{self.base_url}/im/v1/messages" headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } data = { "receive_id": chat_id, "msg_type": msg_type, "content": json.dumps(content) } try: response = requests.post( url, headers=headers, params={"receive_id_type": "chat_id"}, json=data ) return response.json() except Exception as e: return {"code": -1, "msg": f"发送失败: {str(e)}"} # 使用示例 if __name__ == "__main__": # 替换为你的应用凭证 app_id = "your_app_id" app_secret = "your_app_secret" chat_id = "your_chat_id" # 群组ID client = FeishuAPIClient(app_id, app_secret) # 发送文本消息 text_content = {"text": "这是一条通过API发送的测试消息"} result = client.send_message_to_chat(chat_id, "text", text_content) print("API发送结果:", result) ``` ## 4. 完整功能封装类 下面是一个功能更完整的飞书消息发送工具类： ```python import requests import json import time import hmac import hashlib import base64 class FeishuMessenger: """飞书消息发送工具类""" def __init__(self, webhook_url=None, app_id=None, app_secret=None): self.webhook_url = webhook_url self.app_id = app_id self.app_secret = app_secret self.base_url = "https://open.feishu.cn/open-apis" if app_id and app_secret: self.api_client = FeishuAPIClient(app_id, app_secret) def send_text(self, content, chat_id=None): """发送文本消息""" if chat_id and hasattr(self, 'api_client'): # 使用API发送到指定群组 text_content = {"text": content} return self.api_client.send_message_to_chat(chat_id, "text", text_content) elif self.webhook_url: # 使用Webhook发送 bot = FeishuWebhookBot(self.webhook_url) return bot.send_text(content) else: return {"code": -1, "msg": "未配置发送方式"} def send_card_message(self, title, elements, chat_id=None): """发送卡片消息""" card_content = { "config": { "wide_screen_mode": True }, "header": { "title": { "tag": "plain_text", "content": title } }, "elements": elements } if chat_id and hasattr(self, 'api_client'): return self.api_client.send_message_to_chat(chat_id, "interactive", card_content) else: return {"code": -1, "msg": "卡片消息仅支持API方式发送"} def send_image(self, image_key, chat_id=None): """发送图片消息""" image_content = {"image_key": image_key} if chat_id and hasattr(self, 'api_client'): return self.api_client.send_message_to_chat(chat_id, "image", image_content) else: return {"code": -1, "msg": "图片消息仅支持API方式发送"} # 完整使用示例 def main(): # 方式1: 使用Webhook机器人 webhook_bot = FeishuMessenger( webhook_url="https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" ) # 发送文本消息 result1 = webhook_bot.send_text("Webhook机器人测试消息") print("Webhook发送结果:", result1) # 方式2: 使用API api_bot = FeishuMessenger( app_id="your_app_id", app_secret="your_app_secret" ) # 发送到指定群组 result2 = api_bot.send_text( "API发送的测试消息", chat_id="your_chat_id" ) print("API发送结果:", result2) # 发送卡片消息 card_elements = [ { "tag": "div", "text": { "tag": "lark_md", "content": "这是一个卡片消息示例" } }, { "tag": "action", "actions": [ { "tag": "button", "text": { "tag": "plain_text", "content": "点击按钮" }, "type": "primary", "value": {"key": "value"} } ] } ] result3 = api_bot.send_card_message( "卡片消息标题", card_elements, chat_id="your_chat_id" ) print("卡片消息发送结果:", result3) if __name__ == "__main__": main() ``` ## 5. 安全配置说明 ### 5.1 签名校验（如果启用） 如果Webhook机器人启用了签名校验，需要在请求头中添加签名[ref_2]： ```python import time def generate_signature(timestamp, secret): """生成签名""" string_to_sign = f'{timestamp}\n{secret}' hmac_code = hmac.new( string_to_sign.encode("utf-8"), digestmod=hashlib.sha256 ).digest() return base64.b64encode(hmac_code).decode('utf-8') # 在发送消息时添加签名 timestamp = str(int(time.time())) secret = "your_secret" # 机器人安全设置中的签名密钥 signature = generate_signature(timestamp, secret) headers = { 'Content-Type': 'application/json', 'X-Lark-Signature': signature, 'X-Lark-Request-Timestamp': timestamp } ``` ## 6. 应用场景示例 ### 6.1 监控告警通知 ```python def send_monitor_alert(alert_level, alert_message, service_name): """发送监控告警通知""" messenger = FeishuMessenger(webhook_url="your_webhook_url") if alert_level == "CRITICAL": content = f"🚨 严重告警 - {service_name}\n\n{alert_message}" elif alert_level == "WARNING": content = f"⚠️ 警告 - {service_name}\n\n{alert_message}" else: content = f"ℹ️ 信息 - {service_name}\n\n{alert_message}" return messenger.send_text(content) # 使用示例 send_monitor_alert("CRITICAL", "数据库连接失败，请立即处理！", "MySQL服务") ``` ### 6.2 定时任务报告 ```python import schedule def daily_report(): """每日报告发送""" messenger = FeishuMessenger(webhook_url="your_webhook_url") content = "📊 每日系统报告\n\n系统运行正常，所有服务状态良好。" messenger.send_text(content) # 设置定时任务 schedule.every().day.at("09:00").do(daily_report) ``` 通过上述代码示例，您可以快速实现飞书消息的发送功能。根据实际需求选择合适的实现方式：Webhook机器人适合简单的消息推送，而开放平台API则提供更丰富的功能和更灵活的权限控制[ref_2][ref_4]。