diff --git a/2.Design/API.md b/2.Design/API.md index c3a07db..7d4f5d6 100644 --- a/2.Design/API.md +++ b/2.Design/API.md @@ -699,78 +699,69 @@ Query: ## 6.5 外部接入(微信 clawbot) -微信里那个机器人把消息 POST 到本服务,消息就变成任务或知识草稿。 -**这一组接口不带登录态**:机器人不会登我们的账号,它只照着配置好的地址发。 +微信里那个机器人连上本服务之后,发一句话就变成任务或知识草稿。 +走的是腾讯 iLink Bot 协议:**扫码登录 → 后端长轮询收消息 → 用 context_token 回复**。 +所以这一组里只有「扫码」是免登录的(机器人自己不带我们的令牌),其余都要工作区 admin 以上。 -### 6.5.1 建通道(工作区 admin 以上) +### 6.5.1 扫码与通道(工作区 admin 以上) | 方法 | 路径 | 说明 | |---|---|---| -| GET | `/workspaces/{ws}/integrations` | 列出这个工作区的通道 | -| POST | `/workspaces/{ws}/integrations` | `{"name":"微信"}`,**响应里带一次性令牌** | -| PATCH | `/workspaces/{ws}/integrations/{id}` | `{"name"?}`、`{"status":"active"|"disabled"}` | -| POST | `/workspaces/{ws}/integrations/{id}/rotate` | 换令牌,旧地址立刻失效 | -| DELETE | `/workspaces/{ws}/integrations/{id}` | 删通道(历史消息级联删掉) | -| GET | `/workspaces/{ws}/integrations/{id}/messages?limit=20` | 最近的收信记录 | +| GET | `/workspaces/{ws}/integrations` | 列出通道(含 `status`、`connected`、`polling`、`last_seen_at`、`last_error`) | +| POST | `/workspaces/{ws}/integrations/wechat/qr` | 取一张二维码,返回 `{channel_id, qrcode_id, payload, image, expires_in}` | +| GET | `/workspaces/{ws}/integrations/wechat/qr/{qrcode_id}` | 问扫码结果:`waiting`/`scaned`/`confirmed`/`expired` | +| PATCH | `/workspaces/{ws}/integrations/{id}` | `{"status":"active"|"disabled"}`,停用就停轮询 | +| DELETE | `/workspaces/{ws}/integrations/{id}?confirm=true` | 断开:清掉登录态与游标,回到 `pending` | +| GET | `/workspaces/{ws}/integrations/{id}/messages?limit=5` | 最近收到的消息与它变成了什么 | -创建响应: +取二维码的响应(`image` 是可以直接塞进 `` 的 data URL,本地渲染,不是图片地址): ```json { - "code": 0, "message": "ok", - "data": { - "channel": { "id": "…", "kind": "wechat_clawbot", "name": "微信", "status": "active", - "last_seen_at": null, "created_at": "2026-11-05T14:02:11Z" }, - "token": "lS7…(43 字符)", - "webhook_url": "https://example.com/api/v1/integrations/clawbot/lS7…" - } + "channel_id": "4f0a…", + "qrcode_id": "30c97db83291e779217cbe20c5373e91", + "payload": "https://liteapp.weixin.qq.com/q/7GiQu1?qrcode=30c97…&bot_type=3", + "image": "data:image/svg+xml;base64,PD94bWw…", + "expires_in": 300 } ``` -> 令牌只在这里出现一次。库里存的是 `sha256(令牌)`,页面刷新之后连管理员也看不到; -> 丢了就 `rotate` 重新生成。 +扫码成功那一次(`status == "confirmed"`)除了回 `status`,还带 `channel`:登录态已加密存进 +`bot_channel.bot_token_enc`(Fernet,密钥由 `JWT_SECRET` 派生),**不会回给任何接口**,轮询任务同时被拉起。 -### 6.5.2 收消息(clawbot → 本服务) +`status` 的四个值:`pending`(还没连)、`active`(连上了、正在收)、`disabled`(管理员停用)、 +`expired`(登录态失效,需要重新扫码)。`polling` 说明后台任务此刻是否在跑。 -| 方法 | 路径 | 说明 | -|---|---|---| -| GET | `/integrations/clawbot/{token}?echostr=xxx` | 地址校验:有 `echostr` 就原样回纯文本 | -| POST | `/integrations/clawbot/{token}` | 收一条消息,回 `{"reply": "...", "action": "...", "duplicate": false}` | +### 6.5.2 消息怎么变成事项(不分接口,轮询里做) -请求体**宽松解析**(不同机器人的壳子不一样),按顺序找第一个非空字符串: +给机器人发的话这样理解: -- 正文:`text` / `content` / `msg` / `message` / `query` / `body` / `data`(嵌套对象也会往里找 `text`、`content`、`body`、`message`) -- 消息 id:`msg_id` / `message_id` / `msgid` / `id` / `uuid` / `event_id` -- 发送者:`from` / `sender` / `from_user` / `user` / `uid` / `sender_id` - -正文的写法: - -| 开头 | 结果 | +| 开头 | 落成什么 | |---|---| -| 直接说 | 任务(`status=inbox`) | -| `任务:` / `待办:` / `todo:` | 任务 | -| `知识:` / `笔记:` / `note:` / `kb:` | 知识草稿(`doc_type=note`、`status_hint=draft`) | -| `?` / `帮助` / `help` / `用法` | 只回一段用法说明,不建事项 | +| `任务:`/`待办:`/`todo:` 或不带前缀 | 任务(`status = "inbox"`) | +| `知识:`/`笔记:`/`note:`/`kb:` | 知识草稿(`status = "draft"`,`doc_type = "note"`) | +| `帮助`/`?`/`help` | 只回一段用法,不建任何东西 | -首行是标题(截到 180 字),其余行是正文;标题与正文里的 `#标签` 会被摘出来挂到事项上 -(纯数字的 `#123` 不算标签,避免把编号当成标签)。 +正文第一行当标题(超过 180 字截断),其余当正文;`#标签` 会被摘出来加到事项上(纯数字的不算标签)。 +建完之后会用 `context_token` 在微信里回一句,附上链接(`PUBLIC_BASE_URL` 或 CORS 里的第一个来源)。 +同一条消息(按 `message_id`)重复推送只会落一次,重推直接跳过。 -**幂等**:带 `msg_id` 的请求重复到达只处理一次,第二次直接回上次的 `reply` 且 `duplicate: true` -——微信会重推,重推不能变成两条任务。 +图片/文件/视频目前只在 `bot_message.raw` 里留痕,正文不搬;语音如果服务端给了转写就用转写文本。 ### 6.5.3 错误码 -| code | HTTP | 场景 | +| 场景 | code | HTTP | |---|---|---| -| 9002 | 404 | 令牌不存在或已被换掉;对外一律"通道不存在",不区分工作区 | -| 9003 | 403 | 通道被停用(`status=disabled`) | +| 取二维码 / 问状态时微信返回错误 | `9002` | 404 | +| 通道不存在、或不属于这个工作区 | `9002` | 404 | +| 登录态失效(微信 errcode `-14`) | —(通道转 `expired`,接口正常返回) | 200 | -### 6.5.4 租户隔离 +### 6.5.4 表与隔离 + +`bot_channel`(通道、登录态、游标)与 `bot_message`(收到的消息留痕)都挂 `workspace_id`, +走标准 RLS。轮询任务从 `channel_id` 起步、此刻还没有工作区上下文,所以迁移 `0014` +额外给了一条 `app.bot_channel` 的按主键口子(permissive 策略,与原隔离策略 OR)。 -`bot_channel` / `bot_message` 都带 `workspace_id NOT NULL` + RLS。收消息时还不知道它属于哪个 -工作区,所以 `bot_channel` 的策略额外允许 `token_hash = current_setting('app.bot_token_hash')`: -请求先把 `app.bot_token_hash` 设成本次令牌的哈希,策略只放行命中那一行,其余行照旧看不见。 -读到通道之后再 `bind_tenant()` 进这个工作区的上下文,任务是按**通道创建者**记账的。 ## 7. 安卓发布与升级 `/app` diff --git a/2.Design/ROADMAP-M0-M1.md b/2.Design/ROADMAP-M0-M1.md index 3a2e7f6..2f28c0a 100644 --- a/2.Design/ROADMAP-M0-M1.md +++ b/2.Design/ROADMAP-M0-M1.md @@ -749,62 +749,81 @@ ### 13.2 决定 -1. **一条通道一张令牌,令牌在 URL 里**。机器人不会登我们的账号;它只会照着配置好的地址 - POST。令牌是 32 字节随机串,库里只存 `sha256`,页面上"创建/重新生成"时显示一次。 -2. **一个工作区可以有多个通道**,各自能单独停用。塞进 `system_setting` 的 JSON 里就做不到 - "停这个、留那个"。 -3. **收消息用宽松解析**。clawbot 的消息壳子没有公开的固定 schema,与其猜错就 422, - 不如按常见字段名依次摸一遍,摸不到正文就记一条 `ignored` 留痕。 -4. **同一 `msg_id` 只处理一次**。微信会重推,重推变成第二条任务是最难解释的脏数据。 -5. **每条消息都留痕**(`bot_message`,含原始 JSON、处理结果、 reply)。 +1. **走腾讯官方 iLink Bot 协议**:`get_bot_qrcode` 拿二维码 → 手机扫码确认 → + 拿到 `bot_token` → 后端**长轮询** `getupdates` 收消息 → 用 `context_token` 回话。 + 这是 clawbot / openclaw / dsh-im 走的同一条路,扫码是配对的前提。 +2. **不依赖 DSH**:协议自己实现(`backend/app/services/wechat_ilink.py`), + 因为达道Task 是独立项目,不该为了收消息去依赖另一个产品的运行环境。 +3. **登录态加密存库**:`bot_channel.bot_token_enc` 用 Fernet(密钥由 `JWT_SECRET` 派生), + `bot_token` 不落明文、不回调给任何接口、日志里也没有。 +4. **轮询者是我们自己**:每个已连接的通道一个 asyncio 任务,进程启动时按库里的 + `status='active'` 恢复;token 失效(微信 errcode `-14`)就把通道转 `expired`,让用户重新扫。 +5. **一条消息只落一次**:按 `message_id` 去重,微信重推不会变成第二条任务。 +6. **每条消息都留痕**(`bot_message`,含文本、动作、item_id、回复、raw), 出问题时能直接回答"那条任务到底进没进"。 +7. **一个工作区一条微信通道**:同一时间只允许一条 `kind='wechat_clawbot'` 的记录承载登录态; + 断开就是清登录态回到 `pending`,再连就重新扫。 -### 13.3 后端 +### 13.3 协议要点(实现对照) + +| 用途 | 请求 | +|---|---| +| 取二维码 | `POST https://ilinkai.weixin.qq.com/ilink/bot/get_bot_qrcode?bot_type=3` → `{qrcode, qrcode_img_content}` | +| 问扫码结果 | `POST /ilink/bot/get_qrcode_status?qrcode=` → `waiting`/`scaned`/`confirmed`/`expired` | +| 收消息 | `POST /ilink/bot/getupdates` body `{get_updates_buf, base_info:{channel_version}}`,长轮询 70s | +| 回消息 | `POST /ilink/bot/sendmessage` body `{msg:{to_user_id, context_token, item_list:[{type:1,text_item:{text}}]}}` | + +请求头固定带 `iLink-App-ClientVersion: 1` 与 `X-WECHAT-UIN`(随机数 base64); +登录后加 `Authorization: Bearer ` + `AuthorizationType: ilink_bot_token`。 +二维码不是图片地址,`qrcode_img_content` 是待渲染的 payload,服务端用 `qrcode` 库画成 SVG data URL。 + +### 13.4 后端 | 位置 | 内容 | |---|---| -| `backend/app/models_integrations.py` | `BotChannel`(令牌哈希、状态、最后收信时间)、`BotMessage`(文本、动作、item_id、reply、原始 JSON) | -| `backend/alembic/versions/0012_bot_channel.py` | 建两张表 + RLS;`bot_channel` 策略额外放行 `token_hash = current_setting('app.bot_token_hash')` | -| `backend/app/api/integrations.py` | 收消息 `GET/POST /integrations/clawbot/{token}`;管理接口 `/workspaces/{ws}/integrations` | -| `backend/app/errors.py` | `9002` 令牌无效、`9003` 通道停用 | +| `backend/app/services/wechat_ilink.py` | 只封协议:取码、问状态、长轮询、回话、消息解析、token 加解密 | +| `backend/app/services/wechat_intake.py` | 文本 → 事项:前缀识别、`#标签`、标题/正文切分、建任务/知识、回复文案 | +| `backend/app/services/wechat_bot.py` | `BotManager`:起停轮询任务、去重、留痕、游标、`-14` 转 `expired` | +| `backend/app/api/integrations.py` | `POST /workspaces/{ws}/integrations/wechat/qr`、`GET …/qr/{id}`、`PATCH`、`DELETE`、`GET …/messages` | +| `backend/app/models_integrations.py` | `BotChannel`(`bot_token_enc`、`ilink_user_id`、`sync_buf`、`context_token`、`last_seen_at`、`last_error`)、`BotMessage` | +| `backend/alembic/versions/0013_wechat_bot_login.py` | 加登录态相关列,`status` 默认 `pending` | +| `backend/alembic/versions/0014_bot_channel_worker_policy.py` | 轮询入口没有工作区上下文,按 `app.bot_channel` 主键开一条 permissive 策略 | +| `backend/app/main.py` | lifespan 里 `manager.start_all()` / `manager.shutdown()` | 正文解析:首行是标题(180 字封顶),其余是正文;`#标签` 摘出来挂事项上(纯数字不算); -前缀 `知识:`/`笔记:` → 知识草稿,`任务:`/`todo:` → 任务,没有前缀就是任务。 +前缀 `知识:`/`笔记:` → 知识草稿,`任务:`/`todo:` → 任务,没有前缀就是任务,`帮助`/`?` 只回用法。 -### 13.4 Web +### 13.5 Web 新增 `/w/:wsId/integrations`「接入」页(左侧栏最后一项): -- 「怎么用」三步说明; -- 「新建通道」→ 一次性展示**回调地址 + 令牌**,各带复制按钮(clawbot 后台要粘的就是这里); -- 每条通道:状态、最后收信时间、`最近消息`(最近 5 条,带"建了任务/建了知识"标签)、 - 停用/启用、重新生成令牌、删除; -- 只有 owner/admin 看得到写操作。 +- 「扫码连接微信」→ 页面上直接画出二维码,每 2 秒问一次状态(待扫 → 已扫待确认 → 成功/过期); +- 连上之后显示通道卡片:接收中/登录已失效/已停用、最后收信时间、"微信用户 xxx"、 + `最近消息`(最近 5 条,标"建了任务/建了知识")、重新扫码、断开; +- 只有 owner/admin 能扫码与断开,其他成员只读。 ![接入页](screens/web-integrations.png) -### 13.5 验证 +### 13.6 验证 -- `backend/tests/test_clawbot.py` 11 例:纯文本成任务、`知识:`前缀成草稿并带标签、 - `帮助`只回复不建事项、错令牌 404/9002、停用通道 403/9003、同 `msg_id` 幂等、 - GET `echostr` 原样回、令牌只属于一个工作区、换令牌后旧地址失效、普通成员不能建通道、 - 回调地址必须带 `/api/v1`。 -- `web/e2e/integrations.mjs` 7 项(真跑):页面建通道 → 取回调地址 → 冒充 clawbot POST → - 任务出现在任务列表 → 重复推送不建第二条 → 「最近消息」看得到 → 控制台零报错。 +- `backend/tests/test_wechat_bot.py` 9 例:取二维码(SVG data URL、通道 `pending`)、 + 扫码确认后登录态加密入库且能解回、过期二维码回 `expired`、断开清登录态、 + 纯文本成任务、`知识:`前缀成草稿带标签、`帮助`不建事项、同一 `message_id` 只落一条并只回一次、 + 登录态失效后通道转 `expired` 且带原因。 +- `web/e2e/integrations.mjs` 7 项(真跑,拿的是微信真二维码):接入页打开 → 点扫码 → + 二维码是页面里的 SVG → 提示用微信扫 → 后端通道 `pending` → 没登录态不轮询 → 控制台零报错。 +- 全量 `pytest` 149 passed;`ruff check` / `ruff format --check` 干净。 -### 13.6 还没做 +### 13.7 教训:第一版是猜的 -- 只做了"进"(收消息建事项),没做"出"(改完任务回一条微信通知)。 -- 消息不认人:机器人只知道"这是这个工作区的消息",按通道创建者记账;要认人得先做微信身份绑定。 -- 语音、图片、文件只留痕不处理。 +第一版没找到资料,按"webhook + 令牌在 URL 里"自己编了一套(用户看到的回调地址那版)。 +用户当场指出"微信要先扫码配对",一查才发现协议在 +`~/.dsh/profiles/desktop/node_modules/@michengai/dsh-im-connect/src/channels/weixin.ts` 里写着。 +**教训:接入第三方协议,先去已有实现里找协议,别从直觉倒推。** 那一版的迁移 `0012` 保留 +(`bot_channel` / `bot_message` 两张表还在用),`token_hash` 列留着但不再使用。 -### 13.7 上生产时踩的坑 +### 13.8 还没做 -回调地址一开始用 `request.base_url` 拼,本地没问题(Vite 把 `/api/v1` 代理掉了), -生产上应用挂在域名根、前缀只在路由里,拼出来的地址少了 `/api/v1`,clawbot 打过去是 `405`。 -现在从 `base_url` 里截站点根、前缀一律取 `settings.api_prefix`,并补了一条断言钉住。 - -生产实测(工作区「我的备忘」):`POST /api/v1/integrations/clawbot/` -→ `200 code=0 action=task`,任务列表里立刻出现一条。 - -![生产接入页](screens/prod-integrations.png) +- 只做了"进"(收消息建事项),没做"出"(任务有进展回一条微信通知)。 +- 消息不认人:按通道创建者记账;要认人得先做微信身份绑定。 +- 语音只在服务端给了转写时使用,图片/文件/视频只留痕不搬。 diff --git a/2.Design/screens/web-integrations.png b/2.Design/screens/web-integrations.png index 4915267..79eafa6 100644 Binary files a/2.Design/screens/web-integrations.png and b/2.Design/screens/web-integrations.png differ diff --git a/backend/alembic/versions/0013_wechat_bot_login.py b/backend/alembic/versions/0013_wechat_bot_login.py new file mode 100644 index 0000000..0f6f83d --- /dev/null +++ b/backend/alembic/versions/0013_wechat_bot_login.py @@ -0,0 +1,63 @@ +"""微信 clawbot 改成"扫码登录 + 长轮询" + +Revision ID: 0013 +Revises: 0012 +Create Date: 2026-11-05 + +上一版(0012)做成了"clawbot 往我们这 POST webhook"。真实协议是腾讯 iLink Bot: +我们先扫码登录拿 `bot_token`,再由我们长轮询 `getupdates` 拉消息(见 +`app/services/wechat_ilink.py` 文件头的出处说明)。所以这张表要补登录态与游标: + +- `bot_token_enc`:加密后的登录态(密钥由 JWT_SECRET 派生),换 secret 就要重新扫码; +- `ilink_user_id`:扫码的微信号,界面上显示"谁扫的"; +- `base_url`:服务端可能下发新基址; +- `sync_buf`:长轮询游标,丢了会重复收消息; +- `context_token`:回复必须带对方上一条消息的 context_token。 + +`bot_channel.status` 增加 `pending`(等扫码)与 `expired`(登录态失效)两个取值; +`token_hash` 是上一版遗留列,改成可空、不再使用。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +from app.roles import APP_ROLE + +revision: str = "0013" +down_revision: str = "0012" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + +STATEMENTS = ( + "ALTER TABLE bot_channel ADD COLUMN IF NOT EXISTS bot_token_enc text", + "ALTER TABLE bot_channel ADD COLUMN IF NOT EXISTS ilink_user_id varchar(64)", + "ALTER TABLE bot_channel ADD COLUMN IF NOT EXISTS base_url text", + "ALTER TABLE bot_channel ADD COLUMN IF NOT EXISTS sync_buf text", + "ALTER TABLE bot_channel ADD COLUMN IF NOT EXISTS context_token text", + "ALTER TABLE bot_channel ALTER COLUMN token_hash DROP NOT NULL", + "ALTER TABLE bot_channel ALTER COLUMN status SET DEFAULT 'pending'", + "COMMENT ON COLUMN bot_channel.bot_token_enc IS 'iLink 登录态,Fernet 加密(密钥由 JWT_SECRET 派生)'", + "COMMENT ON COLUMN bot_channel.sync_buf IS 'getupdates 游标,丢了会重复收消息'", + f"GRANT SELECT, INSERT, UPDATE, DELETE ON bot_channel TO {APP_ROLE}", +) + +DOWN_STATEMENTS = ( + "ALTER TABLE bot_channel DROP COLUMN IF EXISTS context_token", + "ALTER TABLE bot_channel DROP COLUMN IF EXISTS sync_buf", + "ALTER TABLE bot_channel DROP COLUMN IF EXISTS base_url", + "ALTER TABLE bot_channel DROP COLUMN IF EXISTS ilink_user_id", + "ALTER TABLE bot_channel DROP COLUMN IF EXISTS bot_token_enc", +) + + +def upgrade() -> None: + for statement in STATEMENTS: + op.execute(statement) + + +def downgrade() -> None: + for statement in DOWN_STATEMENTS: + op.execute(statement) diff --git a/backend/alembic/versions/0014_bot_channel_worker_policy.py b/backend/alembic/versions/0014_bot_channel_worker_policy.py new file mode 100644 index 0000000..56a8dda --- /dev/null +++ b/backend/alembic/versions/0014_bot_channel_worker_policy.py @@ -0,0 +1,46 @@ +"""轮询任务读通道需要一条 RLS 口子 + +Revision ID: 0014 +Revises: 0013 +Create Date: 2026-11-05 + +后台轮询是"从 channel_id 开始"的:它还没有工作区上下文,而 `bot_channel` 的策略要求 +`workspace_id = app_ws_id()`,直接 SELECT 会一行都看不到(先有鸡还是先有蛋)。 + +和 0012 那条 `token_hash = app_bot_token_hash()` 同理,这里再放一条**只按主键命中自己** +的口子:任务在读任何东西之前先把 `app.bot_channel` 设成本次要处理的通道 id, +策略允许它看见那一行,其余行照旧不可见。读到 workspace_id 之后立刻 `bind_tenant`, +后面所有查询都回到正常的租户上下文里。 + +两条策略是 permissive 关系(PG 会 OR 起来),所以原来的隔离策略不用改。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +from app.roles import APP_ROLE + +revision: str = "0014" +down_revision: str = "0013" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + +CREATE_POLICY = """ +CREATE POLICY bot_channel_worker ON bot_channel + FOR ALL + USING (id = current_setting('app.bot_channel', true)::uuid) + WITH CHECK (id = current_setting('app.bot_channel', true)::uuid) +""" + + +def upgrade() -> None: + op.execute("DROP POLICY IF EXISTS bot_channel_worker ON bot_channel") + op.execute(CREATE_POLICY) + op.execute(f"GRANT SELECT, INSERT, UPDATE, DELETE ON bot_channel TO {APP_ROLE}") + + +def downgrade() -> None: + op.execute("DROP POLICY IF EXISTS bot_channel_worker ON bot_channel") diff --git a/backend/app/api/integrations.py b/backend/app/api/integrations.py index fdea5eb..728ab6d 100644 --- a/backend/app/api/integrations.py +++ b/backend/app/api/integrations.py @@ -1,407 +1,62 @@ -"""外部接入:微信(clawbot)发来的消息变成任务 / 知识。 +"""外部接入:微信 clawbot 的扫码登录与状态。 -## 一条消息是怎么变成事项的 +真实协议是腾讯 iLink Bot(出处见 `app/services/wechat_ilink.py` 文件头): +**我们**去取二维码、等人扫、拿到登录态后长轮询拉消息。所以这里的接口是三件事: -微信里发一句话给机器人,clawbot 把消息 POST 到 +1. `POST /workspaces/{ws}/integrations/wechat/qr` —— 取一张二维码,页面显示给人扫; +2. `GET /workspaces/{ws}/integrations/wechat/qr/{id}` —— 问"扫了吗",扫成功就存登录态并开始轮询; +3. `DELETE /workspaces/{ws}/integrations/{channel_id}` —— 断开:登录态作废、轮询停止。 - POST /api/v1/integrations/clawbot/{token} +二维码 id 放 Redis 而不是内存:问状态的那一下可能落到另一个进程, +存内存就会"扫完了但没人知道"。key 带工作区,两个工作区互不踩。 -令牌在 URL 里,**不带登录态**:机器人不会登我们的账号,它只会照着配置好的 -地址发。令牌是 32 字节随机串,库里只存 sha256。 - -正文按前缀决定去哪个池子: - - 知识:xxx / 笔记:xxx / note: xxx → 知识草稿 - 任务:xxx / todo: xxx → 任务(默认也是任务) - ?、帮助、help → 回一段用法说明 - -首行是标题,其余行是正文。首行里的 `#标签` 会被摘出来挂到事项上 -(标签是知识的主要检索入口,手机上敲#比打开网页加标签省事)。 - -## 为什么每条消息都留痕 - -微信会重推(网络抖动、机器人重启)。没有留痕就只能靠猜"那条任务到底进没进"。 -`bot_message.external_id` 上有个部分唯一索引:同一个外部消息 id 第二次到达时 -直接回上次的答案,不会变成两条任务。 - -## 安全边界 - -- 令牌错 → 9002(对外一律"通道不存在",不区分"错令牌"和"这个工作区没接")。 -- 通道被停用 → 9003。 -- 消息里不认人:机器人只知道"这是这个工作区的一条消息",落库按通道创建者记账, - 不会因为消息内容去猜是谁发的。 +上一版这里做的是 webhook 接收端(`POST /integrations/clawbot/{token}`), +对 clawbot 的理解是错的,接口与令牌一并删掉了(迁移 0013 的注释记了原因)。 """ from __future__ import annotations -import hashlib -import secrets +import base64 +import io import uuid -from datetime import datetime, timezone -from typing import Annotated, Any +from typing import Any -from fastapi import APIRouter, Body, Query, Request -from fastapi.responses import PlainTextResponse +import qrcode +import qrcode.image.svg +from fastapi import APIRouter, Query +from pydantic import BaseModel from sqlalchemy import select -from sqlalchemy.exc import IntegrityError -from ..config import get_settings from ..db import bind_tenant from ..deps import AdminDep, SessionDep from ..errors import AppError, ErrorCode -from ..models import Member, Workspace from ..models_integrations import BotChannel, BotMessage -from ..models_knowledge import KnowledgeDoc -from ..models_task import Item from ..schemas import ok -from .items import _log, _set_tags +from ..services import wechat_ilink +from ..services.wechat_bot import manager +from ..stores import get_store -webhook_router = APIRouter(prefix="/integrations", tags=["外部接入"]) admin_router = APIRouter(prefix="/workspaces", tags=["外部接入"]) -# 标题在 item.title 上是 varchar(200),留点余量给"…"之类的后缀。 -MAX_TITLE = 180 -MAX_BODY = 20000 - -KNOWLEDGE_PREFIXES = ("知识", "笔记", "note", "knowledge", "kb") -TASK_PREFIXES = ("任务", "待办", "todo", "task") -HELP_PREFIXES = ("?", "?", "帮助", "help", "用法") - -HELP_TEXT = ( - "发什么都能用,写法:\n" - "· 直接说 → 记成任务\n" - "· 知识:内容 → 记成知识草稿(也可以写「笔记:」)\n" - "· 任务:内容 → 明确记成任务\n" - "首行当标题,后面几行当正文;标题里带 #标签 会自动加上。" -) +# iLink 的二维码大约 3 分钟过期,我们按 5 分钟缓存 id,过期就重新取。 +QR_TTL_SECONDS = 300 -def hash_token(token: str) -> str: - return hashlib.sha256(token.encode("utf-8")).hexdigest() +class ChannelPatch(BaseModel): + status: str -def new_token() -> str: - return secrets.token_urlsafe(32) +def _qr_key(ws_id: uuid.UUID) -> str: + return f"wechat:qr:{ws_id}" -def _pick(data: dict[str, Any], *keys: str) -> str | None: - """从任意形状的 payload 里挖字段(clawbot 的消息壳子没定死,宽松一点)。""" - for key in keys: - value = data.get(key) - if isinstance(value, str) and value.strip(): - return value - if isinstance(value, dict): - nested = _pick(value, "text", "content", "body", "message") - if nested: - return nested - return None - - -def extract_text(payload: dict[str, Any]) -> str: - text = _pick( - payload, - "text", - "content", - "msg", - "message", - "query", - "body", - "data", - "payload", - ) - return (text or "").strip() - - -def extract_external_id(payload: dict[str, Any]) -> str | None: - value = _pick(payload, "msg_id", "message_id", "msgid", "id", "uuid", "event_id") - return value[:128] if value else None - - -def extract_sender(payload: dict[str, Any]) -> str | None: - value = _pick(payload, "from", "sender", "from_user", "user", "uid", "sender_id") - return value[:128] if value else None - - -def split_command(text: str) -> tuple[str, str]: - """返回 (kind, 去掉前缀后的正文)。""" - head, _, rest = text.partition("\n") - probe = head.strip().lower() - for prefix in HELP_PREFIXES: - if probe == prefix or probe.startswith(f"{prefix}:") or probe.startswith(f"{prefix}:"): - return "help", rest - for prefix in KNOWLEDGE_PREFIXES: - for sep in (":", ":", " ", "\u3000"): - if probe.startswith(f"{prefix}{sep}"): - offset = len(prefix) + len(sep) - return "knowledge", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip() - for prefix in TASK_PREFIXES: - for sep in (":", ":", " ", "\u3000"): - if probe.startswith(f"{prefix}{sep}"): - offset = len(prefix) + len(sep) - return "task", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip() - return "task", text - - -def split_tags(body: str) -> tuple[str, list[str]]: - """把 `#标签` 摘出来。只认行首和空白后的 #,免得把 "C#" 吃掉。""" - tags: list[str] = [] - kept: list[str] = [] - for line in body.splitlines(): - words = line.split(" ") - rest: list[str] = [] - for word in words: - if word.startswith("#") and len(word) > 1 and not word[1:].isdigit(): - name = word[1:].strip(",,。.;;") - if name and name.lower() not in {t.lower() for t in tags}: - tags.append(name) - continue - rest.append(word) - kept.append(" ".join(rest).rstrip()) - return "\n".join(kept).strip(), tags - - -def split_title(body: str) -> tuple[str, str]: - lines = body.splitlines() - if not lines: - return "", "" - title = lines[0].strip() - rest = "\n".join(lines[1:]).strip() - return title[:MAX_TITLE], rest[:MAX_BODY] - - -async def _creator_id(session: SessionDep, channel: BotChannel) -> uuid.UUID: - """消息记在谁名下:通道创建者;人没了就退回工作区 owner。""" - if channel.created_by is not None: - row = await session.execute( - select(Member.id).where( - Member.workspace_id == channel.workspace_id, - Member.user_id == channel.created_by, - Member.status == "active", - ) - ) - if row.first() is not None: - return channel.created_by - workspace = await session.get(Workspace, channel.workspace_id) - if workspace is None: # 理论上不可能:FK 是 CASCADE - raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在") - return workspace.owner_id - - -async def _create_task( - session: SessionDep, - *, - ws_id: uuid.UUID, - creator_id: uuid.UUID, - title: str, - body: str, - tags: list[str], -) -> Item: - item = Item( - workspace_id=ws_id, - kind="task", - title=title, - body_md=body, - status="inbox", - priority=0, - creator_id=creator_id, - ) - session.add(item) - await session.flush() - if tags: - await _set_tags(session, ws_id, item, tags) - await _log(session, item, creator_id, "created", {"via": "wechat_clawbot"}) - return item - - -async def _create_knowledge( - session: SessionDep, - *, - ws_id: uuid.UUID, - creator_id: uuid.UUID, - title: str, - body: str, - tags: list[str], -) -> Item: - item = Item( - workspace_id=ws_id, - kind="knowledge", - title=title, - body_md=body, - status="draft", - creator_id=creator_id, - ) - session.add(item) - await session.flush() - doc = KnowledgeDoc( - item_id=item.id, - workspace_id=ws_id, - author_id=creator_id, - source_item_id=None, - source_kind="manual", - doc_type="note", - status_hint="draft", - generated_by="human", - ) - session.add(doc) - await session.flush() - if tags: - await _set_tags(session, ws_id, item, tags) - await _log(session, item, creator_id, "knowledge_created", {"via": "wechat_clawbot"}) - return item - - -def _link(request: Request, ws_id: uuid.UUID, item: Item) -> str: - base = str(request.base_url).rstrip("/") - # base_url 形如 http://host:8200/api/v1/,而页面在站点根上 - base = base.split("/api/")[0] - if item.kind == "knowledge": - return f"{base}/w/{ws_id}/knowledge/{item.id}" - return f"{base}/w/{ws_id}?open={item.id}" - - -async def _channel_by_token(session: SessionDep, token: str) -> BotChannel: - from sqlalchemy import text as sql_text - - # 策略要靠这个值才放行那一行,必须先设。 - await session.execute( - sql_text("SELECT set_config('app.bot_token_hash', :h, true)"), {"h": hash_token(token)} - ) - row = ( - await session.execute(select(BotChannel).where(BotChannel.token_hash == hash_token(token))) - ).scalar_one_or_none() - if row is None: - raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在或令牌已失效") - return row - - -@webhook_router.get("/clawbot/{token}") -async def clawbot_verify( - token: str, session: SessionDep, echostr: str | None = Query(default=None) -): - """clawbot 后台通常先 GET 一次确认地址是活的:有 echostr 就原样回。""" - channel = await _channel_by_token(session, token) - if channel.status != "active": - raise AppError(ErrorCode.BOT_DISABLED, "这个通道已停用") - if echostr is not None: - return PlainTextResponse(echostr) - return ok({"channel": channel.name, "status": channel.status}) - - -@webhook_router.post("/clawbot/{token}") -async def clawbot_message( - token: str, - request: Request, - session: SessionDep, - payload: Annotated[dict[str, Any] | None, Body()] = None, -): - channel = await _channel_by_token(session, token) - if channel.status != "active": - raise AppError(ErrorCode.BOT_DISABLED, "这个通道已停用") - - raw: dict[str, Any] = payload if isinstance(payload, dict) else {} - text = extract_text(raw) - external_id = extract_external_id(raw) - sender = extract_sender(raw) - - await bind_tenant(session, channel.workspace_id, channel.created_by) - - # 重推:同一个外部消息 id 再来一次,回上次那句,不再建第二条。 - if external_id: - previous = ( - await session.execute( - select(BotMessage).where( - BotMessage.channel_id == channel.id, - BotMessage.external_id == external_id, - ) - ) - ).scalar_one_or_none() - if previous is not None: - channel.last_seen_at = datetime.now(timezone.utc) - return ok({"reply": previous.reply, "action": previous.action, "duplicate": True}) - - message = BotMessage( - workspace_id=channel.workspace_id, - channel_id=channel.id, - external_id=external_id, - sender=sender, - text=text[:MAX_BODY], - action="ignored", - reply="", - raw=raw if isinstance(raw, dict) else {"value": raw}, - ) - - if not text: - message.action = "ignored" - message.reply = "没收到正文,只收到一个空壳。" - session.add(message) - return ok({"reply": message.reply, "action": "ignored"}) - - kind, body = split_command(text) - creator_id = await _creator_id(session, channel) - - if kind == "help": - message.action = "help" - message.reply = HELP_TEXT - else: - body, tags = split_tags(body) - title, rest = split_title(body) - if not title: - message.action = "ignored" - message.reply = "只有标签没有内容,这条没记。" - elif kind == "knowledge": - item = await _create_knowledge( - session, - ws_id=channel.workspace_id, - creator_id=creator_id, - title=title, - body=rest, - tags=tags, - ) - message.action = "knowledge" - message.item_id = item.id - link = _link(request, channel.workspace_id, item) - message.reply = f"已记成知识草稿:《{item.title}》。去核对:{link}" - else: - item = await _create_task( - session, - ws_id=channel.workspace_id, - creator_id=creator_id, - title=title, - body=rest, - tags=tags, - ) - message.action = "task" - message.item_id = item.id - message.reply = ( - f"已记成任务:《{item.title}》。打开:{_link(request, channel.workspace_id, item)}" - ) - - channel.last_seen_at = datetime.now(timezone.utc) - channel.last_error = None - try: - # 用 SAVEPOINT 兜住唯一索引冲突:另一个并发请求刚插了同一条消息, - # 这里只要认输回同一句就行,不能把整笔事务打挂。 - async with session.begin_nested(): - session.add(message) - await session.flush() - except IntegrityError: - if external_id: - previous = ( - await session.execute( - select(BotMessage).where( - BotMessage.channel_id == channel.id, - BotMessage.external_id == external_id, - ) - ) - ).scalar_one_or_none() - if previous is not None: - return ok({"reply": previous.reply, "action": previous.action, "duplicate": True}) - raise AppError(ErrorCode.BOT_TOKEN_INVALID, "同一条消息重复到达,已忽略") from None - - return ok({"reply": message.reply, "action": message.action, "duplicate": False}) - - -# ── 工作区里的管理接口(登录 + admin 以上) ───────────────────────────── +def render_qr_svg(payload: str) -> str: + """把二维码内容画成 data URL:服务端画,前端不用再引一个二维码库。""" + image = qrcode.make(payload, image_factory=qrcode.image.svg.SvgPathImage, box_size=10) + buffer = io.BytesIO() + image.save(buffer) + encoded = base64.b64encode(buffer.getvalue()).decode("ascii") + return f"data:image/svg+xml;base64,{encoded}" def _channel_out(channel: BotChannel) -> dict[str, Any]: @@ -410,122 +65,152 @@ def _channel_out(channel: BotChannel) -> dict[str, Any]: "kind": channel.kind, "name": channel.name, "status": channel.status, - "created_by": str(channel.created_by) if channel.created_by else None, + "connected": bool(channel.bot_token_enc) and channel.status == "active", + "polling": manager.running(channel.id), + "ilink_user_id": channel.ilink_user_id, "last_seen_at": channel.last_seen_at, "last_error": channel.last_error, "created_at": channel.created_at, } -def _webhook_url(request: Request, token: str) -> str: - """拼出机器人要打的那个地址。 - - 不能直接用 `request.base_url`:生产上应用挂在域名根、路由带 `/api/v1` 前缀, - base_url 里没有前缀,拼出来的地址会 405(本地被 Vite 代理掩盖过这个差异)。 - 所以站点根从 base_url 里截,前缀一律取 `settings.api_prefix`。 - """ - base = str(request.base_url).split("/api/")[0].rstrip("/") - return f"{base}{get_settings().api_prefix}/integrations/clawbot/{token}" +async def _wechat_channel(session: SessionDep, ws_id: uuid.UUID) -> BotChannel | None: + rows = await session.execute( + select(BotChannel) + .where(BotChannel.workspace_id == ws_id, BotChannel.kind == "wechat_clawbot") + .order_by(BotChannel.created_at.desc()) + ) + return rows.scalars().first() @admin_router.get("/{ws_id}/integrations") async def list_integrations(ws_id: uuid.UUID, ctx: AdminDep, session: SessionDep): - rows = ( - ( - await session.execute( - select(BotChannel) - .where(BotChannel.workspace_id == ws_id) - .order_by(BotChannel.created_at.desc()) - ) + rows = await session.execute( + select(BotChannel) + .where(BotChannel.workspace_id == ws_id) + .order_by(BotChannel.created_at.desc()) + ) + return ok({"items": [_channel_out(c) for c in rows.scalars().all()]}) + + +@admin_router.post("/{ws_id}/integrations/wechat/qr") +async def begin_wechat_login(ws_id: uuid.UUID, ctx: AdminDep, session: SessionDep): + """取一张二维码。已经登录过的通道会被重置成 pending(换号 = 重新扫)。""" + try: + code = await wechat_ilink.get_qrcode() + except wechat_ilink.WechatError as exc: + raise AppError(ErrorCode.BOT_TOKEN_INVALID, f"取二维码失败:{exc}") from exc + + channel = await _wechat_channel(session, ws_id) + if channel is None: + channel = BotChannel( + workspace_id=ws_id, + kind="wechat_clawbot", + name="微信", + status="pending", + created_by=ctx.user.id, ) - .scalars() - .all() - ) - return ok({"items": [_channel_out(row) for row in rows]}) - - -@admin_router.post("/{ws_id}/integrations") -async def create_integration( - ws_id: uuid.UUID, - ctx: AdminDep, - session: SessionDep, - request: Request, - body: Annotated[dict[str, Any] | None, Body()] = None, -): - name = str((body or {}).get("name") or "").strip()[:64] or "微信" - token = new_token() - channel = BotChannel( - workspace_id=ws_id, - kind="wechat_clawbot", - name=name, - token_hash=hash_token(token), - status="active", - created_by=ctx.user.id, - ) - session.add(channel) + session.add(channel) + if channel.bot_token_enc: + await manager.stop(channel.id) + channel.bot_token_enc = None + channel.sync_buf = None + channel.context_token = None + channel.status = "pending" + channel.last_error = None await session.flush() + + await get_store().set(_qr_key(ws_id), code.qrcode_id, QR_TTL_SECONDS) + await session.commit() + return ok( { - "channel": _channel_out(channel), - # 令牌只在这里出现一次,之后只能"重新生成"。 - "token": token, - "webhook_url": _webhook_url(request, token), + "channel_id": str(channel.id), + "qrcode_id": code.qrcode_id, + "payload": code.payload, + "image": render_qr_svg(code.payload), + "expires_in": QR_TTL_SECONDS, } ) +@admin_router.get("/{ws_id}/integrations/wechat/qr/{qrcode_id}") +async def poll_wechat_login(ws_id: uuid.UUID, qrcode_id: str, ctx: AdminDep, session: SessionDep): + """问扫码结果。只有 confirmed 才落登录态并开轮询,其余状态原样回给页面。""" + cached = await get_store().get(_qr_key(ws_id)) + if cached != qrcode_id: + # 页面拿着旧二维码在问:直接说"过期",让它重新取一张。 + return ok({"status": "expired"}) + + try: + status = await wechat_ilink.qrcode_status(qrcode_id) + except wechat_ilink.WechatError as exc: + return ok({"status": "error", "message": str(exc)}) + + if status.status != "confirmed" or not status.bot_token: + return ok({"status": status.status}) + + channel = await _wechat_channel(session, ws_id) + if channel is None: + raise AppError(ErrorCode.BOT_TOKEN_INVALID, "通道不见了,请重新取二维码") + + await bind_tenant(session, ws_id, ctx.user.id) + channel.bot_token_enc = wechat_ilink.encrypt_token(status.bot_token) + channel.ilink_user_id = status.ilink_user_id + channel.base_url = status.base_url + channel.status = "active" + channel.sync_buf = None + channel.context_token = None + channel.last_error = None + channel.created_by = channel.created_by or ctx.user.id + await session.commit() + await get_store().delete(_qr_key(ws_id)) + + await manager.start(channel.id) + return ok({"status": "confirmed", "channel": _channel_out(channel)}) + + @admin_router.patch("/{ws_id}/integrations/{channel_id}") async def update_integration( ws_id: uuid.UUID, channel_id: uuid.UUID, ctx: AdminDep, session: SessionDep, - body: Annotated[dict[str, Any] | None, Body()] = None, + body: ChannelPatch, ): channel = await session.get(BotChannel, channel_id) if channel is None or channel.workspace_id != ws_id: raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在") - data = body or {} - if isinstance(data.get("name"), str): - channel.name = data["name"].strip()[:64] - if isinstance(data.get("status"), str): - if data["status"] not in ("active", "disabled"): - raise AppError(ErrorCode.VALIDATION_ERROR, "状态只能是 active 或 disabled") - channel.status = data["status"] + wanted = body.status.strip() + if wanted not in ("active", "disabled"): + raise AppError(ErrorCode.VALIDATION_FAILED, "status 只能是 active 或 disabled") + if wanted == "active" and not channel.bot_token_enc: + raise AppError(ErrorCode.BOT_TOKEN_INVALID, "这个通道还没扫码登录") + channel.status = wanted + await session.commit() + if wanted == "active": + await manager.start(channel.id) + else: + await manager.stop(channel.id) return ok({"channel": _channel_out(channel)}) -@admin_router.post("/{ws_id}/integrations/{channel_id}/rotate") -async def rotate_integration( - ws_id: uuid.UUID, - channel_id: uuid.UUID, - ctx: AdminDep, - session: SessionDep, - request: Request, -): - channel = await session.get(BotChannel, channel_id) - if channel is None or channel.workspace_id != ws_id: - raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在") - token = new_token() - channel.token_hash = hash_token(token) - return ok( - { - "channel": _channel_out(channel), - "token": token, - "webhook_url": _webhook_url(request, token), - } - ) - - @admin_router.delete("/{ws_id}/integrations/{channel_id}") -async def delete_integration( +async def disconnect_integration( ws_id: uuid.UUID, channel_id: uuid.UUID, ctx: AdminDep, session: SessionDep ): + """断开:停轮询、清登录态。留痕(bot_message)不删,翻旧账用得上。""" channel = await session.get(BotChannel, channel_id) if channel is None or channel.workspace_id != ws_id: raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在") - await session.delete(channel) - return ok({"deleted": True}) + await manager.stop(channel.id) + channel.bot_token_enc = None + channel.sync_buf = None + channel.context_token = None + channel.status = "pending" + channel.last_error = None + await session.commit() + return ok({"channel": _channel_out(channel)}) @admin_router.get("/{ws_id}/integrations/{channel_id}/messages") @@ -534,36 +219,27 @@ async def list_messages( channel_id: uuid.UUID, ctx: AdminDep, session: SessionDep, - limit: int = Query(default=20, ge=1, le=100), + limit: int = Query(default=5, ge=1, le=100), ): - rows = ( - ( - await session.execute( - select(BotMessage) - .where(BotMessage.workspace_id == ws_id, BotMessage.channel_id == channel_id) - .order_by(BotMessage.created_at.desc()) - .limit(limit) - ) - ) - .scalars() - .all() + rows = await session.execute( + select(BotMessage) + .where(BotMessage.channel_id == channel_id, BotMessage.workspace_id == ws_id) + .order_by(BotMessage.created_at.desc()) + .limit(limit) ) return ok( { "items": [ { - "id": str(row.id), - "text": row.text, - "action": row.action, - "item_id": str(row.item_id) if row.item_id else None, - "reply": row.reply, - "sender": row.sender, - "created_at": row.created_at, + "id": str(m.id), + "action": m.action, + "sender": m.sender, + "text": m.text, + "reply": m.reply, + "item_id": str(m.item_id) if m.item_id else None, + "created_at": m.created_at, } - for row in rows + for m in rows.scalars().all() ] } ) - - -__all__ = ["webhook_router", "admin_router"] diff --git a/backend/app/config.py b/backend/app/config.py index 0d67ddb..128f9c0 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -145,6 +145,9 @@ class Settings(BaseSettings): # 生产没有 nginx 时由后端直接托管前端构建产物(同源,/api 无跨域)。 # 指向 web/dist 的绝对路径;留空则只提供 API。 web_dist_dir: str | None = None + # 回复微信消息时用的站点地址(形如 http://192.168.8.188:8200,不带 /api/v1)。 + # 留空则退回 cors_origins 的第一个,本机开发够用,生产务必显式配。 + public_base_url: str | None = None # ── 安卓安装包(OTA)────────────────────────────────── # 放 latest.json + 对应 apk 的目录;留空则用 backend/static/apk。 diff --git a/backend/app/main.py b/backend/app/main.py index 9614c9b..7a0264b 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -32,6 +32,7 @@ from .db import SessionFactory, current_app_role, engine from .errors import AppError, ErrorCode, http_status_for from .logging_setup import get_logger, request_id_ctx, setup_logging from .middleware import IdempotencyMiddleware, RequestContextMiddleware, error_body +from .services.wechat_bot import manager as wechat_manager settings = get_settings() setup_logging("DEBUG" if settings.env == "dev" else "INFO") @@ -50,7 +51,10 @@ async def lifespan(app: FastAPI): "服务启动", extra={"extra_fields": {"env": settings.env, "commit": settings.build_commit}}, ) + # 已登录的微信通道在进程启动时接上,不然重启之后消息就静默收不到了。 + await wechat_manager.start_all() yield + await wechat_manager.shutdown() await engine.dispose() @@ -128,7 +132,6 @@ app.include_router(attachments.router, prefix=settings.api_prefix) app.include_router(app_release.router, prefix=settings.api_prefix) app.include_router(settings_api.router, prefix=settings.api_prefix) app.include_router(integrations.admin_router, prefix=settings.api_prefix) -app.include_router(integrations.webhook_router, prefix=settings.api_prefix) @app.get("/health", tags=["运维"], summary="存活探针") diff --git a/backend/app/models_integrations.py b/backend/app/models_integrations.py index 760e9a3..2d7096b 100644 --- a/backend/app/models_integrations.py +++ b/backend/app/models_integrations.py @@ -1,13 +1,18 @@ -"""外部接入:微信(clawbot)这一类的机器人通道。 +"""外部接入:微信 clawbot 的机器人账号。 为什么单开一张表而不是塞进 `system_setting`: -1. 一个工作区可以挂多个通道(自己的号、公司的号),也随时可能换; +1. 一个工作区可能挂过好几个机器人(自己的号、公司的号),也随时可能换; 塞进 setting 的 JSON 里就没法"停用这一个、保留那一个"。 2. 通道是**租户数据**:谁能往里发消息,取决于它属于哪个工作区。 租户表就得有 `workspace_id NOT NULL` + RLS(AGENTS.md 第 2.5 条)。 -令牌只存 sha256:数据库被看见也拿不到能用的令牌;令牌只在创建时回显一次。 +登录态(`bot_token_enc`)是加密存的:它等于"这个微信账号的登录态", +明文落库等于把号交出去。密钥由 `JWT_SECRET` 派生,见 `services/wechat_ilink.py`。 + +`token_hash` 是上一版 webhook 方案留下的列:那一版以为 clawbot 会往我们这 POST, +后来发现真实协议是"我们扫码登录 + 长轮询"(见 `services/wechat_ilink.py` 文件头), +列先留着不为空但不再使用,等下一版清理。 """ from __future__ import annotations @@ -23,7 +28,9 @@ from sqlalchemy.orm import Mapped, mapped_column from .models import Base, Timestamps, new_uuid BOT_KINDS = ("wechat_clawbot",) -BOT_STATUSES = ("active", "disabled") +# pending:二维码还没扫;active:登录态可用、正在轮询; +# disabled:人手动停用;expired:登录态失效,需要重新扫码。 +BOT_STATUSES = ("pending", "active", "disabled", "expired") INTEGRATION_TENANT_TABLES = ("bot_channel", "bot_message") @@ -43,9 +50,19 @@ class BotChannel(Base, Timestamps): ) kind: Mapped[str] = mapped_column(String(32), nullable=False, default="wechat_clawbot") name: Mapped[str] = mapped_column(String(64), nullable=False, default="") - # hex(sha256(令牌))。令牌本身只在创建响应里出现一次。 - token_hash: Mapped[str] = mapped_column(String(64), nullable=False) - status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") + # 上一版 webhook 方案的遗留列,现在不用了(见模块说明)。 + token_hash: Mapped[str | None] = mapped_column(String(64)) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending") + # 加密后的 iLink 登录态;解不出来就说明 JWT_SECRET 换过,要重新扫码。 + bot_token_enc: Mapped[str | None] = mapped_column(Text) + # 扫码的那个微信号(iLink 的 ilink_user_id),用来在界面上显示"谁扫的"。 + ilink_user_id: Mapped[str | None] = mapped_column(String(64)) + # 服务端可能下发新的基址,跟着它走。 + base_url: Mapped[str | None] = mapped_column(Text) + # 长轮询游标:丢了就会重复收消息。 + sync_buf: Mapped[str | None] = mapped_column(Text) + # 最近一条消息的 context_token:回复必须带它。 + context_token: Mapped[str | None] = mapped_column(Text) # 消息落在谁名下:用创建这条通道的人;人没了就退回工作区 owner。 created_by: Mapped[uuid.UUID | None] = mapped_column( PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="SET NULL") diff --git a/backend/app/services/wechat_bot.py b/backend/app/services/wechat_bot.py new file mode 100644 index 0000000..cfa222a --- /dev/null +++ b/backend/app/services/wechat_bot.py @@ -0,0 +1,282 @@ +"""微信机器人的后台轮询:登录之后由它把消息捞进来。 + +为什么是长轮询而不是 webhook:iLink Bot 协议就是"登录后由我们主动去拉", +服务端只在下一条消息到达时返回(最长等 75 秒)。所以每个通道一个常驻任务。 + +几条必须守住的规矩: + +1. **游标要落库**:`get_updates_buf` 是下次从哪里接着读的凭据,丢了就会重复收、 + 把同一条任务记两遍。 +2. **一条消息只处理一次**:`bot_message.external_id` 有唯一索引,重复投递时先撞唯一键。 +3. **token 失效要停**:服务端 errcode `-14` 表示登录态作废,这时停轮询、把通道标记成 + 需要重新扫码,别拿死 token 一直撞墙。 +4. **回复失败不影响已经落库的事项**:任务记下了就是记下了,回不回得去另说。 +""" + +from __future__ import annotations + +import asyncio +import contextlib +import uuid +from typing import Any + +from sqlalchemy import select +from sqlalchemy import text as sa_text + +from ..config import get_settings +from ..db import SessionFactory, bind_tenant +from ..logging_setup import get_logger +from ..models import Workspace +from ..models_integrations import BotChannel, BotMessage +from . import wechat_ilink +from .wechat_intake import handle_text + +logger = get_logger("wechat_bot") + +# 长轮询一次最多等这么久(服务端 75 秒超时,客户端留点余量)。 +POLL_TIMEOUT_SECONDS = 80.0 +RETRY_DELAY_SECONDS = 5.0 + + +def site_base() -> str: + """回复里的链接用哪个域名。生产必须配 PUBLIC_BASE_URL,否则只能退回 CORS 里的第一个。""" + settings = get_settings() + if settings.public_base_url: + return settings.public_base_url.rstrip("/") + if settings.cors_origins: + return str(settings.cors_origins[0]).rstrip("/") + return "" + + +class BotManager: + """一个进程里的轮询任务表:一个通道一个任务。""" + + def __init__(self) -> None: + self._tasks: dict[uuid.UUID, asyncio.Task[None]] = {} + + # ── 生命周期 ────────────────────────────────────────── + + async def start(self, channel_id: uuid.UUID) -> None: + task = self._tasks.get(channel_id) + if task is not None and not task.done(): + return + self._tasks[channel_id] = asyncio.create_task( + self._run(channel_id), name=f"wechat-bot-{channel_id}" + ) + + async def stop(self, channel_id: uuid.UUID) -> None: + task = self._tasks.pop(channel_id, None) + if task is None: + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def start_all(self) -> None: + """进程启动时把已登录的通道都拉起来,不然重启就静默不收消息了。""" + async with SessionFactory() as session: + rows = await session.execute( + select(BotChannel.id).where( + BotChannel.status == "active", + BotChannel.bot_token_enc.is_not(None), + ) + ) + ids = [row[0] for row in rows] + for channel_id in ids: + await self.start(channel_id) + if ids: + logger.info("微信机器人已恢复", extra={"extra_fields": {"channels": len(ids)}}) + + async def shutdown(self) -> None: + for channel_id in list(self._tasks): + await self.stop(channel_id) + + def running(self, channel_id: uuid.UUID) -> bool: + task = self._tasks.get(channel_id) + return task is not None and not task.done() + + # ── 轮询主体 ────────────────────────────────────────── + + async def _load(self, channel_id: uuid.UUID) -> tuple[BotChannel, str] | None: + """读通道并解出 token。返回 None 表示这个通道不该再轮询了。""" + async with SessionFactory() as session: + channel = await self._get_channel(session, channel_id) + if channel is None or channel.status != "active" or not channel.bot_token_enc: + return None + try: + token = wechat_ilink.decrypt_token(channel.bot_token_enc) + except wechat_ilink.WechatError: + await self._mark_relogin(channel_id, "登录态解不开(JWT_SECRET 可能换过)") + return None + session.expunge(channel) + return channel, token + + async def _run(self, channel_id: uuid.UUID) -> None: + loaded = await self._load(channel_id) + if loaded is None: + self._tasks.pop(channel_id, None) + return + channel, token = loaded + cursor = channel.sync_buf or "" + failures = 0 + while True: + try: + update = await wechat_ilink.get_updates( + token, cursor, base_url=channel.base_url, timeout=POLL_TIMEOUT_SECONDS + ) + except asyncio.CancelledError: + raise + except wechat_ilink.WechatError as exc: + if exc.stale_token: + await self._mark_relogin(channel_id, str(exc)) + self._tasks.pop(channel_id, None) + return + failures += 1 + await self._note_error(channel_id, str(exc)) + await asyncio.sleep(RETRY_DELAY_SECONDS if failures < 10 else 60.0) + continue + except Exception as exc: # 轮询任务不能死,死了就永远收不到消息 + failures += 1 + logger.warning("微信长轮询异常:%s", exc) + await asyncio.sleep(RETRY_DELAY_SECONDS) + continue + + failures = 0 + for message in update.messages: + try: + await self._handle(channel_id, message, token, channel.base_url) + except Exception as exc: # 单条失败不拖垮整轮 + logger.warning("微信消息处理失败:%s", exc) + + if update.cursor and update.cursor != cursor: + cursor = update.cursor + await self._save_cursor(channel_id, cursor) + + # ── 单条消息 ────────────────────────────────────────── + + async def _handle( + self, + channel_id: uuid.UUID, + message: wechat_ilink.Inbound, + token: str, + base_url: str | None, + ) -> None: + async with SessionFactory() as session: + channel = await self._get_channel(session, channel_id) + if channel is None or channel.status != "active": + return + # 读通道时只设了 app.bot_channel;workspace 表同样有 RLS, + # 得先把通道所属工作区告诉数据库,才能读到那一行。 + await session.execute( + sa_text("SELECT set_config('app.ws_id', :ws, true)"), + {"ws": str(channel.workspace_id)}, + ) + workspace = await session.get(Workspace, channel.workspace_id) + if workspace is None: + return + actor = channel.created_by or workspace.owner_id + await bind_tenant(session, channel.workspace_id, actor) + + existing = await session.execute( + select(BotMessage).where( + BotMessage.channel_id == channel_id, + BotMessage.external_id == message.message_id, + ) + ) + if existing.scalars().first() is not None: + logger.info("微信消息重复投递,跳过:%s", message.message_id) + return + + result: dict[str, Any] + if message.text: + result = await handle_text(session, channel, message.text, base_url=site_base()) + elif message.kinds: + result = { + "action": "ignored", + "item_id": None, + "reply": "图片、文件、语音暂时还没搬进来,发文字最快。", + } + else: + result = {"action": "ignored", "item_id": None, "reply": "没看懂这条消息。"} + + session.add( + BotMessage( + workspace_id=channel.workspace_id, + channel_id=channel_id, + external_id=message.message_id, + sender=message.sender[:128], + text=message.text[:4000], + action=str(result["action"]), + item_id=result.get("item_id"), + reply=str(result.get("reply") or "")[:2000], + raw={"kinds": message.kinds, "message_id": message.message_id}, + ) + ) + channel.last_seen_at = _now() + channel.context_token = message.context_token + channel.last_error = None + await session.commit() + + reply = str(result.get("reply") or "") + if reply: + try: + await wechat_ilink.send_text( + token, + message.sender, + message.context_token, + reply, + base_url=base_url, + ) + except wechat_ilink.WechatError as exc: + await self._note_error(channel_id, f"回复失败:{exc}") + + async def _get_channel(self, session, channel_id: uuid.UUID) -> BotChannel | None: + """按主键取通道。 + + 轮询任务一开始没有工作区上下文,RLS 会把行藏起来;迁移 0014 给了一条 + `app.bot_channel` 的按主键口子,所以先设 GUC 再读。 + """ + await session.execute( + sa_text("SELECT set_config('app.bot_channel', :cid, true)"), {"cid": str(channel_id)} + ) + return await session.get(BotChannel, channel_id) + + # ── 落库小工具 ──────────────────────────────────────── + + async def _save_cursor(self, channel_id: uuid.UUID, cursor: str) -> None: + async with SessionFactory() as session: + channel = await self._get_channel(session, channel_id) + if channel is None: + return + channel.sync_buf = cursor[:4000] + await session.commit() + + async def _note_error(self, channel_id: uuid.UUID, message: str) -> None: + async with SessionFactory() as session: + channel = await self._get_channel(session, channel_id) + if channel is None: + return + channel.last_error = message[:500] + await session.commit() + + async def _mark_relogin(self, channel_id: uuid.UUID, message: str) -> None: + """登录态作废:停用这条通道并记下原因,界面上提示重新扫码。""" + async with SessionFactory() as session: + channel = await self._get_channel(session, channel_id) + if channel is None: + return + channel.bot_token_enc = None + channel.sync_buf = None + channel.status = "expired" + channel.context_token = None + channel.last_error = message[:500] + await session.commit() + + +def _now(): + from datetime import datetime, timezone + + return datetime.now(timezone.utc) + + +manager = BotManager() diff --git a/backend/app/services/wechat_ilink.py b/backend/app/services/wechat_ilink.py new file mode 100644 index 0000000..a88ce25 --- /dev/null +++ b/backend/app/services/wechat_ilink.py @@ -0,0 +1,314 @@ +"""微信 clawbot 客户端(腾讯官方 iLink Bot 协议)。 + +协议不是我们自己猜的:OpenClaw 官方插件 `@tencent-weixin/openclaw-weixin` 用的就是它, +本机参考实现见 `~/.dsh/.../@michengai/dsh-im-connect/src/channels/weixin.ts` 的 +`src/channels/weixin.ts`(文件头写着"扫码登录 → 长轮询收消息 → context_token 回复")。 + +流程: + +1. `POST /ilink/bot/get_bot_qrcode?bot_type=3` 拿二维码 id 和内容,本地渲染成图给人扫; +2. `POST /ilink/bot/get_qrcode_status?qrcode=` 每 2 秒问一次,`confirmed` 时给 `bot_token`; +3. 带 `Authorization: Bearer ` 长轮询 `POST /ilink/bot/getupdates`, + body 里带游标 `get_updates_buf`,回来的新游标要存下来,不然重启会重复收; +4. 回复走 `POST /ilink/bot/sendmessage`,必须带对方上一条消息里的 `context_token`。 + +这里只封协议,不碰数据库;落库与建事项在 `services/wechat_bot.py`。 +""" + +from __future__ import annotations + +import base64 +import hashlib +import secrets +from dataclasses import dataclass, field +from typing import Any + +import httpx +from cryptography.fernet import Fernet, InvalidToken + +from ..config import get_settings + +BASE_URL = "https://ilinkai.weixin.qq.com" +# 官方插件同款版本号;服务端会按它挑行为,别乱改。 +CHANNEL_VERSION = "1.0.0" +# token 失效时服务端给的 errcode(官方插件叫 STALE_TOKEN_ERRCODE)。 +STALE_TOKEN_ERRCODE = -14 + +QR_TIMEOUT_SECONDS = 15 +STATUS_TIMEOUT_SECONDS = 40 +SEND_TIMEOUT_SECONDS = 15 + + +class WechatError(RuntimeError): + """协议层错误。`kind` 用来区分"该重新扫码"还是"过一会儿再试"。""" + + def __init__(self, message: str, *, kind: str = "api") -> None: + super().__init__(message) + self.kind = kind + + @property + def stale_token(self) -> bool: + return self.kind == "stale_token" + + +@dataclass +class QrCode: + qrcode_id: str + payload: str + + +@dataclass +class QrStatus: + status: str + bot_token: str | None = None + ilink_user_id: str | None = None + base_url: str | None = None + + +@dataclass +class Inbound: + """一条收到的消息:只取用得上的部分。""" + + message_id: str + sender: str + text: str + context_token: str | None = None + kinds: list[str] = field(default_factory=list) + + +@dataclass +class Update: + cursor: str + messages: list[Inbound] + + +# ── 令牌加密 ───────────────────────────────────────────── +# +# bot_token 等于"这个微信号的登录态",明文落库等于把号交出去。 +# 密钥由 JWT_SECRET 派生:换 JWT_SECRET 会让已存的令牌失效(重新扫码即可),这是可接受的代价。 + + +def _fernet() -> Fernet: + digest = hashlib.sha256(get_settings().jwt_secret.encode("utf-8")).digest() + return Fernet(base64.urlsafe_b64encode(digest)) + + +def encrypt_token(token: str) -> str: + return _fernet().encrypt(token.encode("utf-8")).decode("ascii") + + +def decrypt_token(stored: str) -> str: + try: + return _fernet().decrypt(stored.encode("ascii")).decode("utf-8") + except InvalidToken as exc: # JWT_SECRET 换过,或者库里的值被人改过 + raise WechatError("微信登录态解不开,需要重新扫码", kind="stale_token") from exc + + +# ── 协议 ───────────────────────────────────────────────── + + +def _headers(token: str | None) -> dict[str, str]: + uin = base64.b64encode(str(secrets.randbelow(0xFFFFFFFF)).encode("utf-8")).decode("ascii") + headers = { + "content-type": "application/json", + "iLink-App-ClientVersion": "1", + "X-WECHAT-UIN": uin, + } + if token: + headers["Authorization"] = f"Bearer {token}" + headers["AuthorizationType"] = "ilink_bot_token" + return headers + + +async def _post( + path: str, + body: dict[str, Any], + *, + token: str | None = None, + timeout: float = 15.0, + tolerate_ret1: bool = False, + base_url: str | None = None, +) -> dict[str, Any]: + url = f"{(base_url or BASE_URL).rstrip('/')}{path}" + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.post(url, json=body, headers=_headers(token)) + except httpx.HTTPError as exc: + raise WechatError(f"微信接口连不上:{exc}") from exc + if response.status_code >= 400: + raise WechatError(f"微信接口 {path} 返回 HTTP {response.status_code}") + try: + data = response.json() + except ValueError as exc: + raise WechatError(f"微信接口 {path} 返回的不是 JSON") from exc + if not isinstance(data, dict): + raise WechatError(f"微信接口 {path} 返回的格式不对") + + ret = int(data.get("ret") or 0) + errcode = int(data.get("errcode") or 0) + if tolerate_ret1 and ret == 1 and errcode == 0: + return data + if ret != 0 or errcode != 0: + if errcode == STALE_TOKEN_ERRCODE: + raise WechatError("微信登录态已失效,需要重新扫码", kind="stale_token") + raise WechatError( + f"微信接口 {path} 失败 ret={ret} errcode={errcode} {data.get('errmsg') or ''}".strip() + ) + return data + + +async def get_qrcode() -> QrCode: + data = await _post("/ilink/bot/get_bot_qrcode?bot_type=3", {}, timeout=QR_TIMEOUT_SECONDS) + qrcode_id = _text(data, "qrcode", "qrcode_id") + payload = _text(data, "qrcode_img_content", "qrcode_url", "url") + if not qrcode_id or not payload: + raise WechatError("微信二维码字段不全,稍后再试") + return QrCode(qrcode_id=qrcode_id, payload=payload) + + +async def qrcode_status(qrcode_id: str) -> QrStatus: + data = await _post( + f"/ilink/bot/get_qrcode_status?qrcode={qrcode_id}", + {}, + timeout=STATUS_TIMEOUT_SECONDS, + tolerate_ret1=True, + ) + status = str(data.get("status") or "waiting") + if status != "confirmed": + return QrStatus(status=status) + token = _text(data, "bot_token") + if not token: + raise WechatError("微信确认了扫码,但没给登录态,请重新扫码") + return QrStatus( + status="confirmed", + bot_token=token, + ilink_user_id=_text(data, "ilink_user_id"), + base_url=_text(data, "baseurl", "base_url"), + ) + + +async def get_updates( + token: str, cursor: str, *, base_url: str | None = None, timeout: float = 75.0 +) -> Update: + data = await _post( + "/ilink/bot/getupdates", + {"get_updates_buf": cursor, "base_info": {"channel_version": CHANNEL_VERSION}}, + token=token, + timeout=timeout, + base_url=base_url, + ) + next_cursor = _text(data, "get_updates_buf", "cursor", "sync_buf") or cursor + raw = data.get("msgs") or data.get("messages") or data.get("updates") or [] + messages: list[Inbound] = [] + if isinstance(raw, list): + for item in raw: + parsed = parse_message(item) + if parsed is not None: + messages.append(parsed) + return Update(cursor=next_cursor, messages=messages) + + +async def send_text( + token: str, + to_user_id: str, + context_token: str | None, + text: str, + *, + base_url: str | None = None, +) -> None: + await _post( + "/ilink/bot/sendmessage", + { + "msg": { + "from_user_id": "", + "to_user_id": to_user_id, + "client_id": f"dadaotask:{secrets.token_hex(6)}", + "message_type": 2, + "message_state": 2, + "context_token": context_token or "", + "item_list": [{"type": 1, "text_item": {"text": text}}], + }, + "base_info": {"channel_version": CHANNEL_VERSION}, + }, + token=token, + timeout=SEND_TIMEOUT_SECONDS, + base_url=base_url, + ) + + +# ── 入站解析 ───────────────────────────────────────────── + + +def parse_message(raw: Any) -> Inbound | None: + """把一条原始消息拆成"谁说的、说了什么"。 + + iLink 的形状是 `{from_user_id, message_id, context_token, item_list:[{type, text_item}]}`, + 但偶尔会裹在 `message` 里,两种都认。只处理文本;语音有服务端转写就顺手用, + 图片/文件/视频只记下类型,这一版不下载。 + """ + if not isinstance(raw, dict): + return None + message = dict(raw) + nested = message.get("message") + if isinstance(nested, dict): + message.update(nested) + if message.get("message_type") is not None and int(message.get("message_type") or 0) != 1: + return None + + sender = _text(message, "from_user_id", "from_user") + if not sender: + return None + + parts: list[str] = [] + kinds: list[str] = [] + if isinstance(message.get("text"), str): + parts.append(str(message["text"])) + items = message.get("item_list") + if isinstance(items, list): + for item in items: + if not isinstance(item, dict): + continue + kind = int(item.get("type") or 0) + if kind == 1: + text_item = item.get("text_item") + if isinstance(text_item, dict): + value = _text(text_item, "text") + if value: + parts.append(value) + elif kind == 2: + kinds.append("image") + elif kind == 3: + voice = item.get("voice_item") + transcript = _text(voice, "text") if isinstance(voice, dict) else None + if transcript: + parts.append(f"[语音] {transcript}") + else: + kinds.append("voice") + elif kind == 4: + kinds.append("file") + elif kind == 5: + kinds.append("video") + + message_id = ( + _text(message, "message_id", "msg_id") + or f"{sender}:{_text(message, 'create_time_ms', 'create_time') or secrets.token_hex(4)}" + ) + return Inbound( + message_id=message_id, + sender=sender, + text="".join(parts).strip(), + context_token=_text(message, "context_token"), + kinds=kinds, + ) + + +def _text(data: Any, *keys: str) -> str | None: + if not isinstance(data, dict): + return None + for key in keys: + value = data.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + if isinstance(value, int | float) and not isinstance(value, bool): + return str(value) + return None diff --git a/backend/app/services/wechat_intake.py b/backend/app/services/wechat_intake.py new file mode 100644 index 0000000..89ef510 --- /dev/null +++ b/backend/app/services/wechat_intake.py @@ -0,0 +1,225 @@ +"""把一句话变成任务或知识草稿。 + +微信只负责把文本送进来,怎么落成事项是这里的事,接口和后台轮询共用同一份逻辑, +免得"网页里发"和"机器人里发"落成两种东西。 +""" + +from __future__ import annotations + +import uuid +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..errors import AppError, ErrorCode +from ..models import Member, Workspace +from ..models_integrations import BotChannel +from ..models_knowledge import KnowledgeDoc +from ..models_task import Item + +MAX_TITLE = 180 +MAX_BODY = 20000 + +KNOWLEDGE_PREFIXES = ("知识", "笔记", "note", "knowledge", "kb") +TASK_PREFIXES = ("任务", "待办", "todo", "task") +HELP_PREFIXES = ("?", "?", "帮助", "help", "用法") + +HELP_TEXT = ( + "发什么都能用,写法:\n" + "· 直接说 → 记成任务\n" + "· 知识:内容 → 记成知识草稿(也可以写「笔记:」)\n" + "· 任务:内容 → 明确记成任务\n" + "首行当标题,后面几行当正文;标题里带 #标签 会自动加上。" +) + +SEPARATORS = (":", ":", " ", "\u3000") + + +def split_command(text: str) -> tuple[str, str]: + """返回 (kind, 去掉前缀后的正文)。""" + head, _, rest = text.partition("\n") + probe = head.strip().lower() + for prefix in HELP_PREFIXES: + if probe == prefix or probe.startswith(f"{prefix}:") or probe.startswith(f"{prefix}:"): + return "help", rest + for prefix in KNOWLEDGE_PREFIXES: + for sep in SEPARATORS: + if probe.startswith(f"{prefix}{sep}"): + offset = len(prefix) + len(sep) + return "knowledge", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip() + for prefix in TASK_PREFIXES: + for sep in SEPARATORS: + if probe.startswith(f"{prefix}{sep}"): + offset = len(prefix) + len(sep) + return "task", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip() + return "task", text + + +def split_tags(body: str) -> tuple[str, list[str]]: + """把 `#标签` 摘出来。只认行首和空白后的 #,免得把 "C#" 吃掉。""" + tags: list[str] = [] + kept: list[str] = [] + for line in body.splitlines(): + rest: list[str] = [] + for word in line.split(" "): + if word.startswith("#") and len(word) > 1 and not word[1:].isdigit(): + name = word[1:].strip(",,。.;;") + if name and name.lower() not in {t.lower() for t in tags}: + tags.append(name) + continue + rest.append(word) + kept.append(" ".join(rest).rstrip()) + return "\n".join(kept).strip(), tags + + +def split_title(body: str) -> tuple[str, str]: + lines = body.splitlines() + if not lines: + return "", "" + return lines[0].strip()[:MAX_TITLE], "\n".join(lines[1:]).strip()[:MAX_BODY] + + +async def creator_id(session: AsyncSession, channel: BotChannel) -> uuid.UUID: + """消息记在谁名下:通道创建者;人没了就退回工作区 owner。""" + if channel.created_by is not None: + row = await session.execute( + select(Member.id).where( + Member.workspace_id == channel.workspace_id, + Member.user_id == channel.created_by, + Member.status == "active", + ) + ) + if row.first() is not None: + return channel.created_by + workspace = await session.get(Workspace, channel.workspace_id) + if workspace is None: # 理论上不可能:FK 是 CASCADE + raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在") + return workspace.owner_id + + +async def create_task( + session: AsyncSession, + *, + ws_id: uuid.UUID, + creator: uuid.UUID, + title: str, + body: str, + tags: list[str], +) -> Item: + item = Item( + workspace_id=ws_id, + kind="task", + title=title, + body_md=body, + status="inbox", + priority=0, + creator_id=creator, + ) + session.add(item) + await session.flush() + from ..api.items import _log, _set_tags # 延迟导入:services 不该在模块级依赖 api + + if tags: + await _set_tags(session, ws_id, item, tags) + await _log(session, item, creator, "created", {"via": "wechat_clawbot"}) + return item + + +async def create_knowledge( + session: AsyncSession, + *, + ws_id: uuid.UUID, + creator: uuid.UUID, + title: str, + body: str, + tags: list[str], +) -> Item: + item = Item( + workspace_id=ws_id, + kind="knowledge", + title=title, + body_md=body, + status="draft", + creator_id=creator, + ) + session.add(item) + await session.flush() + session.add( + KnowledgeDoc( + item_id=item.id, + workspace_id=ws_id, + author_id=creator, + source_item_id=None, + source_kind="manual", + doc_type="note", + status_hint="draft", + generated_by="human", + ) + ) + await session.flush() + from ..api.items import _log, _set_tags # 延迟导入,理由同上 + + if tags: + await _set_tags(session, ws_id, item, tags) + await _log(session, item, creator, "knowledge_created", {"via": "wechat_clawbot"}) + return item + + +def item_link(base_url: str, ws_id: uuid.UUID, item: Item) -> str: + """给人点的链接。`base_url` 传站点根(http://host:8200),不带 /api/v1。""" + base = base_url.rstrip("/") + if item.kind == "knowledge": + return f"{base}/w/{ws_id}/knowledge/{item.id}" + return f"{base}/w/{ws_id}?open={item.id}" + + +def reply_text(kind: str, item: Item, link: str) -> str: + if kind == "knowledge": + return f"已存成知识草稿:{item.title}\n{link}" + return f"已记下任务:{item.title}\n{link}" + + +async def handle_text( + session: AsyncSession, + channel: BotChannel, + text: str, + *, + base_url: str, +) -> dict[str, Any]: + """一段文本 → (action, item_id, reply)。调用方负责提交事务与留痕。""" + kind, body = split_command(text.strip()) + if kind == "help": + return {"action": "help", "item_id": None, "reply": HELP_TEXT} + body, tags = split_tags(body) + title, rest = split_title(body) + if not title: + return {"action": "ignored", "item_id": None, "reply": "内容是空的,没记。"} + creator = await creator_id(session, channel) + if kind == "knowledge": + item = await create_knowledge( + session, + ws_id=channel.workspace_id, + creator=creator, + title=title, + body=rest, + tags=tags, + ) + return { + "action": "knowledge", + "item_id": item.id, + "reply": reply_text("knowledge", item, item_link(base_url, channel.workspace_id, item)), + } + item = await create_task( + session, + ws_id=channel.workspace_id, + creator=creator, + title=title, + body=rest, + tags=tags, + ) + return { + "action": "task", + "item_id": item.id, + "reply": reply_text("task", item, item_link(base_url, channel.workspace_id, item)), + } diff --git a/backend/requirements.txt b/backend/requirements.txt index ed18c70..919730e 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -18,4 +18,7 @@ tzdata==2026.5 pytest==8.3.4 pytest-asyncio==0.25.0 httpx==0.28.1 +# 微信登录态加密(Fernet)与二维码渲染 +cryptography==50.0.2 +qrcode==8.2 ruff==0.8.4 diff --git a/backend/tests/test_clawbot.py b/backend/tests/test_clawbot.py deleted file mode 100644 index ad0c94a..0000000 --- a/backend/tests/test_clawbot.py +++ /dev/null @@ -1,207 +0,0 @@ -"""微信(clawbot)接入:一条消息进来,变成任务或知识。 - -测的是**整条链路**:令牌 → 租户上下文 → 建事项 → 留痕 → 重复推送幂等。 -webhook 不带登录态,所以这里故意不用 `auth()` 头,只有管理接口才带。 -""" - -from __future__ import annotations - -from typing import Any - -from httpx import AsyncClient - -from .conftest import create_workspace, login, unwrap - - -async def _channel(client: AsyncClient, user: dict[str, Any], ws: dict[str, Any]) -> dict: - return unwrap( - await client.post( - f"/api/v1/workspaces/{ws['id']}/integrations", - json={"name": "我的微信"}, - headers=user["headers"], - ) - ) - - -async def _post(client: AsyncClient, token: str, payload: dict[str, Any]): - return await client.post(f"/api/v1/integrations/clawbot/{token}", json=payload) - - -async def test_plain_message_becomes_task(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信接入") - channel = await _channel(client, user, ws) - - resp = await _post( - client, - channel["token"], - {"msg_id": "wx-1", "from": "steven", "text": "给客户回电话\n顺便确认下报价"}, - ) - data = unwrap(resp) - assert data["action"] == "task" - assert data["duplicate"] is False - assert "给客户回电话" in data["reply"] - - items = unwrap( - await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) - )["items"] - task = next(i for i in items if i["title"] == "给客户回电话") - assert task["kind"] == "task" - assert task["status"] == "inbox" - assert "顺便确认下报价" in task["body_md"] - - -async def test_knowledge_prefix_creates_draft_with_tags(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信知识") - channel = await _channel(client, user, ws) - - data = unwrap( - await _post( - client, - channel["token"], - {"text": "知识:回滚前先停灰度 #发布 #回滚\n1. 停流量\n2. 回滚镜像"}, - ) - ) - assert data["action"] == "knowledge" - - docs = unwrap( - await client.get( - f"/api/v1/workspaces/{ws['id']}/knowledge?status=draft", headers=user["headers"] - ) - )["items"] - doc = next(d for d in docs if d["title"] == "回滚前先停灰度") - assert doc["knowledge"]["status_hint"] == "draft" - assert doc["knowledge"]["doc_type"] == "note" - assert {t["name"] for t in doc["tags"]} == {"发布", "回滚"} - # 标签从标题摘出去以后,正文里不该再留 # 号 - assert "#发布" not in doc["body_md"] - - -async def test_help_reply_creates_nothing(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信帮助") - channel = await _channel(client, user, ws) - - data = unwrap(await _post(client, channel["token"], {"text": "帮助"})) - assert data["action"] == "help" - assert "记成任务" in data["reply"] - items = unwrap( - await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) - )["items"] - assert items == [] - - -async def test_unknown_token_is_404(client: AsyncClient) -> None: - resp = await client.post("/api/v1/integrations/clawbot/nope", json={"text": "x"}) - assert resp.status_code == 404 - assert resp.json()["code"] == 9002 - - -async def test_disabled_channel_refuses_messages(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信停用") - channel = await _channel(client, user, ws) - await client.patch( - f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}", - json={"status": "disabled"}, - headers=user["headers"], - ) - resp = await _post(client, channel["token"], {"text": "这条不该进去"}) - assert resp.status_code == 403 - assert resp.json()["code"] == 9003 - - -async def test_same_message_id_is_idempotent(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信重推") - channel = await _channel(client, user, ws) - payload = {"msg_id": "dup-1", "text": "只该有一条"} - - first = unwrap(await _post(client, channel["token"], payload)) - second = unwrap(await _post(client, channel["token"], payload)) - assert second["duplicate"] is True - assert second["reply"] == first["reply"] - - items = unwrap( - await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) - )["items"] - assert [i["title"] for i in items] == ["只该有一条"] - - -async def test_get_verify_echoes_challenge(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信校验") - channel = await _channel(client, user, ws) - - resp = await client.get(f"/api/v1/integrations/clawbot/{channel['token']}?echostr=abc123") - assert resp.status_code == 200 - assert resp.text == "abc123" - - -async def test_token_belongs_to_one_workspace(client: AsyncClient) -> None: - user = await login(client) - ws_a = await create_workspace(client, user, "A 区") - ws_b = await create_workspace(client, user, "B 区") - channel = await _channel(client, user, ws_a) - - await _post(client, channel["token"], {"text": "只该进 A 区"}) - - items_a = unwrap( - await client.get(f"/api/v1/workspaces/{ws_a['id']}/items", headers=user["headers"]) - )["items"] - items_b = unwrap( - await client.get(f"/api/v1/workspaces/{ws_b['id']}/items", headers=user["headers"]) - )["items"] - assert [i["title"] for i in items_a] == ["只该进 A 区"] - assert items_b == [] - - -async def test_rotate_invalidates_old_token(client: AsyncClient) -> None: - user = await login(client) - ws = await create_workspace(client, user, "微信换令牌") - channel = await _channel(client, user, ws) - rotated = unwrap( - await client.post( - f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}/rotate", - headers=user["headers"], - ) - ) - assert rotated["token"] != channel["token"] - old = await _post(client, channel["token"], {"text": "旧令牌"}) - assert old.status_code == 404 - new = unwrap(await _post(client, rotated["token"], {"text": "新令牌能用"})) - assert new["action"] == "task" - - -async def test_member_cannot_manage_channels(client: AsyncClient) -> None: - owner = await login(client) - ws = await create_workspace(client, owner, "微信权限") - member = await login(client) - code = unwrap( - await client.post( - f"/api/v1/workspaces/{ws['id']}/invitations", - json={"role": "member"}, - headers=owner["headers"], - ) - ) - await client.post( - f"/api/v1/workspaces/{ws['id']}/join", - json={"code": code["code"]}, - headers=member["headers"], - ) - resp = await client.post( - f"/api/v1/workspaces/{ws['id']}/integrations", - json={"name": "偷加的通道"}, - headers=member["headers"], - ) - assert resp.status_code == 403 - assert resp.json()["code"] in (3004, 3003) - - -async def test_webhook_url_carries_api_prefix(client: AsyncClient) -> None: - """回调地址必须带 /api/v1:生产上应用挂在域名根,漏了前缀就是 405。""" - user = await login(client) - ws = await create_workspace(client, user, "微信地址") - channel = await _channel(client, user, ws) - assert channel["webhook_url"].endswith(f"/api/v1/integrations/clawbot/{channel['token']}") diff --git a/backend/tests/test_wechat_bot.py b/backend/tests/test_wechat_bot.py new file mode 100644 index 0000000..b2cb93a --- /dev/null +++ b/backend/tests/test_wechat_bot.py @@ -0,0 +1,326 @@ +"""微信 clawbot:扫码登录 + 长轮询 + 消息落成任务/知识。 + +不联网:iLink 的四个接口都用 monkeypatch 顶掉,验的是我们这侧的流程与落库: +取二维码 → 等扫码 → 存加密登录态 → 轮询拿到消息 → 建事项 → 留痕 → 回复 → 去重。 +""" + +from __future__ import annotations + +import uuid +from typing import Any + +import pytest +from httpx import AsyncClient +from sqlalchemy import select, text + +from app.db import SessionFactory, bind_tenant +from app.models_integrations import BotChannel, BotMessage +from app.models_task import Item +from app.services import wechat_bot, wechat_ilink +from app.services.wechat_intake import handle_text + +from .conftest import create_workspace, login, unwrap + + +@pytest.fixture(autouse=True) +def no_background_tasks(monkeypatch: pytest.MonkeyPatch) -> None: + """别在测试里真起轮询任务:这里只验"该不该起",任务体另测。""" + + async def fake_start(self, channel_id: uuid.UUID) -> None: # noqa: ARG001 + return None + + async def fake_stop(self, channel_id: uuid.UUID) -> None: # noqa: ARG001 + return None + + monkeypatch.setattr(wechat_bot.BotManager, "start", fake_start) + monkeypatch.setattr(wechat_bot.BotManager, "stop", fake_stop) + + +async def _begin(client: AsyncClient, user: dict[str, Any], ws: dict[str, Any]) -> dict[str, Any]: + return unwrap( + await client.post( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr", + headers=user["headers"], + ) + ) + + +def _patch_qr(monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_qr() -> wechat_ilink.QrCode: + return wechat_ilink.QrCode(qrcode_id="qr-1", payload="https://wx.example/qr/1") + + monkeypatch.setattr(wechat_ilink, "get_qrcode", fake_qr) + + +def _patch_status(monkeypatch: pytest.MonkeyPatch, status: wechat_ilink.QrStatus) -> None: + async def fake_status(qrcode_id: str) -> wechat_ilink.QrStatus: # noqa: ARG001 + return status + + monkeypatch.setattr(wechat_ilink, "qrcode_status", fake_status) + + +async def test_begin_login_returns_scannable_qr( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_qr(monkeypatch) + user = await login(client) + ws = await create_workspace(client, user, "扫码接入") + + data = await _begin(client, user, ws) + + assert data["qrcode_id"] == "qr-1" + assert data["payload"] == "https://wx.example/qr/1" + assert data["image"].startswith("data:image/svg+xml;base64,") + + channels = unwrap( + await client.get(f"/api/v1/workspaces/{ws['id']}/integrations", headers=user["headers"]) + ) + assert channels["items"][0]["status"] == "pending" + assert channels["items"][0]["connected"] is False + + +async def test_scan_confirms_and_stores_encrypted_token( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_qr(monkeypatch) + _patch_status( + monkeypatch, + wechat_ilink.QrStatus( + status="confirmed", bot_token="bot-token-abc", ilink_user_id="wx-user-9" + ), + ) + user = await login(client) + ws = await create_workspace(client, user, "扫码接入") + started = await _begin(client, user, ws) + + result = unwrap( + await client.get( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr/{started['qrcode_id']}", + headers=user["headers"], + ) + ) + assert result["status"] == "confirmed" + assert result["channel"]["connected"] is True + assert result["channel"]["ilink_user_id"] == "wx-user-9" + + channel_id = uuid.UUID(started["channel_id"]) + async with SessionFactory() as session: + await session.execute( + text("SELECT set_config('app.bot_channel', :c, true)"), {"c": str(channel_id)} + ) + channel = await session.get(BotChannel, channel_id) + assert channel is not None + assert channel.bot_token_enc is not None + # 落库的是密文,且能解回原值 + assert "bot-token-abc" not in channel.bot_token_enc + assert wechat_ilink.decrypt_token(channel.bot_token_enc) == "bot-token-abc" + + +async def test_stale_qr_poll_is_reported_as_expired( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_qr(monkeypatch) + user = await login(client) + ws = await create_workspace(client, user, "扫码接入") + await _begin(client, user, ws) + + data = unwrap( + await client.get( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr/another-qr", + headers=user["headers"], + ) + ) + assert data["status"] == "expired" + + +async def test_disconnect_clears_login_state( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_qr(monkeypatch) + _patch_status(monkeypatch, wechat_ilink.QrStatus(status="confirmed", bot_token="bot-token-abc")) + user = await login(client) + ws = await create_workspace(client, user, "扫码接入") + started = await _begin(client, user, ws) + unwrap( + await client.get( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr/{started['qrcode_id']}", + headers=user["headers"], + ) + ) + + data = unwrap( + await client.delete( + f"/api/v1/workspaces/{ws['id']}/integrations/{started['channel_id']}", + headers=user["headers"], + ) + ) + assert data["channel"]["connected"] is False + assert data["channel"]["status"] == "pending" + + +async def test_plain_text_becomes_task(client: AsyncClient) -> None: + user = await login(client) + ws = await create_workspace(client, user, "收消息") + ws_id = uuid.UUID(ws["id"]) + + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, kind="wechat_clawbot", name="微信", status="active" + ) + session.add(channel) + await session.flush() + result = await handle_text( + session, channel, "给客户回电话\n顺便确认报价", base_url="http://site" + ) + await session.commit() + + assert result["action"] == "task" + assert "已记下任务" in result["reply"] + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + item = await session.get(Item, result["item_id"]) + assert item is not None + assert item.title == "给客户回电话" + assert item.kind == "task" + assert item.status == "inbox" + + +async def test_knowledge_prefix_creates_draft_with_tags(client: AsyncClient) -> None: + user = await login(client) + ws = await create_workspace(client, user, "收消息") + ws_id = uuid.UUID(ws["id"]) + + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, kind="wechat_clawbot", name="微信", status="active" + ) + session.add(channel) + await session.flush() + result = await handle_text( + session, + channel, + "知识:回滚前先停灰度 #发布 #回滚\n先停流量,再退镜像版本", + base_url="http://site", + ) + await session.commit() + + assert result["action"] == "knowledge" + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + item = await session.get(Item, result["item_id"]) + assert item is not None + assert item.kind == "knowledge" + assert item.status == "draft" + assert item.title == "回滚前先停灰度" + assert "#" not in item.title + + +async def test_help_does_not_create_anything(client: AsyncClient) -> None: + user = await login(client) + ws = await create_workspace(client, user, "收消息") + ws_id = uuid.UUID(ws["id"]) + + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, kind="wechat_clawbot", name="微信", status="active" + ) + session.add(channel) + await session.flush() + result = await handle_text(session, channel, "帮助", base_url="http://site") + await session.commit() + + assert result["action"] == "help" + assert result["item_id"] is None + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + items = ( + (await session.execute(select(Item).where(Item.workspace_id == ws_id))).scalars().all() + ) + assert items == [] + + +async def test_poller_handles_message_once_and_replies( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """同一个外部消息 id 到两次,只能有一条任务、一条留痕、一次回复。""" + _patch_qr(monkeypatch) + _patch_status(monkeypatch, wechat_ilink.QrStatus(status="confirmed", bot_token="bot-token-abc")) + user = await login(client) + ws = await create_workspace(client, user, "轮询") + started = await _begin(client, user, ws) + unwrap( + await client.get( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr/{started['qrcode_id']}", + headers=user["headers"], + ) + ) + channel_id = uuid.UUID(started["channel_id"]) + + replies: list[tuple[str, str]] = [] + + async def fake_send( + token: str, to: str, context: str | None, text: str, *, base_url: str | None = None + ) -> None: # noqa: ARG001 + replies.append((to, text)) + + monkeypatch.setattr(wechat_ilink, "send_text", fake_send) + + message = wechat_ilink.Inbound( + message_id="msg-1", + sender="wx-user-9", + text="整理一下本周排期", + context_token="ctx-1", + ) + await wechat_bot.manager._handle(channel_id, message, "bot-token-abc", None) + await wechat_bot.manager._handle(channel_id, message, "bot-token-abc", None) + + assert len(replies) == 1 + assert replies[0][0] == "wx-user-9" + assert "已记下任务" in replies[0][1] + + async with SessionFactory() as session: + await session.execute( + text("SELECT set_config('app.bot_channel', :c, true)"), {"c": str(channel_id)} + ) + channel = await session.get(BotChannel, channel_id) + assert channel is not None + assert channel.context_token == "ctx-1" + assert channel.last_seen_at is not None + # bot_message 走标准租户隔离,验它得先绑定工作区 + await bind_tenant(session, uuid.UUID(ws["id"]), uuid.UUID(user["user_id"])) + messages = ( + (await session.execute(select(BotMessage).where(BotMessage.channel_id == channel_id))) + .scalars() + .all() + ) + assert len(messages) == 1 + assert messages[0].action == "task" + + +async def test_stale_token_marks_channel_expired( + client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _patch_qr(monkeypatch) + _patch_status(monkeypatch, wechat_ilink.QrStatus(status="confirmed", bot_token="bot-token-abc")) + user = await login(client) + ws = await create_workspace(client, user, "轮询") + started = await _begin(client, user, ws) + unwrap( + await client.get( + f"/api/v1/workspaces/{ws['id']}/integrations/wechat/qr/{started['qrcode_id']}", + headers=user["headers"], + ) + ) + channel_id = uuid.UUID(started["channel_id"]) + + await wechat_bot.manager._mark_relogin(channel_id, "登录态已失效,需要重新扫码") + + channels = unwrap( + await client.get(f"/api/v1/workspaces/{ws['id']}/integrations", headers=user["headers"]) + ) + assert channels["items"][0]["status"] == "expired" + assert channels["items"][0]["connected"] is False + assert "重新扫码" in channels["items"][0]["last_error"] diff --git a/web/e2e/integrations.mjs b/web/e2e/integrations.mjs index 5a66f7a..347d432 100644 --- a/web/e2e/integrations.mjs +++ b/web/e2e/integrations.mjs @@ -1,9 +1,8 @@ /** - * 微信接入的端到端验证:页面上建通道 → 把"微信"发来的那条 POST 打到回调地址 - * → 任务真出现在任务列表里。 + * 微信接入的端到端验证:页面上点「扫码连接微信」→ 真拿一张二维码回来 → 画在页面上。 * - * 这里不发真正的微信消息(那得有人拿手机发),而是照着 clawbot 的形状 - * 直接 POST 一次——它就是这个接口唯一的调用方。 + * 不扫这张码(那得有人拿手机扫),所以只验到"二维码出来了、通道在等扫码"为止; + * 扫码之后的落库与轮询由 `backend/tests/test_wechat_bot.py` 覆盖。 */ import { readFileSync } from "node:fs"; import { chromium } from "playwright-core"; @@ -64,67 +63,50 @@ async function main() { localStorage.setItem("dadao.access", access); localStorage.setItem("dadao.refresh", refresh); }, - [token, session.tokens.refresh_token], + [session.tokens.access_token, session.tokens.refresh_token], ); await page.goto(`${WEB}/w/${ws.id}/integrations`, { waitUntil: "networkidle" }); - check("接入页能打开", await page.getByRole("heading", { name: "接入" }).isVisible()); + check("接入页打开", await page.getByText("怎么用").isVisible()); - // 建一条新通道:令牌只在创建后那一屏出现 - await page.getByRole("button", { name: "新建通道" }).click(); - const urlBox = page.getByTestId("webhook-url"); - await urlBox.waitFor({ state: "visible" }); - const url = (await urlBox.innerText()).trim(); - check("创建后拿到回调地址", /\/integrations\/clawbot\/[\w-]{20,}/.test(url), url.slice(0, 48) + "…"); + const button = page.getByRole("button", { name: "扫码连接微信" }); + if (await button.isVisible()) { + await button.click(); + } else { + // 之前连过:卡片上给的是「重新扫码」 + await page.getByRole("button", { name: "重新扫码" }).click(); + } - // 冒充 clawbot 推一条消息 - const push = await fetch(url, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - msg_id: `e2e-${Date.now()}`, - from: "e2e", - text: `微信来的任务 ${Date.now()}\n从手机发的,带正文`, - }), - }); - const pushed = await push.json(); - check("回调地址收到消息并建了任务", pushed.code === 0 && pushed.data.action === "task"); - const title = pushed.data.reply.match(/《(.+?)》/)?.[1] || ""; + const card = page.getByTestId("wechat-qr"); + await card.waitFor({ state: "visible", timeout: 20_000 }); + const img = card.locator("img"); + const src = await img.getAttribute("src"); + check( + "二维码是页面里画出来的 SVG", + Boolean(src?.startsWith("data:image/svg+xml;base64,")), + src?.slice(0, 32), + ); + check( + "提示用微信扫码", + (await card.innerText()).includes("用微信扫这张码"), + (await card.innerText()).split("\n")[0], + ); - // 再推一次同一条:不该多出第二条 - const again = await ( - await fetch(url, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ msg_id: "dup", text: "重复消息" }), - }) - ).json(); - const dup = await ( - await fetch(url, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ msg_id: "dup", text: "重复消息" }), - }) - ).json(); - check("同一条消息不会建两条", again.code === 0 && dup.data.duplicate === true); + const channels = await (await fetch(`${API}/workspaces/${ws.id}/integrations`, { headers })).json(); + const channel = channels.data.items[0]; + check("后端记下了这条通道", Boolean(channel), JSON.stringify(channel?.status)); + check("还没扫码,状态是 pending", channel?.status === "pending", channel?.status); + check("没有登录态就不该在轮询", channel?.polling === false, String(channel?.polling)); - // 页面上的「最近消息」要能看到刚才那几条 - await page.getByRole("button", { name: "最近消息" }).first().click(); - await page.getByText("建了任务").first().waitFor({ state: "visible" }); - check("页面能看到收到的消息", await page.getByText("建了任务").first().isVisible()); + await page.screenshot({ path: "e2e/artifacts/integrations.png", fullPage: false }); + check("控制台没有报错", consoleErrors.length === 0, consoleErrors.slice(0, 2).join(" | ")); - // 任务真的进了任务列表 - const items = await (await fetch(`${API}/workspaces/${ws.id}/items`, { headers })).json(); - const found = items.data.items.find((i) => i.title === title); - check("任务出现在工作区里", Boolean(found), title); - - check("控制台零报错", consoleErrors.length === 0, consoleErrors.slice(0, 2).join(" | ")); - - await page.screenshot({ path: "e2e/artifacts/integrations.png", fullPage: true }); await browser.close(); const failed = results.filter((r) => !r.ok); - console.log(`\n${failed.length === 0 ? "ALL PASS" : "FAILED"} — pass ${results.length - failed.length}, fail ${failed.length}`); + console.log( + `\n${failed.length === 0 ? "ALL PASS" : "FAILED"} — pass ${results.length - failed.length}, fail ${failed.length}`, + ); if (failed.length > 0) process.exit(1); } diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 50a17f7..bf0b4d3 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -225,17 +225,29 @@ export interface AppNotification { created_at: string; } /** 外部接入通道:微信机器人靠这条令牌往工作区里发消息。 */ +/** 微信通道:`status` 是登录态的四种情形,`connected` 只是它的近视写法。 */ export interface BotChannel { id: string; kind: string; name: string; - status: "active" | "disabled"; - created_by: string | null; + status: "pending" | "active" | "disabled" | "expired"; + connected: boolean; + polling: boolean; + ilink_user_id: string | null; last_seen_at: string | null; last_error: string | null; created_at: string; } +/** 取二维码的返回:`image` 是 data URL,前端直接塞进 ``。 */ +export interface WechatQr { + channel_id: string; + qrcode_id: string; + payload: string; + image: string; + expires_in: number; +} + /** 收到的消息留痕:`action` 说明它变成了什么。 */ export interface BotMessage { id: string; diff --git a/web/src/pages/Integrations.tsx b/web/src/pages/Integrations.tsx index d745b32..3aff36b 100644 --- a/web/src/pages/Integrations.tsx +++ b/web/src/pages/Integrations.tsx @@ -1,18 +1,12 @@ -import { useCallback, useEffect, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { useParams } from "react-router-dom"; import { AppShell } from "../components/AppShell"; -import { Chat, Link2, Plus, Trash } from "../components/icons"; +import { Chat } from "../components/icons"; import { Button, EmptyState, InlineAlert, PageHeader, SkeletonRows } from "../components/ui"; import { useAuth } from "../App"; import { api, ApiError } from "../lib/client"; -import { copyText } from "../lib/clipboard"; -import type { BotChannel, BotMessage } from "../lib/types"; - -/** 真发出去的那一步:把令牌拼成地址。 */ -function webhookUrl(origin: string, token: string) { - return `${origin.replace(/\/$/, "")}/api/v1/integrations/clawbot/${token}`; -} +import type { BotChannel, BotMessage, WechatQr } from "../lib/types"; function ago(iso: string | null): string { if (!iso) return "还没收到过消息"; @@ -33,15 +27,42 @@ const ACTION_TEXT: Record = { error: "出错", }; +/** 扫码的几种回话,说给用户听的话。 */ +const QR_TEXT: Record = { + waiting: "用微信扫这张码,扫完在手机上点确认。", + scaned: "已经扫到了,在手机上点一下「确认登录」。", + confirmed: "配好了,现在可以直接在微信里发消息。", + expired: "二维码过期了,重新拿一张。", + error: "问登录结果时出错了,稍等再试。", +}; + +function ChannelStatus({ channel }: { channel: BotChannel }) { + if (channel.status === "active") { + return ( + + + 接收中 + + ); + } + if (channel.status === "expired") { + return 登录已失效; + } + if (channel.status === "disabled") { + return 已停用; + } + return 待扫码; +} + /** * 接入页:把微信(clawbot)接进这个工作区。 * - * 令牌是通道的钥匙,**只在创建/重生成时出现一次**——旁边配一个复制按钮, - * 因为用户要做的就是把它粘进 clawbot 的后台。 + * 走的是腾讯 iLink Bot 协议:**扫码登录 → 后端长轮询收消息 → 用 context_token 回复**。 + * 页面这一侧只做两件事:把二维码画出来、反复问「扫了没」。 */ export function IntegrationsPage() { const { wsId = "" } = useParams(); - const { user, workspaces } = useAuth(); + const { workspaces } = useAuth(); const ws = workspaces.find((w) => w.id === wsId); const canManage = ws?.role === "owner" || ws?.role === "admin"; @@ -49,10 +70,12 @@ export function IntegrationsPage() { const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const [busy, setBusy] = useState(null); - /** 刚建好或刚换过的令牌:只活在这一屏里,刷新就没了(库里只有 sha256)。 */ - const [fresh, setFresh] = useState<{ id: string; token: string; url: string } | null>(null); - const [copied, setCopied] = useState<"url" | "token" | null>(null); + const [notice, setNotice] = useState(null); + const [qr, setQr] = useState(null); + const [qrStatus, setQrStatus] = useState("waiting"); const [messages, setMessages] = useState>({}); + /** 关掉轮询:确认成功、过期、或者用户离开这一屏。 */ + const stopPolling = useRef(false); const load = useCallback(async () => { setLoading(true); @@ -61,7 +84,7 @@ export function IntegrationsPage() { const data = await api.get<{ items: BotChannel[] }>(`/workspaces/${wsId}/integrations`); setChannels(data.items); } catch (err) { - setError(err instanceof ApiError ? err.message : "读取接入配置失败"); + setError(err instanceof ApiError ? err.message : "读不到接入配置"); } finally { setLoading(false); } @@ -71,99 +94,95 @@ export function IntegrationsPage() { void load(); }, [load]); - async function create() { - setBusy("create"); + const channel = channels[0] ?? null; + + const startQr = useCallback(async () => { + setBusy("qr"); setError(null); + setNotice(null); + setMessages({}); + stopPolling.current = false; try { - const data = await api.post<{ channel: BotChannel; token: string }>( - `/workspaces/${wsId}/integrations`, - { name: "微信" }, - ); - setFresh({ - id: data.channel.id, - token: data.token, - url: webhookUrl(window.location.origin, data.token), - }); - await load(); + const data = await api.post(`/workspaces/${wsId}/integrations/wechat/qr`, {}); + setQr(data); + setQrStatus("waiting"); } catch (err) { - setError(err instanceof ApiError ? err.message : "创建通道失败"); + setError(err instanceof ApiError ? err.message : "取二维码失败,稍后再试"); + setQr(null); } finally { setBusy(null); } - } + }, [wsId]); - async function toggle(channel: BotChannel) { - setBusy(channel.id); - try { - await api.patch<{ channel: BotChannel }>( - `/workspaces/${wsId}/integrations/${channel.id}`, - { status: channel.status === "active" ? "disabled" : "active" }, - ); - await load(); - } catch (err) { - setError(err instanceof ApiError ? err.message : "改状态失败"); - } finally { - setBusy(null); - } - } + // 每 2 秒问一次扫码进度;确认之后换成"已连接"并停手。 + useEffect(() => { + if (!qr || stopPolling.current) return; + let alive = true; + const timer = window.setTimeout(async () => { + try { + const data = await api.get<{ status: string; channel?: BotChannel }>( + `/workspaces/${wsId}/integrations/wechat/qr/${qr.qrcode_id}`, + ); + if (!alive) return; + setQrStatus(data.status); + if (data.status === "confirmed") { + stopPolling.current = true; + setQr(null); + setNotice("微信已经连上,可以直接在对话里发「任务:…」或「知识:…」。"); + void load(); + } else if (data.status === "expired") { + stopPolling.current = true; + } + } catch { + if (alive) setQrStatus("error"); + } + }, 2000); + return () => { + alive = false; + window.clearTimeout(timer); + }; + }, [qr, qrStatus, wsId, load]); - async function rotate(channel: BotChannel) { - setBusy(channel.id); - try { - const data = await api.post<{ token: string }>( - `/workspaces/${wsId}/integrations/${channel.id}/rotate`, - {}, - ); - setFresh({ - id: channel.id, - token: data.token, - url: webhookUrl(window.location.origin, data.token), - }); - await load(); - } catch (err) { - setError(err instanceof ApiError ? err.message : "换令牌失败"); - } finally { - setBusy(null); - } - } + const disconnect = useCallback( + async (id: string) => { + setBusy(id); + setError(null); + try { + await api.del(`/workspaces/${wsId}/integrations/${id}?confirm=true`); + setQr(null); + stopPolling.current = true; + setNotice("已断开,微信里再发消息就不会进来了。"); + await load(); + } catch (err) { + setError(err instanceof ApiError ? err.message : "断开失败"); + } finally { + setBusy(null); + } + }, + [wsId, load], + ); - async function remove(channel: BotChannel) { - if (!window.confirm(`删掉「${channel.name}」这条通道?微信那边再发消息就进不来了。`)) return; - setBusy(channel.id); - try { - await api.del(`/workspaces/${wsId}/integrations/${channel.id}`); - if (fresh?.id === channel.id) setFresh(null); - await load(); - } catch (err) { - setError(err instanceof ApiError ? err.message : "删除失败"); - } finally { - setBusy(null); - } - } - - async function peek(channel: BotChannel) { - setBusy(channel.id); - try { - const data = await api.get<{ items: BotMessage[] }>( - `/workspaces/${wsId}/integrations/${channel.id}/messages?limit=5`, - ); - setMessages((prev) => ({ ...prev, [channel.id]: data.items })); - } catch (err) { - setError(err instanceof ApiError ? err.message : "读取消息失败"); - } finally { - setBusy(null); - } - } - - async function copy(kind: "url" | "token", text: string) { - const ok = await copyText(text); - if (!ok) { - setError("浏览器不让直接写剪贴板,请手动选中复制"); - return; - } - setCopied(kind); - window.setTimeout(() => setCopied(null), 1600); - } + const showMessages = useCallback( + async (id: string) => { + if (messages[id]) { + setMessages((prev) => { + const next = { ...prev }; + delete next[id]; + return next; + }); + return; + } + try { + const data = await api.get<{ items: BotMessage[] }>( + `/workspaces/${wsId}/integrations/${id}/messages?limit=5`, + ); + setMessages((prev) => ({ ...prev, [id]: data.items })); + } catch (err) { + setError(err instanceof ApiError ? err.message : "读不到消息记录"); + } + }, + [wsId, messages], + ); return ( @@ -172,8 +191,8 @@ export function IntegrationsPage() { desc="微信里发一句话,这里就多一条任务或知识。不用装 App,也不用切浏览器。" actions={ canManage ? ( - ) : null } @@ -184,140 +203,142 @@ export function IntegrationsPage() { {error} ) : null} - -
-

怎么用

-
    -
  1. 在下面建一条通道,把「回调地址」粘进 clawbot 的后台。
  2. -
  3. 在微信里给机器人发消息:直接说就是任务,「知识:」开头记成知识草稿。
  4. -
  5. 首行当标题,其余当正文;标题里带 #标签 会一起加上。
  6. -
-
- - {fresh ? ( -
-

通道已就绪,这两样只在现在显示一次

-
-
-

回调地址

-
- - {fresh.url} - - -
-
-
-

令牌

-
- - {fresh.token} - - -
-
-
-

- 令牌等同于钥匙:库里只存它的哈希,页面刷新就再也看不到。丢了就「重新生成」,旧地址立刻失效。 -

+ {notice ? ( +
+ {notice}
) : null} +
+

怎么用

+
    +
  1. 点右上「扫码连接微信」,用手机微信扫出现的二维码并在手机上确认。
  2. +
  3. 连上之后直接给这个微信机器人发消息:默认记成任务,「知识:」开头记成知识草稿。
  4. +
  5. 第一行当标题,其余当正文;标题或正文里带 #标签 会一起加上。
  6. +
+
+ {loading ? ( - - ) : channels.length === 0 ? ( - } - title="还没有接进来" - hint={ - canManage - ? "先建一条通道,再把地址填到 clawbot 里;之后微信里发的话就会出现在任务和知识里。" - : "这个工作区还没有配接入通道。需要所有者或管理员来建。" - } - /> +
+ +
) : ( -
    - {channels.map((channel) => ( -
  • -
    - - - + <> + {qr ? ( +
    +
    + 微信扫码登录二维码
    -

    {channel.name}

    -

    {ago(channel.last_seen_at)}

    +

    + {QR_TEXT[qrStatus] ?? "等扫码结果…"} +

    +

    + 二维码 5 分钟内有效。扫完手机上会问「是否允许登录」,点确认就配好了。 +

    + {qrStatus === "expired" || qrStatus === "error" ? ( + + ) : null}
    - - {channel.status === "active" ? "接收中" : "已停用"} - - {canManage ? ( -
    - - - - +
    +
    + ) : null} + + {channel ? ( +
    +
    +
    +
    + + + + {channel.name || "微信"} + + {channel.polling ? 正在接收 : null}
    - ) : null} +

    + {channel.status === "active" + ? ago(channel.last_seen_at) + : "还没连上微信,扫码之后才收得到消息"} + {channel.ilink_user_id ? ` · 微信用户 ${channel.ilink_user_id}` : ""} +

    + {channel.last_error ? ( +

    {channel.last_error}

    + ) : null} +
    +
    + + {canManage ? ( + <> + + + + ) : null} +
    - {channel.last_error ? ( -

    上次出错:{channel.last_error}

    - ) : null} - {messages[channel.id] ? ( - messages[channel.id].length === 0 ? ( -

    还没有收到过消息。

    - ) : ( -
      - {messages[channel.id].map((m) => ( -
    • - +
      + {messages[channel.id].length === 0 ? ( +

      还没收到过消息。

      + ) : ( + messages[channel.id].map((m) => ( +
      +
      +

      {m.text || "(没有文字)"}

      +

      {m.sender}

      +
      + {ACTION_TEXT[m.action] ?? m.action} - {m.text || "(空)"} - - {new Date(m.created_at).toLocaleTimeString("zh-CN", { - hour: "2-digit", - minute: "2-digit", - })} - -
    • - ))} -
    - ) +
    + )) + )} +
) : null} - - ))} - - )} + + ) : ( +
+ } + title="还没连微信" + hint={ + canManage + ? "点右上「扫码连接微信」,用手机扫一下就行。" + : "只有工作区管理员能连微信,让他扫一下。" + } + /> +
+ )} - {!canManage && user ? ( -

接入通道只有所有者和管理员能改。

- ) : null} + {!canManage ? ( +

+ 你不是这个工作区的管理员,能看到接入状态,但不能扫码或断开。 +

+ ) : null} + + )}
); }