wings-bot 0.1.0

基于 lark-oapi FeishuChannel 的装饰器风格飞书 Bot 开发框架

FeishuBot — 飞书 Bot 开发框架

基于 lark-oapi SDK 的 FeishuChannel 构建的装饰器风格飞书机器人框架。

特性

• 装饰器注册 — @bot.on_message() / @bot.on_event() / @bot.use()
• 命令路由 — @cmds.command("ping") 自动解析 /ping 等命令
• 正则匹配 — @cmds.match(r"^(hi|hey)\b") 灵活匹配消息
• 洋葱中间件 — @bot.use() AOP 风格中间件链，支持前置/后置逻辑
• 定时任务 — @bot.scheduler.cron("*/5 * * * *") cron 表达式调度
• 插件系统 — plugins/ 目录下的 .py 文件自动发现和加载
• 扩展机制 — bot.install() 安装可选扩展

快速开始

1. 环境准备

# 克隆项目
git clone <repo-url> && cd FeishuBot

# 安装依赖（需要 uv）
uv sync

2. 配置飞书应用

在飞书开放平台创建应用，获取 APP_ID 和 APP_SECRET，写入 .env：

APPID=cli_xxxxxxxxxxxxx
APPSEC=xxxxxxxxxxxxxxxxxxxxxxxx

确保应用已开启以下能力：

• 机器人能力（Bot）
• 事件订阅 → 消息接收（im.message.receive_v1）
• 事件订阅方式选择「使用长连接接收」

3. 启动

uv run python main.py

项目结构

FeishuBot/
├── main.py                  # 入口：初始化 Bot → 加载插件 → 启动
├── .env                     # APPID / APPSEC 配置
├── pyproject.toml
├── bot/                     # 框架核心
│   ├── __init__.py          # 统一导出
│   ├── app.py               # FeishuBot 主类
│   ├── context.py           # MessageContext 消息上下文
│   ├── middleware.py         # 洋葱模型中间件链
│   ├── scheduler.py          # 定时任务调度器
│   ├── plugin.py             # 插件自动发现与加载
│   └── ext/
│       ├── __init__.py
│       └── commands.py       # 命令路由 + 正则匹配扩展
└── plugins/                  # 插件目录（自动加载）
    ├── __init__.py
    ├── echo.py               # 示例：消息回显 + 命令
    └── hello.py              # 示例：正则路由 + 定时任务 + 事件

---

核心概念

FeishuBot

框架入口类，持有飞书连接、处理器注册表和调度器。

from bot import FeishuBot

bot = FeishuBot(
    app_id="cli_xxx",
    app_secret="xxx",
    # 以下参数透传给 FeishuChannel，按需配置
    # log_level=lark.LogLevel.DEBUG,
)

属性	类型	说明
bot.channel	FeishuChannel	底层 SDK 实例，可直接调用 send()、stream() 等高级 API
bot.scheduler	Scheduler	定时任务调度器

MessageContext

所有消息处理器和中间件接收的统一上下文对象。

async def handler(ctx: MessageContext):
    # 只读属性
    ctx.text            # str   — 消息纯文本
    ctx.sender_id       # str   — 发送者 open_id
    ctx.sender_name     # str   — 发送者显示名称
    ctx.chat_id         # str   — 当前聊天 chat_id
    ctx.chat_type       # str   — "p2p" 或 "group"
    ctx.message_type    # str   — "text" / "image" / "interactive" 等
    ctx.message_id      # str   — 消息 ID
    ctx.is_mentioned    # bool  — Bot 是否被 @
    ctx.mentions        # list  — @提及 列表
    ctx.reply_to_id     # str|None — 如果是回复消息，被回复的消息 ID
    ctx.parent_message  # ReplyRef|None — 被回复的消息引用
    ctx.content         # MessageContent — 归一化内容对象
    ctx.raw             # InboundMessage — 原始 SDK 对象

    # 操作方法
    await ctx.reply("你好")                          # 回复纯文本
    await ctx.reply("**粗体**", msg_type="markdown") # 回复 markdown
    await ctx.reply({"text": "自定义内容"})           # 回复自定义内容
    await ctx.reply_card({"config": {}, ...})         # 回复卡片
    await ctx.send(chat_id, "主动发消息")             # 向指定聊天发送

    # 共享状态
    ctx.state           # dict  — 中间件和处理器之间传递数据

消息类型判断

ctx.content 是归一化的 discriminated union，通过 kind 字段判断：

if ctx.content.kind == "text":
    print(ctx.content.text)
elif ctx.content.kind == "image":
    print(ctx.content.image_key)
elif ctx.content.kind == "file":
    print(ctx.content.file_key, ctx.content.file_name)
elif ctx.content.kind == "interactive":
    print(ctx.content.card)  # 卡片 JSON

支持的 kind 值：text、post、image、file、audio、media、sticker、interactive、share_chat、share_user、system、location、folder、hongbao、general_calendar、share_calendar_event、video_chat、calendar、vote、todo、merge_forward、unknown。

---

消息处理

@bot.on_message()

注册消息处理器，支持过滤条件和优先级。

@bot.on_message()
async def handle_all(ctx):
    """处理所有消息。"""
    await ctx.reply(f"收到: {ctx.text}")

过滤条件

# 仅处理私聊文本消息
@bot.on_message(chat_type="p2p", message_type="text")
async def dm_text(ctx):
    await ctx.reply("这是私聊文本")

# 仅处理群聊中被 @ 的消息
@bot.on_message(chat_type="group", mentioned=True)
async def group_mention(ctx):
    await ctx.reply("你在群里 @ 了我")

# 高优先级处理器（数值越大越先执行）
@bot.on_message(priority=100)
async def high_priority(ctx):
    pass

参数	类型	说明
chat_type	str	过滤聊天类型："p2p" 或 "group"
message_type	str	过滤消息类型："text"、"image" 等
mentioned	bool	过滤是否被 @：True = 被 @，False = 未被 @
priority	int	优先级，数值越大越先执行，默认 10

处理流程

1. 所有注册的 handler 按 priority 降序排列
2. 遍历 handler，检查过滤条件是否匹配
3. 第一个匹配的 handler 执行（经过中间件链）
4. 如果 handler 执行成功，处理结束
5. 如果 handler 设置了 ctx._matched = False（如命令扩展未命中），继续尝试下一个 handler

---

命令路由

通过 CommandsExtension 实现，以指定前缀解析文本消息为命令。

安装

from bot.ext.commands import CommandsExtension

cmds = CommandsExtension(prefix="/")
bot.install(cmds)

@cmds.command()

@cmds.command("echo", description="回复你发送的内容")
async def echo(ctx, args: list[str]):
    await ctx.reply(" ".join(args))

用户发送 /echo hello world 时，args = ["hello", "world"]。

别名

@cmds.command("help", aliases=["h", "?"], description="显示帮助")
async def help_cmd(ctx, args):
    await ctx.reply(cmds.generate_help())

/help、/h、/? 都会触发同一个处理器。

自动帮助

cmds.generate_help()
# 输出：
# 📋 可用命令：
#
#   /echo — 回复你发送的内容
#   /help (别名: /h, /?) — 显示帮助

@cmds.match()

正则表达式匹配，按注册顺序尝试，第一个匹配即执行。

@cmds.match(r"^(hi|hey|hello)\b")
async def greet(ctx, match: re.Match):
    await ctx.reply(f"{match.group(1)}！有什么可以帮你的？")

@cmds.match(r"^(时间|几点|what\s*time)")
async def show_time(ctx, match: re.Match):
    from datetime import datetime
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    await ctx.reply(f"现在是 {now}")

处理器签名：async def handler(ctx: MessageContext, match: re.Match)。

匹配优先级

安装 CommandsExtension 后，文本消息的处理顺序：

1. 命令匹配 — 文本以 / 开头，尝试匹配已注册命令
2. 正则匹配 — 按注册顺序尝试所有正则表达式
3. 穿透 — 都未命中时，消息继续流入普通的 @bot.on_message() 处理器

---

中间件

@bot.use()

洋葱模型中间件，按注册顺序包裹 handler。

@bot.use()
async def logging_mw(ctx, next):
    print(f"→ 收到: {ctx.text}")
    await next()               # 调用下一个中间件或最终 handler
    print(f"← 处理完成")

中断链

不调用 await next() 即可阻止后续处理：

@bot.use()
async def auth_mw(ctx, next):
    if ctx.sender_id not in ALLOWED_USERS:
        await ctx.reply("你没有权限使用此机器人")
        return                 # 不调用 next()，链中断
    await next()

共享状态

中间件可以通过 ctx.state 传递数据给下游：

@bot.use()
async def user_info_mw(ctx, next):
    ctx.state["user_role"] = get_user_role(ctx.sender_id)
    ctx.state["start_time"] = time.time()
    await next()

@bot.on_message()
async def handler(ctx):
    role = ctx.state.get("user_role", "guest")
    await ctx.reply(f"你的角色是: {role}")

执行顺序

注册了 3 个中间件 + 1 个 handler 时：

mw_1 前置 → mw_2 前置 → mw_3 前置 → handler
                                        ↓
mw_1 后置 ← mw_2 后置 ← mw_3 后置 ← ──┘

---

定时任务

@bot.scheduler.cron()

# 同步函数
@bot.scheduler.cron("*/5 * * * *")
def heartbeat():
    print("Bot is alive")

# 异步函数（通过 FeishuChannel.schedule() 执行）
@bot.scheduler.cron("0 9 * * 1-5")
async def morning_report():
    await bot.channel.send(chat_id, {"text": "早上好！这是今日报告。"})

Cron 表达式格式

5 字段标准格式：分 时 日 月 周

表达式	说明
*/5 * * * *	每 5 分钟
0 * * * *	每小时整点
0 9 * * 1-5	工作日早上 9 点
30 14 * * *	每天 14:30
0 0 1 * *	每月 1 日零点

调度行为

• 后台守护线程，对齐整分钟触发
• 协程任务通过 FeishuChannel.schedule() 在事件循环中执行
• 同步任务直接在后台线程中执行
• 调度器随 bot.run() 自动启动，随 bot.stop() 自动停止

---

事件处理

@bot.on_event()

注册非消息事件的处理器。支持的事件类型：

事件名	说明	handler 参数
"message"	收到消息（通常用 @bot.on_message() 代替）	InboundMessage
"cardAction"	卡片交互（按钮点击等）	CardActionEvent
"reaction"	表情回应	ReactionEvent
"botAdded"	Bot 被添加到群	BotAddedEvent
"botLeave"	Bot 被移出群	BotLeaveEvent
"messageRead"	消息已读	MessageReadEvent
"comment"	文档评论	Comment event
"raw"	原始事件	原始 payload
"error"	错误事件	Exception
"reconnecting"	WS 重连中	无参数
"reconnected"	WS 已重连	无参数

@bot.on_event("botAdded")
async def on_added(event):
    print(f"Bot 被添加到群: {event.chat_id}")

@bot.on_event("cardAction")
async def on_card(event):
    action = event.action
    print(f"卡片按钮: tag={action.tag}, value={action.value}")

@bot.on_event("reaction")
async def on_reaction(event):
    if event.action == "added":
        print(f"用户 {event.operator.open_id} 添加了 {event.emoji_type}")

@bot.on_event("error")
async def on_error(exc):
    logger.exception("Bot 发生错误", exc_info=exc)

---

插件系统

约定

• 每个插件是 plugins/ 目录下的一个 .py 文件
• 文件必须定义 register(bot, **kwargs) 函数
• 以 _ 开头的文件会被跳过（如 __init__.py）

编写插件

# plugins/weather.py

def register(bot, cmds=None):
    """天气插件。"""

    if cmds is not None:
        @cmds.command("weather", description="查询天气")
        async def weather(ctx, args):
            city = " ".join(args) if args else "北京"
            # ... 查询天气 API ...
            await ctx.reply(f"{city}: 晴，25°C")

        @cmds.match(r"今天.*天气")
        async def weather_nlp(ctx, match):
            await ctx.reply("今天天气不错！")

    @bot.scheduler.cron("0 8 * * *")
    async def daily_weather():
        await bot.channel.send(
            monitor_chat_id,
            {"text": "今日天气：晴，25°C"}
        )

自动加载

# main.py 中
from bot import load_plugins

load_plugins(bot, cmds=cmds)
# 扫描 plugins/ 目录，按文件名字典序加载

禁用插件

load_plugins(bot, cmds=cmds, disabled=["weather", "debug"])

传递额外参数

load_plugins() 的 **kwargs 会透传给每个插件的 register() 函数：

# main.py
load_plugins(bot, cmds=cmds, db=db_connection)

# plugins/data.py
def register(bot, cmds=None, db=None):
    # db 是从 main.py 传过来的
    ...

---

完整示例

main.py

import logging
import os

from dotenv import load_dotenv

from bot import FeishuBot, load_plugins
from bot.ext.commands import CommandsExtension

load_dotenv()
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)

bot = FeishuBot(
    app_id=os.environ["APPID"],
    app_secret=os.environ["APPSEC"],
)

cmds = CommandsExtension(prefix="/")
bot.install(cmds)

load_plugins(bot, cmds=cmds)

if __name__ == "__main__":
    bot.run()

plugins/echo.py

import time
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from bot.context import MessageContext

def register(bot, cmds=None):
    @bot.use()
    async def logging_mw(ctx: MessageContext, next):
        start = time.time()
        print(f"→ [{ctx.chat_type}] {ctx.sender_name}: {ctx.text}")
        await next()
        print(f"← ({time.time() - start:.3f}s)")

    @bot.on_message(message_type="text")
    async def echo_text(ctx: MessageContext):
        await ctx.reply(ctx.text)

    if cmds is not None:
        @cmds.command("ping")
        async def ping(ctx: MessageContext, args):
            await ctx.reply("Pong!")

        @cmds.command("help")
        async def help_cmd(ctx: MessageContext, args):
            await ctx.reply(cmds.generate_help())

        @cmds.match(r"^(hi|hello)\b")
        async def greet(ctx: MessageContext, match):
            await ctx.reply("你好！")

---

高级用法

直接使用 FeishuChannel

bot.channel 暴露了完整的 FeishuChannel API：

# 发送 markdown
await bot.channel.send(chat_id, {"markdown": "**粗体** 和 [链接](https://...)"})

# 发送卡片
await bot.channel.send(chat_id, {
    "config": {"wide_screen_mode": True},
    "header": {"title": {"tag": "plain_text", "content": "标题"}},
    "elements": [
        {"tag": "markdown", "content": "卡片内容"}
    ]
})

# 流式输出（打字机效果）
await bot.channel.stream(chat_id, {"markdown": my_async_generator})

# 下载图片
data = await bot.channel.download_resource(file_key, "image")

# 编辑消息
await bot.channel.edit_message(message_id, "新内容")

自定义扩展

任何实现了 _install(bot) 方法的对象都可以作为扩展安装：

class MyExtension:
    def __init__(self, config):
        self.config = config

    def _install(self, bot):
        @bot.on_message(priority=50)
        async def my_handler(ctx):
            # 自定义处理逻辑
            pass

bot.install(MyExtension(config={}))

自定义 FeishuChannel 参数

FeishuBot 的 **channel_kwargs 透传给 FeishuChannel：

from lark_oapi.core.enum import LogLevel

bot = FeishuBot(
    app_id="...",
    app_secret="...",
    log_level=LogLevel.DEBUG,
    domain="https://open.larksuite.com",  # 海外版 Lark
)

---

事件流全景

用户发送消息
    ↓
飞书服务器推送事件
    ↓
WebSocket 长连接接收
    ↓
FeishuChannel 自动处理
    ├── 消息解析（19 种类型 → InboundMessage）
    ├── 去重（WS 重传保护）
    ├── 策略检查（黑白名单、@门控）
    └── 安全管道（批处理、串行队列）
    ↓
FeishuBot._on_message(msg)
    ↓
构建 MessageContext(ctx)
    ↓
中间件链（洋葱模型）
    ↓
处理器匹配
    ├── priority=100 → CommandsExtension（命令/正则）
    ├── priority=20  → 群聊 @处理器
    ├── priority=10  → 通用处理器（默认）
    └── ...穿透...
    ↓
handler 执行 → ctx.reply() / ctx.send()
    ↓
FeishuChannel.send() → 飞书 API → 消息送达