Pywayne Lark Bot Listener — Pywayne Lark 机器人 列出ener
v0.1.0Feishu/Lark message 列出ener for real-time message processing via 网页Socket. Use when users need to 列出en for incoming Feishu messages (text, image, file, post) with automatic deduplication and a同步 handling. Provides convenient decorators (text_处理器, image_处理器, file_处理器) that handle message 下载/清理up and optional automatic reply.
运行时依赖
安装命令
点击复制技能文档
Pywayne Lark 机器人 列出ener
飞书消息监听器,通过 网页Socket 实时接收和处理飞书消息。
Quick 启动 from pywayne.lark_机器人_列出ener 导入 Lark机器人列出ener
# 初始化监听器 列出ener = Lark机器人列出ener( 应用_id="your_应用_id", 应用_secret="your_应用_secret", message_expiry_time=60 # 消息去重过期时间(秒) )
# 处理文本消息 @列出ener.text_处理器() a同步 def handle_text(text: str): print(f"收到消息: {text}")
# 启动监听 列出ener.运行()
Decorators - 消息处理装饰器 text_处理器
文本消息处理装饰器,直接传递文本内容。
@列出ener.text_处理器(group_only=False, user_only=False) a同步 def handle_text(text: str, chat_id: str, is_group: bool, group_name: str, user_name: str): print(f"收到来自 {user_name} 的消息: {text}") 列出ener.发送_message(chat_id, f"已收到:{text}")
参数说明(除 text 外均为可选):
参数 类型 说明 text str 文本内容(必需) chat_id str 会话 ID is_group bool 是否群组消息 group_name str 群组名称(私聊时为空) user_name str 发送消息的用户姓名
装饰器参数:
group_only=True: 只处理群组消息 user_only=True: 只处理私聊消息 image_处理器
图片消息处理装饰器,自动下载图片到临时文件并清理。
from pathlib 导入 Path 导入 cv2 导入 tempfile
@列出ener.image_处理器() a同步 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 发送消息的用户姓名
返回值:
返回 Path: 自动上传并发送新图片 返回 None: 不发送任何图片 file_处理器
文件消息处理装饰器,自动下载文件到临时文件并清理。
@列出ener.file_处理器() a同步 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 发送消息的用户姓名 列出en - 通用消息监听器
用于监听任意类型消息(包括富文本 post)。
@列出ener.列出en(message_type="post") a同步 def handle_post(ctx: Message上下文): print(f"收到富文本消息: {ctx.content}")
Message上下文 属性:
属性 类型 说明 chat_id str 会话 ID user_id str 用户 ID message_type str 消息类型 content str 消息内容(文本消息为字符串,其他类型为 JSON 字符串) is_group bool 是否群组消息 chat_type str 会话类型 message_id str 消息 ID 发送_message - 发送消息
发送 Markdown 格式的消息到飞书。
列出ener.发送_message(chat_id, "这是加粗文本") 列出ener.发送_message(chat_id, "普通文本\n链接")
完整示例 from pywayne.lark_机器人_列出ener 导入 Lark机器人列出ener from pathlib 导入 Path
列出ener = Lark机器人列出ener( 应用_id="your_应用_id", 应用_secret="your_应用_secret" )
# 文本消息 - AI 回复示例 @列出ener.text_处理器() a同步 def handle_text(text: str, chat_id: str, user_name: str): # 处理逻辑... 列出ener.发送_message(chat_id, f"收到来自 {user_name} 的消息")
# 图片消息 - 自动下载和清理 @列出ener.image_处理器() a同步 def handle_image(image_path: Path): # image_path 是临时文件,处理完会自动清理 print(f"处理图片: {image_path}")
# 文件消息 - 自动下载和清理 @列出ener.file_处理器(group_only=True) a同步 def handle_file(file_path: Path, group_name: str): print(f"收到文件: {file_path} 来自 {group_name}")
# 富文本消息 @列出ener.列出en(message_type="post") a同步 def handle_post(ctx: Message上下文): 导入 json post_content = json.loads(ctx.content) print(f"收到富文本: {post_content}")
# 启动监听 列出ener.运行()
注意事项 异步处理: 所有处理函数使用 a同步/awAIt 语法 消息去重: 每个处理函数独立去重,默认 60 秒过期 临时文件: 图片和文件下载到 系统临时目录/lark_机器人_temp,处理完自动清理 错误隔离: 每个处理函数异常独立捕获,不影响其他函数 多注册: 同一消息可被多个处理函数处理