当你需要在自己的应用中集成消息推送时,飞书(Lark)提供了一套非常强大的API。今天,让我们一起来看看如何使用Python和飞书API实现智能消息推送功能。不用担心复杂的过程,我会一步步带你解析这段代码,让你了解每个环节的作用。

1. 准备工作

首先,确保你已经有了飞书开发者账户,以及创建了一个应用,这样你才能获得必须的 app_id app_secret 。接着,我们需要安装 lark_oapi 库,这样才能方便地调用飞书提供的API接口。

2. 功能点拆解

2.1 初始化客户端

client = lark.Client.builder() \
    .app_id(app_id) \
    .app_secret(app_secret) \
    .log_level(lark.LogLevel.ERROR) \
    .build()

在这里,我们初始化了一个lark客户端,配置了应用的app_idapp_secret,这是我们身份的凭证。同时,我们设置了日志级别,这样只有错误信息会被记录,让日志更清爽。

2.2 获取用户ID

request: BatchGetIdUserRequest = BatchGetIdUserRequest.builder() \
    .user_id_type("open_id") \
    .request_body(BatchGetIdUserRequestBody.builder()
        .mobiles([receive_phone_number])
        .include_resigned(False)
        .build()) \
    .build()
response: BatchGetIdUserResponse = client.contact.v3.user.batch_get_id(request)

在这段代码中,我们通过用户的手机号获取其在飞书上的open_id。这个open_id是推送消息的关键,它是每个用户的标识。

2.3 消息推送

request: CreateMessageRequest = CreateMessageRequest.builder() \
    .receive_id_type("open_id") \
    .request_body(CreateMessageRequestBody.builder()
        .receive_id(receive_open_id)
        .msg_type(msg_type)
        .content(msg_content)
        .build()) \
    .build()
response: CreateMessageResponse = client.im.v1.message.create(request)

接下来,我们构建了一个消息推送请求,指定了消息的类型和内容,然后调用client.im.v1.message.create方法进行消息推送。

3. 完整示例代码

import lark_oapi as lark
from lark_oapi.api.im.v1 import *
from lark_oapi.api.contact.v3 import *
# 1. 用户消息推送
# 2. 群组消息推送
# https://open.feishu.cn/document/server-docs/im-v1/message/create
# 3. 批量消息推送
# https://open.feishu.cn/document/server-docs/im-v1/batch_message/send-messages-in-batches
def app_push_msg(
        app_id: str,
        app_secret: str,
        receive_phone_number: str,
        msg_type: str,
        msg_content: str,
    应用用户消息推送
    :param app_id: 应用ID (App ID)
    :param app_secret: App Secret (应用密钥)
    :param receive_phone_number: 消息接收方手机号列表 (仅支持单个用户推送)
    :param msg_type: 消息类型
    :param msg_content: 消息内容
    # 创建client
    client = lark.Client.builder() \
        .app_id(app_id) \
        .app_secret(app_secret) \
        .log_level(lark.LogLevel.ERROR) \
        .build()
    # 通过手机号获取飞书用户open_id
    request: BatchGetIdUserRequest = BatchGetIdUserRequest.builder() \
        .user_id_type("open_id") \
        .request_body(BatchGetIdUserRequestBody.builder()
            .mobiles([receive_phone_number])
            .include_resigned(False)  # 是否获取已离职员工ID
            .build()) \
        .build()
    # 发起请求
    response: BatchGetIdUserResponse = client.contact.v3.user.batch_get_id(request)
    # 处理请求结果
    if response.success():
        result = lark.JSON.marshal(response.data, indent=4)
        lark.logger.info(result)
    else:
        result = f"client.contact.v3.user.batch_get_id failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        lark.logger.error(result)
        return result
    # 解析open_id
    receive_open_id = response.data.user_list[0].user_id
    # 推送消息
    request: CreateMessageRequest = CreateMessageRequest.builder() \
        .receive_id_type("open_id") \
        .request_body(CreateMessageRequestBody.builder()
            .receive_id(receive_open_id)
            .msg_type(msg_type)
            .content(msg_content)
            .build()) \
        .build()
    # 发起请求
    response: CreateMessageResponse = client.im.v1.message.create(request)
    # 处理结果
    if response.success():
        result = lark.JSON.marshal(response.data, indent=4)
        lark.logger.info(result)
    else:
        result = f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        lark.logger.error(result)
    return result
if __name__ == "__main__":
    app_push_msg(
        'cli_aXXXXXXXXXXXXXXX',
        'bkZXXXXXXXXXXXXXXXXXXXXXXXXXXXXEuvbF',
        "156XXXXXXXX",
        "interactive",
        """{\"type\": \"template\", \"data\": { \"template_id\": \"ctp_AAXXXXXXXXXX\", \"template_variable\": {
            \"alarm_title\": \"赚他一个亿\",
            \"alarm_error_content\": \"叮铃,微信到账1亿元\",
            \"alarm_error_datetime\": \"2023-1-30 15:27:00\",
            \"alarm_error_rpc\": \"RPA888\",
            \"alarm_error_rrb\": \"RPA888@nb\",
            \"im_chat_link\": \"https://applink.feishu.cn/client/chat/open?openId=ou_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\",
            \"pm_chat_link\": \"https://applink.feishu.cn/client/chat/open?openId=ou_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\"
            } } }

推送的卡片消息示例:
卡片示例

通过上述步骤,我们已经实现了一个简单的消息推送功能。虽然这只是冰山一角,但足以让你开始探索飞书API的强大功能。

希望你在实现自己的智能消息推送时也能有所启发,不过不要忘了,实际操作中还需要考虑很多细节,比如安全性、错误处理和用户体验等。开发是一门技术,也是一门艺术,代码的每一次迭代都是一次创造。

| Larksuite(Overseas) 如果使用的是飞书,请看 ,飞书Larksuite使用的域名不一样,引用的文档地址也是不同的。(If you are using FeiShu, please see , Feishu and larksuite use different domain names and reference different document addresses.) LarkSuite open api SDK Overview Larksuite open platform facilitates the integration of enterprise applications and larksuite, making collaboration and management more efficient. Larksuite developme
首先建群,建机器人,不多说,自行操作,获取webhook,通过webhook发送信息,原api地址: https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN 1、发送普通消息,代码如下 import requests import json def get_data(): data = { "msg_type": "text", "content": { "text
pip install -r requirements.txt * 如果引入其他新的依赖,可以执行冻结第三方库,就是将所有第三方库及版本号保存到requirements.txt文本文件中 ```shell pip freeze > requirements.txt * 如果pip不起作用,可以从pypi上下载最新的源码包(https://pypi.python.org/pypi/)进行安装: ```shell python setup.py install 1、该资源内项目代码都经过测试运行成功,功能ok的情况下才上传的,请放心下载使用!有问题请及时沟通交流。 2、适用人群:计算机相关专业(如计科、信息安全、数据科学与大数据技术、人工智能、通信、物联网、自动化、电子信息等)在校学生、专业老师或者企业员工下载使用。 3、用途:项目具有较高的学习借鉴价值,不仅适用于小白学习入门进阶。也可作为毕设项目、课程设计、大作业、初期项目立项演示等。 4、如果基础还行,或热爱钻研,亦可在此项目代码基础上进行修改添加,实现其他不同功能。 欢迎下载,沟通交流,互相学习,共同进步! 列表解析(list comprehension)提供了一种优雅的生成列表的方法,能用一行代码代替十几行代码,而且不损失任何可读性。而且,性能还快很多很多。 2. 将字符串拆分为列表 >>> string = "hello my friend !" >>> a = string.split() ['hello',...
对于 Python 飞书消息推送,你可以使用飞书开放平台提供的 API实现。下面是一个简单的示例: 首先,你需要在飞书开放平台创建一个应用并获取到相应的 App ID 和 App Secret。 然后,你可以使用 Python 的 requests 库来发送 HTTP 请求,调用飞书开放平台的 API。 ```python import requests def send_feishu_message(app_id, app_secret, user_id, message): # 获取 access_token token_url = f"https://open.feishu.cn/open-apis/auth/v3/app_access_token" token_payload = { "app_id": app_id, "app_secret": app_secret token_response = requests.post(token_url, json=token_payload) access_token = token_response.json()["app_access_token"] # 发送消息 message_url = f"https://open.feishu.cn/open-apis/message/v4/send/" message_payload = { "msg_type": "text", "content": { "text": message "user_id": user_id headers = { "Authorization": f"Bearer {access_token}" message_response = requests.post(message_url, json=message_payload, headers=headers) return message_response.status_code == 200 # 使用示例 app_id = "your_app_id" app_secret = "your_app_secret" user_id = "user_id_to_receive_message" message = "Hello, World!" send_feishu_message(app_id, app_secret, user_id, message) 在上述示例中,`send_feishu_message` 函数接受应用的 App ID、App Secret、接收消息的用户 ID 和要发送的消息作为参数。函数内部通过调用飞书开放平台的 API 获取 access_token,并使用 access_token 发送消息。 需要注意的是,该示例只是一个简单的推送文本消息的例子。如果你需要发送其他类型的消息,可以参考飞书开放平台的文档,调用相应的 API 进行操作。
# 扩大边距 new_media_box = (media_box.getLowerLeft_x() - 10, # 扩大左边距 media_box.getLowerLeft_y() - 10, # 扩大下边距 media_box.getUpperRight_x() + 10, # 扩大右边距 media_box.getUpperRight_y() + 10) # 扩大上边距 # 设置新的页面大小和位置 _this_page.cropBox = new_media_box _this_page.mediaBox = new_media_box [/code] 这样子试试呢
PyPDF2:项目实战源码分享(PDF裁剪) @wzy: 请问写入时能不能加大顶部边距呢 Python批量清理文件夹历史文件(Windows) CSDN-Ada助手: 恭喜您写了第19篇博客!看到标题我就感觉这篇文章对我来说非常有用,因为我经常需要清理文件夹。希望您能继续分享关于Python的实用技巧,比如如何自动化处理文件夹中的其他任务,或者如何利用Python优化Windows系统的其他功能。期待您的下一篇博客!谢谢您的分享! Xpath以控件标题定位控件本体 CSDN-Ada助手: 恭喜您写了第20篇博客!标题“Xpath以控件标题定位控件本体”听起来很有深度,感谢您分享这样有用的内容。我期待着看到您在未来更多关于Xpath定位的博客,也希望您可以考虑分享一些实际案例或者应用技巧,让读者更容易理解和应用您的知识。继续加油!