Pywayne Lark Bot Listener

Pywayne Lark Bot Listener

飞书消息监听器，通过 WebSocket 实时接收和处理飞书消息。

Quick Start

from pywayne.lark_bot_listener import LarkBotListener
初始化监听器
listener = LarkBotListener(
    app_id="your_app_id",
    app_secret="your_app_secret",
    message_expiry_time=60  # 消息去重过期时间（秒）
)
处理文本消息
@listener.text_handler()
async def handle_text(text: str):
    print(f"收到消息: {text}")
启动监听
listener.run()

Decorators - 消息处理装饰器

text_handler

文本消息处理装饰器，直接传递文本内容。

@listener.text_handler(group_only=False, user_only=False)
async def handle_text(text: str, chat_id: str, is_group: bool, group_name: str, user_name: str):
    print(f"收到来自 {user_name} 的消息: {text}")
    listener.send_message(chat_id, f"已收到：{text}")

参数说明（除 text 外均为可选）：

| 参数 | 类型 | 说明 | |------|------|------| | text | str | 文本内容（必需） | | chat_id | str | 会话 ID | | is_group | bool | 是否群组消息 | | group_name | str | 群组名称（私聊时为空） | | user_name | str | 发送消息的用户姓名 |

装饰器参数：

image_handler

图片消息处理装饰器，自动下载图片到临时文件并清理。

from pathlib import Path
import cv2
import tempfile
@listener.image_handler()
async def handle_image(image_path: Path, user_name: str) -> Path:
    # 处理图片
    img = cv2.imread(str(image_path))
    # ...处理逻辑...
    # 返回新图片路径会自动发送回去，返回 None 则不发送
    return image_path

参数说明（除 image_path 外均为可选）：

| 参数 | 类型 | 说明 | |------|------|------| | image_path | Path | 临时图片文件路径（必需） | | chat_id | str | 会话 ID | | is_group | bool | 是否群组消息 | | group_name | str | 群组名称 | | user_name | str | 发送消息的用户姓名 |

返回值：

file_handler

文件消息处理装饰器，自动下载文件到临时文件并清理。

@listener.file_handler()
async def handle_file(file_path: Path, user_name: str) -> Path:
    # 处理文件
    with open(file_path, 'r') as f:
        content = f.read()
    # ...处理逻辑...
    return file_path  # 返回文件路径会自动发送回去

参数说明（除 file_path 外均为可选）：

| 参数 | 类型 | 说明 | |------|------|------| | file_path | Path | 临时文件路径（必需） | | chat_id | str | 会话 ID | | is_group | bool | 是否群组消息 | | group_name | str | 群组名称 | | user_name | str | 发送消息的用户姓名 |

listen - 通用消息监听器

用于监听任意类型消息（包括富文本 post）。

@listener.listen(message_type="post")
async def handle_post(ctx: MessageContext):
    print(f"收到富文本消息: {ctx.content}")

MessageContext 属性：

| 属性 | 类型 | 说明 | |------|------|------| | chat_id | str | 会话 ID | | user_id | str | 用户 ID | | message_type | str | 消息类型 | | content | str | 消息内容（文本消息为字符串，其他类型为 JSON 字符串） | | is_group | bool | 是否群组消息 | | chat_type | str | 会话类型 | | message_id | str | 消息 ID |

send_message - 发送消息

发送 Markdown 格式的消息到飞书。

listener.send_message(chat_id, "这是加粗文本")
listener.send_message(chat_id, "普通文本\n链接")

完整示例

from pywayne.lark_bot_listener import LarkBotListener
from pathlib import Path
listener = LarkBotListener(
    app_id="your_app_id",
    app_secret="your_app_secret"
)
文本消息 - AI 回复示例
@listener.text_handler()
async def handle_text(text: str, chat_id: str, user_name: str):
    # 处理逻辑...
    listener.send_message(chat_id, f"收到来自 {user_name} 的消息")
图片消息 - 自动下载和清理
@listener.image_handler()
async def handle_image(image_path: Path):
    # image_path 是临时文件，处理完会自动清理
    print(f"处理图片: {image_path}")
文件消息 - 自动下载和清理
@listener.file_handler(group_only=True)
async def handle_file(file_path: Path, group_name: str):
    print(f"收到文件: {file_path} 来自 {group_name}")
富文本消息
@listener.listen(message_type="post")
async def handle_post(ctx: MessageContext):
    import json
    post_content = json.loads(ctx.content)
    print(f"收到富文本: {post_content}")
启动监听
listener.run()

注意事项

1. 异步处理: 所有处理函数使用 async/await 语法 2. 消息去重: 每个处理函数独立去重，默认 60 秒过期 3. 临时文件: 图片和文件下载到 系统临时目录/lark_bot_temp，处理完自动清理 4. 错误隔离: 每个处理函数异常独立捕获，不影响其他函数 5. 多注册: 同一消息可被多个处理函数处理

from pywayne.lark_bot_listener import LarkBotListener
初始化监听器
listener = LarkBotListener(
    app_id="your_app_id",
    app_secret="your_app_secret",
    message_expiry_time=60  # 消息去重过期时间（秒）
)
处理文本消息
@listener.text_handler()
async def handle_text(text: str):
    print(f"收到消息: {text}")
启动监听
listener.run()