Compare commits

...

6 Commits

Author SHA1 Message Date
46bc573777 Release v1.0.66: table default_sort_json for data management
All checks were successful
技能自动化发布 / release (push) Successful in 10s
2026-08-02 12:17:55 +08:00
0002e86729 docs: columnMapping for replaces create/update lookup reuse
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-07-31 15:14:44 +08:00
caf0191366 feat: document optional replaces for CRUD mutation binding
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-07-31 14:43:15 +08:00
ba284922e5 docs: data-management column lookup via options_json
All checks were successful
技能自动化发布 / release (push) Successful in 8s
2026-07-31 13:37:51 +08:00
7cf139dfd2 docs: Skill Action interaction direct/confirm/form
All checks were successful
技能自动化发布 / release (push) Successful in 7s
2026-07-31 11:15:51 +08:00
6c1ffc73d3 Release v1.0.61: desktop RPA Name/Value and activate fixes
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-25 17:02:07 +08:00
32 changed files with 4179 additions and 86 deletions

View File

@@ -9,6 +9,37 @@
- 小节正文写用户能看懂的变化(新能力、修复、注意事项),避免堆砌内部实现细节 - 小节正文写用户能看懂的变化(新能力、修复、注意事项),避免堆砌内部实现细节
- 打 tag 前**必须**为即将发布的版本新增一节;找不到匹配小节时,本次发布不会更新 changelog 字段 - 打 tag 前**必须**为即将发布的版本新增一节;找不到匹配小节时,本次发布不会更新 changelog 字段
## 1.0.66
- 数据管理支持表级默认排序:`_jiangchang_tables.default_sort_json`(多字段 JSON);宿主优先采用,未配置则仍按 updated_at → created_at → 主键
- 更新 SCHEMA.md 与 `upsert_table_metadata`;旧库幂等补列
## 1.0.65
- `replaces` create/update 可复用表列 lookup:字段名对齐列名,或使用 `bind.columnMapping` / 字段 `column`
- 更新 ACTIONS.md 与 skill-actions.schema.json
## 1.0.64
- Skill Action 新增可选 `replaces: create|update|delete`:接管数据管理默认新增/编辑/删除,无配置时宿主仍直写 SQLite
- 更新 ACTIONS.md 与 skill-actions.schema.json;`replaces` 须配合 `bind.tables`,可与空 `placements` 并用
## 1.0.63
- 数据管理支持字段级 lookup:`options_json` 可声明同库关联表(`ref_table` / `value_column` / `label_column`),宿主新建/编辑时渲染下拉
- 更新 SCHEMA.md 与元数据校验,兼容静态枚举数组与 lookup 对象两种形态
## 1.0.62
- Skill Action 新增 `interaction` / `interactionByPlacement`:声明数据管理点击是直通、轻确认还是参数表单(与 `executionProfile` 正交)
- 明确 `confirmation` 只负责风险文案;cron 触发与 agent 忽略 `interaction`
- 补充 ACTIONS.md 示例(打开文件位置 = direct,打开浏览器 = confirm)
## 1.0.61
- 桌面 RPA 标准反哺:明确 Name≠Value、主窗最大化与禁止误 `SW_RESTORE`、同名叶子须绑父容器
- 更新 `real_desktop_rpa` example:activate/Value 定位、element-map 补全「新的客户/添加」、README 与契约测试
## 1.0.60 ## 1.0.60
- 强化日志规范:验收铁律、固定事件词表、失败须先写文件日志再 print - 强化日志规范:验收铁律、固定事件词表、失败须先写文件日志再 print

View File

@@ -1,7 +1,7 @@
--- ---
name: 技能开发模板(通用业务版) name: 技能开发模板(通用业务版)
description: "OpenClaw 通用业务技能开发模板,供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。" description: "OpenClaw 通用业务技能开发模板,供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。"
version: 1.0.60 version: 1.0.66
author: 深圳匠厂科技有限公司 author: 深圳匠厂科技有限公司
metadata: metadata:
openclaw: openclaw:

View File

@@ -28,6 +28,11 @@
"enum": ["toolbar", "row", "batch", "cron", "agent", "skill-detail"], "enum": ["toolbar", "row", "batch", "cron", "agent", "skill-detail"],
"description": "稳定支持 toolbar/row/batch/cron/agent/skill-detail(对齐匠厂宿主)。placements 与 executionProfile 正交,不做条件限制。" "description": "稳定支持 toolbar/row/batch/cron/agent/skill-detail(对齐匠厂宿主)。placements 与 executionProfile 正交,不做条件限制。"
}, },
"mutationKind": {
"type": "string",
"enum": ["create", "update", "delete"],
"description": "可选。接管数据管理默认按钮:create=新增、update=编辑、delete=删除。须配合非空 bind.tables;无配置时宿主仍直写 SQLite。"
},
"scalarArg": { "scalarArg": {
"type": ["string", "number", "boolean"] "type": ["string", "number", "boolean"]
}, },
@@ -52,6 +57,12 @@
"readOnly": { "readOnly": {
"type": "boolean", "type": "boolean",
"description": "宿主数据管理表单只读展示该字段(仍会随提交传入 entrypoint);用于行内/勾选预填的主键等" "description": "宿主数据管理表单只读展示该字段(仍会随提交传入 entrypoint);用于行内/勾选预填的主键等"
},
"column": {
"type": "string",
"minLength": 1,
"pattern": "^[a-z][a-z0-9_]*$",
"description": "可选。绑定业务表物理列名;replaces create/update 时宿主可复用该列的中文名与 lookup 下拉"
} }
} }
}, },
@@ -120,9 +131,18 @@
"type": "object", "type": "object",
"additionalProperties": { "type": "string", "minLength": 1 }, "additionalProperties": { "type": "string", "minLength": 1 },
"description": "可选。将 $row.$pk / $row.col / $selection.ids 等映射到 inputSchema 字段名,供 row/batch 预填。" "description": "可选。将 $row.$pk / $row.col / $selection.ids 等映射到 inputSchema 字段名,供 row/batch 预填。"
},
"columnMapping": {
"type": "object",
"additionalProperties": {
"type": "string",
"minLength": 1,
"pattern": "^[a-z][a-z0-9_]*$"
},
"description": "可选。将 inputSchema 属性名映射到业务表物理列名,供 replaces create/update 抽屉复用列 displayName / lookup options。"
} }
}, },
"description": "表级绑定;toolbar 必填。可选 inputMapping 供行内/批量预填。concurrency/locks 勿写入。" "description": "表级绑定;toolbar 必填。可选 inputMapping / columnMapping。concurrency/locks 勿写入。"
}, },
"action": { "action": {
"type": "object", "type": "object",
@@ -144,10 +164,11 @@
}, },
"placements": { "placements": {
"type": "array", "type": "array",
"minItems": 1, "minItems": 0,
"items": { "$ref": "#/$defs/placement" }, "items": { "$ref": "#/$defs/placement" },
"description": "决定 Action 出现在哪里;与 executionProfile 正交" "description": "决定 Action 出现在哪里;与 executionProfile 正交。仅 replaces 接管默认 CRUD 时可为空数组。"
}, },
"replaces": { "$ref": "#/$defs/mutationKind" },
"entrypoint": { "$ref": "#/$defs/entrypoint" }, "entrypoint": { "$ref": "#/$defs/entrypoint" },
"inputSchema": { "$ref": "#/$defs/inputSchema" }, "inputSchema": { "$ref": "#/$defs/inputSchema" },
"confirmation": { "$ref": "#/$defs/confirmation" }, "confirmation": { "$ref": "#/$defs/confirmation" },
@@ -156,6 +177,27 @@
"enum": ["sync", "async"], "enum": ["sync", "async"],
"description": "sync=调用方等待结构化结果(不建 Job);async=宿主建后台 Job 并进任务中心。必须显式写出;长耗时建议 async,但不限制 placements。" "description": "sync=调用方等待结构化结果(不建 Job);async=宿主建后台 Job 并进任务中心。必须显式写出;长耗时建议 async,但不限制 placements。"
}, },
"interaction": {
"type": "string",
"enum": ["auto", "direct", "confirm", "form"],
"description": "数据管理等可点击入口的交互方式(与 executionProfile 正交)。direct=一点即跑;confirm=轻量确认(文案用 confirmation.message);form=参数表单;auto/省略=宿主按 inputSchema+confirmation 推断。cron 触发与 agent 忽略本字段。"
},
"interactionByPlacement": {
"type": "object",
"additionalProperties": false,
"properties": {
"toolbar": { "type": "string", "enum": ["direct", "confirm", "form"] },
"row": { "type": "string", "enum": ["direct", "confirm", "form"] },
"batch": { "type": "string", "enum": ["direct", "confirm", "form"] },
"cron": { "type": "string", "enum": ["direct", "confirm", "form"] },
"agent": { "type": "string", "enum": ["direct", "confirm", "form"] },
"skill-detail": { "type": "string", "enum": ["direct", "confirm", "form"] },
"create": { "type": "string", "enum": ["direct", "confirm", "form"] },
"update": { "type": "string", "enum": ["direct", "confirm", "form"] },
"delete": { "type": "string", "enum": ["direct", "confirm", "form"] }
},
"description": "按 placement / replaces 覆盖 interaction(只能写 concrete 模式,不能写 auto)。"
},
"bind": { "$ref": "#/$defs/bind" } "bind": { "$ref": "#/$defs/bind" }
}, },
"if": { "if": {
@@ -169,7 +211,17 @@
"then": { "then": {
"required": ["bind"] "required": ["bind"]
}, },
"description": "concurrency/locks 为未稳定预留字段,勿写入新技能 manifest。placements 含 toolbar 时必须声明 bind.tables。" "allOf": [
{
"if": {
"required": ["replaces"]
},
"then": {
"required": ["bind"]
}
}
],
"description": "concurrency/locks 为未稳定预留字段,勿写入新技能 manifest。placements 含 toolbar 或声明 replaces 时必须声明 bind.tables。无 replaces 时 placements 建议非空(模板测试强制)。"
} }
} }
} }

View File

@@ -158,7 +158,7 @@ scripts/
``` ```
上述 import 来自宿主共享 runtime 安装的 `jiangchang-platform-kit`,不是技能目录副本。 上述 import 来自宿主共享 runtime 安装的 `jiangchang-platform-kit`,不是技能目录副本。
4. **单测/CI 须 mock 离线可过**;业务 `.env` 默认 `real_rpa`,须真实跑通。**不得只 mock 通就交活**;须 release 到匠厂宿主验收(见 §15,含用户体验)。多账号或矩阵能力还须按 `ACCOUNT_MANAGER.md` 覆盖至少两个账号、租约冲突、关闭重开保持登录和异常释放,不能以单账号一两次成功代替规模验证。 4. **单测/CI 须 mock 离线可过**;业务 `.env` 默认 `real_rpa`,须真实跑通。**不得只 mock 通就交活**;须 release 到匠厂宿主验收(见 §15,含用户体验)。多账号或矩阵能力还须按 `ACCOUNT_MANAGER.md` 覆盖至少两个账号、租约冲突、关闭重开保持登录和异常释放,不能以单账号一两次成功代替规模验证。
5. **Windows 桌面**:技术路线已由 `add-friend-wecom` 验证。先读 `examples/real_desktop_rpa/README.md`,复用窗口唯一性、UIA作用域、probe、dry-run/assisted 和失败留证范式;目标软件 selector 必须重新拾取,禁止照抄企业微信控件名。 5. **Windows 桌面**:技术路线已由 `add-friend-wecom` 全链路验证。先读 `examples/real_desktop_rpa/README.md` 与 `development/RPA_DESKTOP.md`(含 Name/Value、最大化激活、同名叶子父容器),复用窗口唯一性、UIA 作用域、probe、dry-run/assisted 和失败留证范式;目标软件 selector 必须重新拾取,禁止照抄企业微信控件名。
6. **手机**:先按 `RPA_MOBILE.md` 完成设备、context、element map 和验收设计。当前 `examples/mobile_rpa/` 只是完成规格,尚无真实成功案例;不得把骨架宣传为已跑通实现。 6. **手机**:先按 `RPA_MOBILE.md` 完成设备、context、element map 和验收设计。当前 `examples/mobile_rpa/` 只是完成规格,尚无真实成功案例;不得把骨架宣传为已跑通实现。
**技术路线红线:**正常界面或获得授权的官方 API 无法完成闭环时,停止并向负责人说明;禁止 AI 或开发者擅自改用破解、反编译、逆向、私有协议、Hook / 注入、验证码破解、设备指纹伪造、自动换号或无限重试。 **技术路线红线:**正常界面或获得授权的官方 API 无法完成闭环时,停止并向负责人说明;禁止 AI 或开发者擅自改用破解、反编译、逆向、私有协议、Hook / 注入、验证码破解、设备指纹伪造、自动换号或无限重试。

View File

@@ -15,14 +15,16 @@
## 1. 已验证基线 ## 1. 已验证基线
真实成功来源:`D:\OpenClaw\client-marketing\add-friend-wecom`。 真实成功来源:`D:\OpenClaw\client-marketing\add-friend-wecom`(含通讯录 → 新的客户 → 添加 → 搜索 → 发送全链路)。
该案例验证了: 该案例验证了:
- 通过进程/窗口类名/标题找到企业微信顶层窗口。 - 通过进程/窗口类名/标题找到企业微信顶层窗口。
- 从 HWND 获取 UIA 根控件。 - 主窗激活后**最大化**,避免侧栏折叠导致控件不可见或坐标漂移。
- 在父窗口和弹窗范围内按 UIA 属性定位。 - 从 HWND 获取 UIA 根控件;在**子业务 pane**(如「企业微信-通讯录」)内缩小作用域。
- 真实激活窗口、点击、逐字输入和轮询状态。 - 区分 UIA `Name` 与 `Value`:可见中文常在 Value / 子 Label,不在 Name。
- 通用叶子名(`item` / `status_layout` / `wbtn_container`)必须绑业务父容器,并用文案校验。
- 真实点击后校验后置状态(标题、专用容器、子窗口)。
- 处理独立弹窗与嵌套弹窗。 - 处理独立弹窗与嵌套弹窗。
- 提供 `probe-ui` 诊断和 `dry-run / assisted / auto` 三档。 - 提供 `probe-ui` 诊断和 `dry-run / assisted / auto` 三档。
@@ -34,31 +36,37 @@ example 只提炼通用模式;企业微信名称、窗口类、控件名、列
操作步骤: 操作步骤:
1. 启动目标软件,登录并手工进入准确页面。 1. 启动目标软件,登录并手工进入准确页面;需要完整侧栏时先**最大化**。
2. 记录目标进程名、PID、主窗口标题和 Window Class。 2. 记录目标进程名、PID、主窗口标题和 Window Class。
3. 用检查器点选目标元素,记录: 3. 用检查器点选目标元素,记录:
- AutomationId - AutomationId
- Name - **Name**(内部 id,可能是 `org_op_new` / `status_layout`)
- ControlType - **Value**(可见中文文案常在这里,如「通讯录」「新的客户」)
- ClassName - ControlType / ClassName / FrameworkId
- FrameworkId - IsEnabled / IsOffscreen / BoundingRectangle
- IsEnabled / IsOffscreen - 父容器链(含业务子窗/pane 标题)
- BoundingRectangle
- 父容器链
4. 保存截图和原始 UIA 树/拾取器导出文件。 4. 保存截图和原始 UIA 树/拾取器导出文件。
5. 从长路径中找出稳定窗口、业务父容器和叶子属性。 5. 从长路径中找出稳定窗口、业务父容器和叶子属性。
6. 删除仅反映当前布局的深层 index;确需 index 时写明业务依据。 6. 删除仅反映当前布局的深层 index;确需 index 时写明业务依据,且优先用文案定位。
7. 写入 element map。 7. 写入 element map(Name 与 Value 分开登记)。
8. 运行 `probe-ui`,验证关键窗口和元素。 8. 运行 `probe-ui`(建议先激活并最大化),验证关键窗口和元素。
9. 关闭重开软件、切换一次页面后再次验证。 9. 关闭重开软件、切换一次页面后再次验证。
### Name / Value(易错)
- 影刀/检查器里的**选择器显示名**(如「新的客户」)≠ 控件 UIA `Name`。
- 许多国产客户端:`Name` 是内部 id,**界面中文在 `Value` 或子控件 Value**。
- 拾取时两者都记;定位时勿把业务中文当成 `Name=` 去搜。
### 从拾取结果提炼定位器 ### 从拾取结果提炼定位器
影刀等工具可能导出很长的 UIA 路径。不要整条照抄: 影刀等工具可能导出很长的 UIA 路径。不要整条照抄:
```text ```text
原始:主窗口 → 多层布局 → 列表 → item[2] → status_layout 原始:主窗口 → 通讯录 pane → wlist_view → item[2] → status_layout
提炼:进程/窗口 → 通讯录列表父容器 → 业务文案或稳定叶子属性 → … → main_chattext(Value=新的客户)
提炼:通讯录 pane → wlist_view → 行内 main_chattext.Value=新的客户 → 点击该行 status_layout
禁止:全树第一个 Name=status_layout(常为「群聊」等其它行)
``` ```
长路径用于排障和理解结构;生产定位器使用最短、唯一、可解释的稳定组合。 长路径用于排障和理解结构;生产定位器使用最短、唯一、可解释的稳定组合。
@@ -67,25 +75,28 @@ example 只提炼通用模式;企业微信名称、窗口类、控件名、列
从高到低: 从高到低:
1. `Process/PID + WindowClass + WindowTitle` 1. `Process/PID + WindowClass + WindowTitle`(及稳定业务子 pane)
2. 目标窗口内 `AutomationId + ControlType` 2. 目标窗口内 `AutomationId + ControlType`
3. 稳定 `Name + ControlType` 3. 稳定内部 `Name + ControlType`(如 `org_op_new`、`searchedit`)
4. 业务父容器 + 稳定子元素属性 4. 业务父容器 + 稳定子元素属性
5. 可见业务文案 + 相对关系 5. **可见业务文案(优先读 Value / 子 Label)+ 相对关系**
6. `ClassName/FrameworkId` 辅助组合 6. `ClassName/FrameworkId` 辅助组合
7. 有业务依据的 index 7. 有业务依据的 index(仅兜底;须写清父容器与序号含义,禁止「过滤后的列表 [2]」与「父节点 child index=2」混用)
8. UIA 元素实时 BoundingRectangle 派生坐标 8. UIA 元素实时 BoundingRectangle 派生坐标
9. 局部截图模板/OCR 9. 局部截图模板/OCR
10. 写死绝对坐标:禁止作为生产方案 10. 写死绝对坐标:禁止作为生产方案
顶层窗口或关键按钮出现多个候选时,返回 `ERROR:ELEMENT_AMBIGUOUS`,不得默认使用第一个。 顶层窗口或关键按钮出现多个候选时,返回 `ERROR:ELEMENT_AMBIGUOUS`,不得默认使用第一个。
同名叶子(如多处 `wbtn_container`)必须限制在业务父容器内(如 `search_external`),并可用子文案(如 Value=`添加`)校验;禁止点搜索旁通用「+」。
## 4. 窗口与控件规则 ## 4. 窗口与控件规则
### 窗口 ### 窗口
- 校验窗口属于预期进程和当前用户会话。 - 校验窗口属于预期进程和当前用户会话。
- 主窗口、搜索窗、确认窗分别建立作用域。 - 主窗口、**应用内业务子 pane**、搜索窗、确认窗分别建立作用域。
- 进入依赖完整侧栏/工具栏的页面时:**激活后最大化**;`activate` **禁止**对已最大化窗口无条件调用 `SW_RESTORE`(会还原成普通窗,导致折叠与坐标漂移)。仅最小化时才 Restore。
- 应用可能重建窗口;页面切换后重新取 HWND 和 UIA root。 - 应用可能重建窗口;页面切换后重新取 HWND 和 UIA root。
- 自动化进程与目标进程权限级别必须兼容;不要为了绕过权限注入目标进程。 - 自动化进程与目标进程权限级别必须兼容;不要为了绕过权限注入目标进程。
- 多开场景必须由账号、PID或明确窗口信息选择,歧义时停止。 - 多开场景必须由账号、PID或明确窗口信息选择,歧义时停止。
@@ -102,11 +113,13 @@ example 只提炼通用模式;企业微信名称、窗口类、控件名、列
### 点击 ### 点击
1. 激活目标窗口。 1. 激活目标窗口(主窗按需最大化;子对话框不要误 Maximize)。
2. 重新定位元素。 2. 重新定位元素。
3. 检查边界矩形。 3. 检查边界矩形。
4. 使用 UIA Click/Invoke 或正常鼠标点击。 4. 使用 UIA Click/Invoke 或正常鼠标点击。
5. 等待后置窗口/控件/业务状态。 5. 等待后置窗口/控件/业务状态(如标题文案、专用容器、子窗口 Class)。
关键导航建议日志:`locate`(策略/文案)→ `click` → `verify`(后置)。这与 [`LOGGING.md`](LOGGING.md)「不必每个 Click 一行」不冲突:要求的是**关键跳转可观测**,不是像素级刷屏。
UIA 点击失败时可以用该元素刚读取的 BoundingRectangle 中心点降级,但必须记录降级、确认窗口前台并验证后置状态;不得静默改点写死坐标。 UIA 点击失败时可以用该元素刚读取的 BoundingRectangle 中心点降级,但必须记录降级、确认窗口前台并验证后置状态;不得静默改点写死坐标。
@@ -133,10 +146,11 @@ UIA 点击失败时可以用该元素刚读取的 BoundingRectangle 中心点降
先检查: 先检查:
1. 权限级别和 Windows 会话是否一致。 1. 权限级别和 Windows 会话是否一致。
2. 是否选错进程或子窗口。 2. 是否选错进程或子窗口;主窗是否未最大化导致侧栏未实例化。
3. 控件是否离屏或尚未实例化。 3. 控件是否离屏或尚未实例化。
4. 应用是否为 Electron(可评估 Playwright)。 4. 应用是否为 Electron(可评估 Playwright)。
5. 是否为自绘、游戏引擎、远程桌面或安全控件。 5. 是否为自绘、游戏引擎、远程桌面或安全控件。
6. 空树/树抖动时:最大化主窗;用 Inspect/影刀附着一次后再 probe(远程桌面工具可能干扰无障碍树)。
只有控件树确认不可用时才使用图像: 只有控件树确认不可用时才使用图像:
@@ -153,11 +167,15 @@ U盾、人脸、动态口令和安全键盘交给人工,不用 OCR、Hook 或
桌面 `health` / `probe-ui` 至少输出: 桌面 `health` / `probe-ui` 至少输出:
- UIA 依赖是否可用。 - UIA 依赖是否可用。
- 目标进程和主窗口是否存在。 - 目标进程和主窗口是否存在;是否已最大化(若业务需要)。
- 关键窗口/控件是否各唯一匹配。 - 关键窗口/控件是否各唯一匹配。
- 当前窗口标题、Class、PID。 - 当前窗口标题、Class、PID。
- DPI/缩放诊断。 - DPI/缩放诊断。
example 里的最小 `probe`(只探主窗+个别入口)**不等于**生产 probe 完整度;业务技能应对关键跳转节点分别探测或做深探。
手开 PowerShell 跑 `stats` / `health` / dump 时,必须显式对齐宿主的 `CLAW_USER_ID` / 数据根,否则会读到默认用户空库,误判熔断/冷却。
定位失败保存: 定位失败保存:
- 目标窗口和全屏截图。 - 目标窗口和全屏截图。

View File

@@ -155,7 +155,7 @@ verified_at: "YYYY-MM-DD"
1. 先阅读本总标准和对应分端手册。 1. 先阅读本总标准和对应分端手册。
2. 先输出业务步骤表和元素清单,再生成代码。 2. 先输出业务步骤表和元素清单,再生成代码。
3. 只使用真实拾取得到的属性;缺少属性时明确停止并索要界面快照。 3. 只使用真实拾取得到的属性;缺少属性时明确停止并索要界面快照。桌面端须同时记录 **Name 与 Value**(可见中文常在 Value;见 [`RPA_DESKTOP.md`](RPA_DESKTOP.md) §2)。
4. 把目标系统 selector 集中到 element map/selector catalog,不散落在业务流程中。 4. 把目标系统 selector 集中到 element map/selector catalog,不散落在业务流程中。
5. 为每个动作实现前置状态、定位、动作和后置状态。 5. 为每个动作实现前置状态、定位、动作和后置状态。
6. 把诊断探针、界面树导出和失败截图作为实现的一部分。 6. 把诊断探针、界面树导出和失败截图作为实现的一部分。

View File

@@ -5,24 +5,24 @@
## 状态 ## 状态
- **技术路线已真实验证**:Windows UIA + Python `uiautomation` - **技术路线已真实验证**:Windows UIA + Python `uiautomation`
- **来源业务已真实跑通**:企业微信添加客户 - **来源业务已真实跑通**:企业微信添加客户(通讯录 → 新的客户 → 添加 → 搜索 → 发送)
- **本目录默认安全模式**:只提供 probe、dry-run、assisted 范式,不默认发送真实邀请 - **本目录默认安全模式**:只提供 probe、dry-run、assisted 范式,不默认发送真实邀请
- **不是完整 skill 仓库**:按 Copy map 选择性复制 - **不是完整 skill 仓库**:按 Copy map 选择性复制
## 先读 ## 先读
1. [`../../development/RPA_ELEMENT_LOCATING.md`](../../development/RPA_ELEMENT_LOCATING.md) 1. [`../../development/RPA_ELEMENT_LOCATING.md`](../../development/RPA_ELEMENT_LOCATING.md)
2. [`../../development/RPA_DESKTOP.md`](../../development/RPA_DESKTOP.md) 2. [`../../development/RPA_DESKTOP.md`](../../development/RPA_DESKTOP.md)(含 Name/Value、最大化激活、同名叶子)
3. 本 README 的“原始拾取到生产定位器” 3. 本 README 的“原始拾取到生产定位器”
## 成功案例验证了什么 ## 成功案例验证了什么
```text ```text
枚举目标顶层窗口 枚举目标顶层窗口
→ 按进程/窗口类/标题确认作用域 → 激活并最大化主窗(勿无条件 SW_RESTORE)
→ 从 HWND 获取 UIA root → 从 HWND 获取 UIA root / 业务子 pane
→ 在父容器中定位元素 → 在父容器中定位(Name 与 Value 分开看)
→ 激活窗口并真实点击/输入 → 真实点击并校验后置状态
→ 等待子窗口或业务结果 → 等待子窗口或业务结果
→ 处理独立/嵌套弹窗 → 处理独立/嵌套弹窗
→ probe-ui 诊断 → probe-ui 诊断
@@ -30,6 +30,8 @@
来源案例还提供 `dry-run / assisted / auto` 三档、频率控制和业务记忆。本 example 只提炼 UIA 元素工程;账号、风控和数据库仍由具体业务技能实现。 来源案例还提供 `dry-run / assisted / auto` 三档、频率控制和业务记忆。本 example 只提炼 UIA 元素工程;账号、风控和数据库仍由具体业务技能实现。
本目录 `probe()` 是**最小探针**(主窗 + 通讯录入口,可选尝试新的客户行),**不等于**生产技能完整 probe。
## 目录结构 ## 目录结构
```text ```text
@@ -51,25 +53,39 @@ real_desktop_rpa/
## 原始拾取到生产定位器 ## 原始拾取到生产定位器
来源案例的影刀 `selectorsV2` 导出通常包含: ### 搜索框(添加客户窗)
```text ```text
Window(class=SearchExternalsWnd,title=添加客户) 原始:Window(SearchExternalsWnd) → 多层 Layout/index → search_layout → searchedit
→ 多层 Layout/index 提炼:窗口 → search_layout → Name=searchedit
→ search_layout ```
→ searchedit
### 新的客户(易错)
```text
原始:通讯录 pane → wlist_view → item[2] → status_layout → main_chattext
提炼:wlist_view 内找 子控件 main_chattext.Value=新的客户 的那一行,再点该行 status_layout
禁止:全树第一个 Name=status_layout(常为「群聊」)
禁止:把选择器标题「新的客户」当成 UIA Name 去搜
```
### 添加按钮
```text
提炼:search_external → wbtn_container(子文案 Value=添加)
禁止:搜索旁 plus_button 下的同名 wbtn_container
``` ```
正确提炼方式: 正确提炼方式:
1. 保留稳定顶层窗口:进程、Window Class、标题。 1. 保留稳定顶层窗口:进程、Window Class、标题;需要时加上业务子 pane。
2. 保留最小业务父容器:如 `search_layout`。 2. 保留最小业务父容器。
3. 使用叶子元素稳定属性:如 `Name=searchedit`。 3. 叶子用稳定 Name **和/或** Value / 子控件 Value。
4. 删除只反映当前布局的深层 `index`。 4. 删除只反映当前布局的深层 index;index 仅作已验证兜底并写清含义。
5. 验证父容器内候选数量为 `1`。 5. 验证父容器内候选数量为 `1`。
6. 重启应用后再次验证。 6. 重启应用后再次验证。
不要把完整 XML 路径或 `item[0]` 原样变成生产定位器。 不要把完整 XML 路径或未校验的 `item[0]` 原样变成生产定位器。
## Element map ## Element map
@@ -79,8 +95,9 @@ Window(class=SearchExternalsWnd,title=添加客户)
- 窗口不唯一时失败,不取 `hwnds[0]`。 - 窗口不唯一时失败,不取 `hwnds[0]`。
- 控件不唯一时失败,不取第一个匹配。 - 控件不唯一时失败,不取第一个匹配。
- `item`、`status_layout`、`wbtn_container` 等通用 Name 必须放进稳定父容器。 - `item`、`status_layout`、`wbtn_container` 等通用 Name 必须放进稳定父容器,并用文案校验。
- 行顺序不是身份;用业务文案或稳定属性选择结果行。 - 行顺序不是身份;用业务文案(Value)选择结果行。
- 主窗依赖完整侧栏时先 maximize;`activate(maximize=False)` 不得冲掉已最大化状态。
- 只有主定位成功、窗口前台且矩形有效时,才允许从 UIA BoundingRectangle 降级坐标点击。 - 只有主定位成功、窗口前台且矩形有效时,才允许从 UIA BoundingRectangle 降级坐标点击。
## 三种实施模式 ## 三种实施模式
@@ -97,8 +114,8 @@ Window(class=SearchExternalsWnd,title=添加客户)
| 来源 | 目标 | 说明 | | 来源 | 目标 | 说明 |
|---|---|---| |---|---|---|
| `scripts/service/desktop_session.py` | `scripts/service/desktop_session.py` | 顶层窗口枚举、唯一性、激活 | | `scripts/service/desktop_session.py` | `scripts/service/desktop_session.py` | 顶层窗口枚举、唯一性、激活/最大化 |
| `scripts/service/uia_locator.py` | `scripts/service/uia_locator.py` | 作用域内唯一定位、点击、输入 | | `scripts/service/uia_locator.py` | `scripts/service/uia_locator.py` | 作用域内唯一定位(含 Value)、点击、输入 |
| `scripts/util/selectors.py` | `scripts/util/<platform>_selectors.py` | element spec;必须全部替换 | | `scripts/util/selectors.py` | `scripts/util/<platform>_selectors.py` | element spec;必须全部替换 |
| `scripts/service/wecom_reference_flow.py` | `scripts/service/<business>_flow.py` | 只复制流程分层,不复制企微 selector | | `scripts/service/wecom_reference_flow.py` | `scripts/service/<business>_flow.py` | 只复制流程分层,不复制企微 selector |
| `assets/element-map.sample.yaml` | `assets/<platform>-element-map.yaml` | 真实拾取后填写 | | `assets/element-map.sample.yaml` | `assets/<platform>-element-map.yaml` | 真实拾取后填写 |
@@ -118,14 +135,14 @@ Window(class=SearchExternalsWnd,title=添加客户)
## 禁止照抄 ## 禁止照抄
- `WXWork`、`WeWorkWindow`、`SearchExternalsWnd` - `WXWork`、`WeWorkWindow`、`SearchExternalsWnd`
- `org_op_new`、`searchedit`、`wbtn_container` - `org_op_new`、`searchedit`、`wbtn_container`、`main_chattext`
- 企业微信列表第 0/1 行含义 - 企业微信列表第 0/1/2 行含义或固定 index
- 来源案例的本机路径、账号和业务数据 - 来源案例的本机路径、账号和业务数据
- 任何未经目标软件真实拾取验证的 fallback - 任何未经目标软件真实拾取验证的 fallback
## 验收 ## 验收
- `probe` 能明确报告依赖、窗口和关键元素 - `probe` 能明确报告依赖、窗口和关键元素(生产技能覆盖关键跳转节点)
- 每个关键元素在目标作用域内唯一 - 每个关键元素在目标作用域内唯一
- 关闭重开软件后仍能定位 - 关闭重开软件后仍能定位
- 多窗口时不会误选 - 多窗口时不会误选

View File

@@ -2,6 +2,10 @@ source_case: add-friend-wecom
status: verified_source_redacted_example status: verified_source_redacted_example
surface: windows surface: windows
driver: uiautomation driver: uiautomation
notes:
- 选择器显示名 ≠ UIA Name;可见中文常在 Value / 子 Label
- 禁止全树第一个 status_layout / wbtn_container
- 主窗流程先 maximize;activate 勿无条件 SW_RESTORE
elements: elements:
- logical_name: main_window - logical_name: main_window
purpose: 企业微信主窗口 purpose: 企业微信主窗口
@@ -11,7 +15,7 @@ elements:
window_class: WeWorkWindow window_class: WeWorkWindow
title: 企业微信 title: 企业微信
expected_count: 1 expected_count: 1
postcondition: UIA root 可读取 postcondition: UIA root 可读取;业务需要时已最大化
replace_in_new_skill: true replace_in_new_skill: true
- logical_name: contacts_entry - logical_name: contacts_entry
@@ -21,8 +25,57 @@ elements:
primary_locator: primary_locator:
name: org_op_new name: org_op_new
control_type: TreeItem control_type: TreeItem
value_hint: 通讯录
expected_count: 1 expected_count: 1
postcondition: 通讯录业务面板出现 postcondition: 出现企业微信-通讯录业务 pane
replace_in_new_skill: true
- logical_name: contacts_pane
purpose: 通讯录业务子层作用域
scope:
logical_parent: main_window
primary_locator:
name_contains: 通讯录
expected_count: 1
postcondition: 可枚举 wlist_view
replace_in_new_skill: true
- logical_name: new_customer_row
purpose: 侧栏进入「新的客户」
scope:
logical_parent: contacts_pane
ancestor_name: wlist_view
primary_locator:
row_leaf_name: status_layout
child_name: main_chattext
child_value: 新的客户
expected_count: 1
postcondition: panel_title_label Value=新的客户 或出现 search_external
failure_note: 禁止点全树第一个 status_layout(常为群聊)
replace_in_new_skill: true
- logical_name: panel_title
purpose: 校验已进入新的客户页
scope:
logical_parent: contacts_pane
primary_locator:
name: panel_title_label
value: 新的客户
expected_count: 1
replace_in_new_skill: true
- logical_name: add_customer_button
purpose: 打开添加客户窗口
scope:
logical_parent: contacts_pane
ancestor_name: search_external
primary_locator:
name: wbtn_container
child_name: ww_btn_text
child_value: 添加
expected_count: 1
postcondition: 出现 SearchExternalsWnd
failure_note: 禁止点搜索旁 plus_button 下的 wbtn_container
replace_in_new_skill: true replace_in_new_skill: true
- logical_name: add_customer_window - logical_name: add_customer_window

View File

@@ -10,3 +10,17 @@
<uia name="inner_search_layout" index="1" /> <uia name="inner_search_layout" index="1" />
<uia name="searchedit" index="1" /> <uia name="searchedit" index="1" />
</selector> </selector>
<!--
新的客户:长路径示意 → 生产应改为
wlist_view 内 child main_chattext.Value=新的客户 的行,再点该行 status_layout。
禁止全树第一个 status_layout。
-->
<selector name="new_customer_row" source="inspector-export">
<window process="WXWork" class="WeWorkWindow" title="企业微信" />
<pane name="企业微信-通讯录" />
<uia name="wlist_view" />
<uia name="item" index="2" />
<uia name="status_layout" />
<uia name="main_chattext" value="新的客户" />
</selector>

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import ctypes import ctypes
import ntpath import ntpath
import time
from ctypes import wintypes from ctypes import wintypes
from dataclasses import dataclass from dataclasses import dataclass
from typing import Iterable from typing import Iterable
@@ -31,6 +32,7 @@ class WindowCandidate:
user32 = ctypes.windll.user32 user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32 kernel32 = ctypes.windll.kernel32
SW_RESTORE = 9 SW_RESTORE = 9
SW_MAXIMIZE = 3
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
@@ -105,7 +107,23 @@ def resolve_unique_window(
return matched[0] return matched[0]
def activate_window(window: WindowCandidate) -> None: def activate_window(window: WindowCandidate, *, maximize: bool = False) -> None:
user32.ShowWindow(window.hwnd, SW_RESTORE) """激活窗口。
if not user32.SetForegroundWindow(window.hwnd):
raise DesktopRpaError("WINDOW_ACTIVATE_FAILED", f"无法激活 hwnd={window.hwnd}") maximize=True:主窗进入依赖完整侧栏的流程时使用。
maximize=False:只前置;**禁止**对已最大化窗口无条件 SW_RESTORE。
"""
hwnd = window.hwnd
iconic = bool(user32.IsIconic(hwnd))
zoomed = bool(user32.IsZoomed(hwnd))
if maximize:
if iconic:
user32.ShowWindow(hwnd, SW_RESTORE)
if not zoomed:
user32.ShowWindow(hwnd, SW_MAXIMIZE)
else:
if iconic:
user32.ShowWindow(hwnd, SW_RESTORE)
if not user32.SetForegroundWindow(hwnd):
raise DesktopRpaError("WINDOW_ACTIVATE_FAILED", f"无法激活 hwnd={hwnd}")
time.sleep(0.35 if maximize else 0.2)

View File

@@ -22,6 +22,15 @@ def root_from_window(window: WindowCandidate):
return auto.ControlFromHandle(window.hwnd) return auto.ControlFromHandle(window.hwnd)
def control_value(control: Any) -> str:
try:
auto = require_uiautomation()
raw = control.GetPropertyValue(auto.PropertyId.ValueValueProperty)
return str(raw or "").strip()
except Exception:
return ""
def _walk(root: Any, *, depth: int = 0, max_depth: int = 25) -> Iterator[Any]: def _walk(root: Any, *, depth: int = 0, max_depth: int = 25) -> Iterator[Any]:
if depth > max_depth: if depth > max_depth:
return return
@@ -36,9 +45,21 @@ def _walk(root: Any, *, depth: int = 0, max_depth: int = 25) -> Iterator[Any]:
yield from _walk(child, depth=depth + 1, max_depth=max_depth) yield from _walk(child, depth=depth + 1, max_depth=max_depth)
def _child_has_value(control: Any, child_name: str, expected_value: str, *, max_depth: int = 10) -> bool:
for node in _walk(control, max_depth=max_depth):
if (getattr(node, "Name", None) or "") != child_name:
continue
if control_value(node) == expected_value:
return True
return False
def _matches(control: Any, spec: ElementSpec) -> bool: def _matches(control: Any, spec: ElementSpec) -> bool:
if spec.name is not None and (getattr(control, "Name", None) or "") != spec.name: if spec.name is not None and (getattr(control, "Name", None) or "") != spec.name:
return False return False
if spec.value is not None and not spec.child_value_name:
if control_value(control) != spec.value:
return False
if spec.automation_id is not None and ( if spec.automation_id is not None and (
getattr(control, "AutomationId", None) or "" getattr(control, "AutomationId", None) or ""
) != spec.automation_id: ) != spec.automation_id:
@@ -52,6 +73,9 @@ def _matches(control: Any, spec: ElementSpec) -> bool:
getattr(control, "ClassName", None) or "" getattr(control, "ClassName", None) or ""
) != spec.class_name: ) != spec.class_name:
return False return False
if spec.child_value_name and spec.value:
if not _child_has_value(control, spec.child_value_name, spec.value):
return False
return True return True
@@ -64,7 +88,9 @@ def locate_unique(root: Any, spec: ElementSpec, *, max_depth: int = 25):
) )
search_root = locate_unique(root, ancestor_spec, max_depth=max_depth) search_root = locate_unique(root, ancestor_spec, max_depth=max_depth)
matched = [control for control in _walk(search_root, max_depth=max_depth) if _matches(control, spec)] matched = [
control for control in _walk(search_root, max_depth=max_depth) if _matches(control, spec)
]
if not matched: if not matched:
raise DesktopRpaError("ELEMENT_NOT_FOUND", f"{spec.logical_name} 未找到") raise DesktopRpaError("ELEMENT_NOT_FOUND", f"{spec.logical_name} 未找到")
if len(matched) != spec.expected_count: if len(matched) != spec.expected_count:
@@ -75,8 +101,13 @@ def locate_unique(root: Any, spec: ElementSpec, *, max_depth: int = 25):
return matched[0] return matched[0]
def click_control(control: Any, *, window: WindowCandidate) -> None: def click_control(
activate_window(window) control: Any,
*,
window: WindowCandidate,
maximize: bool = False,
) -> None:
activate_window(window, maximize=maximize)
rect = control.BoundingRectangle rect = control.BoundingRectangle
if rect.width() <= 0 or rect.height() <= 0 or rect.left < -10000: if rect.width() <= 0 or rect.height() <= 0 or rect.left < -10000:
raise DesktopRpaError("ELEMENT_NOT_READY", "目标元素不可见或边界矩形无效") raise DesktopRpaError("ELEMENT_NOT_READY", "目标元素不可见或边界矩形无效")
@@ -89,7 +120,7 @@ def click_control(control: Any, *, window: WindowCandidate) -> None:
def type_text(control: Any, text: str, *, window: WindowCandidate) -> None: def type_text(control: Any, text: str, *, window: WindowCandidate) -> None:
auto = require_uiautomation() auto = require_uiautomation()
click_control(control, window=window) click_control(control, window=window, maximize=False)
auto.SendKeys("{Ctrl}a", waitTime=0.05) auto.SendKeys("{Ctrl}a", waitTime=0.05)
auto.SendKeys("{Delete}", waitTime=0.05) auto.SendKeys("{Delete}", waitTime=0.05)
for char in text: for char in text:

View File

@@ -5,28 +5,49 @@ from __future__ import annotations
from dataclasses import asdict from dataclasses import asdict
from typing import Any from typing import Any
from service.desktop_session import resolve_unique_window from service.desktop_session import activate_window, resolve_unique_window
from service.uia_locator import locate_unique, root_from_window, type_text from service.uia_locator import locate_unique, root_from_window, type_text
from util.selectors import ADD_CUSTOMER_WINDOW, CONTACTS_ENTRY, MAIN_WINDOW, SEARCH_INPUT from util.selectors import (
ADD_CUSTOMER_WINDOW,
CONTACTS_ENTRY,
MAIN_WINDOW,
NEW_CUSTOMER_ROW,
SEARCH_INPUT,
)
def probe() -> dict[str, Any]: def probe() -> dict[str, Any]:
"""最小 probe:主窗 + 通讯录入口。
生产技能应对「新的客户 / 添加」等关键节点做更深探测;本函数不等于生产完整度。
调用前会 maximize 主窗,避免侧栏折叠导致 org_op_new 抖动。
"""
main = resolve_unique_window(MAIN_WINDOW) main = resolve_unique_window(MAIN_WINDOW)
activate_window(main, maximize=True)
root = root_from_window(main) root = root_from_window(main)
contacts = locate_unique(root, CONTACTS_ENTRY) contacts = locate_unique(root, CONTACTS_ENTRY)
return { payload: dict[str, Any] = {
"success": True, "success": True,
"minimized_probe": True,
"main_window": asdict(main), "main_window": asdict(main),
"contacts_entry": { "contacts_entry": {
"name": getattr(contacts, "Name", ""), "name": getattr(contacts, "Name", ""),
"control_type": getattr(contacts, "ControlTypeName", ""), "control_type": getattr(contacts, "ControlTypeName", ""),
}, },
} }
# 可选:若已在通讯录页,尝试按 Value 定位「新的客户」行(失败不视为 probe 失败)
try:
row = locate_unique(root, NEW_CUSTOMER_ROW)
payload["new_customer_row"] = {"name": getattr(row, "Name", "")}
except Exception:
payload["new_customer_row"] = None
return payload
def fill_search_dry_run(target: str) -> dict[str, Any]: def fill_search_dry_run(target: str) -> dict[str, Any]:
"""要求操作者已打开“添加客户”窗口;只填写搜索框,不提交。""" """要求操作者已打开“添加客户”窗口;只填写搜索框,不提交。"""
search_window = resolve_unique_window(ADD_CUSTOMER_WINDOW) search_window = resolve_unique_window(ADD_CUSTOMER_WINDOW)
activate_window(search_window, maximize=False)
root = root_from_window(search_window) root = root_from_window(search_window)
search_input = locate_unique(root, SEARCH_INPUT) search_input = locate_unique(root, SEARCH_INPUT)
type_text(search_input, target, window=search_window) type_text(search_input, target, window=search_window)

View File

@@ -18,10 +18,16 @@ class WindowSpec:
class ElementSpec: class ElementSpec:
logical_name: str logical_name: str
name: Optional[str] = None name: Optional[str] = None
"""UIA Name(内部 id,未必是界面中文)。"""
value: Optional[str] = None
"""UIA Value(可见中文文案常在此)。"""
automation_id: Optional[str] = None automation_id: Optional[str] = None
control_type: Optional[str] = None control_type: Optional[str] = None
class_name: Optional[str] = None class_name: Optional[str] = None
ancestor_name: Optional[str] = None ancestor_name: Optional[str] = None
"""先定位到该 Name 的祖先,再在其下搜索。"""
child_value_name: Optional[str] = None
"""在候选控件子树中查找该 Name 的子控件,并用 value 匹配其 Value。"""
expected_count: int = 1 expected_count: int = 1
@@ -37,6 +43,19 @@ CONTACTS_ENTRY = ElementSpec(
name="org_op_new", name="org_op_new",
control_type="TreeItem", control_type="TreeItem",
) )
# 示意:用子控件 main_chattext 的 Value 定位「新的客户」行(禁止全树第一个 status_layout)
NEW_CUSTOMER_ROW = ElementSpec(
"new_customer_row",
name="status_layout",
ancestor_name="wlist_view",
child_value_name="main_chattext",
value="新的客户",
)
ADD_BUTTON = ElementSpec(
"add_customer_button",
name="wbtn_container",
ancestor_name="search_external",
)
SEARCH_INPUT = ElementSpec( SEARCH_INPUT = ElementSpec(
"search_input", "search_input",
name="searchedit", name="searchedit",

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import sys import sys
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
@@ -11,9 +12,11 @@ sys.path.insert(0, str(ROOT / "scripts"))
from service.desktop_session import ( # noqa: E402 from service.desktop_session import ( # noqa: E402
DesktopRpaError, DesktopRpaError,
WindowCandidate, WindowCandidate,
activate_window,
resolve_unique_window, resolve_unique_window,
) )
from util.selectors import MAIN_WINDOW # noqa: E402 from service import uia_locator # noqa: E402
from util.selectors import ElementSpec, MAIN_WINDOW # noqa: E402
class SelectorContractTest(unittest.TestCase): class SelectorContractTest(unittest.TestCase):
@@ -39,6 +42,58 @@ class SelectorContractTest(unittest.TestCase):
resolve_unique_window(MAIN_WINDOW, candidates=[hidden]) resolve_unique_window(MAIN_WINDOW, candidates=[hidden])
self.assertEqual(ctx.exception.code, "WINDOW_NOT_FOUND") self.assertEqual(ctx.exception.code, "WINDOW_NOT_FOUND")
def test_activate_without_maximize_skips_restore_when_zoomed(self) -> None:
window = WindowCandidate(42, 10, "WXWork", "WeWorkWindow", "企业微信", True)
with patch("service.desktop_session.user32") as user32:
user32.IsIconic.return_value = False
user32.IsZoomed.return_value = True
user32.SetForegroundWindow.return_value = True
activate_window(window, maximize=False)
user32.ShowWindow.assert_not_called()
user32.SetForegroundWindow.assert_called_once_with(42)
def test_activate_maximize_when_not_zoomed(self) -> None:
window = WindowCandidate(42, 10, "WXWork", "WeWorkWindow", "企业微信", True)
with patch("service.desktop_session.user32") as user32:
user32.IsIconic.return_value = False
user32.IsZoomed.return_value = False
user32.SetForegroundWindow.return_value = True
activate_window(window, maximize=True)
user32.ShowWindow.assert_called()
user32.SetForegroundWindow.assert_called_once_with(42)
def test_locate_unique_matches_child_value(self) -> None:
row = MagicMock()
row.Name = "status_layout"
row.AutomationId = ""
row.ControlTypeName = "TreeControl"
row.ClassName = ""
label = MagicMock()
label.Name = "main_chattext"
label.GetChildren.return_value = []
row.GetChildren.return_value = [label]
root = MagicMock()
root.Name = "wlist_view"
root.AutomationId = ""
root.ControlTypeName = "TreeControl"
root.ClassName = ""
root.GetChildren.return_value = [row]
with patch.object(uia_locator, "control_value", side_effect=lambda c: "新的客户" if c is label else ""):
spec = ElementSpec(
"new_customer_row",
name="status_layout",
ancestor_name="wlist_view",
child_value_name="main_chattext",
value="新的客户",
)
# ancestor is root itself when searching from root with name wlist_view
found = uia_locator.locate_unique(root, ElementSpec("wlist", name="wlist_view"))
self.assertIs(found, root)
found_row = uia_locator.locate_unique(root, spec)
self.assertIs(found_row, row)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@@ -80,9 +80,11 @@ Action 是**可选能力**:
| 字段 / 层 | 只决定什么 | 不决定什么 | | 字段 / 层 | 只决定什么 | 不决定什么 |
|-----------|------------|------------| |-----------|------------|------------|
| **`placements`** | Action **出现在哪里**(toolbar / row / batch / cron / agent / skill-detail) | 不决定 sync/async | | **`placements`** | Action **出现在哪里**(toolbar / row / batch / cron / agent / skill-detail) | 不决定 sync/async,不决定点了是否弹窗 |
| **`replaces`** | **可选**:接管数据管理默认「新增 / 编辑 / 删除」 | 不强制;无配置时宿主仍直写 SQLite |
| **`bind.tables`** | 数据管理里出现在**哪些业务表** | 不决定业务实现 | | **`bind.tables`** | 数据管理里出现在**哪些业务表** | 不决定业务实现 |
| **`executionProfile`** | 宿主是**同步等待**还是**异步 Job** | 不限制可展示入口 | | **`executionProfile`** | 宿主是**同步等待**还是**异步 Job** | 不限制可展示入口,不决定是否弹窗 |
| **`interaction`** | 可点击入口(数据管理 toolbar/row/batch、默认 CRUD 接管、未来 skill-detail)**点了怎么交互** | 不决定 sync/async;**cron 触发与 agent 忽略** |
| **`entrypoint`** | 实际执行哪个 CLI 命令与参数 | 不写业务逻辑 | | **`entrypoint`** | 实际执行哪个 CLI 命令与参数 | 不写业务逻辑 |
| **domain service** | **真正完成什么业务** | 不感知 toolbar / cron / agent 来源 | | **domain service** | **真正完成什么业务** | 不感知 toolbar / cron / agent 来源 |
@@ -93,13 +95,16 @@ Action 是**可选能力**:
| 层级 | 稳定字段 / 能力 | | 层级 | 稳定字段 / 能力 |
|------|----------------| |------|----------------|
| 顶层 | `schemaVersion`、`skill`、`actions` | | 顶层 | `schemaVersion`、`skill`、`actions` |
| action | `id`、`label`、`description`、`placements`、`entrypoint`、**`executionProfile`(必填)**;可选 `inputSchema`、`confirmation`、**`bind`** | | action | `id`、`label`、`description`、`placements`、`entrypoint`、**`executionProfile`(必填)**;可选 `inputSchema`、`confirmation`、**`interaction`**、**`interactionByPlacement`**、**`bind`**、**`replaces`** |
| entrypoint | `type = cli`、`command`、`args`(**必填** JSON 数组,项为 `string` / `number` / `boolean`) | | entrypoint | `type = cli`、`command`、`args`(**必填** JSON 数组,项为 `string` / `number` / `boolean`) |
| placements | `toolbar`、`row`、`batch`、`skill-detail`、`agent`、`cron` | | placements | `toolbar`、`row`、`batch`、`skill-detail`、`agent`、`cron`;仅 `replaces` 时可为空数组 |
| **replaces** | 可选 `"create"` \| `"update"` \| `"delete"` — 接管默认新增/编辑/删除(须 `bind.tables`) |
| **bind** | `{ "tables": ["business_table"] }`;可选 `inputMapping`(行内/批量预填,见下节) | | **bind** | `{ "tables": ["business_table"] }`;可选 `inputMapping`(行内/批量预填,见下节) |
| inputSchema | 根 `type: object`;properties 使用 `string` / `number` / `boolean`;支持 `enum`、`default`、`required`、`sensitive`、`readOnly` | | inputSchema | 根 `type: object`;properties 使用 `string` / `number` / `boolean`;支持 `enum`、`default`、`required`、`sensitive`、`readOnly` |
| confirmation | `{ "message": "..." }` | | confirmation | `{ "message": "..." }` — **风险文案**(confirm/form 弹窗黄条;agent 元数据);**不是**「要不要弹窗」的开关 |
| **executionProfile** | `"sync"` \| `"async"`(**必须显式写出**,禁止依赖缺省) | | **executionProfile** | `"sync"` \| `"async"`(**必须显式写出**,禁止依赖缺省) |
| **interaction** | `"auto"` \| `"direct"` \| `"confirm"` \| `"form"`(可选;省略≈`auto`) |
| **interactionByPlacement** | 可选;按 placement 覆盖为 `direct`/`confirm`/`form`(不可写 `auto`) |
## `bind.tables`(表级绑定) ## `bind.tables`(表级绑定)
@@ -122,6 +127,125 @@ Action 是**可选能力**:
- **`inputMapping`(可选)**:把行/勾选映射到 `inputSchema` 字段,例如 `"leadId": "$row.$pk"`、`"ids": "$selection.ids"`。适合 `row` / `batch`。宿主仍会提交这些字段;可用 `inputSchema.properties.*.readOnly: true` 在表单中只读展示。 - **`inputMapping`(可选)**:把行/勾选映射到 `inputSchema` 字段,例如 `"leadId": "$row.$pk"`、`"ids": "$selection.ids"`。适合 `row` / `batch`。宿主仍会提交这些字段;可用 `inputSchema.properties.*.readOnly: true` 在表单中只读展示。
- `concurrency` / `locks` 仍未稳定,**不要**写入新技能 manifest。 - `concurrency` / `locks` 仍未稳定,**不要**写入新技能 manifest。
## `replaces`(可选:接管默认新增 / 编辑 / 删除)
有领域副作用的表(建目录、加密、级联清理)不要依赖宿主直写 SQLite。此时可为该表声明:
```json
{
"id": "account-add-web",
"label": "新增账号",
"description": "登记网页账号并初始化 profile 目录。",
"replaces": "create",
"placements": [],
"executionProfile": "sync",
"interaction": "form",
"bind": { "tables": ["accounts"] },
"inputSchema": {
"type": "object",
"required": ["platform_key", "account_label"],
"properties": {
"platform_key": { "type": "string", "title": "平台" },
"account_label": { "type": "string", "title": "账号名称" }
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["add-web", "--platform", "{{platform_key}}", "--label", "{{account_label}}"]
}
}
```
规则:
- **全可选**:无 `replaces` 时,宿主默认按钮仍直写库,行为与以前一致。
- 取值:`create`(新增)/ `update`(编辑)/ `delete`(删除)。
- **必须**非空 `bind.tables`;同一技能同一表同一 `replaces` 至多一个 Action。
- 与 `placements` 正交:可只接管默认按钮(`placements: []`),不必再挂 toolbar「第二个新增」。
- 点默认按钮时走该 Action 的 `interaction`(create/update 常用 `form`;delete 常用 `confirm`)。
- `update` / `delete` 可用 `inputMapping`(`$row.$pk`、`$selection.ids`)预填主键。
- 宿主导入仍直写:有 `replaces:create` 的表慎用宿主导入,或另做 import Action。
- 整表 `_jiangchang_tables.readonly=1` 时仍隐藏写按钮(优先于 `replaces`)。
- **列元数据复用(lookup)**:create/update 抽屉可将 `inputSchema` 字段对齐业务列,复用中文名与 lookup 下拉:
- 优先让属性名 = 物理列名(如 `platform_key`);
- 或 `bind.columnMapping`:`{ "platform": "platform_key" }`;
- 或字段上写 `"column": "platform_key"`。
- **不要**在 actions.json 里写死会过期的全量平台 `enum`。
## `interaction`(点击交互方式)
与 `executionProfile` **正交**:一个 Action 可以是 `async` + `direct`(一点就进任务中心),也可以是 `sync` + `form`(填参后当场返回)。
| 值 | 数据管理体验 | 典型场景 |
|----|--------------|----------|
| `direct` | **一点即跑**,无弹窗 | 打开文件位置、无参只读、行内主键已映射 |
| `confirm` | 轻量确认弹窗(无参数区;文案优先 `confirmation.message`) | 打开浏览器、有副作用但无可编辑入参 |
| `form` | 完整参数表单(现状默认路径) | 导入、查询条件、需用户改写的字段 |
| `auto` / 省略 | 宿主推断:有**可编辑**参数 → `form`;仅有 `confirmation` → `confirm`;否则 → `direct` | 旧技能兼容;新技能建议显式写出 |
```json
{
"id": "reveal-profile",
"label": "打开文件位置",
"description": "在资源管理器中打开 Profile 目录。",
"placements": ["row"],
"executionProfile": "sync",
"interaction": "direct",
"bind": {
"tables": ["accounts"],
"inputMapping": { "accountId": "$row.$pk" }
},
"inputSchema": {
"type": "object",
"required": ["accountId"],
"properties": {
"accountId": { "type": "number", "title": "账号 ID", "readOnly": true }
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["reveal-profile", "{{accountId}}"]
}
}
```
```json
{
"id": "open-browser",
"label": "打开浏览器",
"placements": ["row"],
"executionProfile": "async",
"interaction": "confirm",
"confirmation": { "message": "将用该账号 Profile 打开浏览器,仅供查看。" },
"bind": {
"tables": ["accounts"],
"inputMapping": { "accountId": "$row.$pk" }
},
"inputSchema": {
"type": "object",
"required": ["accountId"],
"properties": {
"accountId": { "type": "number", "readOnly": true }
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["open", "{{accountId}}"]
}
}
```
可选 `interactionByPlacement`:同一 Action 在不同入口用不同交互(例如 `row: "direct"`、`toolbar: "form"`)。值只能是 `direct`/`confirm`/`form`,不能写 `auto`。
**边界:**
- **cron 配置**:仍展示可编辑参数;`confirmation.message` 可作为配置页风险提示(由宿主决定)。
- **cron 触发 / agent**:忽略 `interaction`(无人值守 / 模型填参);`confirmation` 对 agent 仍作 `requiresConfirmation` 元数据。
- 不要用 `interaction: "direct"` 掩盖需要用户填写的可编辑字段——宿主在 `auto` 下会因可编辑参数回落到 `form`;显式 `direct` + 可编辑字段时参数取 default/预填,作者须自觉。
默认模板 `actions.json` 仅含 `health` / `version` / `config-path`,不放在 toolbar/row/batch,故无需 `bind` 示例污染默认 manifest。业务示例: 默认模板 `actions.json` 仅含 `health` / `version` / `config-path`,不放在 toolbar/row/batch,故无需 `bind` 示例污染默认 manifest。业务示例:
```json ```json

View File

@@ -78,6 +78,8 @@ metadata:
不要把「外部来源只读」错误理解成「技能本地数据库一定只读」。技能通过 `_jiangchang_tables.readonly` 与 `_jiangchang_columns.editable` 声明本地表的实际写权限。 不要把「外部来源只读」错误理解成「技能本地数据库一定只读」。技能通过 `_jiangchang_tables.readonly` 与 `_jiangchang_columns.editable` 声明本地表的实际写权限。
若创建/更新/删除带有领域副作用(自动建目录、加密、级联清理),不要依赖宿主默认直写:在 `assets/actions.json` 用可选字段 `replaces: create|update|delete` 接管默认按钮,详见 [`ACTIONS.md`](ACTIONS.md)。
## 标准数据同步语义(外源 → 本地库) ## 标准数据同步语义(外源 → 本地库)
适用于「同步 / refresh / collect」类能力。使用通用名称表述,勿把具体平台表名写进模板实现。 适用于「同步 / refresh / collect」类能力。使用通用名称表述,勿把具体平台表名写进模板实现。
@@ -131,6 +133,38 @@ metadata:
| `sort_order` | 否 | 兼容保留字段,不用于替代物理字段顺序 | | `sort_order` | 否 | 兼容保留字段,不用于替代物理字段顺序 |
| `visible` | 是 | `1` 在宿主数据管理中展示 / `0` 隐藏 | | `visible` | 是 | `1` 在宿主数据管理中展示 / `0` 隐藏 |
| `readonly` | 是 | `1` 整张表只读;`0` 允许宿主根据字段权限执行写操作 | | `readonly` | 是 | `1` 整张表只读;`0` 允许宿主根据字段权限执行写操作 |
| `default_sort_json` | 否 | 数据管理**默认行排序**(见下节);空则宿主用内置默认 |
#### 表级默认排序 `default_sort_json`(宿主支持)
技能可声明该表在数据管理中的默认排序;宿主优先采用,未声明或声明无效时回退:
`updated_at DESC` → `created_at DESC` → 主键 DESC。
JSON 数组(可多字段,按顺序):
```json
[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]
```
规则:
- `column` 必须是该表真实列名;敏感列会被宿主忽略
- `direction` 仅 `asc` / `desc`(缺省按 `desc`)
- 用户在 UI 上手动改排序后,以用户选择为准;声明只影响未选择时的默认
- 通过 `upsert_table_metadata(..., default_sort_json=...)` 写入;旧库由 `init_display_metadata_tables` 幂等 `ALTER` 补列
示例(消息时间线):
```python
upsert_table_metadata(
cursor,
table_name="messages",
display_name="消息",
readonly=0,
default_sort_json='[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]',
)
```
禁止: 禁止:
@@ -354,7 +388,11 @@ CREATE TABLE demo_items (
## 枚举字段 options_json ## 枚举字段 options_json
`options_json` 为 JSON 数组,每项包含 `value` 与 `label`: `options_json` 支持两种形态:
### 1)静态枚举(数组)
每项包含 `value` 与 `label`:
```json ```json
[ [
@@ -366,6 +404,31 @@ CREATE TABLE demo_items (
宿主据此渲染选择框;技能侧负责保证 JSON 合法且与物理字段取值一致。 宿主据此渲染选择框;技能侧负责保证 JSON 合法且与物理字段取值一致。
### 2)表关联 lookup(对象,推荐用于外键式字段)
当字段取值依赖同库另一张业务表时,`display_type` 设为 `lookup`,`options_json` 写对象(**不是**数组):
```json
{
"ref_table": "platforms",
"value_column": "platform_key",
"label_column": "display_name",
"filter_column": "enabled",
"filter_value": 1
}
```
| 字段 | 必填 | 说明 |
|---|---|---|
| `ref_table` | 是 | 同库被引用表名(snake_case 标识符) |
| `value_column` | 是 | 写入当前字段的值列 |
| `label_column` | 是 | 下拉展示名列 |
| `filter_column` / `filter_value` | 否 | 等值过滤(如只列出 `enabled=1`) |
宿主在加载表 schema 时查询 `ref_table`,把结果展开成 `{value,label}` 下拉选项;**物理列仍存标量**(如 `platform_key` 文本、`account_id` 整数),不做嵌套子表。
示例:账号表的平台字段 → 平台信息表;凭据表的账号编号 → 账号信息表。
## 幂等初始化标准 ## 幂等初始化标准
- `_jiangchang_tables` 和 `_jiangchang_columns` 必须通过 upsert 初始化 - `_jiangchang_tables` 和 `_jiangchang_columns` 必须通过 upsert 初始化

View File

@@ -24,7 +24,8 @@ CREATE TABLE IF NOT EXISTS _jiangchang_tables (
description TEXT, description TEXT,
sort_order INTEGER, sort_order INTEGER,
visible INTEGER NOT NULL DEFAULT 1, visible INTEGER NOT NULL DEFAULT 1,
readonly INTEGER NOT NULL DEFAULT 0 readonly INTEGER NOT NULL DEFAULT 0,
default_sort_json TEXT
) )
""" """
@@ -82,23 +83,50 @@ TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
) )
def _ensure_jiangchang_tables_columns(cursor) -> None:
"""旧库补列:CREATE IF NOT EXISTS 不会给已有表加新字段。"""
cursor.execute("PRAGMA table_info('_jiangchang_tables')")
cols = {str(row[1]) for row in cursor.fetchall()}
if "default_sort_json" not in cols:
cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN default_sort_json TEXT")
def init_display_metadata_tables(cursor) -> None: def init_display_metadata_tables(cursor) -> None:
cursor.execute(_JIANGCHANG_TABLES_DDL) cursor.execute(_JIANGCHANG_TABLES_DDL)
cursor.execute(_JIANGCHANG_COLUMNS_DDL) cursor.execute(_JIANGCHANG_COLUMNS_DDL)
_ensure_jiangchang_tables_columns(cursor)
def upsert_table_metadata(cursor, *, table_name: str, display_name: str, description: str | None = None, visible: int = 1, readonly: int = 0) -> None: def upsert_table_metadata(
cursor,
*,
table_name: str,
display_name: str,
description: str | None = None,
visible: int = 1,
readonly: int = 0,
default_sort_json: str | None = None,
) -> None:
"""写入表级元数据。
default_sort_json: 宿主数据管理默认排序,JSON 数组,例如
'[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]'
为空则宿主回退到 updated_at → created_at → 主键。
"""
cursor.execute( cursor.execute(
""" """
INSERT INTO _jiangchang_tables (table_name, display_name, description, visible, readonly) INSERT INTO _jiangchang_tables (
VALUES (?, ?, ?, ?, ?) table_name, display_name, description, visible, readonly, default_sort_json
)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(table_name) DO UPDATE SET ON CONFLICT(table_name) DO UPDATE SET
display_name = excluded.display_name, display_name = excluded.display_name,
description = excluded.description, description = excluded.description,
visible = excluded.visible, visible = excluded.visible,
readonly = excluded.readonly readonly = excluded.readonly,
default_sort_json = excluded.default_sort_json
""", """,
(table_name, display_name, description, visible, readonly), (table_name, display_name, description, visible, readonly, default_sort_json),
) )

View File

@@ -196,7 +196,7 @@ def _default_uses_unixepoch(default_value: object) -> bool:
def validate_options_json(options_json: str | None) -> tuple[list[str], list[str]]: def validate_options_json(options_json: str | None) -> tuple[list[str], list[str]]:
"""校验枚举 options_json 可被解析为 [{value, label}, ...] 结构。""" """校验 options_json:静态枚举数组,或 lookup 对象。"""
errors: list[str] = [] errors: list[str] = []
warnings: list[str] = [] warnings: list[str] = []
if options_json is None or str(options_json).strip() == "": if options_json is None or str(options_json).strip() == "":
@@ -206,8 +206,24 @@ def validate_options_json(options_json: str | None) -> tuple[list[str], list[str
except json.JSONDecodeError as exc: except json.JSONDecodeError as exc:
errors.append(f"options_json 不是合法 JSON:{exc}") errors.append(f"options_json 不是合法 JSON:{exc}")
return errors, warnings return errors, warnings
if isinstance(parsed, dict):
for key in ("ref_table", "value_column", "label_column"):
if not str(parsed.get(key) or "").strip():
errors.append(f"options_json lookup 必须包含非空 {key}")
ident = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
for key in ("ref_table", "value_column", "label_column", "filter_column"):
raw = parsed.get(key)
if raw is None or raw == "":
continue
if not ident.match(str(raw)):
errors.append(f"options_json.{key} 必须是安全标识符,当前为 {raw!r}")
if parsed.get("filter_column") not in (None, "") and "filter_value" not in parsed:
warnings.append("options_json lookup 声明了 filter_column 但未提供 filter_value")
return errors, warnings
if not isinstance(parsed, list): if not isinstance(parsed, list):
errors.append("options_json 必须是 JSON 数组") errors.append("options_json 必须是 JSON 数组(静态枚举)或对象(lookup)")
return errors, warnings return errors, warnings
for index, item in enumerate(parsed): for index, item in enumerate(parsed):
if not isinstance(item, dict): if not isinstance(item, dict):

View File

@@ -1,6 +1,6 @@
"""技能标识、版本与平台公共库约束(复制后请修改 slug/version/logger)。""" """技能标识、版本与平台公共库约束(复制后请修改 slug/version/logger)。"""
SKILL_SLUG = "your-skill-slug" SKILL_SLUG = "your-skill-slug"
SKILL_VERSION = "1.0.60" SKILL_VERSION = "1.0.66"
LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug" LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug"
PLATFORM_KIT_MIN_VERSION = "1.2.2" PLATFORM_KIT_MIN_VERSION = "1.2.2"

View File

@@ -190,7 +190,21 @@ class TestActionsManifest(unittest.TestCase):
placements = action.get("placements") placements = action.get("placements")
self.assertIsInstance(placements, list, msg=f"{action_id} placements must be array") self.assertIsInstance(placements, list, msg=f"{action_id} placements must be array")
self.assertGreater(len(placements), 0, msg=f"{action_id} placements empty") replaces = action.get("replaces")
if replaces is not None:
self.assertIn(
replaces,
{"create", "update", "delete"},
msg=f"{action_id} invalid replaces {replaces!r}",
)
bind = action.get("bind") or {}
tables = bind.get("tables") or []
self.assertTrue(
isinstance(tables, list) and len(tables) > 0,
msg=f"{action_id} replaces requires non-empty bind.tables",
)
else:
self.assertGreater(len(placements), 0, msg=f"{action_id} placements empty")
for placement in placements: for placement in placements:
self.assertIn(placement, _ALLOWED_PLACEMENTS, msg=f"{action_id} invalid placement {placement!r}") self.assertIn(placement, _ALLOWED_PLACEMENTS, msg=f"{action_id} invalid placement {placement!r}")
@@ -216,6 +230,31 @@ class TestActionsManifest(unittest.TestCase):
self.assertEqual(bind_def.get("required"), ["tables"]) self.assertEqual(bind_def.get("required"), ["tables"])
self.assertIs(bind_def.get("additionalProperties"), False) self.assertIs(bind_def.get("additionalProperties"), False)
def test_schema_defines_interaction_fields(self) -> None:
schema = _load_actions_schema()
action_def = schema["$defs"]["action"]
props = action_def["properties"]
self.assertIn("interaction", props)
self.assertEqual(
set(props["interaction"].get("enum") or []),
{"auto", "direct", "confirm", "form"},
)
self.assertIn("interactionByPlacement", props)
by_placement = props["interactionByPlacement"]
self.assertIs(by_placement.get("additionalProperties"), False)
for key in ("toolbar", "row", "batch", "cron", "agent", "skill-detail", "create", "update", "delete"):
self.assertIn(key, by_placement.get("properties") or {})
self.assertEqual(
set((by_placement["properties"][key].get("enum") or [])),
{"direct", "confirm", "form"},
)
self.assertIn("replaces", props)
self.assertEqual(props["replaces"].get("$ref"), "#/$defs/mutationKind")
self.assertEqual(
set(schema["$defs"]["mutationKind"].get("enum") or []),
{"create", "update", "delete"},
)
def test_manifest_validates_against_json_schema(self) -> None: def test_manifest_validates_against_json_schema(self) -> None:
try: try:
import jsonschema import jsonschema
@@ -469,6 +508,8 @@ class TestActionsManifest(unittest.TestCase):
"additionalProperties", "additionalProperties",
"placements", "placements",
"executionProfile", "executionProfile",
"interaction",
"confirmation",
): ):
self.assertIn(marker, text, msg=f"ACTIONS.md missing {marker!r}") self.assertIn(marker, text, msg=f"ACTIONS.md missing {marker!r}")
for forbidden in ( for forbidden in (

View File

@@ -1,5 +1,14 @@
# skill-template 工具 # skill-template 工具
## 桌面 UIA 拾取(开发期)
```powershell
cd D:\OpenClaw\client-commons\skill-template
python tools/ui_inspector/run.py --scene demo
```
说明见 [`ui_inspector/README.md`](ui_inspector/README.md)。企微暖树依赖本机 `jiangchang-platform-kit>=1.2.5`(与技能同调 `ensure_ui_ready`)。
## 业务仓也可以直接 clone(来源 A) ## 业务仓也可以直接 clone(来源 A)
若项目经理已在**公司自建 Gitea**([https://git.jc2009.com/](https://git.jc2009.com/),基于开源 Gitea 自建,**不是** Gitea 官网)开好业务技能仓库并授权,技术人员应直接: 若项目经理已在**公司自建 Gitea**([https://git.jc2009.com/](https://git.jc2009.com/),基于开源 Gitea 自建,**不是** Gitea 官网)开好业务技能仓库并授权,技术人员应直接:

5
tools/ui_inspector/.gitignore vendored Normal file
View File

@@ -0,0 +1,5 @@
# 仅本工具目录使用;不修改 skill-template 根 .gitignore
_captures/
__pycache__/
*.pyc
.pytest_cache/

View File

@@ -0,0 +1,49 @@
# OpenClaw UI Inspector(桌面试验版 v0.8.1)
> **有焦点的窗口 = 采集目标**(与大小无关)。
> 企微暖树:**只调用** `jiangchang_desktop_wake.ensure_ui_ready`(与技能同路径),不复制注入逻辑。
> 主窗 → profile `wecom`;搜索浮层 `SearchResultWindow2` → profile `wecom_search`(需 kit≥1.2.6)。
## 启动
```powershell
cd D:\OpenClaw\client-commons\skill-template
# 建议使用已安装 jiangchang-platform-kit>=1.2.5 的 Python(与宿主共享 venv 一致)
python tools/ui_inspector/run.py --scene demo
```
依赖:
```powershell
pip install -r tools/ui_inspector/requirements.txt
# 企微自动暖树(强烈建议)
pip install "jiangchang-platform-kit>=1.2.5" # 或指向你们 Gitea/本地 index
```
## 规则
1. 焦点在谁就采谁(排除 Inspector 自身 hwnd)
2. **自动拾取默认勾选**;点「开始流程」后先 **5 秒倒数**,请在此期间点击目标窗
3. Inspector **保持置顶**便于看进度;置顶 ≠ 焦点,不因此改采本面板
4. **不强制最大化**目标窗,保持操作者当时的大小/位置(仅最小化时会还原)
5. 不会因为窗口小 / FindAll 少而改采别的软件
6. **企微自动暖树(无感)**:识别到 `WeWorkWindow` 时,在准备/拾取前静默调用
`ensure_ui_ready("wecom")`;已暖则快速返回。结果写入每步 `meta.json` → `dump_stats.kit_wake`
## 空壳说明
- **已装 kit**:冷企微应被 n2-wake 暖到 `findall` 较大、`provider_cold=false`。若仍冷,看状态栏 `kit_wake: FAILED …`。
- **未装 kit**:行为与旧版相同,可能 `findall≈2`、`provider_cold=true`;状态栏会提示未安装 platform-kit。
## 验收建议(企微)
1. 打开并登录桌面企业微信
2. 启动 Inspector →「开始流程」→ 倒数内点到企微主窗
3. 看详情区 / 第一步 `meta.json`:`kit_wake.warm=true`,`findall` 通常 ≫ 80
4. 再搜联系人、开会话,确认 `groupbuddyname` 的 Value 能采到
## 冒烟(不弹 GUI)
```powershell
python tools/ui_inspector/smoke_check.py
```

View File

@@ -0,0 +1,3 @@
"""Desktop UI Inspector — development-only UIA capture tool (additive, disposable)."""
__version__ = "0.8.1"

View File

@@ -0,0 +1,12 @@
"""python -m tools.ui_inspector (from skill-template root, if tools is a package)
Prefer: python tools/ui_inspector/run.py
"""
from __future__ import annotations
import runpy
from pathlib import Path
if __name__ == "__main__":
runpy.run_path(str(Path(__file__).resolve().parent / "app.py"), run_name="__main__")

784
tools/ui_inspector/app.py Normal file
View File

@@ -0,0 +1,784 @@
"""Tk panel v0.8.1: 开始流程 → 手动拾取 / 自动串行循环 → 结束流程。
企微主窗 / 搜索浮层在识别/拾取前静默调用 kit ensure(wecom / wecom_search)。
"""
from __future__ import annotations
import argparse
import os
import subprocess
import sys
import threading
from pathlib import Path
import tkinter as tk
from tkinter import messagebox, simpledialog, ttk
_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import artifacts as art # noqa: E402
import desktop_uia as uia # noqa: E402
try:
from . import __version__
except Exception:
try:
import importlib.util
_spec = importlib.util.spec_from_file_location(
"ui_inspector_pkg", _HERE / "__init__.py"
)
_mod = importlib.util.module_from_spec(_spec) # type: ignore[arg-type]
assert _spec and _spec.loader
_spec.loader.exec_module(_mod)
__version__ = getattr(_mod, "__version__", "0.8.1")
except Exception:
__version__ = "0.8.1"
# Auto loop: after each dump finishes, wait this long, then dump again.
_AUTO_GAP_MS = 2000
# After frontmost target is resolved, short delay before first dump.
_AUTO_START_DELAY_MS = 400
# After「开始流程」with auto-pick: countdown so user can focus the target app.
_START_COUNTDOWN_SEC = 5
_FRONTMOST_WAIT_SEC = 15.0
class InspectorApp:
def __init__(
self,
*,
captures_root: Path,
scene: str = "desktop",
max_depth: int = 16,
max_nodes: int = 8000,
) -> None:
self.captures_root = captures_root
self.scene = scene
self.max_depth = max_depth
self.max_nodes = max_nodes
self.flow_dir: Path | None = None
self.flow: dict | None = None
self.last_window: dict | None = None
self.sticky_pid: int | None = None
self._warmed_pids: set[int] = set()
self._busy = False
self._auto_loop_active = False
self._pending_end = False
self._auto_after_id: str | None = None
self._countdown_after_id: str | None = None
self._countdown_left = 0
self.root = tk.Tk()
self.root.title(f"OpenClaw UI Inspector v{__version__} (desktop)")
self.root.geometry("840x660")
self.root.minsize(720, 540)
self.root.attributes("-topmost", True)
self.flow_name_var = tk.StringVar(value=self.scene)
self.auto_pick_var = tk.BooleanVar(value=True)
self.flow_var = tk.StringVar(value="流程:未开始(请先点「开始流程」)")
self.target_var = tk.StringVar(value="尚未拾取")
self.status_var = tk.StringVar(
value="用法:开始流程 →(默认自动拾取)5 秒倒数 → 串行 dump。"
"企微将自动 ensure_ui_ready(需 platform-kit≥1.2.5);选窗看焦点,不强制放大。"
)
self._build_ui()
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
def _build_ui(self) -> None:
pad = {"padx": 8, "pady": 4}
top = ttk.Frame(self.root)
top.pack(fill=tk.X, **pad)
ttk.Label(top, text="流程名").pack(side=tk.LEFT)
ttk.Entry(top, textvariable=self.flow_name_var, width=28).pack(side=tk.LEFT, padx=4)
ttk.Button(top, text="打开输出目录", command=self.on_open_out).pack(side=tk.LEFT, padx=8)
btns = ttk.Frame(self.root)
btns.pack(fill=tk.X, **pad)
self.start_btn = ttk.Button(btns, text="1. 开始流程", command=self.on_start_flow)
self.start_btn.pack(side=tk.LEFT, padx=2)
self.pick_btn = ttk.Button(btns, text="2. 拾取当前页", command=self.on_pick_page)
self.pick_btn.pack(side=tk.LEFT, padx=2)
self.end_btn = ttk.Button(btns, text="3. 结束流程", command=self.on_end_flow)
self.end_btn.pack(side=tk.LEFT, padx=2)
self.auto_chk = ttk.Checkbutton(
btns,
text="自动拾取(串行·步间2秒)",
variable=self.auto_pick_var,
command=self.on_auto_toggle,
)
self.auto_chk.pack(side=tk.LEFT, padx=12)
info = ttk.LabelFrame(self.root, text="状态")
info.pack(fill=tk.X, **pad)
ttk.Label(info, textvariable=self.flow_var, wraplength=800, justify=tk.LEFT).pack(
anchor=tk.W, padx=6, pady=2
)
ttk.Label(info, textvariable=self.target_var, wraplength=800, justify=tk.LEFT).pack(
anchor=tk.W, padx=6, pady=2
)
ttk.Label(info, textvariable=self.status_var, wraplength=800, justify=tk.LEFT).pack(
anchor=tk.W, padx=6, pady=2
)
tree_box = ttk.LabelFrame(self.root, text="本步拾取结果")
tree_box.pack(fill=tk.BOTH, expand=True, **pad)
scroll = ttk.Scrollbar(tree_box)
scroll.pack(side=tk.RIGHT, fill=tk.Y)
self.detail = tk.Text(
tree_box,
height=24,
wrap=tk.WORD,
yscrollcommand=scroll.set,
)
self.detail.pack(fill=tk.BOTH, expand=True, padx=6, pady=6)
scroll.config(command=self.detail.yview)
self.detail.insert(
tk.END,
"1) 填写流程名,点「开始流程」——新建一份流程文档\n"
"2) 默认已勾选「自动拾取」:先 5 秒倒数(请点目标窗到最前端),\n"
" 然后串行:dump → 保存 → 等 2 秒 → 再 dump … 直到「结束流程」\n"
" 本窗口保持置顶方便看进度;选目标看焦点并排除自身\n"
" 不强制最大化目标窗(保持你当时的大小/位置)\n"
"3) 也可取消自动,改为手动点「拾取当前页」\n"
"4) 点「结束流程」——退出循环并关闭文档\n\n"
"完整内容在 _captures/<流程>/flow.json、SUMMARY.md、steps/\n",
)
self.detail.configure(state=tk.DISABLED)
tip = (
"一个流程 = 一份文档;多次拾取 = 多个 steps。"
"自动拾取严格串行:上一拍完全结束才会开始下一拍。"
"给 AI 时优先丢 SUMMARY.md + 各步 locatable.json。"
)
ttk.Label(self.root, text=tip, wraplength=800, foreground="#444").pack(
fill=tk.X, padx=8, pady=6
)
self._refresh_button_states()
def _flow_is_open(self) -> bool:
return bool(self.flow_dir and self.flow and self.flow.get("status") == "open")
def _in_start_countdown(self) -> bool:
return self._countdown_after_id is not None or self._countdown_left > 0
def _refresh_button_states(self) -> None:
open_flow = self._flow_is_open()
counting = self._in_start_countdown()
# Manual pick only when open, idle, and not in auto loop / countdown
can_manual = (
open_flow
and not self._busy
and not self._auto_loop_active
and not counting
)
self.pick_btn.configure(state=tk.NORMAL if can_manual else tk.DISABLED)
# End must stay clickable during a long dump so the loop can be stopped
self.end_btn.configure(state=tk.NORMAL if open_flow else tk.DISABLED)
self.start_btn.configure(
state=tk.DISABLED if (self._busy or counting) else tk.NORMAL
)
# Don't flip checkbox mid-dump in a confusing way; allow toggle to stop loop
self.auto_chk.configure(state=tk.NORMAL)
def _set_status(self, text: str) -> None:
self.status_var.set(text)
def _set_detail(self, text: str) -> None:
self.detail.configure(state=tk.NORMAL)
self.detail.delete("1.0", tk.END)
self.detail.insert(tk.END, text)
self.detail.see("1.0")
self.detail.configure(state=tk.DISABLED)
def _show_picking_placeholder(self, *, auto: bool) -> None:
n = int((self.flow or {}).get("step_count") or 0) + 1
mode = "自动" if auto else "手动"
self.target_var.set(f"{mode}拾取中…(即将写入第 {n} 步)")
self._set_status(f"正在{mode}拾取:已清空本框上次结果,请稍候…")
self._set_detail(
"(已清空上次本步结果)\n\n"
f"正在读取控件树,完成后写入当前流程第 {n} 步…\n"
)
def _exclude_hwnds(self) -> set[int]:
try:
return {int(self.root.winfo_id())}
except Exception:
return set()
def _update_flow_label(self) -> None:
if not self.flow_dir or not self.flow:
self.flow_var.set("流程:未开始(请先点「开始流程」)")
return
if self._auto_loop_active:
auto = "LOOP"
elif self.auto_pick_var.get() and self._flow_is_open():
auto = "armed"
else:
auto = "off"
self.flow_var.set(
f"流程:{self.flow.get('flow_name')} | status={self.flow.get('status')} | "
f"steps={self.flow.get('step_count')} | auto={auto} | {self.flow_dir}"
)
def _cancel_auto_timer(self) -> None:
if self._auto_after_id is not None:
try:
self.root.after_cancel(self._auto_after_id)
except Exception:
pass
self._auto_after_id = None
def _cancel_countdown(self) -> None:
if self._countdown_after_id is not None:
try:
self.root.after_cancel(self._countdown_after_id)
except Exception:
pass
self._countdown_after_id = None
self._countdown_left = 0
def _stop_auto_loop(self) -> None:
self._auto_loop_active = False
self._cancel_auto_timer()
self._cancel_countdown()
def _start_auto_loop(self, *, first_delay_ms: int = _AUTO_START_DELAY_MS) -> None:
self._auto_loop_active = True
self._pending_end = False
self._update_flow_label()
self._refresh_button_states()
self._schedule_auto_next(first_delay_ms)
def _set_inspector_topmost(self, enabled: bool) -> None:
try:
self.root.attributes("-topmost", bool(enabled))
except Exception:
pass
def _ensure_inspector_visible(self) -> None:
"""Keep panel on-screen for progress; do not force-focus (avoids stealing FG)."""
try:
if str(self.root.state()) == "iconic":
self.root.deiconify()
except Exception:
pass
def _arm_auto_with_countdown(self, *, seconds: int = _START_COUNTDOWN_SEC) -> None:
"""Visible countdown, then resolve frontmost and enter auto loop."""
if not self._flow_is_open():
return
# Stop prior loop/timers without clearing the countdown we are about to arm.
self._auto_loop_active = False
self._cancel_auto_timer()
self._cancel_countdown()
self._ensure_inspector_visible()
# Stay topmost for progress; capture still uses focus + exclude self.
self._set_inspector_topmost(True)
self._countdown_left = max(1, int(seconds))
self._update_flow_label()
self._set_status(
f"倒计时 {self._countdown_left} 秒:请把要采集的窗口点到最前端"
"(本面板置顶看进度;不强制放大目标窗)。"
)
self._set_detail(
f"流程:{self.flow_dir}\n\n"
f"- {self._countdown_left} 秒内请点击目标窗口(给它焦点)\n"
f"- 本 Inspector 保持置顶,方便看进度;选窗排除自身\n"
f"- 倒计时结束后:识别最前端 → dump → 等 2 秒 → 再 dump …\n"
f"- 目标窗保持你当前的大小/位置(不强制最大化)\n"
f"- 可点「结束流程」或取消「自动拾取」中止\n"
)
self._refresh_button_states()
self._countdown_after_id = self.root.after(1000, self._countdown_tick)
def _countdown_tick(self) -> None:
self._countdown_after_id = None
if not self._flow_is_open():
self._countdown_left = 0
self._refresh_button_states()
return
if not self.auto_pick_var.get():
self._countdown_left = 0
self._refresh_button_states()
self._set_status("已取消自动拾取倒计时;可手动点「拾取当前页」。")
return
self._countdown_left -= 1
if self._countdown_left > 0:
self._set_status(
f"倒计时 {self._countdown_left} 秒:请把要采集的窗口点到最前端…"
)
self._refresh_button_states()
self._countdown_after_id = self.root.after(1000, self._countdown_tick)
return
self._countdown_left = 0
self._refresh_button_states()
self._set_status("倒计时结束,正在识别最前端窗口…")
self._begin_auto_with_probe()
def _begin_auto_with_probe(self) -> None:
"""Resolve frontmost target (not Inspector), warmup best-effort, then loop."""
if self._busy:
return
self._busy = True
self._auto_loop_active = False
self._refresh_button_states()
# Keep topmost for progress; target = focused window (Inspector hwnd excluded).
self._set_inspector_topmost(True)
self._ensure_inspector_visible()
self._set_status(
"自动拾取:正在识别最前端窗口(焦点窗=采集目标,已排除本面板)…"
)
self._set_detail(
"识别最前端窗口中…\n\n"
"- 规则:谁有焦点就采谁(与窗口大小无关)\n"
"- 本面板可置顶;选目标时排除 Inspector 自身\n"
"- 不强制最大化目标窗\n"
"- 若是企微:自动调用 kit ensure_ui_ready(与技能同路径)\n"
"- FindAll 多少只作提示,不会因此改采别的软件\n"
)
def work() -> None:
try:
uia.ensure_com_sta()
report = uia.prepare_frontmost_for_capture(
exclude_hwnds=self._exclude_hwnds(),
timeout_sec=_FRONTMOST_WAIT_SEC,
)
self.root.after(0, lambda r=report: self._on_frontmost_ready(r))
except Exception as exc:
err = exc
self.root.after(0, lambda e=err: self._on_frontmost_fail(e))
threading.Thread(target=work, daemon=True).start()
def _on_frontmost_ready(self, report: dict) -> None:
self._busy = False
self._refresh_button_states()
win = report.get("window") or {}
if win.get("pid"):
self.sticky_pid = int(win["pid"])
self._warmed_pids.add(int(win["pid"]))
import kit_wake
kit = report.get("kit_wake") or {}
kit_line = kit_wake.format_kit_wake_line(kit)
sparse_note = ""
if report.get("provider_cold") or report.get("uia_sparse"):
if kit.get("reason") == "kit_not_installed":
sparse_note = (
f"(仍偏稀 FindAll={report.get('findall')}:未安装 platform-kit,"
"企微无法走 n2-wake)"
)
elif kit.get("applicable") and not kit.get("ok"):
sparse_note = (
f"(kit 暖树未成功 FindAll={report.get('findall')} "
f"code={kit.get('error_code')};仍按此窗采集)"
)
else:
sparse_note = (
f"(UIA 仍偏稀 FindAll={report.get('findall')};仍按此前端窗采集)"
)
self._set_status(
f"已锁定最前端:{win.get('title')!r} / {win.get('class_name')!r} "
f"findall={report.get('findall')} — 开始自动串行拾取。{sparse_note}"
)
self._set_detail(
f"采集目标(最前端):\n"
f" title={win.get('title')!r}\n"
f" class={win.get('class_name')!r}\n"
f" hwnd={win.get('hwnd')} pid={win.get('pid')}\n"
f" findall={report.get('findall')} sparse={report.get('uia_sparse')} "
f"cold={report.get('provider_cold')}\n"
f" {kit_line}\n\n"
"即将进入自动循环(步间 2 秒)…\n"
)
self._set_inspector_topmost(True)
self._ensure_inspector_visible()
self._start_auto_loop(first_delay_ms=_AUTO_START_DELAY_MS)
def _on_frontmost_fail(self, exc: BaseException) -> None:
self._busy = False
self._auto_loop_active = False
# Keep auto checked (default on); user can retry or end flow.
self._set_inspector_topmost(True)
self._ensure_inspector_visible()
self._refresh_button_states()
self._set_status(f"未能锁定最前端目标: {exc}")
messagebox.showerror("UI Inspector", f"未能锁定最前端目标:\n{exc}")
def _schedule_auto_next(self, delay_ms: int) -> None:
self._cancel_auto_timer()
if not self._auto_loop_active or not self._flow_is_open():
return
self._auto_after_id = self.root.after(delay_ms, self._auto_loop_tick)
def _auto_loop_tick(self) -> None:
self._auto_after_id = None
if not self._auto_loop_active or not self._flow_is_open():
return
if self._busy:
# Should not happen in serial design; re-check soon.
self._schedule_auto_next(200)
return
self.on_pick_page(auto=True)
def _continue_auto_loop_after_pick(self) -> None:
if self._pending_end:
self._pending_end = False
self._stop_auto_loop()
self._finish_end_flow()
return
if self._auto_loop_active and self._flow_is_open():
self._set_status(
f"自动循环:第 {self.flow.get('step_count') if self.flow else '?'} 步已处理;"
f"{_AUTO_GAP_MS // 1000} 秒后开始下一拍。点「结束流程」退出循环。"
)
self._schedule_auto_next(_AUTO_GAP_MS)
def on_auto_toggle(self) -> None:
if self.auto_pick_var.get():
if (
self._flow_is_open()
and not self._auto_loop_active
and not self._busy
and not self._in_start_countdown()
):
self._arm_auto_with_countdown()
else:
self._update_flow_label()
else:
was = self._auto_loop_active or self._in_start_countdown()
self._stop_auto_loop()
self._update_flow_label()
self._refresh_button_states()
if was and self._flow_is_open():
self._set_status("已停止自动循环/倒计时;仍可手动点「拾取当前页」。")
elif self._flow_is_open():
self._set_status("自动拾取已关闭;可手动点「拾取当前页」。")
def on_open_out(self) -> None:
path = self.flow_dir if self.flow_dir else self.captures_root
path.mkdir(parents=True, exist_ok=True)
try:
os.startfile(str(path)) # type: ignore[attr-defined]
except Exception:
subprocess.Popen(["explorer", str(path)])
def on_start_flow(self) -> None:
if self._busy:
return
if self.flow and self.flow.get("status") == "open":
if not messagebox.askyesno(
"UI Inspector",
"当前已有进行中的流程。\n是否结束旧流程并开始新流程?",
):
return
self._stop_auto_loop()
self._pending_end = False
try:
art.end_flow(self.flow_dir, self.flow) # type: ignore[arg-type]
except Exception:
pass
name = (self.flow_name_var.get() or "").strip()
if not name:
name = simpledialog.askstring("流程名", "请输入流程名称:", initialvalue="desktop-flow")
if not name:
return
self.flow_name_var.set(name)
try:
self.flow_dir = art.create_flow(self.captures_root, flow_name=name)
self.flow = art.load_flow(self.flow_dir)
self._pending_end = False
self._warmed_pids.clear()
self._update_flow_label()
self._refresh_button_states()
if self.auto_pick_var.get():
# No modal / no minimize: keep panel visible for countdown + progress.
self._arm_auto_with_countdown(seconds=_START_COUNTDOWN_SEC)
else:
self._stop_auto_loop()
self._ensure_inspector_visible()
self._set_status(
f"已开始流程:{self.flow_dir.name}。请翻到目标页后点「拾取当前页」。"
)
self._set_detail(
f"流程已创建:{self.flow_dir}\n\n"
f"- flow.json / SUMMARY.md / steps/\n"
f"- 手动点「拾取当前页」追加一步;或勾选「自动拾取」进入串行循环。\n"
)
except Exception as exc:
messagebox.showerror("UI Inspector", str(exc))
def _finish_end_flow(self) -> None:
if not self.flow_dir or not self.flow:
return
if self.flow.get("status") != "open":
self._refresh_button_states()
return
try:
self.flow = art.end_flow(self.flow_dir, self.flow)
self._update_flow_label()
self._refresh_button_states()
n = self.flow.get("step_count") or 0
self._set_status(f"流程已结束,共 {n} 步。可再点「开始流程」开新文档。")
self._set_detail(
f"流程已结束:{self.flow_dir}\n\n"
f"- 共 {n} 步\n"
f"- 请打开 SUMMARY.md 查看总览\n"
)
messagebox.showinfo("UI Inspector", f"流程已结束\n共 {n} 步\n{self.flow_dir}")
except Exception as exc:
messagebox.showerror("UI Inspector", str(exc))
def on_end_flow(self) -> None:
if not self.flow_dir or not self.flow:
messagebox.showwarning("UI Inspector", "没有进行中的流程。")
return
if self.flow.get("status") != "open":
messagebox.showinfo("UI Inspector", "流程已经结束。")
return
self._stop_auto_loop()
# Keep「自动拾取」勾选状态(默认开),方便下一轮直接开始。
if self._busy:
self._pending_end = True
self._set_status("已请求结束:等待当前这一拍 dump/保存完成…")
self._refresh_button_states()
return
self._pending_end = False
self._finish_end_flow()
def on_pick_page(self, *, auto: bool = False) -> None:
if self._busy:
return
if not self._flow_is_open():
if not auto:
messagebox.showwarning("UI Inspector", "请先点「开始流程」。")
return
if auto and not self._auto_loop_active:
return
self._busy = True
self._refresh_button_states()
self._show_picking_placeholder(auto=auto)
self.root.update_idletasks()
def work() -> None:
try:
uia.ensure_com_sta()
# Keep panel topmost for progress; selection excludes Inspector hwnd.
self.root.after(0, lambda: self._set_inspector_topmost(True))
flow = art.load_flow(self.flow_dir) # type: ignore[arg-type]
if flow.get("status") != "open":
raise RuntimeError("流程已结束,取消本步拾取。")
info, cands = uia.find_capture_target(
exclude_hwnds=self._exclude_hwnds(),
prefer_pid=None if auto else self.sticky_pid,
)
# Prefer light after PID already warmed; dump_window_tree escalates
# to full poke if locatable still sparse (still no force maximize).
warmup_mode = (
"light" if int(info.pid) in self._warmed_pids else "full"
)
tree, index, stats, locatable = uia.dump_window_tree(
info.hwnd,
max_depth=self.max_depth,
max_nodes=self.max_nodes,
only_visible=False,
warmup_mode=warmup_mode,
merge_same_pid_if_sparse=True,
)
self._warmed_pids.add(int(info.pid))
window = info.to_dict()
label = art.resolve_step_label(
flow,
window_title=info.title or "",
window_class=info.class_name or "",
locatable=locatable,
)
step_dir, step_idx, step_name = art.add_step_dir(
self.flow_dir, # type: ignore[arg-type]
flow,
window_title=info.title or "",
window_class=info.class_name or "",
locatable=locatable,
label=label,
)
meta = art.build_meta(
window=window,
scene=flow.get("flow_name") or "flow",
extra={
"flow_dir": str(self.flow_dir),
"step_index": step_idx,
"step_name": step_name,
"step_label": label,
"auto_pick": auto,
"dump_stats": stats,
"selection": {
"strategy": "z-order+last-active-popup",
"sticky_pid": self.sticky_pid,
"candidate_count": len(cands),
},
},
)
art.write_json(step_dir / "meta.json", meta)
art.write_json(step_dir / "tree.json", tree)
art.write_jsonl(step_dir / "index.jsonl", index)
art.write_json(step_dir / "locatable.json", locatable)
png_ok = art.capture_window_png(info.hwnd, step_dir / "screen.png")
noteworthy = art.noteworthy_from_index(index)
art.write_step_summary(
step_dir, meta=meta, stats=stats, noteworthy=noteworthy
)
art.register_step(
self.flow_dir, # type: ignore[arg-type]
flow,
step_index=step_idx,
step_name=step_name,
window=window,
stats=stats,
label=label,
)
flow = art.load_flow(self.flow_dir) # type: ignore[arg-type]
loc_text = uia.format_locatable_text(locatable)
tree_text = uia.format_tree_text(tree)
import kit_wake
warm = stats.get("warmup") or {}
kit = stats.get("kit_wake") or {}
kit_line = kit_wake.format_kit_wake_line(kit)
sparse_note = ""
if stats.get("uia_sparse"):
sparse_note = "\n⚠ uia_sparse=true:可定位候选偏少,请结合 screen.png。\n"
mode = "自动" if auto else "手动"
header = (
f"流程步骤: #{step_idx} `{step_name}`({mode})\n"
f"窗口: title={info.title!r} class={info.class_name!r} "
f"pid={info.pid} hwnd={info.hwnd}\n"
f"kit: {kit_line}\n"
f"软poke: findall {warm.get('findall_before')} → {warm.get('findall_after')} "
f"woke={warm.get('woke')}\n"
f"统计: kept={stats.get('nodes_kept')} findall={stats.get('findall_count')} "
f"locatable={stats.get('locatable_count')} png={png_ok} "
f"sparse={stats.get('uia_sparse')} cold={stats.get('provider_cold')}\n"
f"已写入: {step_dir}\n"
f"流程文档: {self.flow_dir}\n"
f"{sparse_note}"
f"{'=' * 60}\n"
)
body = header + "\n" + loc_text + "\n\n" + ("=" * 60) + "\n控件树:\n" + tree_text
self.root.after(
0,
lambda i=info, f=flow, t=body, sn=step_name, a=auto: self._on_pick_ok(
i, f, t, sn, auto=a
),
)
except Exception as exc:
err = exc
self.root.after(0, lambda e=err, a=auto: self._on_fail(e, auto=a))
threading.Thread(target=work, daemon=True).start()
def _on_pick_ok(
self,
info: uia.WindowInfo,
flow: dict,
text: str,
step_name: str,
*,
auto: bool = False,
) -> None:
self._busy = False
self.flow = flow
self.last_window = info.to_dict()
self.sticky_pid = int(info.pid)
self._update_flow_label()
self._refresh_button_states()
self.target_var.set(
f"本步目标: title={info.title!r} class={info.class_name!r} | step=`{step_name}`"
)
self._set_detail(text)
if not auto:
self._set_status(
f"第 {flow.get('step_count')} 步已写入。可换页继续拾取,或点「结束流程」。"
)
try:
self.root.lift()
self.root.attributes("-topmost", True)
except Exception:
pass
if self._pending_end:
self._pending_end = False
self._finish_end_flow()
return
# auto: either pending end, or continue loop after gap
self._continue_auto_loop_after_pick()
def _on_fail(self, exc: BaseException, *, auto: bool = False) -> None:
self._busy = False
self._refresh_button_states()
self._set_status(f"失败: {exc}")
if not auto:
messagebox.showerror("UI Inspector", str(exc))
if self._pending_end:
self._pending_end = False
self._finish_end_flow()
return
# Auto: do not pop modal; keep looping after gap (unless ending)
self._continue_auto_loop_after_pick()
def _on_close(self) -> None:
self._stop_auto_loop()
self._pending_end = False
self.root.destroy()
def run(self) -> None:
self.captures_root.mkdir(parents=True, exist_ok=True)
self.root.mainloop()
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p = argparse.ArgumentParser(
description="OpenClaw desktop UI Inspector v0.6(流程文档:开始/拾取/结束/自动串行)"
)
p.add_argument("--out", type=Path, default=None, help="输出根目录")
p.add_argument("--scene", default="desktop", help="默认流程名")
p.add_argument("--max-depth", type=int, default=16)
p.add_argument("--max-nodes", type=int, default=8000)
return p.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
out = args.out.resolve() if args.out else art.default_captures_root(_HERE)
try:
uia.ensure_uia()
except Exception as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
app = InspectorApp(
captures_root=out,
scene=args.scene,
max_depth=args.max_depth,
max_nodes=args.max_nodes,
)
app.run()
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,694 @@
"""Flow-oriented capture artifacts: one flow document, many steps."""
from __future__ import annotations
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
try:
from . import __version__
except ImportError:
try:
import importlib.util
from pathlib import Path
_spec = importlib.util.spec_from_file_location(
"ui_inspector_pkg", Path(__file__).resolve().parent / "__init__.py"
)
_mod = importlib.util.module_from_spec(_spec) # type: ignore[arg-type]
assert _spec and _spec.loader
_spec.loader.exec_module(_mod)
__version__ = getattr(_mod, "__version__", "0.8.0")
except Exception:
__version__ = "0.8.0"
def _local_ts() -> str:
now = datetime.now().astimezone()
return now.strftime("%Y%m%dT%H%M%S%z")
def safe_slug(text: str, *, fallback: str = "session") -> str:
text = (text or "").strip()
if not text:
return fallback
text = re.sub(r"\s+", "-", text)
text = re.sub(r"[^\w\u4e00-\u9fff\-]+", "", text, flags=re.UNICODE)
text = text.strip("-_") or fallback
return text[:80]
def _clean_ui_text(text: str) -> str:
"""Strip WeCom-style rich markup and collapse whitespace."""
s = (text or "").strip()
if not s:
return ""
s = re.sub(r"<[^>]+>", "", s)
s = re.sub(r"\s+", " ", s).strip()
return s
# Name/AutomationId keys that usually mark "what page am I on"
_PAGE_HINT_KEYS: dict[str, int] = {
"panel_title_label": 100,
"title": 95,
"ww_btn_text": 70,
"main_chattext": 45,
}
_LAYOUT_NOISE = {
"bg",
"navigation_bar",
"outer_layout",
"content_container",
"title_layout",
"closebtn",
"VerticalLayoutUI",
"HorizontalLayoutUI",
"wbtn_container",
"search_layout",
"search_layout_border",
"inner_search_layout",
"ControlUI",
"LabelUI",
"item",
"externals",
"status_layout",
"search_button_region",
"search_icon",
"ww_btn_icon",
"wbadge_label",
"lasttime",
"groupbuddydescription",
"bottom_text",
"chattext",
"pane",
"Pane",
}
_CHROME_TEXT = {
"关闭",
"最小化",
"最大化",
"还原",
"取消",
"确定",
"关闭窗口",
}
# Left-rail / chrome values that appear on every WeCom page — not page identity
_GLOBAL_NAV_TEXT = {
"消息",
"邮件",
"文档",
"日程",
"待办",
"会议",
"智能文档",
"通讯录",
"微盘",
"高级功能",
"工作台",
"收藏",
"智能总结",
"回到最新位置",
}
def _row_key(r: dict[str, Any]) -> str:
return _clean_ui_text(str(r.get("name") or r.get("automation_id") or ""))
def _looks_like_search_dialog(
window_title: str, window_class: str, rows: list[dict[str, Any]]
) -> bool:
t = window_title or ""
c = window_class or ""
if "Search" in c or "搜索" in t or "添加客户" in t or "添加联系人" in t:
return True
for r in rows:
if _row_key(r) == "title":
v = _clean_ui_text(str(r.get("value") or ""))
if "添加" in v or "搜索" in v:
return True
return False
def _short_entity(text: str, *, limit: int = 16) -> str:
"""Clean contact/chat title: strip markup leftovers and @微信 suffix."""
s = _clean_ui_text(text)
if not s:
return ""
s = re.sub(r"@微信\s*$", "", s)
s = re.sub(r"@WeCom\s*$", "", s, flags=re.I)
s = s.strip(" -_|")
if len(s) > limit:
s = s[:limit].rstrip("-_")
return s
def _raw_value(r: dict[str, Any]) -> str:
return str(r.get("value") or "")
def _is_selected_buddy(raw_value: str) -> bool:
# WeCom marks the active conversation with semi-transparent white color markup.
return "#99ffffff" in (raw_value or "").lower()
def derive_step_label(
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
max_hints: int = 2,
) -> str:
"""
Build a human-readable step label from window + page-specific UIA hints.
Prefers: search box text, page title, selected chat, compose draft, action buttons.
Avoids bare「会话列表」when a concrete chat/search/input signal exists.
"""
rows = list(locatable or [])
base = _clean_ui_text(window_title)
if not base:
for r in rows:
if (r.get("control_type") or "") == "WindowControl":
base = _clean_ui_text(str(r.get("name") or ""))
if base:
break
if not base and window_class:
base = window_class
has_panel_title = any(
_row_key(r) == "panel_title_label" and _clean_ui_text(str(r.get("value") or ""))
for r in rows
)
has_groupbuddy = any(_row_key(r) == "groupbuddyname" for r in rows)
has_search_result = any(_row_key(r) == "search_result" for r in rows)
has_searchedit = any(_row_key(r) in {"searchedit", "search_edit"} for r in rows)
search_dialog = _looks_like_search_dialog(window_title, window_class, rows)
scored: list[tuple[int, str]] = []
searchedit_value = ""
selected_buddy = ""
first_buddy = ""
for r in rows:
key = _row_key(r)
if not key or key in _LAYOUT_NOISE:
continue
raw = _raw_value(r)
value = _clean_ui_text(raw)
name_disp = _clean_ui_text(str(r.get("name") or ""))
ctype = str(r.get("control_type") or "")
if key in {"searchedit", "search_edit"} and value:
searchedit_value = _short_entity(value, limit=20)
# Main-window search AND add-friend dialog both matter.
scored.append((98, f"搜索{searchedit_value}"))
continue
if key == "send_rich_edit" and value:
draft = _short_entity(value, limit=12)
if draft:
scored.append((86, f"输入{draft}"))
continue
if key == "groupbuddyname" and value:
entity = _short_entity(value, limit=14)
if not entity:
continue
if _is_selected_buddy(raw) and not selected_buddy:
selected_buddy = entity
scored.append((92, entity))
elif not first_buddy and not r.get("is_offscreen"):
first_buddy = entity
continue
pri = _PAGE_HINT_KEYS.get(key)
if pri is not None and value:
if value in _CHROME_TEXT or value in _GLOBAL_NAV_TEXT:
continue
if key == "title" and value == base:
continue
if key == "main_chattext" and (has_panel_title or has_groupbuddy):
continue
entity = _short_entity(value, limit=16)
if not entity:
continue
scored.append((pri, entity))
continue
if ctype in {
"ButtonControl",
"EditControl",
"TabItemControl",
"TextControl",
"HyperlinkControl",
"MenuItemControl",
"CheckBoxControl",
"RadioButtonControl",
"ComboBoxControl",
"ListItemControl",
}:
label = _short_entity(value or name_disp, limit=16)
if (
not label
or label in _CHROME_TEXT
or label in _LAYOUT_NOISE
or label in _GLOBAL_NAV_TEXT
):
continue
scored.append((40, label))
# Conversation context when no selected buddy / search text yet
if (
has_groupbuddy
and not selected_buddy
and first_buddy
and not has_panel_title
and not searchedit_value
):
scored.append((78, first_buddy))
strong = any(p >= 90 for p, _ in scored)
if search_dialog and has_searchedit and not searchedit_value and not strong:
scored.append((75, "搜索框"))
elif search_dialog and has_search_result and searchedit_value and not strong:
scored.append((50, "搜索结果"))
elif (
has_groupbuddy
and not has_panel_title
and not strong
and not first_buddy
and not selected_buddy
):
scored.append((60, "会话列表"))
scored.sort(key=lambda x: (-x[0], x[1]))
hints: list[str] = []
seen: set[str] = set()
for _pri, text in scored:
if not text or text == base or text in seen:
continue
if text in _CHROME_TEXT or text in _GLOBAL_NAV_TEXT:
continue
compact = (
text[2:] if text.startswith(("搜索", "输入")) and len(text) > 2 else text
)
if compact and any(
compact == s or s.endswith(compact) or compact in s for s in seen
):
continue
seen.add(text)
if compact:
seen.add(compact)
hints.append(text)
if len(hints) >= max_hints:
break
parts = [p for p in [base, *hints] if p]
if not parts:
parts = ["step"]
return "_".join(parts)
def _label_slug_key(label: str) -> str:
return safe_slug(label, fallback="step")
def _unique_step_dirname(steps_root: Path, idx: int, label: str) -> str:
"""NN_label, with -2/-3 suffix if the folder already exists."""
slug = _label_slug_key(label)
base = f"{idx:02d}_{slug}"
if not (steps_root / base).exists():
return base
n = 2
while (steps_root / f"{base}-{n}").exists():
n += 1
return f"{base}-{n}"
def _existing_label_slugs(flow: dict[str, Any]) -> set[str]:
out: set[str] = set()
for s in flow.get("steps") or []:
lab = s.get("label") or ""
name = str(s.get("name") or "")
if lab:
out.add(_label_slug_key(str(lab)))
m = re.match(r"^\d{2}_(.+?)(?:-\d+)?$", name)
if m:
out.add(m.group(1))
return out
def resolve_step_label(
flow: dict[str, Any],
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
) -> str:
"""Derive label; if it collides with earlier steps, add more hints."""
used = _existing_label_slugs(flow)
label = derive_step_label(
window_title=window_title,
window_class=window_class,
locatable=locatable,
max_hints=2,
)
if _label_slug_key(label) not in used:
return label
label3 = derive_step_label(
window_title=window_title,
window_class=window_class,
locatable=locatable,
max_hints=3,
)
if _label_slug_key(label3) not in used:
return label3
# Last resort: keep richest label; folder uniquifier adds -2
return label3
def default_captures_root(tool_dir: Optional[Path] = None) -> Path:
base = tool_dir or Path(__file__).resolve().parent
return base / "_captures"
def write_json(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(data, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as fh:
for row in rows:
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
def read_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def build_meta(
*,
window: dict[str, Any],
scene: str,
skill_root: Optional[str] = None,
extra: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
now = datetime.now().astimezone()
meta = {
"schema": "openclaw.ui_inspector.capture.v1",
"surface": "windows",
"captured_at": now.isoformat(),
"scene": scene,
"skill_root": skill_root,
"app": {
"title": window.get("title"),
"class_name": window.get("class_name"),
"process_name": window.get("process_name"),
"pid": window.get("pid"),
"hwnd": window.get("hwnd"),
},
"environment": {
"os": "windows",
"tz": now.tzname(),
},
"tool": {
"name": "openclaw-ui-inspector",
"version": __version__,
"driver": "uiautomation",
},
}
if extra:
meta["extra"] = extra
return meta
def noteworthy_from_index(index: list[dict[str, Any]], *, limit: int = 80) -> list[dict[str, Any]]:
rows = [
r
for r in index
if (r.get("name") or r.get("value") or r.get("automation_id"))
]
rows.sort(
key=lambda r: (
0 if r.get("value") else 1,
0 if r.get("name") else 1,
int(r.get("depth") or 0) if isinstance(r.get("depth"), int) and r.get("depth") >= 0 else 99,
)
)
return rows[:limit]
def capture_window_png(hwnd: int, dest: Path) -> bool:
try:
from PIL import ImageGrab
except ImportError:
return False
try:
import ctypes
from ctypes import wintypes
user32 = ctypes.windll.user32
rect = wintypes.RECT()
if not user32.GetWindowRect(int(hwnd), ctypes.byref(rect)):
return False
bbox = (int(rect.left), int(rect.top), int(rect.right), int(rect.bottom))
if bbox[2] <= bbox[0] or bbox[3] <= bbox[1]:
return False
img = ImageGrab.grab(bbox=bbox, all_screens=True)
dest.parent.mkdir(parents=True, exist_ok=True)
img.save(str(dest))
return True
except Exception:
try:
img = ImageGrab.grab(all_screens=True)
dest.parent.mkdir(parents=True, exist_ok=True)
img.save(str(dest))
return True
except Exception:
return False
# --- Flow document API -------------------------------------------------
def create_flow(
captures_root: Path,
*,
flow_name: str = "desktop-flow",
) -> Path:
"""开始流程:新建一份流程文档目录。"""
captures_root.mkdir(parents=True, exist_ok=True)
slug = safe_slug(flow_name, fallback="flow")
flow_dir = captures_root / f"{_local_ts()}_{slug}"
flow_dir.mkdir(parents=True, exist_ok=False)
(flow_dir / "steps").mkdir(parents=True, exist_ok=True)
now = datetime.now().astimezone().isoformat()
flow = {
"schema": "openclaw.ui_inspector.flow.v1",
"flow_name": flow_name,
"status": "open",
"created_at": now,
"updated_at": now,
"ended_at": None,
"step_count": 0,
"steps": [],
"tool": {
"name": "openclaw-ui-inspector",
"version": __version__,
},
}
write_json(flow_dir / "flow.json", flow)
write_flow_summary(flow_dir, flow)
return flow_dir
def load_flow(flow_dir: Path) -> dict[str, Any]:
return read_json(flow_dir / "flow.json")
def save_flow(flow_dir: Path, flow: dict[str, Any]) -> None:
flow["updated_at"] = datetime.now().astimezone().isoformat()
write_json(flow_dir / "flow.json", flow)
write_flow_summary(flow_dir, flow)
def next_step_index(flow: dict[str, Any]) -> int:
return int(flow.get("step_count") or 0) + 1
def add_step_dir(
flow_dir: Path,
flow: dict[str, Any],
*,
window_title: str = "",
window_class: str = "",
locatable: Optional[list[dict[str, Any]]] = None,
label: Optional[str] = None,
) -> tuple[Path, int, str]:
"""Allocate steps/NN_<page-fingerprint>/ under an open flow."""
if flow.get("status") != "open":
raise RuntimeError("当前流程已结束,请先「开始流程」。")
idx = next_step_index(flow)
if not label:
label = resolve_step_label(
flow,
window_title=window_title,
window_class=window_class,
locatable=locatable,
)
steps_root = flow_dir / "steps"
steps_root.mkdir(parents=True, exist_ok=True)
step_name = _unique_step_dirname(steps_root, idx, label)
step_dir = steps_root / step_name
step_dir.mkdir(parents=True, exist_ok=False)
return step_dir, idx, step_name
def register_step(
flow_dir: Path,
flow: dict[str, Any],
*,
step_index: int,
step_name: str,
window: dict[str, Any],
stats: dict[str, Any],
label: Optional[str] = None,
) -> dict[str, Any]:
entry = {
"index": step_index,
"name": step_name,
"label": label or step_name,
"captured_at": datetime.now().astimezone().isoformat(),
"window_title": window.get("title"),
"window_class": window.get("class_name"),
"pid": window.get("pid"),
"hwnd": window.get("hwnd"),
"findall": stats.get("findall_count"),
"locatable": stats.get("locatable_count"),
"kept": stats.get("nodes_kept"),
"uia_sparse": stats.get("uia_sparse"),
"path": f"steps/{step_name}",
}
steps = list(flow.get("steps") or [])
steps.append(entry)
flow["steps"] = steps
flow["step_count"] = len(steps)
save_flow(flow_dir, flow)
return entry
def end_flow(flow_dir: Path, flow: dict[str, Any]) -> dict[str, Any]:
if flow.get("status") == "ended":
return flow
flow["status"] = "ended"
flow["ended_at"] = datetime.now().astimezone().isoformat()
save_flow(flow_dir, flow)
return flow
def write_flow_summary(flow_dir: Path, flow: dict[str, Any]) -> None:
lines = [
f"# 流程:{flow.get('flow_name')}",
"",
f"- status: `{flow.get('status')}`",
f"- created_at: `{flow.get('created_at')}`",
f"- updated_at: `{flow.get('updated_at')}`",
f"- ended_at: `{flow.get('ended_at')}`",
f"- step_count: `{flow.get('step_count')}`",
"",
"## 步骤",
"",
]
steps = flow.get("steps") or []
if not steps:
lines.append("(尚无拾取步骤。点「拾取当前页」追加。)")
lines.append("")
else:
for s in steps:
lines.append(f"### {s.get('index')}. `{s.get('name')}`")
lines.append("")
if s.get("label"):
lines.append(f"- label: `{s.get('label')}`")
lines.append(
f"- window: `{s.get('window_title')}` / `{s.get('window_class')}`"
)
lines.append(
f"- stats: findall=`{s.get('findall')}` locatable=`{s.get('locatable')}` "
f"kept=`{s.get('kept')}` sparse=`{s.get('uia_sparse')}`"
)
lines.append(f"- path: `{s.get('path')}/`")
lines.append("")
lines.extend(
[
"## 说明",
"",
"- 一个流程对应本目录一份文档(`flow.json` + `SUMMARY.md` + `steps/`)。",
"- 每拾取一页,在 `steps/` 下新增一步;给 AI 时优先读本 `SUMMARY.md` 与各步 `locatable.json`。",
"- Name ≠ 界面中文时请看 Value。",
"",
]
)
(flow_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8")
def write_step_summary(
step_dir: Path,
*,
meta: dict[str, Any],
stats: Optional[dict[str, Any]] = None,
noteworthy: Optional[list[dict[str, Any]]] = None,
) -> None:
app = meta.get("app") or {}
lines = [
"# Step Summary",
"",
f"- captured_at: `{meta.get('captured_at')}`",
f"- window: `{app.get('title')}` / `{app.get('class_name')}` pid=`{app.get('pid')}`",
"",
]
if stats:
lines.extend(
[
"## Stats",
"",
f"- kept: `{stats.get('nodes_kept')}`",
f"- findall: `{stats.get('findall_count')}`",
f"- locatable: `{stats.get('locatable_count')}`",
f"- uia_sparse: `{stats.get('uia_sparse')}`",
"",
]
)
if noteworthy:
lines.extend(["## Notable", ""])
for item in noteworthy[:40]:
lines.append(
f"- `{item.get('control_type')}` Name=`{item.get('name')}` "
f"Value=`{item.get('value')}`"
)
lines.append("")
lines.extend(
[
"## Files",
"",
"- `meta.json` / `tree.json` / `index.jsonl` / `locatable.json` / `screen.png`",
"",
]
)
(step_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8")
# Back-compat aliases used by older smoke scripts
def create_session_dir(captures_root: Path, *, scene: str = "desktop") -> Path:
return create_flow(captures_root, flow_name=scene)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,192 @@
"""调用 jiangchang_desktop_wake(只调不复制注入逻辑)。
- 主窗 WeWorkWindow → ``ensure_ui_ready("wecom")``
- 搜索浮层 SearchResultWindow2 → ``ensure_ui_ready("wecom_search")``
未安装 kit 时降级为跳过,不阻断采集。
"""
from __future__ import annotations
import time
from typing import Any, Optional
_CACHE_TTL_SEC = 45.0
# profile -> cache
_caches: dict[str, dict[str, Any]] = {}
_PROFILE_WECOM = "wecom"
_PROFILE_WECOM_SEARCH = "wecom_search"
_THRESHOLDS = {
_PROFILE_WECOM: 80,
_PROFILE_WECOM_SEARCH: 20,
}
def resolve_wake_profile(*, class_name: str = "", title: str = "") -> Optional[str]:
"""返回应调用的 kit profile id;非企微相关则 None。"""
cls = (class_name or "").strip()
title = title or ""
if cls == "SearchResultWindow2":
return _PROFILE_WECOM_SEARCH
if cls == "WeWorkWindow":
return _PROFILE_WECOM
if "WeWork" in cls and ("企业微信" in title or "WeCom" in title or "wecom" in title.lower()):
return _PROFILE_WECOM
return None
def looks_like_wecom(*, class_name: str = "", title: str = "") -> bool:
return resolve_wake_profile(class_name=class_name, title=title) is not None
def kit_available() -> bool:
try:
from jiangchang_desktop_wake import ensure_ui_ready # noqa: F401
return True
except ImportError:
return False
def auto_ensure_if_wecom(
*,
class_name: str = "",
title: str = "",
hwnd: Optional[int] = None,
force: bool = False,
findall_hint: Optional[int] = None,
) -> dict[str, Any]:
"""若目标像企微主窗/搜索浮层则静默 ensure;否则跳过。"""
profile = resolve_wake_profile(class_name=class_name, title=title)
if profile is None:
return {
"applicable": False,
"skipped": True,
"reason": "not_wecom",
"available": kit_available(),
}
warm_min = int(_THRESHOLDS.get(profile, 80))
cache = _caches.setdefault(
profile, {"t": 0.0, "warm": False, "hwnd": 0, "findall": 0}
)
now = time.monotonic()
hint = int(findall_hint) if findall_hint is not None else -1
if (
not force
and cache.get("warm")
and (now - float(cache.get("t") or 0.0)) < _CACHE_TTL_SEC
and (hint < 0 or hint >= warm_min)
):
return {
"applicable": True,
"available": True,
"skipped": True,
"reason": "cache_warm",
"ok": True,
"warm": True,
"wake_applied": False,
"findall": int(cache.get("findall") or 0),
"hwnd": int(cache.get("hwnd") or 0),
"profile": profile,
}
try:
from jiangchang_desktop_wake import ensure_ui_ready
from jiangchang_desktop_wake.errors import DesktopWakeError
except ImportError:
return {
"applicable": True,
"available": False,
"skipped": True,
"reason": "kit_not_installed",
"ok": False,
"profile": profile,
"hint": "请安装 jiangchang-platform-kit>=1.2.6(含 wecom_search)后再采企微",
}
t0 = time.perf_counter()
try:
result = ensure_ui_ready(
profile,
hwnd=int(hwnd) if hwnd else None,
raise_on_cold=True,
)
except DesktopWakeError as exc:
raw = (getattr(exc, "code", None) or "").strip()
code = raw[6:] if raw.startswith("ERROR:") else (raw or "WAKE_FAILED")
return {
"applicable": True,
"available": True,
"skipped": False,
"ok": False,
"warm": False,
"wake_applied": False,
"error_code": code,
"message": str(getattr(exc, "message", None) or exc),
"elapsed_ms": int(round((time.perf_counter() - t0) * 1000)),
"profile": profile,
}
except Exception as exc:
return {
"applicable": True,
"available": True,
"skipped": False,
"ok": False,
"warm": False,
"error_code": "ENSURE_EXCEPTION",
"message": f"{type(exc).__name__}: {exc}",
"elapsed_ms": int(round((time.perf_counter() - t0) * 1000)),
"profile": profile,
}
after = result.probe_after
findall = int(after.findall_count) if after is not None else 0
report = {
"applicable": True,
"available": True,
"skipped": False,
"ok": bool(result.warm),
"warm": bool(result.warm),
"wake_applied": bool(result.wake_applied),
"findall": findall,
"hwnd": int(result.hwnd) if result.hwnd else 0,
"elapsed_ms": int(round((time.perf_counter() - t0) * 1000)),
"profile": profile,
"markers_found": list(after.markers_found) if after is not None else [],
}
if result.warm:
cache["t"] = now
cache["warm"] = True
cache["hwnd"] = report["hwnd"]
cache["findall"] = findall
else:
cache["warm"] = False
return report
def format_kit_wake_line(report: Optional[dict[str, Any]]) -> str:
if not report:
return "kit_wake: (none)"
if not report.get("applicable"):
return "kit_wake: n/a (非企微)"
if report.get("reason") == "kit_not_installed":
return "kit_wake: 未安装 platform-kit(企微可能冷树)"
prof = report.get("profile") or "?"
if report.get("skipped") and report.get("reason") == "cache_warm":
return (
f"kit_wake[{prof}]: cache warm=True findall={report.get('findall')} "
f"hwnd={report.get('hwnd')}"
)
if report.get("ok"):
return (
f"kit_wake[{prof}]: warm={report.get('warm')} "
f"wake_applied={report.get('wake_applied')} "
f"findall={report.get('findall')} elapsed_ms={report.get('elapsed_ms')}"
)
return (
f"kit_wake[{prof}]: FAILED code={report.get('error_code')} "
f"msg={report.get('message')}"
)

View File

@@ -0,0 +1,5 @@
uiautomation>=2.0.18
# 可选:导出 screen.png 需要 Pillow;没有也能 dump JSON
Pillow>=10.0.0
# 企微自动暖树:与技能同款 ensure_ui_ready(请用宿主共享 venv 或本机安装)
# jiangchang-platform-kit>=1.2.5

10
tools/ui_inspector/run.py Normal file
View File

@@ -0,0 +1,10 @@
#!/usr/bin/env python3
"""Entry: python tools/ui_inspector/run.py [--out DIR] [--scene NAME]"""
from __future__ import annotations
import runpy
from pathlib import Path
if __name__ == "__main__":
runpy.run_path(str(Path(__file__).resolve().parent / "app.py"), run_name="__main__")

View File

@@ -0,0 +1,132 @@
"""Smoke test for flow document API + step labeling + kit_wake helpers (v0.8.0)."""
from __future__ import annotations
import sys
import tempfile
from pathlib import Path
_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import artifacts as art
import kit_wake
def _assert_kit_wake_detect() -> None:
assert kit_wake.looks_like_wecom(class_name="WeWorkWindow", title="企业微信")
assert kit_wake.resolve_wake_profile(class_name="SearchResultWindow2") == "wecom_search"
assert not kit_wake.looks_like_wecom(class_name="Chrome_WidgetWin_1", title="Google")
r = kit_wake.auto_ensure_if_wecom(class_name="NotWeCom", title="x")
assert r.get("skipped") and r.get("reason") == "not_wecom"
print("OK kit_wake detect")
def _assert_label() -> None:
msg_page = [
{"control_type": "TreeItemControl", "name": "groupbuddyname", "value": "某群"},
{"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "你好"},
{"control_type": "TreeItemControl", "name": "con_op", "value": "消息"},
]
searching = [
{"control_type": "TreeItemControl", "name": "groupbuddyname", "value": "徐书建"},
{"control_type": "TreeControl", "name": "searchedit", "value": "陈德炼"},
{"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "陈德炼"},
]
selected = [
{
"control_type": "TreeItemControl",
"name": "groupbuddyname",
"value": "陈德炼<c #99ffffff>@微信</c>",
},
{"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "您好"},
]
contacts = [
{"control_type": "TreeItemControl", "name": "panel_title_label", "value": "RPA之家"},
{"control_type": "TreeItemControl", "name": "main_chattext", "value": "新的客户"},
]
new_customer = [
{"control_type": "TreeItemControl", "name": "panel_title_label", "value": "新的客户"},
{"control_type": "TreeItemControl", "name": "ww_btn_text", "value": "添加"},
]
add_empty = [
{"control_type": "TreeItemControl", "name": "title", "value": "添加客户"},
{"control_type": "TreeControl", "name": "searchedit", "value": ""},
{"control_type": "WindowControl", "name": "添加客户", "value": ""},
]
add_result = [
{"control_type": "TreeItemControl", "name": "title", "value": "添加客户"},
{"control_type": "TreeControl", "name": "searchedit", "value": "13316463772"},
{"control_type": "TreeItemControl", "name": "main_chattext", "value": "Nancy"},
{"control_type": "TreeItemControl", "name": "ww_btn_text", "value": "添加为"},
{"control_type": "TreeControl", "name": "search_result", "value": ""},
]
a = art.derive_step_label(window_title="企业微信", locatable=msg_page)
b = art.derive_step_label(window_title="企业微信", locatable=searching)
c = art.derive_step_label(window_title="企业微信", locatable=selected)
d = art.derive_step_label(window_title="企业微信", locatable=contacts)
e = art.derive_step_label(window_title="企业微信", locatable=new_customer)
f = art.derive_step_label(window_title="添加客户", locatable=add_empty)
g = art.derive_step_label(window_title="添加客户", locatable=add_result)
assert "输入你好" in a or "某群" in a, a
assert "搜索陈德炼" in b, b
assert "陈德炼" in c and "输入您好" in c, c
assert "RPA之家" in d, d
assert "新的客户" in e and "添加" in e, e
assert f.startswith("添加客户"), f
assert "13316463772" in g or "搜索13316463772" in g, g
assert "会话列表" not in a and "会话列表" not in b and "会话列表" not in c
print(f"OK labels:\n {a}\n {b}\n {c}\n {d}\n {e}\n {f}\n {g}")
def main() -> int:
_assert_kit_wake_detect()
_assert_label()
with tempfile.TemporaryDirectory(prefix="ui_flow_smoke_") as tmp:
root = Path(tmp)
flow_dir = art.create_flow(root, flow_name="demo-flow")
flow = art.load_flow(flow_dir)
loc = [
{
"control_type": "TreeItemControl",
"name": "groupbuddyname",
"value": "陈德炼<c #99ffffff>@微信</c>",
},
{"control_type": "TreeItemControl", "name": "send_rich_edit", "value": "您好"},
]
label = art.resolve_step_label(
flow, window_title="企业微信", window_class="WeWorkWindow", locatable=loc
)
step_dir, idx, name = art.add_step_dir(
flow_dir,
flow,
window_title="企业微信",
window_class="WeWorkWindow",
locatable=loc,
label=label,
)
assert idx == 1
assert "陈德炼" in name, name
art.write_json(step_dir / "meta.json", {"ok": True})
art.register_step(
flow_dir,
flow,
step_index=idx,
step_name=name,
window={"title": "企业微信", "class_name": "WeWorkWindow", "pid": 1, "hwnd": 2},
stats={"findall_count": 10, "locatable_count": 5, "nodes_kept": 3, "uia_sparse": False},
label=label,
)
flow = art.end_flow(flow_dir, art.load_flow(flow_dir))
assert flow["status"] == "ended"
print(f"OK flow={flow_dir.name} step={name}")
print("OK ui_inspector flow smoke passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())