飞书单聊应用机器人

def push_text_message(
    appid: str = typer.Option(..., help="飞书应用ID", envvar="FEISHU_APP_ID"),
    appsecret: str = typer.Option(..., help="飞书应用密钥", envvar="FEISHU_APP_SECRET"),
    log_level: int = typer.Option(logging.DEBUG, help="日志级别", envvar="FEISHU_LOG_LEVEL"),
    receive_id_type: Literal["open_id", "union_id", "user_id", "email", "chat_id"] = typer.Option("open_id", help="和 receive_id 一起使用"),
    receive_id: str = typer.Option(..., help="和 receive_id_type 一起使用"),
    msg_type: Literal["text", "post", "image", "file", "audio", "media", "sticker", "interactive", "share_chat", "share_user", "system"] = typer.Option("text", help="和 content 一起使用"),
    content: str = typer.Argument(..., help="推送文本消息"),
):
    """
    飞书推送文本消息.

    https://open.feishu.cn/document/server-docs/im-v1/message/create
    """
    logging.basicConfig(level=log_level)
    typer.echo(helptext)
    client = init_client(appid, appsecret)
    if client.im is None:
        raise Exception("client.im is None")
    if client.im.v1 is None:
        raise Exception("client.im.v1 is None")
    request: CreateMessageRequest = (
        CreateMessageRequest.builder()
        .receive_id_type(receive_id_type)
        .request_body(CreateMessageRequestBody.builder().receive_id(receive_id).msg_type(msg_type).content(json.dumps({"text": content})).build())
        .build()
    )

    # 发起请求
    response: CreateMessageResponse = client.im.v1.message.create(request)
)

def init_ws_client(appid: str, appsecret: str) -> lark.ws.Client:
    global client
    event_handler = lark.EventDispatcherHandler.builder("", "").register_p2_im_message_receive_v1(do_p2_im_message_receive_v1).build()

    client = lark.Client.builder().app_id(appid).app_secret(appsecret).build()
    wsClient = lark.ws.Client(
        appid,
        appsecret,
        event_handler=event_handler,
        log_level=lark.LogLevel.DEBUG,
    )
    return wsClient

# https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/events/receive
# https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/create
def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    global client
    if client is None:
        raise Exception("client is None")
    if client.im is None:
        raise Exception("client.im is None")
    res_content = ""
    if data.event is None or data.event.message is None or data.event.sender is None or data.event.sender.sender_id is None or data.event.sender.sender_id.open_id is None:
        typer.echo("解析消息失败, 请发送文本消息")
        raise Exception("解析消息失败, 请发送文本消息")
    if data.event.message.message_type == "text" and data.event.message.content is not None:
        text = json.loads(data.event.message.content)["text"]
        res_content = (
            f"{text}\nchat_type: {data.event.message.chat_type}\nchat_id: {data.event.message.chat_id}"
            f"\nmessage_type: {data.event.message.message_type}\nmessage_id: {data.event.message.message_id}"
            f"\nsender_openid: {data.event.sender.sender_id.open_id}\nsender_type: {data.event.sender.sender_type}"
        )
        typer.echo(f"res_content: {res_content}")
    else:
        res_content = "解析消息失败, 请发送文本消息"

    content = json.dumps({"text": "收到你发送的消息:" + res_content})
    if data.event.message.chat_type == "p2p":
        receive_id = data.event.message.chat_id
        if receive_id is None:
            raise Exception("receive_id is None")
        p2p_request: CreateMessageRequest = (
            CreateMessageRequest.builder().receive_id_type("chat_id").request_body(CreateMessageRequestBody.builder().receive_id(receive_id).msg_type("text").content(content).build()).build()
        )

        response: CreateMessageResponse = client.im.v1.message.create(p2p_request)

        if not response.success():
            raise Exception(f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")
    else:
        msg_id = data.event.message.message_id
        if msg_id is None:
            raise Exception("msg_id is None")
        other_request: ReplyMessageRequest = ReplyMessageRequest.builder().message_id(msg_id).request_body(ReplyMessageRequestBody.builder().content(content).msg_type("text").build()).build()
        reply_response: ReplyMessageResponse = client.im.v1.message.reply(other_request)
        if not reply_response.success():
            raise Exception(f"client.im.v1.message.reply failed, code: {reply_response.code}, msg: {reply_response.msg}, log_id: {reply_response.get_log_id()}")

wsClient = init_ws_client(appid, appsecret)
wsClient.start()