diff --git a/.gitignore b/.gitignore index debd63b..7385731 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,8 @@ col.ps1 # brief skill 临时样例输出 (可由 skill 重新生成, 不入库) .brief_out/ + +# ClawBot 接入探测临时产物 (二维码图 / 测试文件, 探测时重新生成, 不入库; +# 探测脚本 scripts/probe_clawbot*.py 保留作参考与复测) +scripts/clawbot_qr*.png +scripts/zcbot_filetest.txt diff --git a/DESIGN.md b/DESIGN.md index 7a4a1dd..8cde5a6 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -656,6 +656,83 @@ create index on usage_events (model_profile, created_at); **取舍**:重构对三 profile 各渲前后 diff `word/document.xml` **字节一致**(零回归);brief 不强并进 manuscript 路径(引文/配色差异大,只共用叶子原语,降回归面)。 +### 8.7 微信接入(双渠道:ClawBot 个人微信 + 企业微信自建应用)(2026-06-23 设计,status=design) + +**诉求**:把 zcbot 送进用户**个人微信**——简报/任务结果主动推过来,且能在微信里直接跟它对话。用户体感 = 微信通讯录里多一个叫「微信 ClawBot」的**联系人**,像加了个好友一样聊。 + +> **⚠️ 实测结论(2026-06-23,`scripts/probe_clawbot*.py`,真机端到端;关键是 `client_id`):ClawBot 可双向对话 + 可主动推送(有前提)。** +> ① 灰度可用(扫码 `confirmed` 拿 `bot_token` + `baseurl`);② **入站通**(`getupdates` 长轮询收用户消息,带 `from_user_id` + `context_token`);③ **多条/流式回复成立**——同一 `context_token` 连发多条,**每条 `msg` 必须带唯一 `client_id`**(漏它则只有第一条送达——前几轮误判"单条/纯被动"的真因),中间块 `message_state=1`(GENERATING)、末块 `=2`(FINISH),按 ~1000 字分块、各块间隔 ~300ms;④ **主动推送成立**——发完 FINISH 后隔 30s 复用同一 `context_token`(+ 新 `client_id`)仍送达,**`context_token` 有效期约 24h、可复用**。 +> **故「定时简报主动推送」(本节最初核心诉求)在 ClawBot 上可行**,前提:用户**先开口过一次**(冷启动无 token 不能凭空推),且距上次互动在 token 有效期(~24h)内——**每条入站消息刷新该用户的 `context_token`**;超期未互动则需用户再开口(或退邮件兜底)。冷推(从未开口)仍不可能。 + +**选型:三条路,选官方 ClawBot(详见对话调研 2026-06-23)**: +- **wechaty / hook(非官方个微)** —— 逆向/注入,违反腾讯 ToS,**封号率高**(hook >80%、web 协议被大量封),要养号/同省 IP/限速。**排除**。 +- **企业微信自建应用** —— 官方、稳定;①只触达**企业微信成员**(非个人微信);②要企业**管理员**建应用 + 配可信域名;③双向对话要回调 + AES + 5s ACK,重。但**主动推送无条件**(不挑用户活跃度、不依赖灰度)→ 定时简报"必达"首选。**与 ClawBot 并列为第二渠道(本节一并设计,见下「渠道 B」),共用渠道抽象。** +- **微信 ClawBot(iLink Bot API)** —— 腾讯 2026-03-22 官方上线,跑在官方 iLink 协议 + 官方服务器 `ilinkai.weixin.qq.com`,**零封号**;腾讯定位"管道",**后端接谁都行**(可接 zcbot)。**采用**。 + +**为什么先实现 ClawBot(企业微信紧随)**:零管理员(用户自扫,不建应用/不配域名)→ 能立即跑通验证(协议已真机实测全通);企业微信要等管理员建应用 + 配可信域名的资源到位。企业微信随后补上,用其**无条件推送**补 ClawBot 的"24h 活跃才可推"短板。 + +**渠道抽象(两渠道共用,加渠道不改 scheduler / 工具主体)**: +- **绑定**:per-user 记"绑了哪些渠道 + 各自凭据/标识"(ClawBot:`bot_token`+`latest_context_token`;企业微信:`wecom_userid`,应用凭据走全局 env)。 +- **统一发送**:`send_to_user(user_id, text, file?)` → 解析该用户已绑渠道 → 各渠道实现各自发;`scheduler.deliver_notify`、`WechatPushTool` 都调这层,不感知具体渠道。 +- **推送择优**:简报这类"必达" → 优先企业微信(无条件);ClawBot 作个人微信触达 + 聊天;两者都绑可多投或按用户偏好。 + +**第一期两处已定决策(评审通过)**: +- **入站对话 → 每用户一条 persistent「微信」task**(聊天要连续性;token 增长靠 §8.2 context 压缩;打标签与网页 task 区分)。**两渠道入站都落到这条 task**。 +- **敏感凭据入库一律加密列**(`bot_token`/`latest_context_token`;企业微信 secret 走 env 不入库)——env `ZCBOT_WECHAT_SECRET_KEY` 派生密钥;绝不进沙箱/日志/API 响应(§3.4)。 + +**唯一现实卡点 = 微信灰度可用性**:仅**国内个人微信**、需 **8.0.70+** 且功能灰度推送中(设置→插件),**不支持企业微信**(`bot_type=3`)。目标用户没有插件入口就用不了——落地前要先核实目标用户在灰度内。腾讯另保留**限频 / 决定可连哪些 AI / 随时终止**的权力(政策风险)。 + +**注册门槛 ≈ 零**:`get_bot_qrcode` **无需任何预置 app_id/凭据/审核/费用**,任何后端直接调即可生成二维码;`bot_token` 纯靠用户扫码下发。**能完全脱离 OpenClaw 自实现**协议客户端(社区 `weixin-ClawBot-API` 已证)。 + +**绑定模型(沿用前版已对的 per-user 扫码骨架)**: +- 每个 zcbot 用户**扫一次码** → 后端拿到**该用户专属 `bot_token`**(Bot ID `xxx@im.bot` / User ID `xxx@im.wechat`)→ 存库 → 之后按用户收发。**1 个 bot_token 对应 1 个微信账号**(扫码者)。 +- 这与"每个用户连自己的微信"天然吻合,且**零管理员**(对比企业微信省掉建应用 + 可信域名)。 +- ⚠️ **待核实**:`bot_token` 是 1:1(每用户一条、各自一条长轮询)还是 1:N(单 token 多用户、靠消息内 `@im.wechat` 区分,Telegram 式)。设计**按更确定的 1:1** 落,若实测为 1:N 则简化为单循环。 + +**扫码绑定流程(iLink)**: +1. zcbot 网页"绑定微信" → 后端 `GET get_bot_qrcode?bot_type=3` → `{qrcode, qrcode_img_content}`,前端展示二维码。 +2. 后端 `GET get_qrcode_status?qrcode=`(长轮询,单连 hold ≤35s,循环续)→ 用户用**个人微信**扫码确认 → 返回 `{status:'confirmed', bot_token, baseurl}`。 +3. 把当前登录 zcbot user 与返回的 `bot_token/baseurl/user_im_id` upsert 进 `wechat_bot_bindings`。前端轮询自己的绑定状态翻转。 + +**数据模型(新表 `wechat_bot_bindings`,独立加表 → 公测兼容)**: +`user_id(PK/FK→users), bot_token(敏感,长期), bot_im_id, user_im_id(xxx@im.wechat), base_url, latest_context_token(敏感,入站时刷新), context_token_at(时间戳,判 24h 推送窗口), status(active|revoked), created_at, updated_at`。migration `0012_wechat_bot_bindings`。`bot_token` / `latest_context_token` 是**敏感凭据**:加密列存或至少**绝不进沙箱 / 不落日志**(沿用 §3.4 密钥隔离);不进 run_python。 + +**协议要点(自实现客户端,2026-06-23 实测验证)**:base = 绑定返回的 `base_url`(实测 `https://ilinkai.weixin.qq.com`)。所有请求 header:`Content-Type: application/json` + `AuthorizationType: ilink_bot_token` + **`X-WECHAT-UIN` 每请求变**(`base64(随机uint32)`,反重放);除取码/查状态外加 `Authorization: Bearer `。 +- **取码/绑定**:`GET /ilink/bot/get_bot_qrcode?bot_type=3`(无需任何预置凭据)→ `{qrcode, qrcode_img_content}`,`qrcode_img_content` 是**微信深链**(`liteapp.weixin.qq.com/q/...`),需**自渲成二维码**(非图片直链);`GET /ilink/bot/get_qrcode_status?qrcode=`(长轮询)→ `{status: wait|confirmed|expired, bot_token, baseurl}`。二维码 TTL 短(~1min),实现要**过期自动换码**。 +- **收**:`POST /ilink/bot/getupdates`,body `{get_updates_buf:<游标,首次空>, base_info:{channel_version:"1.0.2"}}`(长轮询 hold ≤35s)→ `{msgs:[{from_user_id, context_token, item_list:[{type:1,text_item:{text}}]}], get_updates_buf}`。 +- **发**:`POST /ilink/bot/sendmessage`,body `{msg:{to_user_id, client_id:<每条唯一>, message_type:2, message_state:1|2, context_token, item_list:[...]}, base_info:{channel_version:"1.0.2"}}`。**`client_id` 必带且每条唯一**(否则同 token 后续消息被丢);多条/长文 → 中间块 `message_state=1`、末块 `=2`,~1000 字/块、间隔 ~300ms。成功返回 HTTP 200 + 空 body `{}`(无 ret,不能据 body 判成败,以实投为准)。 +- **token 生命周期**:`context_token` 有效期 ~24h、可复用(发完 FINISH 仍可再发)→ 主动推送靠它;**每条入站消息刷新**该用户 token(存最新值 + 时间戳)。`bot_token` 长期 per-user 凭据(扫码下发)。 +- **文件发送(2026-06-23 实测通,`scripts/probe_clawbot_file.py`)**:①`POST /ilink/bot/getuploadurl`(body `{filekey:随机16B的hex, media_type:3(FILE)/1(IMAGE), to_user_id, rawsize, rawfilemd5, filesize:PKCS7填充后大小, aeskey:随机16B的hex, no_need_thumb:true, base_info}`)→ 返回 `{upload_param}`;② 本地用该 aeskey 做 **AES-128-ECB + PKCS7** 加密文件;③ `POST {cdn_base}/upload?encrypted_query_param=&filekey=`(`cdn_base=https://novac2c.cdn.weixin.qq.com/c2c`,body=密文、`application/octet-stream`)→ **响应头 `x-encrypted-param`** = 下载引用(漏 `&filekey=` 会 400 `filekey mismatch`);④ `sendmessage` 带 `item_list:[{type:4, file_item:{media:{encrypt_query_param:<上一步 x-encrypted-param>, aes_key:base64(aeskey.hex()的ascii字节), encrypt_type:1}, file_name, len:str(rawsize)}}]`。**docx/pdf 简报可原生直推为可打开附件**,无须退下载链接。 +- ⚠️ **仍待核实**:富文本(markdown)渲染支持度(源码有 `markdown-filter.ts`,暂按纯文本正文 + 文件直推设计);限频数值(腾讯保留限速);媒体大小上限(暂沿用 20MB)。 + +**架构:入站与出站一体(第一期一起做)** —— **主动推送依赖 `context_token`,而 token 只能从入站消息拿**,故"只出站不入站"不成立;getupdates 长轮询既收对话、又负责刷新 token。 +- **入站长轮询管理器**(lifespan 起,仿 §8.4 `_disk_scanner` plain-asyncio):每个 active binding 一条 `getupdates`(hold ≤35s 循环续)。收到消息 → 按 `bot_token`→binding→zcbot `user_id` 定位是谁 → **刷新该 binding 的 `latest_context_token` + 时间戳** → 映射到该用户的微信对话 task(默认一条 persistent「微信」task 保连续性,§8.5 会话模式)→ 复用 `_run_agent_bg` 跑 → 结果按 ~1000 字分块 `sendmessage`(每块新 `client_id`、中间 `state=1` 末 `state=2`)带 `context_token` 回。**无 5s ACK 约束**,长 run 天然 OK——相对企业微信回调的根本简化。 +- **出站主动推送**(scheduler 简报 / 任务结果 / `WechatPushTool`):用库里该用户 `latest_context_token`,**距上次入站 <~24h** 则直接 `sendmessage`(文本 + docx/pdf 文件直推);**超期 / 从未开口** → 推不出,退邮件兜底(§8.5)或挂起待用户下次开口刷新 token。即"用户开口过、且近 24h 活跃 → 可主动推"。 +- **scale**:N 个 active binding = N 条长轮询;公测期 N 小可接受;放大时视 1:1/1:N 实测结果改为单循环轮询多 token。 + +**接入面(复用现有范式)**: +1. `tools/wechat_bot.py`:ClawBot 客户端(`get_bot_qrcode/get_qrcode_status/getupdates/sendmessage` + AES 媒体)+ `wechat_bot_enabled()`(开关在才挂工具,沿用 §3.4)+ `resolve_wechat_target(user_id)`→`bot_token` + `WechatPushTool`(agent 可调,按当前 run 的 user_id 解析)。HTTP 走已有 httpx。 +2. `core/scheduler.py` `deliver_notify` 加 `channel=="wechat"` 分支,与 email 并列 → 定时简报**把最新产物文件直推**本人微信(取 `_newest_artifact`,≤上限 `sendmessage` 文件、超限退"点此下载"链接;**不改 job schema**——通道是 notify 字段的值)。 +3. `web/app.py`:`POST /v1/wechat/bind/qrcode`(起二维码)、`GET /v1/wechat/bind/status`(轮询绑定结果)、`DELETE /v1/wechat/bind`(解绑)、`POST /v1/wechat/test`(自检发一条);**lifespan 起入站长轮询管理器**(见上"架构");前端设置加"绑定微信"扫码 UI。 + +**渠道 B:企业微信自建应用(紧随 ClawBot 实现,共用渠道抽象)** +- **应用凭据(全局 env,需管理员建应用)**:`WECOM_CORPID / WECOM_AGENTID / WECOM_SECRET`;secret 仅 host 进程读、不进沙箱(同 ClawBot / `send_email`)。host 直连 `qyapi.weixin.qq.com`。 +- **扫码绑定(OAuth 网页授权)**:网页"绑定企业微信" → `open.weixin.qq.com/connect/oauth2/authorize?...scope=snsapi_base&agentid=&state=#wechat_redirect` → 桌面出二维码扫 / 企业微信内静默 → 回调 `cgi-bin/auth/getuserinfo?code=` 拿 `wecom_userid` → 写绑定。**需管理员另配「网页授权可信域名」指向 zcbot 域名**。 +- **推送(无条件主动推,核心优势)**:`gettoken(corpid,secret)` → `access_token`(2h 进程内缓存 + 提前刷新 + 线程安全锁)→ `message/send` 发 text/markdown/**file**(file 先 `media/upload?type=file` 换临时 `media_id`,≤20MB)。**不需用户先开口、无 24h 窗口** → 定时简报必达首选。 +- **入站对话(第二期)**:自建应用配「接收消息回调」+ AES 解密(`WXBizMsgCrypt`)+ 5s ACK + 异步跑 agent + 主动回推;复杂度高,后置。 +- **数据**:绑定抽象加企业微信侧(`wecom_userid`;多企业留 nullable `corpid/permanent_code` 走服务商 ISV,additive);migration `0013_wecom_binding`(或并入同表多列)。 +- **触达**:仅企业成员;**品牌可自定义**(应用名/头像,区别于 ClawBot 统一名)。 + +**取舍(不选)**: +- **不用 wechaty/hook**:违规 + 高封号 + 养号运维,机构产品不可接受。 +- **第一期不锁企业微信**:企业微信触达面窄(仅成员)、要管理员、双向重;ClawBot 触达个人微信 + 零管理员 + 双向轻。企业微信留作"机构身份 / 不依赖灰度"的后续备选,与本通道正交、绑定表/推送抽象可平行扩。 +- **bot_token 落库但隔离**:它是长期 per-user 凭据,必须持久化(不同于企业微信 2h `access_token` 可纯内存);安全靠加密列 + 不进沙箱,不靠不落库。 +- **富排版不强求卡片**:个微富文本能力存疑,统一走"正文纯文本 + 产物文件直推",规避平台差异。 + +**改动面(第一期,含入站+出站)**:1 张新表 + migration `0012_wechat_bot_bindings`;`tools/wechat_bot.py`(iLink 客户端 + `WechatPushTool` + 绑定/token 服务);**1 个 lifespan 入站长轮询管理器 + 消息→user/task 映射**(复用 `_run_agent_bg`);`core/scheduler.py` `deliver_notify` 加 `wechat` 分支;`web/app.py` 4 端点 + 前端扫码 UI;agent_builder 注册(开关在才挂)。env:`ZCBOT_WECHAT_BOT_ENABLED`(+ 可选 `ZCBOT_WECHAT_BASE_URL` 覆盖)+ `ZCBOT_WECHAT_SECRET_KEY`(凭据加密)——**无全局 app secret**(凭据是 per-user `bot_token`,扫码下发)。**不动** loop/llm/capabilities/现有 schema。 + +**渠道 B(企业微信,紧随)改动面**:env `WECOM_CORPID/AGENTID/SECRET`;`tools/wecom_push.py`(access_token 缓存 + `message/send` + `media/upload` + 渠道实现);`send_to_user` / `deliver_notify` 接 wecom 渠道;绑定抽象加 wecom 侧 + migration `0013`;OAuth 起始/回调 2 端点 + 前端"绑定企业微信"。**两渠道共用 `send_to_user` 抽象与绑定层**,故渠道 B 主要是"多一个渠道实现 + 一种绑定方式",不重写主体。 + --- ## 附录:DeepSeek V4 关键事实(2026-04-24) diff --git a/PROGRESS.md b/PROGRESS.md index b8de89f..23ef918 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-06-23(平台渲染层 rendering/:三 skill docx 统一 + chromium md→pdf + bump 0.21.0) +最后更新:2026-06-24(微信接入第一期 ClawBot 后端 + 自包含绑定页;双渠道设计 §8.7 + bump 0.22.0) --- @@ -21,6 +21,15 @@ ## 已完成关键能力 +### 2026-06-24 / 微信接入第一期:ClawBot 个人微信(后端完成,bump 0.22.0) + +- 需求:把 zcbot 送进用户**个人微信**——能对话、能推简报/结果。调研三条路:wechaty/hook(违规高封号,排除)、企业微信自建应用(官方但要管理员+仅企业成员)、**微信 ClawBot**(腾讯 2026-03 官方个人号 Bot API,iLink 协议,零封号,后端接谁都行)。选 ClawBot 先行。详 DESIGN §8.7。 +- **协议全程真机实测**(`scripts/probe_clawbot*.py`,本人微信号在灰度内):① 扫码绑定拿 `bot_token`;② `getupdates` 长轮询收消息;③ `sendmessage` **每条 `client_id` 必唯一**(漏则同 token 后续被丢——前几轮误判"纯被动"的真因),多条/长文中间块 `state=1` 末块 `state=2`;④ `context_token` 24h 可复用 → **主动推送成立**(需用户先开口一次);⑤ 文件:`getuploadurl`→AES-128-ECB(PKCS7)→CDN(URL 带 `filekey`,漏则 400 mismatch)→`file_item`,docx/pdf 原生直推。 +- **关键设计决策**:入站对话→每用户一条 persistent「微信」task(连续性,token 靠 §8.2 压缩);凭据(bot_token/context_token)加密列(env `ZCBOT_WECHAT_SECRET_KEY`),绝不进沙箱/日志;**入站出站一体**——主动推送依赖入站给的 context_token,故 getupdates 长轮询常驻(既收对话又刷新 24h 窗口)。 +- **文件**(后端全部 import/编译自测过):`core/wechat/{ilink.py 协议客户端, crypto.py 凭据加密, service.py 绑定CRUD+推送+send_to_user 渠道抽象, inbound.py 长轮询管理器+回复提取}`;`core/storage/models.py` 加 `WeChatBotBinding` + migration `0012_wechat_bot_bindings`;`tools/wechat_bot.py` `WechatPushTool` + `core/agent_builder.py` 注册(有开关才挂);`core/scheduler.py` `deliver_notify` 加 `wechat` 通道(未送达退邮件兜底);`web/app.py` lifespan 起入站管理器 + `_run_wechat_message` 回调 + 5 端点(`/v1/wechat/bind/qrcode|status`、`/v1/wechat/bind` GET/DELETE、`/v1/wechat/test`);`web/static/wechat_bind.html` 自包含绑定页;`requirements.txt` 加 segno+cryptography。 +- **env**:`ZCBOT_WECHAT_BOT_ENABLED=1`(渠道开关)+ `ZCBOT_WECHAT_SECRET_KEY=<串>`(凭据加密,缺则退明文标记)+ 可选 `ZCBOT_WECHAT_BASE_URL`。 +- **待办(部署后联调)**:migration `0012` 上库;起 web 进程端到端验(扫码绑定→对话→主动推→定时简报推);**SPA 集成**绑定 UI(当前是独立 `/static/wechat_bind.html`,后续并入主 SPA 设置,仿 crons.js modal 范式);**渠道 B 企业微信**(无条件推送,补 ClawBot 24h 窗口短板)按 §8.7「渠道 B」实现。 + ### 2026-06-23 / 平台渲染层 rendering/:三 skill docx 统一 + chromium md→pdf(bump 0.21.0) - 背景:线上 `简报` task 用户要"输出为pdf",模型因 brief 无 PDF 路径而临场即兴——试 `apt install libreoffice`(只读 fs 失败)→ `pip install weasyprint markdown` 手搓 md→HTML→weasyprint;容器空闲回收后包不持久,二次导出又重装一遍。深挖发现两个问题:① skill 缺 PDF 路径、weasyprint 不在镜像;② `_CHEM_RE` 化学式白名单在 brief/paper/proposal **三份 render_docx.py 逐字重复**(改一处易漏改),patent/standard 还复用 proposal 那份。 diff --git a/RUN.md b/RUN.md index 774f697..fd96c75 100644 --- a/RUN.md +++ b/RUN.md @@ -55,9 +55,14 @@ # ZCBOT_DISABLE_SCHEDULER=1 # 可选,整体关掉调度(对照 Claude Code CLAUDE_CODE_DISABLE_CRON) # ZCBOT_SCHEDULER_TICK_SECONDS=10 # 可选,扫描间隔,默 10s(只决定最坏延迟≤1tick,不影响会否漏) # ZCBOT_SCHEDULER_CONCURRENCY=4 # 可选,并发跑的定时 run 上限,默 4 + # 微信接入(ClawBot 个人微信,DESIGN §8.7):可选。开关在才挂 wechat_push tool + 起入站长轮询。 + # ZCBOT_WECHAT_BOT_ENABLED=1 # 渠道总开关;开启后 lifespan 起入站管理器,用户可扫码绑定 + # ZCBOT_WECHAT_SECRET_KEY=<随机串> # 凭据(bot_token/context_token)列加密密钥;缺则退明文标记(公测兜底) + # ZCBOT_WECHAT_BASE_URL=... # 可选,覆盖 iLink base(默 https://ilinkai.weixin.qq.com) ``` > litellm 在 import 时副作用加载 .env;入口走 `main.py`,`.env` 自动生效。直跑 `python -c "from core.storage import ..."` 不经 litellm 链路时记得自己 `import litellm` 触发,或手动 `export ZCBOT_DB_URL=...`。 -- **依赖**:`pip install -r requirements.txt`(已在 `.venv` 里;含 `bcrypt`)。 +- **依赖**:`pip install -r requirements.txt`(已在 `.venv` 里;含 `bcrypt`、`segno`、`cryptography`)。 +- **微信接入(ClawBot,§8.7)**:① `main.py db upgrade head` 带上 migration `0012`;② `.env` 设 `ZCBOT_WECHAT_BOT_ENABLED=1` + `ZCBOT_WECHAT_SECRET_KEY=<串>`;③ 用户登录后开 `/static/wechat_bind.html` 扫码绑定(需个人微信 8.0.70+ 且灰度到 ClawBot 插件)。绑定后在微信「微信 ClawBot」对话即走 zcbot;**主动推送需用户近 24h 在微信开口过一次**(冷启动/超期推不出,退邮件兜底)。绑定页目前独立、未并入主 SPA。 - **PG**:`ZCBOT_DB_URL` 必填。本地 docker compose / 远端 dev / 生产任选;未设置时启动清晰报错,不引导 docker(§7.4)。 - **Auth env**:`PLATFORM_KEY` + `JWT_SECRET` 任一缺失 web 启动 fail-fast。生成随机串:`python -c "import secrets; print(secrets.token_urlsafe(48))"`。 - **用户管理**(`users.email/password_hash/role`,0005 UNIQUE(email)、0009 role):dev SPA 登录后端。发用户两条路径任选:CLI `main.py user add`(下方),或在登录页右下角"+ 管理员添加用户"链接(需先设 `ZCBOT_ADMIN_TOKEN` env,弹窗输入 email/密码/管理员口令/角色)。撤用户 `DELETE FROM users WHERE email=...`(先 DELETE 该 user 的 tasks)。**用户自助改密**:登录后顶栏「改密码」按钮(走 `POST /v1/auth/change_password`,需知道旧密码);改邮箱 / 用户忘了旧密码无法自助 → 手动 SQL(见故障兜底)。 diff --git a/core/__init__.py b/core/__init__.py index d75f543..8e3fcca 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.21.0" +__version__ = "0.22.0" diff --git a/core/agent_builder.py b/core/agent_builder.py index 472289e..6aeb4b0 100644 --- a/core/agent_builder.py +++ b/core/agent_builder.py @@ -60,6 +60,7 @@ from tools.schedule import ( ScheduleCancelTool, ScheduleCreateTool, ScheduleListTool, ScheduleUpdateTool, ) from tools.send_email import SendEmailTool, smtp_configured +from tools.wechat_bot import WechatPushTool, wechat_push_available from core.ark_client import ArkConfig from core.bocha_client import BochaConfig @@ -568,6 +569,12 @@ def build_agent( se = SendEmailTool(base_dir=tool_base, user_root=ur_path) tools[se.name] = se + # 微信主动推送(§8.7 渠道抽象):仅当微信渠道开关在才挂(沿用"有开关才注册")。 + # 交互与定时 run 都可用(定时简报可主动推回用户微信,24h 窗口内)。user_id ctor 注入。 + if wechat_push_available(): + wp = WechatPushTool(uid, base_dir=tool_base, user_root=ur_path) + tools[wp.name] = wp + if caps.enable_run_python: rp = RunPythonTool(base_dir=tool_base, user_root=ur_path) tools[rp.name] = rp diff --git a/core/scheduler.py b/core/scheduler.py index 0a28c96..007144a 100644 --- a/core/scheduler.py +++ b/core/scheduler.py @@ -182,27 +182,8 @@ def _newest_artifact(working_dir: Path) -> Optional[Path]: return best -def deliver_notify( - notify: Optional[dict[str, Any]], - *, - job_name: str, - working_dir: Path, - tz: str, -) -> None: - """job 配了 notify 就确定性补发(不靠 agent 记性)。目前仅 email 通道: - 把工作目录最新产物当附件,套固定模板发。无产物则发纯文本告知已执行。 - - 阻塞 IO(smtplib),由编排层放进 run_in_executor 调。失败抛异常,编排层吞掉记日志。 - """ - if not notify or notify.get("channel") != "email": - return - to = notify.get("to") - if not to: - return +def _notify_email(to, job_name: str, when: str, artifact: Optional[Path]) -> None: from tools.send_email import send_email_smtp # 延迟导入,避免 core→tools 顶层环依赖 - - when = datetime.now(_tzinfo(tz)).strftime("%Y-%m-%d %H:%M") - artifact = _newest_artifact(working_dir) if artifact is not None: subject = f"[定时任务] {job_name} · {when}" body = f"定时任务「{job_name}」已于 {when} 执行,产物见附件:{artifact.name}。" @@ -213,6 +194,51 @@ def deliver_notify( send_email_smtp(to, subject, body) +def deliver_notify( + notify: Optional[dict[str, Any]], + *, + job_name: str, + working_dir: Path, + tz: str, + user_id: Optional[Any] = None, +) -> None: + """job 配了 notify 就确定性补发(不靠 agent 记性)。通道: + - `email`:把工作目录最新产物当附件发到 notify.to。 + - `wechat`:把最新产物 + 一句话主动推到该用户已绑微信(§8.7);未送达(超 24h 窗口 / + 未绑 / 未开口)且 notify 配了 `to`(邮箱)+ SMTP 在 → 退邮件兜底,否则抛错。 + + 阻塞 IO(smtplib / httpx),由编排层放进 run_in_executor 调。失败抛异常,编排层吞掉记日志。 + """ + if not notify: + return + channel = notify.get("channel") + when = datetime.now(_tzinfo(tz)).strftime("%Y-%m-%d %H:%M") + artifact = _newest_artifact(working_dir) + + if channel == "email": + to = notify.get("to") + if to: + _notify_email(to, job_name, when, artifact) + return + + if channel == "wechat": + if user_id is None: + return + from core.wechat.service import send_to_user # 延迟导入,避免顶层环依赖 + from tools.send_email import smtp_configured + + text = (f"定时任务「{job_name}」已于 {when} 执行" + + (f",产物:{artifact.name}" if artifact else ",本次未产生文件产物。")) + report = send_to_user(user_id, text, str(artifact) if artifact else None) + if report.delivered: + return + fb = notify.get("to") # 可选 fallback 邮箱 + if fb and smtp_configured(): + _notify_email(fb, job_name, when, artifact) + return + raise RuntimeError("微信推送未送达: " + ", ".join(r.reason for r in report.results)) + + # ───────────── CRUD 服务层(对话工具 + REST 端点共用,DESIGN §8.5)───────────── # # tools/schedule.py(对话)与 web/app.py 的 /v1/schedules(前端只读+停用/删除)都调 diff --git a/core/storage/models.py b/core/storage/models.py index 99fb173..d324922 100644 --- a/core/storage/models.py +++ b/core/storage/models.py @@ -225,3 +225,44 @@ class ScheduledJob(Base): ) +class WeChatBotBinding(Base): + """ClawBot 个人微信绑定(0012,DESIGN §8.7 渠道 A)。 + + 一行 = 一个 zcbot 用户绑定其个人微信「微信 ClawBot」。PK=user_id → 1 用户 1 绑定。 + - `bot_token`:扫码下发的长期 per-user 凭据。 + - `latest_context_token` + `context_token_at`:每条入站消息刷新;主动推送靠它, + 仅在 ~24h 有效期内可用(冷启动 / 超期则推不出,退邮件兜底,§8.5)。 + 两个 token 列存**密文**(core/wechat/crypto.py;无 ZCBOT_WECHAT_SECRET_KEY 时退明文标记)。 + 绝不进沙箱 / 日志 / API 响应(§3.4)。 + """ + + __tablename__ = "wechat_bot_bindings" + + user_id: Mapped[UUID] = mapped_column( + PG_UUID(as_uuid=True), + ForeignKey("users.user_id", ondelete="CASCADE"), + primary_key=True, + ) + bot_token: Mapped[str] = mapped_column(Text, nullable=False) # 密文 + bot_im_id: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # xxx@im.bot + user_im_id: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # xxx@im.wechat + base_url: Mapped[str] = mapped_column( + Text, nullable=False, server_default="https://ilinkai.weixin.qq.com" + ) + latest_context_token: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # 密文,入站刷新 + context_token_at: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # 该用户的微信对话常驻 task(persistent,§8.7 决策);两渠道入站都落这条 + chat_task_id: Mapped[Optional[UUID]] = mapped_column( + PG_UUID(as_uuid=True), ForeignKey("tasks.task_id", ondelete="SET NULL"), nullable=True + ) + status: Mapped[str] = mapped_column(Text, nullable=False, server_default="active") # active|revoked + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + diff --git a/core/wechat/__init__.py b/core/wechat/__init__.py new file mode 100644 index 0000000..f42ed8a --- /dev/null +++ b/core/wechat/__init__.py @@ -0,0 +1,6 @@ +"""微信接入(DESIGN §8.7)。 + +渠道 A = ClawBot 个人微信 iLink Bot API(`ilink.py`,协议已真机实测,见 +`scripts/probe_clawbot*.py`);渠道 B = 企业微信自建应用(后续 `wecom.py`)。 +本包只放协议客户端等纯逻辑,与 DB / agent 编排解耦。 +""" diff --git a/core/wechat/crypto.py b/core/wechat/crypto.py new file mode 100644 index 0000000..7f951e9 --- /dev/null +++ b/core/wechat/crypto.py @@ -0,0 +1,59 @@ +"""敏感凭据的列加密(DESIGN §8.7:bot_token / latest_context_token 加密入库)。 + +- env `ZCBOT_WECHAT_SECRET_KEY` 在 → 用其派生的 Fernet 密钥加密,密文带 `v1:` 前缀。 +- env 不在 → 退「明文标记」`plain:`(公测兜底,日志/沙箱/API 仍绝不带这两列; + 正式部署应配 key)。`enc()`/`dec()` 对两种前缀都可逆,换 key 不影响存量明文行。 + +只在 host 进程(绑定服务 / 入站管理器 / push)用;绝不进沙箱 / run_python。 +""" +from __future__ import annotations + +import base64 +import hashlib +import os +from typing import Optional + +from cryptography.fernet import Fernet, InvalidToken + +_PREFIX_ENC = "v1:" +_PREFIX_PLAIN = "plain:" + + +def _fernet() -> Optional[Fernet]: + key = os.getenv("ZCBOT_WECHAT_SECRET_KEY", "").strip() + if not key: + return None + # 任意口令 → 32B → urlsafe-base64 Fernet 密钥(确定性,免单独管 Fernet key) + digest = hashlib.sha256(key.encode("utf-8")).digest() + return Fernet(base64.urlsafe_b64encode(digest)) + + +def enc(plaintext: Optional[str]) -> Optional[str]: + """明文 → 入库串。配了 key 走密文(v1:),否则明文标记(plain:)。None 透传。""" + if plaintext is None: + return None + f = _fernet() + if f is None: + return _PREFIX_PLAIN + plaintext + token = f.encrypt(plaintext.encode("utf-8")).decode("ascii") + return _PREFIX_ENC + token + + +def dec(stored: Optional[str]) -> Optional[str]: + """入库串 → 明文。识别 v1:/plain: 前缀;v1: 需 key 且匹配。None 透传。""" + if stored is None: + return None + if stored.startswith(_PREFIX_PLAIN): + return stored[len(_PREFIX_PLAIN):] + if stored.startswith(_PREFIX_ENC): + f = _fernet() + if f is None: + raise RuntimeError( + "密文需要 ZCBOT_WECHAT_SECRET_KEY 才能解密,但 env 未配置" + ) + try: + return f.decrypt(stored[len(_PREFIX_ENC):].encode("ascii")).decode("utf-8") + except InvalidToken as e: + raise RuntimeError("ZCBOT_WECHAT_SECRET_KEY 与密文不匹配(key 变了?)") from e + # 无前缀:历史/手填的裸明文,容错原样返回 + return stored diff --git a/core/wechat/ilink.py b/core/wechat/ilink.py new file mode 100644 index 0000000..7385e8c --- /dev/null +++ b/core/wechat/ilink.py @@ -0,0 +1,297 @@ +"""ClawBot 个人微信 iLink Bot API 客户端(DESIGN §8.7 渠道 A)。 + +协议全部经真机实测(`scripts/probe_clawbot*.py`,2026-06-23): +- 绑定:`get_bot_qrcode`(无凭据,出深链 → 自渲二维码)→ 轮询 `get_qrcode_status` + (TTL ~1min,过期换码)→ `confirmed` 得 `bot_token` + `baseurl`。 +- 收:`getupdates` 长轮询(hold ≤35s),消息带 `from_user_id` + `context_token`。 +- 发:`sendmessage`,**每条 `client_id` 必唯一**(漏则同 token 后续被丢);多条/长文 + 按 ~1000 字分块,中间 `message_state=GENERATING(1)`、末块 `FINISH(2)`,间隔 ~300ms。 +- `context_token` 有效期 ~24h、可复用 → 主动推送靠它(用户须先开口拿到 token)。 +- 文件:`getuploadurl` → AES-128-ECB(PKCS7)加密 → POST 密文到 CDN 拿 `x-encrypted-param` + → `sendmessage` 带 `file_item`。 + +纯协议客户端,不碰 DB / agent 编排。阻塞 IO(httpx 同步),调用方放 to_thread / executor。 +""" +from __future__ import annotations + +import base64 +import hashlib +import os +import time +import uuid +from dataclasses import dataclass +from typing import Any, Optional +from urllib.parse import quote + +import httpx +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +DEFAULT_BASE = "https://ilinkai.weixin.qq.com" +CDN_BASE = "https://novac2c.cdn.weixin.qq.com/c2c" +CHANNEL_VERSION = "1.0.2" +BOT_TYPE_PERSONAL = 3 + +# 协议枚举(源码 @tencent-weixin/openclaw-weixin src/api/types.ts,已实测) +MSG_TYPE_BOT = 2 +STATE_GENERATING = 1 +STATE_FINISH = 2 +ITEM_TEXT = 1 +ITEM_IMAGE = 2 +ITEM_FILE = 4 +UPLOAD_MEDIA_FILE = 3 +UPLOAD_MEDIA_IMAGE = 1 + +# 分块:长文按 ~1000 字切,块间隔防丢 +CHUNK_CHARS = 1000 +CHUNK_DELAY_S = 0.3 +MAX_FILE_BYTES = 20 * 1024 * 1024 + + +def _uin_header() -> str: + """X-WECHAT-UIN:base64(随机 uint32 的十进制字符串),反重放,每请求变。""" + n = int.from_bytes(os.urandom(4), "big") + return base64.b64encode(str(n).encode()).decode() + + +def _headers(bot_token: Optional[str] = None) -> dict[str, str]: + h = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": _uin_header(), + } + if bot_token: + h["Authorization"] = f"Bearer {bot_token}" + return h + + +def _base_info() -> dict[str, str]: + return {"channel_version": CHANNEL_VERSION} + + +def _new_client_id() -> str: + return f"openclaw-weixin-{uuid.uuid4().hex}" + + +def _aes_ecb_pkcs7(plaintext: bytes, key: bytes) -> bytes: + padder = padding.PKCS7(128).padder() + padded = padder.update(plaintext) + padder.finalize() + enc = Cipher(algorithms.AES(key), modes.ECB()).encryptor() + return enc.update(padded) + enc.finalize() + + +# ─────────────────────────── 绑定(无 token)─────────────────────────── + +@dataclass +class QrCode: + qrcode_id: str + deeplink: str # liteapp.weixin.qq.com/q/...,调用方自渲成二维码图片 + + +def get_bot_qrcode(base_url: str = DEFAULT_BASE, *, timeout: float = 20.0) -> QrCode: + """取一张绑定二维码。无需任何预置凭据。`deeplink` 需自渲成二维码让用户扫。""" + with httpx.Client(timeout=timeout) as c: + r = c.get( + f"{base_url}/ilink/bot/get_bot_qrcode", + params={"bot_type": BOT_TYPE_PERSONAL}, + headers=_headers(), + ) + r.raise_for_status() + d = r.json() + return QrCode(qrcode_id=d.get("qrcode", ""), deeplink=d.get("qrcode_img_content", "")) + + +@dataclass +class BindResult: + status: str # wait | confirmed | expired + bot_token: Optional[str] = None + base_url: Optional[str] = None + + +def poll_qrcode_status( + qrcode_id: str, base_url: str = DEFAULT_BASE, *, timeout: float = 40.0 +) -> BindResult: + """单次轮询扫码状态(服务端长轮询,hold 数十秒)。调用方循环调用, + 遇 `expired` 重新 `get_bot_qrcode` 换码。`confirmed` 时返回 bot_token + base_url。""" + with httpx.Client(timeout=timeout) as c: + r = c.get( + f"{base_url}/ilink/bot/get_qrcode_status", + params={"qrcode": qrcode_id}, + headers=_headers(), + ) + r.raise_for_status() + d = r.json() + return BindResult( + status=d.get("status", ""), + bot_token=d.get("bot_token"), + base_url=d.get("baseurl") or d.get("base_url"), + ) + + +# ─────────────────────────── 收发(带 token)─────────────────────────── + +@dataclass +class InboundMessage: + from_user_id: str # xxx@im.wechat + context_token: str # 回复 / 24h 内主动推须带回 + text: str + raw: dict[str, Any] + + +class ILinkClient: + """绑定后按用户持有 `bot_token` + `base_url`,收发该用户消息。""" + + def __init__(self, bot_token: str, base_url: str = DEFAULT_BASE) -> None: + self.bot_token = bot_token + self.base_url = base_url or DEFAULT_BASE + + # —— 收 —— + def get_updates( + self, cursor: str = "", *, timeout: float = 45.0 + ) -> tuple[list[InboundMessage], str]: + """长轮询拉新消息。返回 (消息列表, 新游标);游标传回下次调用。""" + with httpx.Client(timeout=timeout) as c: + r = c.post( + f"{self.base_url}/ilink/bot/getupdates", + json={"get_updates_buf": cursor, "base_info": _base_info()}, + headers=_headers(self.bot_token), + ) + r.raise_for_status() + d = r.json() + msgs: list[InboundMessage] = [] + for m in d.get("msgs", []) or []: + text = "".join( + (it.get("text_item", {}) or {}).get("text", "") + for it in m.get("item_list", []) or [] + ) + msgs.append(InboundMessage( + from_user_id=m.get("from_user_id", ""), + context_token=m.get("context_token", ""), + text=text, + raw=m, + )) + return msgs, d.get("get_updates_buf", cursor) + + # —— 发(底层单条)—— + def _send( + self, to_user_id: str, context_token: str, item: dict, *, state: int + ) -> None: + body = { + "msg": { + "from_user_id": "", + "to_user_id": to_user_id, + "client_id": _new_client_id(), + "message_type": MSG_TYPE_BOT, + "message_state": state, + "context_token": context_token, + "item_list": [item], + }, + "base_info": _base_info(), + } + with httpx.Client(timeout=30.0) as c: + r = c.post( + f"{self.base_url}/ilink/bot/sendmessage", + json=body, + headers=_headers(self.bot_token), + ) + # 成功为 HTTP 200 + 空 body {};非 200 抛错(空 body 不代表失败) + r.raise_for_status() + + # —— 发文本(自动分块,长文不丢)—— + def send_text(self, to_user_id: str, context_token: str, text: str) -> None: + text = text or "" + chunks = [text[i:i + CHUNK_CHARS] for i in range(0, len(text), CHUNK_CHARS)] or [""] + last = len(chunks) - 1 + for i, chunk in enumerate(chunks): + self._send( + to_user_id, context_token, + {"type": ITEM_TEXT, "text_item": {"text": chunk}}, + state=STATE_FINISH if i == last else STATE_GENERATING, + ) + if i != last: + time.sleep(CHUNK_DELAY_S) + + # —— 发文件(getuploadurl → AES-128-ECB → CDN → file_item)—— + def _upload_file(self, to_user_id: str, data: bytes) -> dict[str, Any]: + rawsize = len(data) + rawmd5 = hashlib.md5(data).hexdigest() + aeskey = os.urandom(16) + filekey = os.urandom(16).hex() + ciphertext = _aes_ecb_pkcs7(data, aeskey) + filesize = len(ciphertext) + + with httpx.Client(timeout=30.0) as c: + ru = c.post( + f"{self.base_url}/ilink/bot/getuploadurl", + json={ + "filekey": filekey, + "media_type": UPLOAD_MEDIA_FILE, + "to_user_id": to_user_id, + "rawsize": rawsize, + "rawfilemd5": rawmd5, + "filesize": filesize, + "no_need_thumb": True, + "aeskey": aeskey.hex(), + "base_info": _base_info(), + }, + headers=_headers(self.bot_token), + ) + ru.raise_for_status() + uj = ru.json() + full = (uj.get("upload_full_url") or uj.get("uploadFullUrl") + or uj.get("full_url") or uj.get("url")) + param = (uj.get("upload_param") or uj.get("uploadParam") or uj.get("param")) + if full: + cdn_url = full + elif param: + cdn_url = (f"{CDN_BASE}/upload?encrypted_query_param={quote(param)}" + f"&filekey={quote(filekey)}") + else: + raise RuntimeError(f"getuploadurl 无 upload url/param: {uj}") + + rc = c.post(cdn_url, content=ciphertext, + headers={"Content-Type": "application/octet-stream"}) + download_param = rc.headers.get("x-encrypted-param") + if rc.status_code != 200 or not download_param: + raise RuntimeError( + f"CDN 上传失败 http={rc.status_code} " + f"err={rc.headers.get('x-error-message')}" + ) + return { + "encrypt_query_param": download_param, + "aes_key": base64.b64encode(aeskey.hex().encode()).decode(), + "rawsize": rawsize, + } + + def send_file( + self, + to_user_id: str, + context_token: str, + file_path: str | os.PathLike, + *, + file_name: Optional[str] = None, + ) -> None: + data = _read_file_capped(file_path) + name = file_name or os.path.basename(str(file_path)) + up = self._upload_file(to_user_id, data) + item = { + "type": ITEM_FILE, + "file_item": { + "media": { + "encrypt_query_param": up["encrypt_query_param"], + "aes_key": up["aes_key"], + "encrypt_type": 1, + }, + "file_name": name, + "len": str(up["rawsize"]), + }, + } + self._send(to_user_id, context_token, item, state=STATE_FINISH) + + +def _read_file_capped(file_path: str | os.PathLike) -> bytes: + size = os.path.getsize(file_path) + if size > MAX_FILE_BYTES: + raise ValueError(f"文件超过 {MAX_FILE_BYTES // (1024*1024)}MB 上限") + with open(file_path, "rb") as f: + return f.read() diff --git a/core/wechat/inbound.py b/core/wechat/inbound.py new file mode 100644 index 0000000..ce386a1 --- /dev/null +++ b/core/wechat/inbound.py @@ -0,0 +1,144 @@ +"""入站长轮询管理器(DESIGN §8.7):收用户消息 → 跑 agent → 回复发回。 + +- 每个 active 绑定一条 `getupdates` 长轮询(ilink 同步,放 to_thread);收到消息: + ① `service.refresh_context_token` 刷新 24h 推送窗口;② 调注入的 `handle_message` + (app.py 提供:解析/建该用户常驻「微信」task → 抢 run 锁 → `_run_agent_bg` → 取回复); + ③ 用本轮新鲜 `context_token` 分块发回。 +- 每绑定 loop **串行**处理(收→跑→回→再收):天然避免同用户并发 run 锁冲突;不同用户并发。 +- 管理器周期性对账 active 绑定:新增起 loop、撤销/revoke 停 loop。 + +`handle_message` 注入解耦 app.py 内部(broker / run 锁 / _run_agent_bg);本模块只管协议循环 +与回复提取(`extract_last_assistant_text` 纯函数可测)。 +""" +from __future__ import annotations + +import asyncio +from typing import Any, Awaitable, Callable, Optional +from uuid import UUID + +from sqlalchemy import select + +from core.storage import session_scope +from core.storage.models import Message +from core.wechat import service +from core.wechat.ilink import ILinkClient +from core.wechat.service import BindingSnapshot + +# app.py 注入:跑该用户的微信对话 task,返回 assistant 回复文本(可空) +HandleMessage = Callable[[UUID, str], Awaitable[str]] + + +def _content_to_text(content: Any) -> str: + """OpenAI 风格 content → 纯文本(str 直返;content blocks 拼 text 段)。""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for b in content: + if isinstance(b, dict) and b.get("type") in (None, "text"): + parts.append(b.get("text", "")) + return "".join(parts) + return "" + + +def extract_last_assistant_text(task_id: UUID, *, scan: int = 20) -> str: + """取该 task 最后一条**有正文**的 assistant 消息文本(跳过纯 tool_calls 行)。""" + with session_scope() as s: + rows = s.execute( + select(Message.payload) + .where(Message.task_id == task_id) + .order_by(Message.idx.desc()) + .limit(scan) + ).all() + for (payload,) in rows: + if not isinstance(payload, dict) or payload.get("role") != "assistant": + continue + text = _content_to_text(payload.get("content")) + if text.strip(): + return text + return "" + + +async def _poll_binding( + snap: BindingSnapshot, handle_message: HandleMessage, stop: asyncio.Event +) -> None: + """单个绑定的长轮询循环。异常退避重试,直到 stop。""" + client = ILinkClient(snap.bot_token, snap.base_url) + cursor = "" + backoff = 2 + while not stop.is_set(): + try: + msgs, cursor = await asyncio.to_thread(client.get_updates, cursor) + backoff = 2 + except Exception as e: # noqa: BLE001 + print(f"[wechat-inbound] {str(snap.user_id)[:8]} getupdates err: " + f"{type(e).__name__}: {e}; retry in {backoff}s") + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60) + continue + for m in msgs: + if stop.is_set(): + break + if not m.text.strip(): + continue + # ① 刷新该用户推送窗口(主动推靠它续命) + await asyncio.to_thread( + service.refresh_context_token, snap.user_id, m.from_user_id, m.context_token + ) + # ② 跑 agent 取回复 + try: + reply = await handle_message(snap.user_id, m.text) + except Exception as e: # noqa: BLE001 + reply = f"[出错] {type(e).__name__}: {e}" + # ③ 用本轮新鲜 token 分块回 + if reply and reply.strip(): + try: + await asyncio.to_thread( + client.send_text, m.from_user_id, m.context_token, reply + ) + except Exception as e: # noqa: BLE001 + print(f"[wechat-inbound] {str(snap.user_id)[:8]} reply send err: " + f"{type(e).__name__}: {e}") + + +async def run_inbound_manager( + handle_message: HandleMessage, + stop: asyncio.Event, + *, + reconcile_seconds: int = 60, +) -> None: + """常驻管理器:周期对账 active 绑定,起/停 per-binding 长轮询循环。""" + loops: dict[UUID, asyncio.Task] = {} + try: + while not stop.is_set(): + try: + active = await asyncio.to_thread(service.list_active_bindings) + except Exception as e: # noqa: BLE001 + print(f"[wechat-inbound] list bindings err: {type(e).__name__}: {e}") + active = [] + active_ids = {s.user_id for s in active} + # 起新增 + for snap in active: + t = loops.get(snap.user_id) + if t is None or t.done(): + loops[snap.user_id] = asyncio.create_task( + _poll_binding(snap, handle_message, stop), + name=f"wechat-poll-{str(snap.user_id)[:8]}", + ) + # 清撤销 / 已结束 + for uid in list(loops): + if uid not in active_ids: + loops.pop(uid).cancel() + elif loops[uid].done(): + loops.pop(uid) + await _wait_stop(stop, reconcile_seconds) # 等 stop 或到下次对账 + finally: + for t in loops.values(): + t.cancel() + + +async def _wait_stop(stop: asyncio.Event, timeout: float) -> None: + try: + await asyncio.wait_for(stop.wait(), timeout=timeout) + except asyncio.TimeoutError: + pass diff --git a/core/wechat/service.py b/core/wechat/service.py new file mode 100644 index 0000000..d81f2aa --- /dev/null +++ b/core/wechat/service.py @@ -0,0 +1,194 @@ +"""微信渠道服务层(DESIGN §8.7):绑定 CRUD + 主动推送 + `send_to_user` 渠道抽象。 + +- 绑定行的 `bot_token` / `latest_context_token` 经 `crypto` 加解密;快照(BindingSnapshot) + 脱离 session、含明文 token,**仅 host 进程内用,绝不外泄/进沙箱**。 +- 主动推送 24h 窗口:`context_token` 仅在末次入站 ~24h 内可用;超期/未开口 → 推不出, + 返回 reason 给调用方退邮件兜底(§8.5)。 +- `send_to_user` 是渠道抽象:scheduler / WechatPushTool 调它,不感知 ClawBot/企业微信; + 企业微信(渠道 B)后续在此追加一路。 + +阻塞 IO(DB + httpx),调用方放 to_thread / executor。 +""" +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Optional +from uuid import UUID + +from sqlalchemy import select + +from core.storage import session_scope +from core.storage.models import WeChatBotBinding +from core.wechat import crypto +from core.wechat.ilink import DEFAULT_BASE, ILinkClient + +CONTEXT_TOKEN_TTL = timedelta(hours=24) + + +def clawbot_enabled() -> bool: + """ClawBot 渠道总开关(沿用「有开关才挂」范式,§3.4)。""" + return os.getenv("ZCBOT_WECHAT_BOT_ENABLED", "").strip().lower() in ( + "1", "true", "yes", "on", + ) + + +# ─────────────────────────── 绑定快照 / CRUD ─────────────────────────── + +@dataclass +class BindingSnapshot: + user_id: UUID + bot_token: str # 明文(已解密) + base_url: str + user_im_id: Optional[str] + context_token: Optional[str] # 明文(已解密) + context_token_at: Optional[datetime] + chat_task_id: Optional[UUID] + status: str + + +def _snap(row: WeChatBotBinding) -> BindingSnapshot: + return BindingSnapshot( + user_id=row.user_id, + bot_token=crypto.dec(row.bot_token) or "", + base_url=row.base_url or DEFAULT_BASE, + user_im_id=row.user_im_id, + context_token=crypto.dec(row.latest_context_token), + context_token_at=row.context_token_at, + chat_task_id=row.chat_task_id, + status=row.status, + ) + + +def get_binding(user_id: UUID) -> Optional[BindingSnapshot]: + with session_scope() as s: + row = s.get(WeChatBotBinding, user_id) + return _snap(row) if row else None + + +def list_active_bindings() -> list[BindingSnapshot]: + """入站长轮询管理器用:所有 active 绑定(含明文 bot_token)。""" + with session_scope() as s: + rows = ( + s.execute( + select(WeChatBotBinding).where(WeChatBotBinding.status == "active") + ) + .scalars() + .all() + ) + return [_snap(r) for r in rows] + + +def upsert_clawbot_binding( + user_id: UUID, bot_token: str, base_url: str, *, bot_im_id: Optional[str] = None +) -> None: + """扫码 confirmed 后写/更新绑定。bot_token 加密入库。""" + now = datetime.now(timezone.utc) + with session_scope() as s: + row = s.get(WeChatBotBinding, user_id) + if row is None: + row = WeChatBotBinding(user_id=user_id) + s.add(row) + row.bot_token = crypto.enc(bot_token) + row.base_url = base_url or DEFAULT_BASE + if bot_im_id: + row.bot_im_id = bot_im_id + row.status = "active" + row.updated_at = now + + +def refresh_context_token(user_id: UUID, user_im_id: str, context_token: str) -> None: + """每条入站消息刷新该用户的 context_token(+时间戳)——主动推送窗口靠它续命。""" + now = datetime.now(timezone.utc) + with session_scope() as s: + row = s.get(WeChatBotBinding, user_id) + if row is None: + return + if user_im_id: + row.user_im_id = user_im_id + row.latest_context_token = crypto.enc(context_token) + row.context_token_at = now + row.updated_at = now + + +def set_chat_task(user_id: UUID, task_id: UUID) -> None: + now = datetime.now(timezone.utc) + with session_scope() as s: + row = s.get(WeChatBotBinding, user_id) + if row is not None: + row.chat_task_id = task_id + row.updated_at = now + + +def unbind(user_id: UUID) -> bool: + """解绑(标 revoked,不物理删 → 保留轨迹)。返回是否有绑定被改。""" + now = datetime.now(timezone.utc) + with session_scope() as s: + row = s.get(WeChatBotBinding, user_id) + if row is None: + return False + row.status = "revoked" + row.updated_at = now + return True + + +# ─────────────────────────── 推送 ─────────────────────────── + +@dataclass +class PushResult: + ok: bool + channel: str = "clawbot" + # sent | no_binding | never_opened | token_stale | error:<...> + reason: str = "" + + +def _token_fresh(snap: BindingSnapshot) -> bool: + if not snap.context_token or snap.context_token_at is None: + return False + at = snap.context_token_at + if at.tzinfo is None: + at = at.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - at) < CONTEXT_TOKEN_TTL + + +def push_clawbot( + user_id: UUID, text: str = "", file_path: Optional[str] = None +) -> PushResult: + """主动推一条到用户个人微信。仅在 24h 窗口内可用,否则返回 reason 供兜底。""" + snap = get_binding(user_id) + if snap is None or snap.status != "active": + return PushResult(False, reason="no_binding") + if not snap.user_im_id or not snap.context_token: + return PushResult(False, reason="never_opened") # 冷启动:用户从未开口 + if not _token_fresh(snap): + return PushResult(False, reason="token_stale") # 超 24h 未互动 + client = ILinkClient(snap.bot_token, snap.base_url) + try: + if text: + client.send_text(snap.user_im_id, snap.context_token, text) + if file_path: + client.send_file(snap.user_im_id, snap.context_token, file_path) + except Exception as e: # noqa: BLE001 —— 调用方据 reason 决定兜底 + return PushResult(False, reason=f"error:{type(e).__name__}") + return PushResult(True, reason="sent") + + +@dataclass +class DeliveryReport: + results: list[PushResult] = field(default_factory=list) + + @property + def delivered(self) -> bool: + return any(r.ok for r in self.results) + + +def send_to_user( + user_id: UUID, text: str = "", file_path: Optional[str] = None +) -> DeliveryReport: + """渠道抽象:按用户已绑渠道投递。当前仅 ClawBot;企业微信(渠道 B)后续追加。""" + report = DeliveryReport() + if clawbot_enabled(): + report.results.append(push_clawbot(user_id, text, file_path)) + # TODO 渠道 B:if wecom_configured(): report.results.append(push_wecom(user_id, text, file_path)) + return report diff --git a/db/migrations/versions/20260624_1000_0012_wechat_bot_bindings.py b/db/migrations/versions/20260624_1000_0012_wechat_bot_bindings.py new file mode 100644 index 0000000..4216fb8 --- /dev/null +++ b/db/migrations/versions/20260624_1000_0012_wechat_bot_bindings.py @@ -0,0 +1,63 @@ +"""wechat_bot_bindings 表(ClawBot 个人微信绑定,DESIGN §8.7 渠道 A). + +Revision ID: 0012 +Revises: 0011 +Create Date: 2026-06-24 + +新增独立表 wechat_bot_bindings —— 不碰现有 schema(公测兼容)。一行 = 一个用户绑定其 +个人微信 ClawBot。bot_token / latest_context_token 存密文(core/wechat/crypto.py)。 +入站长轮询管理器按 status='active' 拉绑定起 getupdates 循环;主动推送用 latest_context_token +(24h 内有效)。详 DESIGN §8.7 / core/storage/models.py。 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import UUID as PG_UUID + + +revision: str = "0012" +down_revision: Union[str, None] = "0011" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "wechat_bot_bindings", + sa.Column( + "user_id", PG_UUID(as_uuid=True), + sa.ForeignKey("users.user_id", ondelete="CASCADE"), primary_key=True, + ), + sa.Column("bot_token", sa.Text(), nullable=False), + sa.Column("bot_im_id", sa.Text(), nullable=True), + sa.Column("user_im_id", sa.Text(), nullable=True), + sa.Column( + "base_url", sa.Text(), nullable=False, + server_default="https://ilinkai.weixin.qq.com", + ), + sa.Column("latest_context_token", sa.Text(), nullable=True), + sa.Column("context_token_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "chat_task_id", PG_UUID(as_uuid=True), + sa.ForeignKey("tasks.task_id", ondelete="SET NULL"), nullable=True, + ), + sa.Column("status", sa.Text(), nullable=False, server_default="active"), + sa.Column( + "created_at", sa.DateTime(timezone=True), + server_default=sa.func.now(), nullable=False, + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), + server_default=sa.func.now(), nullable=False, + ), + ) + # 入站管理器扫 active 绑定起长轮询 + op.create_index( + "ix_wechat_bot_bindings_active", "wechat_bot_bindings", ["status"], + ) + + +def downgrade() -> None: + op.drop_index("ix_wechat_bot_bindings_active", table_name="wechat_bot_bindings") + op.drop_table("wechat_bot_bindings") diff --git a/requirements.txt b/requirements.txt index f4215d0..1a2be8c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -19,6 +19,10 @@ html2text>=2024.0 # 定时任务(§8.5 scheduled_jobs):cron 串 → next_run_at 计算,正确处理 dom/dow OR 语义 + 时区 croniter>=2.0 +# 微信接入(§8.7 ClawBot):segno 渲绑定二维码;cryptography 做凭据列加密 + 文件 AES-128-ECB +segno>=1.6 +cryptography>=42.0 + # §7 B 阶段: Storage 落 PG sqlalchemy>=2.0.0 psycopg[binary]>=3.1.0 diff --git a/scripts/probe_clawbot.py b/scripts/probe_clawbot.py new file mode 100644 index 0000000..3a86c60 --- /dev/null +++ b/scripts/probe_clawbot.py @@ -0,0 +1,150 @@ +"""一次性探测:微信 ClawBot 灰度是否覆盖某个微信号。 + +只做两件事(不碰 zcbot 主体、不落库): + 1. GET get_bot_qrcode 拿二维码 -> 存 qr.png 并自动打开 + 2. 轮询 get_qrcode_status 等扫码确认 -> 报告 status + +判读: + - 接口连不通 / 非 200 -> 本机到 ilinkai 网络不通,换网或在有网机器跑 + - 出码成功、手机扫得动确认 -> 该微信号在灰度内,ClawBot 可用 + - 出码成功、扫了报"不支持" -> 版本不够或未灰度到该号 + +ASCII-only 输出(Windows GBK 控制台)。 +""" +from __future__ import annotations + +import base64 +import os +import random +import sys +import time +import webbrowser + +import httpx + +BASE = "https://ilinkai.weixin.qq.com" +QR_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "clawbot_qr.png") + + +def _uin_header() -> str: + # X-WECHAT-UIN: base64(String(randomUint32())) + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers() -> dict: + return { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": _uin_header(), + } + + +def _save_qr(img_content: str, qrcode_id: str) -> bool: + """实测:qrcode_img_content 是微信深链(https://liteapp.weixin.qq.com/q/...), + 需把该 URL **编码成二维码** 让微信扫,而非当图片下载。 + 兜底:若哪天返回的是真图片字节(data-uri / base64 PNG)则直接存。 + """ + try: + if not img_content: + print(f"[hint] no img content; encode this id manually: {qrcode_id}") + return False + # 情况 A:真图片字节 + if img_content.startswith("data:image"): + data = base64.b64decode(img_content.split(",", 1)[1]) + with open(QR_PATH, "wb") as f: + f.write(data) + print(f"[ok] QR (image) saved -> {QR_PATH}") + return True + # 情况 B(实测):深链 / 任意字符串 -> 自己渲染成二维码 + import segno + print(f"[info] encoding deep-link into QR: {img_content}") + segno.make(img_content, error="m").save(QR_PATH, scale=8, border=3) + print(f"[ok] QR (rendered from deep-link) saved -> {QR_PATH}") + return True + except Exception as e: + print(f"[warn] could not build QR: {type(e).__name__}: {e}") + print(f"[hint] deep-link to scan manually: {img_content}") + return False + + +def main() -> int: + print("[step1] GET get_bot_qrcode ...") + try: + with httpx.Client(timeout=20) as c: + r = c.get( + f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, + headers=_headers(), + ) + except Exception as e: + print(f"[FAIL] network error to {BASE}: {type(e).__name__}: {e}") + print("[judge] host cannot reach ilinkai.weixin.qq.com -> try another network.") + return 2 + + print(f"[http] status={r.status_code}") + body_preview = r.text[:600] + print(f"[body] {body_preview}") + if r.status_code != 200: + print("[judge] non-200 from get_bot_qrcode -> endpoint/params may be wrong or blocked.") + return 3 + + try: + data = r.json() + except Exception: + print("[FAIL] response not JSON; see body above.") + return 3 + + qrcode_id = data.get("qrcode") or data.get("qrcode_id") or "" + img = data.get("qrcode_img_content") or data.get("qrcode_img") or "" + if not qrcode_id: + print("[FAIL] no 'qrcode' field in response; field names differ -> inspect body above.") + return 3 + + if _save_qr(img, qrcode_id): + try: + webbrowser.open("file://" + QR_PATH.replace("\\", "/")) + except Exception: + pass + print("[action] QR opened. Scan it with your phone WeChat NOW.") + else: + print("[action] QR image unavailable; cannot open. See hint above.") + + poll_secs = int(sys.argv[1]) if len(sys.argv) > 1 else 100 + print(f"[step2] polling get_qrcode_status (up to ~{poll_secs}s; Ctrl-C to stop)...") + deadline = time.time() + poll_secs + last = "" + with httpx.Client(timeout=40) as c: + while time.time() < deadline: + try: + r = c.get( + f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qrcode_id}, + headers=_headers(), + ) + st = "" + try: + st = (r.json() or {}).get("status", "") + except Exception: + st = f"(non-json http {r.status_code})" + if st != last: + print(f"[poll] status={st!r}") + last = st + if st == "confirmed": + j = r.json() + tok = j.get("bot_token", "") + base_url = j.get("baseurl") or j.get("base_url") or "" + masked = (tok[:6] + "..." + tok[-4:]) if len(tok) > 12 else "(short)" + print("[SUCCESS] scan confirmed -> this WeChat account IS in the ClawBot rollout.") + print(f"[SUCCESS] bot_token={masked} baseurl={base_url}") + print("[note] token masked on purpose; it is a per-user credential.") + return 0 + except Exception as e: + print(f"[poll] error: {type(e).__name__}: {e}") + time.sleep(2) + print("[timeout] no confirmation within window. Either not scanned in time, or") + print("[timeout] your WeChat lacks the ClawBot entry (version <8.0.70 or not gray-rolled).") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_clawbot_chat.py b/scripts/probe_clawbot_chat.py new file mode 100644 index 0000000..7c650b7 --- /dev/null +++ b/scripts/probe_clawbot_chat.py @@ -0,0 +1,181 @@ +"""探测二:微信 ClawBot 的【对话】与【主动推送】能力(命门验证)。 + +流程(都在一次运行里,不落库): + 1. 扫码绑定拿 bot_token(同探测一) + 2. getupdates 长轮询,等你给「微信 ClawBot」联系人发一条消息 + 3. 收到后,依次测三种发送,逐一报 ret: + A. 带 context_token 回复 -> 验「被动回复」是否通 + B. 等 25s 后,用【同一个】context_token 再发 -> 验「开口一次后能否延迟主动推」 + C. context_token 置空再发 -> 验「冷推(无 token)」是否被拒 + 判读: + A 通 = 双向对话成立 + B 通 = 用户开口一次后可后续推送(简报可走"先开口、后定时推"的弱化版) + C 通 = 可冷推(几乎不可能,但要验) + B/C 都不通 = ClawBot 纯被动回复,定时主动推送这条路不成立 + +ASCII-only 输出。bot_token 不打印。 +""" +from __future__ import annotations + +import base64 +import os +import random +import sys +import time + +import httpx +import segno + +BASE = "https://ilinkai.weixin.qq.com" +QR_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "clawbot_qr.png") +CHANNEL_VER = "1.0.2" + + +def _uin() -> str: + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers(token: str | None = None) -> dict: + h = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": _uin(), + } + if token: + h["Authorization"] = f"Bearer {token}" + return h + + +def bind() -> tuple[str, str] | None: + print("[bind] GET get_bot_qrcode ...") + with httpx.Client(timeout=20) as c: + r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, headers=_headers()) + if r.status_code != 200: + print(f"[FAIL] get_bot_qrcode http {r.status_code}: {r.text[:300]}") + return None + d = r.json() + qid = d.get("qrcode", "") + link = d.get("qrcode_img_content", "") + segno.make(link, error="m").save(QR_PATH, scale=8, border=3) + try: + import webbrowser + webbrowser.open("file://" + QR_PATH.replace("\\", "/")) + except Exception: + pass + print(f"[bind] QR opened -> {QR_PATH} SCAN IT NOW with phone WeChat.") + deadline = time.time() + 180 + with httpx.Client(timeout=40) as c: + last = "" + while time.time() < deadline: + try: + r = c.get(f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qid}, headers=_headers()) + j = r.json() + st = j.get("status", "") + if st != last: + print(f"[bind] status={st!r}") + last = st + if st == "confirmed": + print("[bind] confirmed.") + return j.get("bot_token", ""), (j.get("baseurl") or BASE) + if st == "expired": + print("[bind] QR expired before scan.") + return None + except Exception as e: + print(f"[bind] poll err: {type(e).__name__}: {e}") + time.sleep(2) + print("[bind] timeout waiting for scan.") + return None + + +def _send(client: httpx.Client, token: str, to_user: str, text: str, + context_token: str) -> dict: + body = { + "msg": { + "to_user_id": to_user, + "message_type": 2, + "message_state": 2, + "context_token": context_token, + "item_list": [{"type": 1, "text_item": {"text": text}}], + } + } + r = client.post(f"{BASE}/ilink/bot/sendmessage", + json=body, headers=_headers(token)) + try: + return {"http": r.status_code, "json": r.json()} + except Exception: + return {"http": r.status_code, "text": r.text[:300]} + + +def main() -> int: + b = bind() + if not b: + return 2 + token, base_url = b + global BASE + BASE = base_url or BASE + + print("[chat] now SEND a message (e.g. 'hi') to the WeChat ClawBot contact on your phone.") + print("[chat] waiting via getupdates (up to ~150s)...") + buf = "" + deadline = time.time() + 150 + got = None + with httpx.Client(timeout=40) as c: + while time.time() < deadline and got is None: + try: + r = c.post(f"{BASE}/ilink/bot/getupdates", + json={"get_updates_buf": buf, + "base_info": {"channel_version": CHANNEL_VER}}, + headers=_headers(token)) + j = r.json() + buf = j.get("get_updates_buf", buf) + for m in j.get("msgs", []) or []: + txt = "" + for it in m.get("item_list", []) or []: + txt += (it.get("text_item", {}) or {}).get("text", "") + print(f"[chat] <- from={m.get('from_user_id')} text={txt!r}") + got = m + break + except Exception as e: + print(f"[chat] getupdates err: {type(e).__name__}: {e}") + time.sleep(2) + if got is None: + print("[chat] no message received in window. Re-run and send promptly after scan.") + return 1 + + to_user = got.get("from_user_id", "") + ctx = got.get("context_token", "") + print(f"[chat] captured to_user={to_user} context_token_len={len(ctx)}") + + with httpx.Client(timeout=30) as c: + print("\n[testA] reply WITH context_token ...") + ra = _send(c, token, to_user, "[zcbot 测试A] 收到你的消息,这是带 token 的回复。", ctx) + print(f"[testA] result={ra}") + + print("\n[testB] wait 25s, then push again with the SAME context_token (delayed proactive)...") + time.sleep(25) + rb = _send(c, token, to_user, "[zcbot 测试B] 这是25秒后用同一token的延迟主动推送。", ctx) + print(f"[testB] result={rb}") + + print("\n[testC] push with EMPTY context_token (cold push) ...") + rc = _send(c, token, to_user, "[zcbot 测试C] 这是空token的冷推送。", "") + print(f"[testC] result={rc}") + + def ok(r): + j = r.get("json") or {} + return r.get("http") == 200 and j.get("ret", -1) == 0 + + print("\n========== VERDICT ==========") + print(f"A reply(with token) : {'OK' if ok(ra) else 'FAIL'}") + print(f"B delayed push(same token) : {'OK' if ok(rb) else 'FAIL'}") + print(f"C cold push(empty token) : {'OK' if ok(rc) else 'FAIL'}") + print("Interpretation:") + print(" - A only -> reply-only; scheduled PROACTIVE push NOT possible.") + print(" - A+B -> after user opens chat once, delayed push works (weak push OK).") + print(" - C -> true cold push works (unlikely).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_clawbot_clientid.py b/scripts/probe_clawbot_clientid.py new file mode 100644 index 0000000..26e9c54 --- /dev/null +++ b/scripts/probe_clawbot_clientid.py @@ -0,0 +1,158 @@ +"""探测五(决定性):补上 client_id(每条唯一)+ base_info,重验两件事。 + A. 流式多条:同一 context_token 连发 3 块(client_id 各异,state 1/1/2,间隔300ms) + -> 三块都到 = 多条/长简报可行 + B. finish 后复用:发完 FINISH,等30s,用【同一 context_token】+新 client_id 再发一条(state=2) + -> 到 = context_token 24h 内可复用 -> "用户开口一次后可主动推" 成立(简报推送复活) + +之前失败的最大嫌疑:缺 client_id(后续块无法路由被丢)。需要你发【一条】消息触发。 +ASCII-only,bot_token 不打印。 +""" +from __future__ import annotations + +import base64 +import os +import random +import sys +import time +import uuid + +import httpx +import segno + +BASE = "https://ilinkai.weixin.qq.com" +QR_DIR = os.path.dirname(os.path.abspath(__file__)) +CHANNEL_VER = "1.0.2" + + +def _uin() -> str: + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers(token=None) -> dict: + h = {"Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", "X-WECHAT-UIN": _uin()} + if token: + h["Authorization"] = f"Bearer {token}" + return h + + +def _new_qr(): + with httpx.Client(timeout=20) as c: + r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, headers=_headers()) + if r.status_code != 200: + print(f"[FAIL] http {r.status_code}"); return None + d = r.json() + uniq = os.path.join(QR_DIR, f"clawbot_qr_{int(time.time())}.png") + segno.make(d.get("qrcode_img_content", ""), error="m").save(uniq, scale=8, border=3) + try: + os.startfile(uniq) + except Exception: + pass + print(f"[bind] FRESH QR -> {uniq}") + return d.get("qrcode", "") + + +def bind(): + print("[bind] auto-refresh on expiry; scan whenever ready.") + qid = _new_qr() + if not qid: + return None + deadline = time.time() + 300 + with httpx.Client(timeout=40) as c: + last = "" + while time.time() < deadline: + try: + j = c.get(f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qid}, headers=_headers()).json() + st = j.get("status", "") + if st != last: + print(f"[bind] status={st!r}"); last = st + if st == "confirmed": + return j.get("bot_token", ""), (j.get("baseurl") or BASE) + if st == "expired": + nq = _new_qr() + if not nq: + return None + qid, last = nq, "" + except Exception as e: + print(f"[bind] err {e}") + time.sleep(2) + return None + + +def send(c, token, to_user, text, ctx, state, tag): + cid = uuid.uuid4().hex + body = { + "msg": { + "to_user_id": to_user, + "client_id": cid, + "message_type": 2, + "message_state": state, + "context_token": ctx, + "item_list": [{"type": 1, "text_item": {"text": text}}], + }, + "base_info": {"channel_version": CHANNEL_VER}, + } + r = c.post(f"{BASE}/ilink/bot/sendmessage", json=body, headers=_headers(token)) + try: + j = r.json() + except Exception: + j = r.text[:160] + print(f"[send {tag}] state={state} client_id={cid[:8]} -> http={r.status_code} body={j}") + + +def wait_msg(c, token): + deadline = time.time() + 150 + buf = "" + while time.time() < deadline: + try: + j = c.post(f"{BASE}/ilink/bot/getupdates", + json={"get_updates_buf": buf, + "base_info": {"channel_version": CHANNEL_VER}}, + headers=_headers(token)).json() + buf = j.get("get_updates_buf", buf) + for m in j.get("msgs", []) or []: + txt = "".join((it.get("text_item", {}) or {}).get("text", "") + for it in m.get("item_list", []) or []) + print(f"[recv] <- {txt!r}") + return m + except Exception as e: + print(f"[recv] err {e}"); time.sleep(2) + return None + + +def main() -> int: + b = bind() + if not b: + return 2 + token, base_url = b + global BASE + BASE = base_url or BASE + print("[bind] confirmed.\n[A] SEND one message now (e.g. 'go') ...") + with httpx.Client(timeout=30) as c: + m = wait_msg(c, token) + if not m: + print("no msg; abort."); return 1 + to_user, ctx = m.get("from_user_id", ""), m.get("context_token", "") + + print("[A] streaming 3 chunks WITH client_id (state 1,1,2, 300ms apart)...") + send(c, token, to_user, "[A1] client_id+流式第一段(state=1)", ctx, 1, "A1") + time.sleep(0.3) + send(c, token, to_user, "[A2] client_id+流式第二段(state=1)", ctx, 1, "A2") + time.sleep(0.3) + send(c, token, to_user, "[A3] client_id+末段(state=2 FINISH)", ctx, 2, "A3") + + print("\n[B] wait 30s, then reuse SAME context_token + new client_id (state=2)...") + time.sleep(30) + send(c, token, to_user, "[B] finish后30秒,复用同token主动推(若到=24h可复用)", ctx, 2, "B") + + print("\n========== CHECK YOUR PHONE ==========") + print("Report which arrived:") + print(" [A1]/[A2]/[A3] -> all three = multi-message/streaming OK (need client_id)") + print(" [B] -> arrived = token reusable after finish => PROACTIVE PUSH revives") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_clawbot_file.py b/scripts/probe_clawbot_file.py new file mode 100644 index 0000000..2223200 --- /dev/null +++ b/scripts/probe_clawbot_file.py @@ -0,0 +1,223 @@ +"""探测六:验证 ClawBot 能否发【文件附件】(照官方 @tencent-weixin/openclaw-weixin 协议复刻)。 + +流程(全诊断,每步打印): + 绑定 -> 等你发一条消息(拿 to_user + context_token) -> 造个小 txt -> + md5/随机aeskey(16B)/随机filekey(16B hex) -> AES-128-ECB+PKCS7 加密 -> + POST /ilink/bot/getuploadurl(打印完整返回,字段名不对可据此改) -> + POST 密文到 CDN 拿 header x-encrypted-param -> + sendmessage 带 file_item(type=4) 引用 -> 看手机是否收到文件。 + +字段依据(源码):MessageItemType.FILE=4 / UploadMediaType.FILE=3 / MessageState.FINISH=2, +aes_key = base64(aeskey.hex() 的 ascii 字节)。ASCII-only,bot_token 不打印。 +""" +from __future__ import annotations + +import base64 +import hashlib +import os +import random +import sys +import time +import uuid +from urllib.parse import quote + +import httpx +import segno +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +BASE = "https://ilinkai.weixin.qq.com" +CDN_BASE_DEFAULT = "https://novac2c.cdn.weixin.qq.com/c2c" +QR_DIR = os.path.dirname(os.path.abspath(__file__)) +CHANNEL_VER = "1.0.2" + + +def _uin() -> str: + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers(token=None) -> dict: + h = {"Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", "X-WECHAT-UIN": _uin()} + if token: + h["Authorization"] = f"Bearer {token}" + return h + + +def _new_qr(): + with httpx.Client(timeout=20) as c: + r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, headers=_headers()) + if r.status_code != 200: + print(f"[FAIL] http {r.status_code}"); return None + d = r.json() + uniq = os.path.join(QR_DIR, f"clawbot_qr_{int(time.time())}.png") + segno.make(d.get("qrcode_img_content", ""), error="m").save(uniq, scale=8, border=3) + try: + os.startfile(uniq) + except Exception: + pass + print(f"[bind] FRESH QR -> {uniq}") + return d.get("qrcode", "") + + +def bind(): + print("[bind] auto-refresh on expiry; scan whenever ready.") + qid = _new_qr() + if not qid: + return None + deadline = time.time() + 300 + with httpx.Client(timeout=40) as c: + last = "" + while time.time() < deadline: + try: + j = c.get(f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qid}, headers=_headers()).json() + st = j.get("status", "") + if st != last: + print(f"[bind] status={st!r}"); last = st + if st == "confirmed": + return j.get("bot_token", ""), (j.get("baseurl") or BASE) + if st == "expired": + nq = _new_qr() + if not nq: + return None + qid, last = nq, "" + except Exception as e: + print(f"[bind] err {e}") + time.sleep(2) + return None + + +def wait_msg(c, token): + deadline = time.time() + 150 + buf = "" + while time.time() < deadline: + try: + j = c.post(f"{BASE}/ilink/bot/getupdates", + json={"get_updates_buf": buf, + "base_info": {"channel_version": CHANNEL_VER}}, + headers=_headers(token)).json() + buf = j.get("get_updates_buf", buf) + for m in j.get("msgs", []) or []: + txt = "".join((it.get("text_item", {}) or {}).get("text", "") + for it in m.get("item_list", []) or []) + print(f"[recv] <- {txt!r}") + return m + except Exception as e: + print(f"[recv] err {e}"); time.sleep(2) + return None + + +def aes_ecb_pkcs7(plain: bytes, key: bytes) -> bytes: + padder = padding.PKCS7(128).padder() + padded = padder.update(plain) + padder.finalize() + enc = Cipher(algorithms.AES(key), modes.ECB()).encryptor() + return enc.update(padded) + enc.finalize() + + +def main() -> int: + b = bind() + if not b: + return 2 + token, base_url = b + global BASE + BASE = base_url or BASE + print("[bind] confirmed.\n[file] SEND one message now (e.g. 'file') ...") + + with httpx.Client(timeout=30) as c: + m = wait_msg(c, token) + if not m: + print("no msg; abort."); return 1 + to_user, ctx = m.get("from_user_id", ""), m.get("context_token", "") + + # 1) 造测试文件 + fpath = os.path.join(QR_DIR, "zcbot_filetest.txt") + with open(fpath, "w", encoding="utf-8") as f: + f.write("zcbot 文件发送测试\nClawBot file attachment probe\n" + "x" * 200) + data = open(fpath, "rb").read() + fname = "zcbot_filetest.txt" + rawsize = len(data) + rawmd5 = hashlib.md5(data).hexdigest() + aeskey = random.randbytes(16) + filekey = random.randbytes(16).hex() + cipher = aes_ecb_pkcs7(data, aeskey) + filesize = len(cipher) + print(f"[file] {fname} rawsize={rawsize} md5={rawmd5} filesize(enc)={filesize}") + + # 2) getuploadurl + up_body = { + "filekey": filekey, "media_type": 3, "to_user_id": to_user, + "rawsize": rawsize, "rawfilemd5": rawmd5, "filesize": filesize, + "no_need_thumb": True, "aeskey": aeskey.hex(), + "base_info": {"channel_version": CHANNEL_VER}, + } + ru = c.post(f"{BASE}/ilink/bot/getuploadurl", json=up_body, headers=_headers(token)) + print(f"[getuploadurl] http={ru.status_code}") + try: + uj = ru.json() + except Exception: + print(f"[getuploadurl] non-json: {ru.text[:300]}"); return 3 + print(f"[getuploadurl] resp={uj}") + + # 3) 解析上传 URL(字段名不确定,多名兜底) + full = (uj.get("upload_full_url") or uj.get("uploadFullUrl") + or uj.get("full_url") or uj.get("url")) + param = (uj.get("upload_param") or uj.get("uploadParam") or uj.get("param")) + cdn_base = uj.get("cdn_base_url") or uj.get("cdnBaseUrl") or CDN_BASE_DEFAULT + if full: + cdn_url = full + elif param: + # 源码模板:?encrypted_query_param=&filekey= + cdn_url = (f"{cdn_base}/upload?encrypted_query_param={quote(param)}" + f"&filekey={quote(filekey)}") + else: + print("[FAIL] no upload url/param in resp; inspect resp above to fix field names.") + return 4 + print(f"[upload] POST ciphertext -> {cdn_url[:120]}...") + + # 4) 上传密文到 CDN + rc = c.post(cdn_url, content=cipher, + headers={"Content-Type": "application/octet-stream"}) + download_param = rc.headers.get("x-encrypted-param") + print(f"[upload] http={rc.status_code} x-encrypted-param={download_param!r}") + if not download_param: + print(f"[upload] resp headers={dict(rc.headers)} body={rc.text[:200]}") + print("[FAIL] no x-encrypted-param returned; upload likely rejected.") + return 5 + + # 5) sendmessage 带 file_item + msg_body = { + "msg": { + "from_user_id": "", "to_user_id": to_user, + "client_id": f"openclaw-weixin-{uuid.uuid4().hex}", + "message_type": 2, "message_state": 2, "context_token": ctx, + "item_list": [{ + "type": 4, + "file_item": { + "media": { + "encrypt_query_param": download_param, + "aes_key": base64.b64encode(aeskey.hex().encode()).decode(), + "encrypt_type": 1, + }, + "file_name": fname, + "len": str(rawsize), + }, + }], + }, + "base_info": {"channel_version": CHANNEL_VER}, + } + rs = c.post(f"{BASE}/ilink/bot/sendmessage", json=msg_body, headers=_headers(token)) + try: + sj = rs.json() + except Exception: + sj = rs.text[:200] + print(f"[sendmessage file] http={rs.status_code} body={sj}") + + print("\n========== CHECK YOUR PHONE ==========") + print(f"Did a file '{fname}' arrive in the WeChat ClawBot chat (openable)?") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_clawbot_stream.py b/scripts/probe_clawbot_stream.py new file mode 100644 index 0000000..634911a --- /dev/null +++ b/scripts/probe_clawbot_stream.py @@ -0,0 +1,147 @@ +"""探测四:验证 ClawBot 流式/多条回复(message_state 非 FINISH 是关键)。 + +上轮发现:message_state=2 = FINISH,会"封口"本轮,故第二条被丢。 +本轮:同一 context_token 连发三段——前两段 state=1(未结束),末段 state=2(FINISH), +看手机收到的形态: + - 三条独立气泡 AAA / BBB / CCC -> 支持多条独立消息 + - 一条气泡里 AAABBBCCC(增长) -> 流式增量(delta),拼成一条 + - 只剩 CCC -> 流式覆盖(cumulative,末值胜) +据此定长简报的发法。需要你发【一条】消息触发。bot_token 不打印。ASCII-only。 +""" +from __future__ import annotations + +import base64 +import os +import random +import sys +import time + +import httpx +import segno + +BASE = "https://ilinkai.weixin.qq.com" +QR_DIR = os.path.dirname(os.path.abspath(__file__)) +CHANNEL_VER = "1.0.2" + + +def _uin() -> str: + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers(token: str | None = None) -> dict: + h = {"Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", "X-WECHAT-UIN": _uin()} + if token: + h["Authorization"] = f"Bearer {token}" + return h + + +def _new_qr() -> str | None: + with httpx.Client(timeout=20) as c: + r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, headers=_headers()) + if r.status_code != 200: + print(f"[FAIL] http {r.status_code}: {r.text[:200]}"); return None + d = r.json() + uniq = os.path.join(QR_DIR, f"clawbot_qr_{int(time.time())}.png") + segno.make(d.get("qrcode_img_content", ""), error="m").save(uniq, scale=8, border=3) + try: + os.startfile(uniq) + except Exception: + pass + print(f"[bind] FRESH QR -> {uniq}") + return d.get("qrcode", "") + + +def bind() -> tuple[str, str] | None: + print("[bind] auto-refresh on expiry; scan whenever ready.") + qid = _new_qr() + if not qid: + return None + deadline = time.time() + 300 + with httpx.Client(timeout=40) as c: + last = "" + while time.time() < deadline: + try: + j = c.get(f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qid}, headers=_headers()).json() + st = j.get("status", "") + if st != last: + print(f"[bind] status={st!r}"); last = st + if st == "confirmed": + return j.get("bot_token", ""), (j.get("baseurl") or BASE) + if st == "expired": + print("[bind] expired -> new QR"); + nq = _new_qr() + if not nq: + return None + qid, last = nq, "" + continue + except Exception as e: + print(f"[bind] err {type(e).__name__}: {e}") + time.sleep(2) + return None + + +def send(c, token, to_user, text, ctx, state): + body = {"msg": {"to_user_id": to_user, "message_type": 2, "message_state": state, + "context_token": ctx, + "item_list": [{"type": 1, "text_item": {"text": text}}]}} + r = c.post(f"{BASE}/ilink/bot/sendmessage", json=body, headers=_headers(token)) + try: + j = r.json() + except Exception: + j = r.text[:200] + print(f"[send] state={state} text={text!r} -> http={r.status_code} body={j}") + + +def wait_msg(c, token): + deadline = time.time() + 150 + buf = "" + while time.time() < deadline: + try: + j = c.post(f"{BASE}/ilink/bot/getupdates", + json={"get_updates_buf": buf, + "base_info": {"channel_version": CHANNEL_VER}}, + headers=_headers(token)).json() + buf = j.get("get_updates_buf", buf) + for m in j.get("msgs", []) or []: + txt = "".join((it.get("text_item", {}) or {}).get("text", "") + for it in m.get("item_list", []) or []) + print(f"[recv] <- {txt!r}") + return m + except Exception as e: + print(f"[recv] err {type(e).__name__}: {e}"); time.sleep(2) + return None + + +def main() -> int: + b = bind() + if not b: + return 2 + token, base_url = b + global BASE + BASE = base_url or BASE + print("[bind] confirmed.\n[stream] SEND one message now (e.g. 'go') ...") + with httpx.Client(timeout=30) as c: + m = wait_msg(c, token) + if not m: + print("[stream] no msg; abort."); return 1 + to_user, ctx = m.get("from_user_id", ""), m.get("context_token", "") + print("[stream] sending 3 parts with same token (state 1,1,2)...") + send(c, token, to_user, "AAA-第一段(state=1)", ctx, 1) + time.sleep(1) + send(c, token, to_user, "BBB-第二段(state=1)", ctx, 1) + time.sleep(1) + send(c, token, to_user, "CCC-第三段(state=2,FINISH)", ctx, 2) + print("\n========== CHECK YOUR PHONE ==========") + print("Which form did you get?") + print(" (a) three separate bubbles: AAA / BBB / CCC -> multi-message OK") + print(" (b) one bubble growing: AAABBBCCC -> streaming delta-append") + print(" (c) one bubble only: CCC -> streaming cumulative(last wins)") + print(" (d) only AAA / nothing else -> still single") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_clawbot_token.py b/scripts/probe_clawbot_token.py new file mode 100644 index 0000000..4d0a03c --- /dev/null +++ b/scripts/probe_clawbot_token.py @@ -0,0 +1,168 @@ +"""探测三:钉死 ClawBot 的 context_token 语义(决定拉取式简报 + 长回复可行性)。 + +要回答两个问题: + T1 多发:一条用户消息收到后,用【同一个新鲜 token】连发两条回复 + -> 第二条到不到 = 能否分段/多条回复(长简报关键) + T2 延迟:第二条用户消息收到后,【先不回】,等 25s,再用那条【没用过的】token 回一次 + -> 到不到 = token 是否限时(能否把回复推迟一会儿) + +需要你【先后发两条消息】给「微信 ClawBot」(比如先发 1,再发 2)。 +结果以手机实收为准(接口返空 body 不可信)。bot_token 不打印。ASCII-only。 +""" +from __future__ import annotations + +import base64 +import os +import random +import sys +import time + +import httpx +import segno + +BASE = "https://ilinkai.weixin.qq.com" +QR_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "clawbot_qr.png") +CHANNEL_VER = "1.0.2" + + +def _uin() -> str: + return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode() + + +def _headers(token: str | None = None) -> dict: + h = {"Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": _uin()} + if token: + h["Authorization"] = f"Bearer {token}" + return h + + +def _new_qr() -> str | None: + """拉一张新二维码、弹窗,返回 qrcode id;失败返回 None。""" + with httpx.Client(timeout=20) as c: + r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode", + params={"bot_type": "3"}, headers=_headers()) + if r.status_code != 200: + print(f"[FAIL] get_bot_qrcode http {r.status_code}: {r.text[:200]}") + return None + d = r.json() + qid = d.get("qrcode", "") + uniq = os.path.join(os.path.dirname(QR_PATH), f"clawbot_qr_{int(time.time())}.png") + segno.make(d.get("qrcode_img_content", ""), error="m").save(uniq, scale=8, border=3) + try: + os.startfile(uniq) + except Exception: + try: + import webbrowser + webbrowser.open("file://" + uniq.replace("\\", "/")) + except Exception: + pass + print(f"[bind] FRESH QR -> {uniq} (older windows are stale, ignore them)") + return qid + + +def bind() -> tuple[str, str] | None: + """过期自动换新码,直到扫成功或总超时(5min)。消除扫码时间竞争。""" + print("[bind] GET get_bot_qrcode ... (auto-refresh on expiry; scan whenever ready)") + qid = _new_qr() + if not qid: + return None + deadline = time.time() + 300 + with httpx.Client(timeout=40) as c: + last = "" + while time.time() < deadline: + try: + j = c.get(f"{BASE}/ilink/bot/get_qrcode_status", + params={"qrcode": qid}, headers=_headers()).json() + st = j.get("status", "") + if st != last: + print(f"[bind] status={st!r}"); last = st + if st == "confirmed": + return j.get("bot_token", ""), (j.get("baseurl") or BASE) + if st == "expired": + print("[bind] QR expired -> generating a new one ...") + nq = _new_qr() + if not nq: + return None + qid, last = nq, "" + continue + except Exception as e: + print(f"[bind] err {type(e).__name__}: {e}") + time.sleep(2) + print("[bind] overall timeout (5min)."); return None + + +def send(c, token, to_user, text, ctx): + body = {"msg": {"to_user_id": to_user, "message_type": 2, "message_state": 2, + "context_token": ctx, + "item_list": [{"type": 1, "text_item": {"text": text}}]}} + r = c.post(f"{BASE}/ilink/bot/sendmessage", json=body, headers=_headers(token)) + try: + return {"http": r.status_code, "json": r.json()} + except Exception: + return {"http": r.status_code, "text": r.text[:200]} + + +def wait_msg(c, token, buf): + """阻塞等下一条用户消息,返回 (msg, new_buf)。""" + deadline = time.time() + 150 + while time.time() < deadline: + try: + j = c.post(f"{BASE}/ilink/bot/getupdates", + json={"get_updates_buf": buf, + "base_info": {"channel_version": CHANNEL_VER}}, + headers=_headers(token)).json() + buf = j.get("get_updates_buf", buf) + for m in j.get("msgs", []) or []: + txt = "".join((it.get("text_item", {}) or {}).get("text", "") + for it in m.get("item_list", []) or []) + print(f"[recv] <- {txt!r}") + return m, buf + except Exception as e: + print(f"[recv] err {type(e).__name__}: {e}"); time.sleep(2) + return None, buf + + +def main() -> int: + b = bind() + if not b: + return 2 + token, base_url = b + global BASE + BASE = base_url or BASE + print("[bind] confirmed.\n") + + with httpx.Client(timeout=40) as c: + # ---- T1: 同一 token 连发两条 ---- + print("[T1] SEND your 1st message now (e.g. '1') ...") + m, buf = wait_msg(c, token, "") + if not m: + print("[T1] no msg; abort."); return 1 + to_user, ctx = m.get("from_user_id", ""), m.get("context_token", "") + r1a = send(c, token, to_user, "[T1-a] 同token第一条(立即)", ctx) + r1b = send(c, token, to_user, "[T1-b] 同token第二条(紧接)", ctx) + print(f"[T1] sent two with same token. http: a={r1a.get('http')} b={r1b.get('http')}") + + # ---- T2: 收到后不回,延迟 25s 再用未用过的 token 回一次 ---- + print("\n[T2] SEND your 2nd message now (e.g. '2') ...") + m2, buf = wait_msg(c, token, buf) + if not m2: + print("[T2] no msg; skip."); + else: + to_user2, ctx2 = m2.get("from_user_id", ""), m2.get("context_token", "") + print("[T2] received; NOT replying; waiting 25s...") + time.sleep(25) + r2 = send(c, token, to_user2, "[T2] 延迟25秒,未用过的token回复", ctx2) + print(f"[T2] sent after delay. http={r2.get('http')}") + + print("\n========== CHECK YOUR PHONE ==========") + print("Report which of these arrived in the WeChat ClawBot chat:") + print(" [T1-a] 同token第一条(立即)") + print(" [T1-b] 同token第二条(紧接) <- if arrives: multi-message per turn OK") + print(" [T2] 延迟25秒,未用过的token回复 <- if arrives: token is time-windowed, deferred reply OK") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/wechat_bot.py b/tools/wechat_bot.py new file mode 100644 index 0000000..b57dd60 --- /dev/null +++ b/tools/wechat_bot.py @@ -0,0 +1,84 @@ +"""微信主动推送工具(DESIGN §8.7 渠道抽象)。host-side,按 user_id 注入。 + +agent 在(交互 / 定时)run 里调:把一段文本 + 可选产物文件**主动推**到用户已绑的微信 +(当前 ClawBot 个人微信;企业微信渠道后续在 `service.send_to_user` 内追加)。仅当渠道开关 +在才挂(沿用「有开关才注册」§3.4)。 + +ClawBot 推送有 **24h 窗口**约束:用户超期未在微信里跟机器人说过话 / 从未开口 → 推不出, +工具返回明确 reason,agent 可改走 `send_email`。密钥/凭据只在 host 进程,绝不进沙箱。 +""" +from __future__ import annotations + +from pathlib import Path +from typing import Optional +from uuid import UUID + +from core.wechat import service +from .base import Tool + + +def wechat_push_available() -> bool: + """任一微信渠道可用(当前 = ClawBot 开关在;后续 or 企业微信配齐)。""" + return service.clawbot_enabled() + + +_REASON_HINT = { + "no_binding": "用户还没绑定微信(让其在 zcbot 网页扫码绑定)", + "never_opened": "用户绑了但从未在微信里跟机器人说过话(主动推需先开口一次)", + "token_stale": "距用户上次在微信互动已超 24h,主动推送窗口已过(让其发条消息即可恢复)", +} + + +class WechatPushTool(Tool): + name = "wechat_push" + description = ( + "Proactively push a short text message (and optionally one result file, e.g. a .docx/.pdf " + "report) to the user's bound WeChat. Use when the user asks to send something to their " + "WeChat, or when a scheduled task should deliver its output there. NOTE: WeChat push only " + "works if the user has messaged the bot within the last ~24h; if it returns a window/binding " + "error, fall back to send_email. The file path is relative to the working directory." + ) + parameters = { + "type": "object", + "properties": { + "text": {"type": "string", "description": "Message text to push (plain text)."}, + "file": { + "type": "string", + "description": "Optional path (relative to working dir) of one file to attach, e.g. 'report.docx'.", + }, + }, + "required": ["text"], + } + + def __init__(self, user_id: UUID, base_dir=None, user_root=None) -> None: + super().__init__(base_dir=base_dir, user_root=user_root) + self.user_id = user_id + + def execute(self, text: str = "", file: Optional[str] = None) -> str: + text = (text or "").strip() + if not text and not file: + return "[Error] text 不能为空" + + fpath: Optional[str] = None + if file and file.strip(): + p = self._resolve(file.strip()).resolve() + if self.user_root is not None: + try: + p.relative_to(self.user_root.resolve()) + except ValueError: + return f"[Error] 文件路径越界(必须在工作目录内): {file}" + if not p.is_file(): + return f"[Error] 文件不存在: {file}" + fpath = str(p) + + report = service.send_to_user(self.user_id, text, fpath) + if report.delivered: + n = "(含 1 个文件)" if fpath else "" + return f"[ok] 已推送到微信 {n}".strip() + + if not report.results: + return "[Error] 没有可用的微信渠道(未开启 / 未配置)" + # 取首个失败原因给 agent 决定是否改走 send_email + r = report.results[0] + hint = _REASON_HINT.get(r.reason, r.reason) + return f"[Error] 微信推送未送达({r.channel}: {r.reason})。{hint}。可改用 send_email。" diff --git a/web/app.py b/web/app.py index 51f004a..3ab301c 100644 --- a/web/app.py +++ b/web/app.py @@ -813,6 +813,7 @@ def create_app() -> FastAPI: None, lambda: deliver_notify( snap["notify"], job_name=snap["name"], working_dir=fs_dir, tz=snap["tz"], + user_id=snap["user_id"], ) ) except Exception as e: @@ -848,6 +849,81 @@ def create_app() -> FastAPI: if scheduler_enabled: print(f"[scheduler] enabled (tick={sched_tick}s)") + # ── 微信(ClawBot)入站长轮询管理器(§8.7)── 仅当 ZCBOT_WECHAT_BOT_ENABLED 在。 + # 每个 active 绑定一条 getupdates 长轮询;收到消息 → 跑用户常驻「微信」task → 回复发回。 + from core.wechat.service import clawbot_enabled + wechat_stop = asyncio.Event() + wechat_task = None + + async def _run_wechat_message(uid: UUID, text: str) -> str: + """微信入站一条消息:解析/建用户常驻「微信」task → 抢 run 锁 → _run_agent_bg → 取回复。""" + from core.agent_builder import resolve_workspace, working_dir_from_name + from core.storage.utils import ensure_local_task_row + from core.wechat import service as _wx + from core.wechat.inbound import extract_last_assistant_text + + snap = await asyncio.to_thread(_wx.get_binding, uid) + if snap is None: + return "" + profile, model_id = _resolve_model_profile("") + ws = resolve_workspace(None, _cfg) + tid = snap.chat_task_id + need_create = tid is None + if not need_create: + with session_scope() as s: + exists = s.execute( + select(Task.task_id).where(Task.task_id == tid, Task.deleted_at.is_(None)) + ).first() + if exists is None: + need_create = True + if need_create: + tid = uuid4() + fs_dir = working_dir_from_name(ws, uid, f"wechat-{str(uid)[:8]}") + fs_dir.mkdir(parents=True, exist_ok=True) + ensure_local_task_row( + task_id=tid, name="微信对话", working_dir=to_db_path(fs_dir), + skill="", user_id=uid, model=model_id, model_profile=profile, + description="(微信 ClawBot 对话)", + ) + await asyncio.to_thread(_wx.set_chat_task, uid, tid) + + # 抢 run 锁:正忙 → 提示稍候(同用户串行,inbound loop 本就串行) + with session_scope() as s: + row = s.execute( + select(Task.run_status).where(Task.task_id == tid).with_for_update() + ).first() + if row is None: + return "[出错] 对话 task 不存在" + if row.run_status in ("running", "cancelling"): + return "上一条还在处理中,请稍候再发。" + s.execute(update(Task).where(Task.task_id == tid).values( + run_status="running", run_error=None)) + + broker.start(tid) + runner = asyncio.create_task(asyncio.to_thread( + _run_agent_bg, tid, uid, text, "", "", False, + )) + app.state.inflight[runner] = tid + runner.add_done_callback(lambda t: app.state.inflight.pop(t, None)) + await runner + + with session_scope() as s: + st = s.execute( + select(Task.run_status, Task.run_error).where(Task.task_id == tid) + ).first() + if st is not None and st.run_status == "error": + return f"[出错] {st.run_error}" + reply = await asyncio.to_thread(extract_last_assistant_text, tid) + return reply or "(本轮无文本回复)" + + if clawbot_enabled(): + from core.wechat.inbound import run_inbound_manager + wechat_task = asyncio.create_task( + run_inbound_manager(_run_wechat_message, wechat_stop), + name="wechat-inbound", + ) + print("[wechat] ClawBot inbound manager enabled") + # Sandbox pool(§7.5):仅当 ZCBOT_SANDBOX_BACKEND=docker 时启用。 # 启动钩子:① init_pool(创建 docker network + pool 实例)② shutdown_all 清 # 前驱孤儿(上次进程留下的 zcbot-sandbox-* 容器,内存 _last_active 为空, @@ -945,6 +1021,13 @@ def create_app() -> FastAPI: await scheduler_task except (asyncio.CancelledError, Exception): pass + if wechat_task is not None: + wechat_stop.set() + wechat_task.cancel() + try: + await wechat_task + except (asyncio.CancelledError, Exception): + pass if sandbox_reaper_task is not None: sandbox_reaper_task.cancel() try: @@ -1005,6 +1088,67 @@ def create_app() -> FastAPI: """ return {"user_id": str(user_id), "role": get_user_role(user_id) or "user"} + # ───────────── 微信接入(ClawBot,§8.7)───────────── + + @app.post("/v1/wechat/bind/qrcode", tags=["wechat"]) + async def wechat_bind_qrcode(user_id: UUID = Depends(require_user)): + """起一张 ClawBot 绑定二维码(渲成 PNG data-uri)。前端展示,用户手机微信扫; + 二维码 TTL ~1min,前端轮询到 expired 后重调本端点换码。""" + import base64 as _b64 + import io as _io + + import segno + + from core.wechat import ilink + qr = await asyncio.to_thread(ilink.get_bot_qrcode) + buf = _io.BytesIO() + segno.make(qr.deeplink, error="m").save(buf, kind="png", scale=6, border=3) + data_uri = "data:image/png;base64," + _b64.b64encode(buf.getvalue()).decode() + return {"qrcode_id": qr.qrcode_id, "qr_png": data_uri} + + @app.get("/v1/wechat/bind/status", tags=["wechat"]) + async def wechat_bind_status(qrcode_id: str, user_id: UUID = Depends(require_user)): + """轮询扫码状态(服务端长轮询,hold 数十秒)。confirmed → 写绑定。 + 返回 {status: wait|confirmed|expired};expired 时前端重起二维码。""" + from core.wechat import ilink + from core.wechat import service as _wx + res = await asyncio.to_thread(ilink.poll_qrcode_status, qrcode_id) + if res.status == "confirmed" and res.bot_token: + await asyncio.to_thread( + _wx.upsert_clawbot_binding, user_id, res.bot_token, + res.base_url or ilink.DEFAULT_BASE, + ) + return {"status": res.status} + + @app.get("/v1/wechat/bind", tags=["wechat"]) + def wechat_bind_get(user_id: UUID = Depends(require_user)): + """当前用户的微信绑定状态(不泄露 token)。""" + from core.wechat import service as _wx + snap = _wx.get_binding(user_id) + if snap is None or snap.status != "active": + return {"bound": False} + return { + "bound": True, + "user_im_id": snap.user_im_id, + "can_push": bool(_wx._token_fresh(snap)), # 24h 窗口内可主动推 + "last_active": _iso(snap.context_token_at), + } + + @app.delete("/v1/wechat/bind", status_code=204, tags=["wechat"]) + def wechat_unbind(user_id: UUID = Depends(require_user)): + from core.wechat import service as _wx + _wx.unbind(user_id) + return + + @app.post("/v1/wechat/test", tags=["wechat"]) + async def wechat_test(user_id: UUID = Depends(require_user)): + """自检:给已绑用户推一条测试消息(需用户近 24h 在微信开口过)。""" + from core.wechat import service as _wx + res = await asyncio.to_thread( + _wx.push_clawbot, user_id, "zcbot 测试消息:绑定成功,这条来自你的 zcbot。" + ) + return {"ok": res.ok, "reason": res.reason} + @app.get("/v1/models", tags=["misc"]) def list_models(user_id: UUID = Depends(require_user)): """列出所有可用 LLM 模型(扫 config/models/*.yaml)。 diff --git a/web/static/wechat_bind.html b/web/static/wechat_bind.html new file mode 100644 index 0000000..c894bf4 --- /dev/null +++ b/web/static/wechat_bind.html @@ -0,0 +1,152 @@ + + + + + +绑定微信 · zcbot + + + +
+

绑定微信(ClawBot)

+

绑定后可在个人微信「微信 ClawBot」里直接和 zcbot 对话;近 24h 内有互动时,定时简报 / 结果也能主动推给你。

+ +
加载中…
+ +
+ + + +
+ + + +
    +
  1. 需个人微信 8.0.70+ 且已灰度到「ClawBot」插件(设置→插件)。
    2. +
  2. 绑定成功后,先在微信里给「微信 ClawBot」发一句话,主动推送才会开启(24h 窗口)。
    3. +
+
+ + + +