Metadata-Version: 2.4
Name: wings-bot
Version: 0.1.0
Summary: 基于 lark-oapi FeishuChannel 的装饰器风格飞书 Bot 开发框架
Project-URL: Homepage, https://github.com/isyuah/FeishuBot
Author: isyuah
License-Expression: MIT
Keywords: bot,chatbot,feishu,framework,lark
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Requires-Python: >=3.11
Requires-Dist: croniter>=6.2.2
Requires-Dist: lark-oapi>=1.6.5
Requires-Dist: python-dotenv>=1.2.2
Description-Content-Type: text/markdown

# 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. 环境准备

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

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

### 2. 配置飞书应用

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

```env
APPID=cli_xxxxxxxxxxxxx
APPSEC=xxxxxxxxxxxxxxxxxxxxxxxx
```

确保应用已开启以下能力：
- 机器人能力（Bot）
- 事件订阅 → 消息接收（`im.message.receive_v1`）
- 事件订阅方式选择「使用长连接接收」

### 3. 启动

```bash
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

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

```python
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

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

```python
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` 字段判断：

```python
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()

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

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

#### 过滤条件

```python
# 仅处理私聊文本消息
@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` 实现，以指定前缀解析文本消息为命令。

### 安装

```python
from bot.ext.commands import CommandsExtension

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

### @cmds.command()

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

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

#### 别名

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

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

#### 自动帮助

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

### @cmds.match()

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

```python
@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。

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

### 中断链

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

```python
@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` 传递数据给下游：

```python
@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()

```python
# 同步函数
@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 已重连 | 无参数 |

```python
@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`）

### 编写插件

```python
# 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"}
        )
```

### 自动加载

```python
# main.py 中
from bot import load_plugins

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

### 禁用插件

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

### 传递额外参数

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

```python
# 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

```python
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

```python
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：

```python
# 发送 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)` 方法的对象都可以作为扩展安装：

```python
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`：

```python
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 → 消息送达
```
