Compare commits

...

2 Commits

Author SHA1 Message Date
48a86e56f1 chore: auto release commit (2026-07-03 18:50:07)
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-03 18:50:08 +08:00
7136689efe feat: 固化 verb-noun-platform slug 命名规范与语义校验
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-03 10:24:05 +08:00
41 changed files with 2274 additions and 556 deletions

5
.openclaw-skill-template Normal file
View File

@@ -0,0 +1,5 @@
# 此文件仅存在于 skill-template 源仓库。
# 用 tools/scaffold_skill 创建新技能时,脚本会从目标目录删除本文件。
# 若业务技能仓库中仍存在本文件,说明未正确 scaffoldtests 会失败。
template_slug: your-skill-slug
template_version: "1.0.32"

View File

@@ -1,7 +1,7 @@
--- ---
name: 技能开发模板(通用业务版) name: 技能开发模板(通用业务版)
description: "OpenClaw 通用业务技能开发模板,供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。" description: "OpenClaw 通用业务技能开发模板,供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。"
version: 1.0.14 version: 1.0.32
author: 深圳匠厂科技有限公司 author: 深圳匠厂科技有限公司
metadata: metadata:
openclaw: openclaw:
@@ -38,6 +38,7 @@ allowed-tools:
- **只有当**用户明确询问命令、参数、输出结构、开发、调试、集成或排错时,才读取 `references/``development/` - **只有当**用户明确询问命令、参数、输出结构、开发、调试、集成或排错时,才读取 `references/``development/`
-`README.md``SKILL.md` / `references` 表述冲突:对用户展示与市场说明以 `README.md` 为准对执行契约、CLI、schema、运行约束以 `SKILL.md` / `references` 为准。 -`README.md``SKILL.md` / `references` 表述冲突:对用户展示与市场说明以 `README.md` 为准对执行契约、CLI、schema、运行约束以 `SKILL.md` / `references` 为准。
- 回答用户时**不要**暴露 Playwright、DOM、Python、RPA 等实现细节,除非用户明确询问技术实现。 - 回答用户时**不要**暴露 Playwright、DOM、Python、RPA 等实现细节,除非用户明确询问技术实现。
- 用户未问技术实现时,**不要**讨论 slug 命名规则;开发者问命名或 slug 时,读取 `development/NAMING.md`
## 文档分工 ## 文档分工
@@ -78,14 +79,19 @@ python {baseDir}/scripts/main.py version
## 定制入口 ## 定制入口
1. 复制目录为你的新 skill 仓库,全局替换 `your-skill-slug` 等占位词 1. 从本模板创建新业务技能时,**不得保留模板 `.git`**;优先使用 [`tools/scaffold_skill.ps1`](tools/scaffold_skill.ps1)(见 [`tools/README.md`](tools/README.md)
2. `development/DEVELOPMENT.md` 完成开发与文档定制 2. 业务技能仓库中**不得**存在 `.openclaw-skill-template`(仅 template 源仓库保留)
3. 用户市场说明写入根 `README.md`Agent 调用契约见 `references/CLI.md``references/SCHEMA.md` 3. 复制目录为你的新 skill 仓库,全局替换 `your-skill-slug` 等占位词
4. 同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION` 4. `development/DEVELOPMENT.md` 完成开发与文档定制
5. 本地 SQLite 的中文表名/字段名、标准时间字段与字段顺序规范见 `references/SCHEMA.md`(元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序) 5. 用户市场说明写入根 `README.md`Agent 调用契约见 `references/CLI.md``references/SCHEMA.md`
6. 同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION`
7. 本地 SQLite 的中文表名/字段名、标准时间字段与字段顺序规范见 `references/SCHEMA.md`(元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序)。
## 技能命名与 slug ## 技能命名与 slug
- 标准格式:`{verb}-{noun-phrase}-{platform}`(详见 `development/NAMING.md`)。
- slug / repo / 目录 / `SKILL_SLUG` / DB 文件名:**英文** kebab-case`name``description`、根 `README.md`**中文**。
- 复制为新业务技能前**必须**阅读 `development/NAMING.md`
- `name`frontmatter**用户可见的中文业务名称**`description` 是说明,不是名称。 - `name`frontmatter**用户可见的中文业务名称**`description` 是说明,不是名称。
- `metadata.openclaw.slug`**英文 kebab-case** 机器标识;目录名、默认 `{slug}.db` 使用 slug不使用中文。 - `metadata.openclaw.slug`**英文 kebab-case** 机器标识;目录名、默认 `{slug}.db` 使用 slug不使用中文。
- 复制后全局替换 `your-skill-slug`;正式技能不得保留英文占位 `name` 或 slug。 - 复制后全局替换 `your-skill-slug`;正式技能不得保留英文占位 `name` 或 slug。

View File

@@ -0,0 +1,13 @@
amazon
shopee
xinghang
alibaba
kingdee
icbc
pingpong
worldfirst
paypal
lianlian
single-window
export-rebate
etax

4
assets/naming/scopes.txt Normal file
View File

@@ -0,0 +1,4 @@
all
batch
multi
default

25
assets/naming/verbs.txt Normal file
View File

@@ -0,0 +1,25 @@
scrape
receive
download
export
add
fill
enter
bind
withdraw
reconcile
verify
submit
query
process
reply
generate
draft
review
confirm
track
upload
match
triage
ship
reprice

View File

@@ -37,17 +37,22 @@ scripts/service/<domain>_adapter/
base.py # 数据契约dataclass+ AdapterBase 接口 base.py # 数据契约dataclass+ AdapterBase 接口
mock.py # 离线仿真,给 CI/单测 mock.py # 离线仿真,给 CI/单测
real_api.py # 真实系统 API real_api.py # 真实系统 API
sim_rpa.py # 仿真平台 RPA sim_rpa.py # 仿真平台 RPA adapter委托 *_playwright.py
real_rpa.py # 真实系统 RPA占位谨慎实现 real_rpa.py # 真实系统 RPA占位谨慎实现
scripts/service/
<platform>_playwright.py # RPA 主流程(厚);与薄 sim_rpa.py 并列
account_client.py # account-manager subprocess 唯一封装
browser_session.py # async persistent context 启动
``` ```
模板示例(均在 `examples/`**不在** `scripts/` 模板示例(均在 `examples/`**不在** `scripts/`
- **仿真浏览器 RPA 完整案例**`examples/simulator_browser_rpa/` — mock + simulator_rpa + sandbox 页面 + adapter 分层 + 测试 - **仿真浏览器 RPA 完整案例**`examples/simulator_browser_rpa/` async mock + simulator_rpa + `simulator_playwright.py` + sandbox 页面 + adapter 分层 + 测试
- **真实浏览器 RPA 完整案例**`examples/real_browser_rpa/` — 登录/验证码/滚动采集 - **真实浏览器 RPA 完整案例**`examples/real_browser_rpa/` — 登录/验证码/滚动采集 + `account_client.py`
- **真实 API / 仿真 API**`examples/real_api/``examples/simulator_api/` — 当前为规划占位,尚未沉淀可复制实现 - **真实 API / 仿真 API**`examples/real_api/``examples/simulator_api/` — 当前为规划占位,尚未沉淀可复制实现
复制 adapter 分层时,以 `examples/simulator_browser_rpa/scripts/service/adapter/` 为权威参考,在新技能中创建 `scripts/service/<domain>_adapter/` 并按 README copy map 选择性复制。 复制 adapter 分层时,以修正后的 `examples/simulator_browser_rpa/`(薄 adapter + `simulator_playwright.py`)与 `examples/real_browser_rpa/scripts/service/account_client.py` 为权威参考,在新技能中创建 `scripts/service/<domain>_adapter/` 并按 README copy map 选择性复制。
> 先判断属于四象限哪一种(`real_browser_rpa` / `real_api` / `simulator_browser_rpa` / `simulator_api`),再读对应 `examples/*/README.md`。示例是**参考架构与边界**,不是业务代码原样复制。 > 先判断属于四象限哪一种(`real_browser_rpa` / `real_api` / `simulator_browser_rpa` / `simulator_api`),再读对应 `examples/*/README.md`。示例是**参考架构与边界**,不是业务代码原样复制。
@@ -93,7 +98,7 @@ def get_adapter():
1.`SKILL.md``metadata.openclaw.dependencies.required` 声明。 1.`SKILL.md``metadata.openclaw.dependencies.required` 声明。
2. **普通兄弟技能调用**,优先走统一 **`service.sibling_bridge`**`call_sibling_json`**不要**在 `task_service.py``task_rpa.py` 等业务流程文件中到处散落 `subprocess.run` 2. **普通兄弟技能调用**,优先走统一 **`service.sibling_bridge`**`call_sibling_json`**不要**在 `task_service.py``task_rpa.py` 等业务流程文件中到处散落 `subprocess.run`
3. **account-manager 账号/租约能力**是例外:可参考 `examples/real_browser_rpa/scripts/service/account_client.py`,封装为**单一** `account_client.py`;允许在该文件内部集中通过 subprocess 调 account-manager CLI。 3. **account-manager 账号/租约能力**是例外:可参考 `examples/real_browser_rpa/scripts/service/account_client.py` 与 `examples/simulator_browser_rpa/scripts/service/account_client.py`,封装为**单一** `account_client.py`;允许在该文件内部集中通过 subprocess 调 account-manager CLI。**禁止** import `rpa_helpers` 等 account-manager 内部模块;`simulator_rpa``real_rpa` 均走同一 `pick_web_account` / `release_lease` 模式。
4. **不允许**直接 `import account-manager` 的内部 Python 模块(如 `service/``util/``db/`)。 4. **不允许**直接 `import account-manager` 的内部 Python 模块(如 `service/``util/``db/`)。
5. **pick lease 后必须 `finally release lease`**;进程被 kill 后可能残留 lease需在运维文档说明排查方式查 account-manager lease 列表 / 手动释放)。 5. **pick lease 后必须 `finally release lease`**;进程被 kill 后可能残留 lease需在运维文档说明排查方式查 account-manager lease 列表 / 手动释放)。

View File

@@ -23,6 +23,7 @@ OPENCLAW_TEST_TARGET=mock # mock | simulator_rpa | real_api | real_
TARGET_BASE_URL=https://sandbox.jc2009.com TARGET_BASE_URL=https://sandbox.jc2009.com
# ── 默认账号(仅非敏感标识;密码走 account-manager── # ── 默认账号(仅非敏感标识;密码走 account-manager──
# 浏览器 RPA simulator_rpa 联调账号来自 account-managerplatform + profile_dir不是此处的 DEFAULT_LOGIN_ID
DEFAULT_LOGIN_ID=04110001 DEFAULT_LOGIN_ID=04110001
# ── 浏览器 / RPA ── # ── 浏览器 / RPA ──

View File

@@ -79,11 +79,14 @@
你不应该直接在这个仓库里开发业务,而应该: 你不应该直接在这个仓库里开发业务,而应该:
1. 复制这个目录 0. 按 [`NAMING.md`](NAMING.md) 确定 slug`{verb}-{noun-phrase}-{platform}`
2. 改成新 skill 的目录名 1. **优先**用 [`tools/scaffold_skill.ps1`](../tools/scaffold_skill.ps1) 创建新目录(见 [`tools/README.md`](../tools/README.md)
2. 在新目录内 `git init` 并绑定**本技能**远端(**不得**保留模板 `.git`
3. 把占位内容替换掉 3. 把占位内容替换掉
4. 再开始写业务逻辑 4. 再开始写业务逻辑
> **Git 红线**:禁止资源管理器整文件夹复制后保留模板 `.git``git remote -v` 必须指向新技能仓库,不能仍是 skill-template。
## 2. 新 skill 的标准目录结构 ## 2. 新 skill 的标准目录结构
复制模板后,你应该保留下面这套结构: 复制模板后,你应该保留下面这套结构:
@@ -167,10 +170,13 @@ scripts/
- `examples/real_browser_rpa/scripts/service/task_rpa.py` - `examples/real_browser_rpa/scripts/service/task_rpa.py`
- `examples/real_browser_rpa/scripts/service/account_client.py` - `examples/real_browser_rpa/scripts/service/account_client.py`
- 若是 **仿真浏览器 RPA**(自有 sandbox、表单批量提交、可控 DOM**必须先读** `examples/simulator_browser_rpa/README.md`,再按需复制到 `scripts/service/` - 若是 **仿真浏览器 RPA**(自有 sandbox、表单批量提交、可控 DOM**必须先读** `examples/simulator_browser_rpa/README.md`,再按需复制到 `scripts/service/`
- `examples/simulator_browser_rpa/scripts/service/browser_session.py` - `examples/simulator_browser_rpa/scripts/service/browser_session.py`async
- `examples/simulator_browser_rpa/scripts/service/adapter/` - `examples/simulator_browser_rpa/scripts/service/account_client.py`subprocess对齐 `real_browser_rpa`
- `examples/simulator_browser_rpa/scripts/service/task_service.py` - `examples/simulator_browser_rpa/scripts/service/simulator_playwright.py`RPA 主流程,复制后改名为 `<platform>_playwright.py`
- `examples/simulator_browser_rpa/scripts/service/adapter/`(薄 `simulator_rpa` + mock
- `examples/simulator_browser_rpa/scripts/service/task_service.py`async 编排)
- `sandbox/demo_app.html` 仅留在 examples不进入生产 skill - `sandbox/demo_app.html` 仅留在 examples不进入生产 skill
- **禁止**`rpa_helpers`、sync Playwright 用于完整技能 RPA 主路径、`task_service` 散落 account-manager subprocess
3. **只用共享库,不在 skill 里重写反反爬** 3. **只用共享库,不在 skill 里重写反反爬**
```python ```python
from jiangchang_skill_core import config from jiangchang_skill_core import config
@@ -243,12 +249,52 @@ release workflow 会对 `scripts/` 下的 Python 源码做加密/打包。当前
### 第一步:复制模板并改目录名 ### 第一步:复制模板并改目录名
例如你要开发 `your-skill-slug`(或 `disburse-payroll-icbc` 一类领域 skill 例如你要开发 `disburse-payroll-icbc` 一类领域 skill(目录名 = slug
1. 复制 `skill-template` #### 推荐方式(首选)
2. 新目录改成与 `slug` 一致的名称(如 `your-skill-slug`
3. 初始化为独立 git 仓库 在 **skill-template 仓库根目录**执行:
4. 关联它自己的远端仓库
```powershell
.\tools\scaffold_skill.ps1 -Slug disburse-payroll-icbc -Destination D:\OpenClaw\client-gdcm\disburse-payroll-icbc
cd D:\OpenClaw\client-gdcm\disburse-payroll-icbc
git init
git remote add origin <你的新技能 Gitea/Git 仓库 URL>
git remote -v # 确认 origin 不是 skill-template
```
跨平台可用 `python tools/scaffold_skill.py --slug ... --destination ...`(见 [`tools/README.md`](../tools/README.md))。
脚手架会排除 `.git`、缓存与 `.env`,并删除 `.openclaw-skill-template` 标记;**不会**自动 `git init`。
#### 禁止方式
| 做法 | 后果 |
|------|------|
| ❌ 资源管理器整文件夹复制后不做 Git 清理 | 隐藏 `.git` 被带走push 串到 template |
| ❌ 保留模板 `.git` 只改 `remote url` | 历史、分支、对象库仍属 template |
| ❌ 未删 `.git` 就 `git init` | 嵌套/混乱仓库,难以排查 |
#### 若已手工复制(补救)
```powershell
cd <新技能目录>
Remove-Item -Recurse -Force .git
Remove-Item -Recurse -Force .pytest_cache -ErrorAction SilentlyContinue
Remove-Item -Force .openclaw-skill-template -ErrorAction SilentlyContinue
git init
git remote add origin <新技能仓库 URL>
git remote -v
```
#### AI / 编程代理复制红线
| 禁止 | 说明 |
|------|------|
| 保留模板 `.git` | 必须先 `Remove-Item -Recurse -Force .git` 再 `git init` |
| `origin` 仍指向 skill-template | 复制后必须 `git remote -v` 自检 |
| 保留 `.openclaw-skill-template` | 仅 template 源仓库可有;业务技能不得保留 |
| 复制 `.pytest_cache` / `__pycache__` | 用 scaffold 已排除;手工复制应删 |
目录名要和 skill slug 对齐,后面很多地方都依赖这个命名。 目录名要和 skill slug 对齐,后面很多地方都依赖这个命名。
@@ -682,6 +728,8 @@ uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-relea
每个新 skill 发布前,建议技术人员逐条确认: 每个新 skill 发布前,建议技术人员逐条确认:
- [ ] 目录结构符合当前模板 - [ ] 目录结构符合当前模板
- [ ] slug 符合 [`NAMING.md`](NAMING.md)verb-noun-platform
- [ ] 目录名、`SKILL.md` slug、`constants.SKILL_SLUG` 三者一致
- [ ] `SKILL.md` 中 slug、名称、描述都已替换 - [ ] `SKILL.md` 中 slug、名称、描述都已替换
- [ ] `scripts/util/constants.py` 已修改 - [ ] `scripts/util/constants.py` 已修改
- [ ] `../references/CLI.md` 示例命令已改成真实命令 - [ ] `../references/CLI.md` 示例命令已改成真实命令
@@ -696,6 +744,9 @@ uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-relea
- [ ] 源码/文档/配置为 **UTF-8 without BOM**Python 文件不得含 `U+FEFF` - [ ] 源码/文档/配置为 **UTF-8 without BOM**Python 文件不得含 `U+FEFF`
- [ ] 没有新增默认必跑测试访问真实网络或浏览器 - [ ] 没有新增默认必跑测试访问真实网络或浏览器
- [ ] 如有 integration 测试需求,已写在 `tests/integration/` 下并保持 `.sample` 后缀 - [ ] 如有 integration 测试需求,已写在 `tests/integration/` 下并保持 `.sample` 后缀
- [ ] 本仓库**不是** skill-template 的误复制(根目录**无** `.openclaw-skill-template`
- [ ] `git remote -v` 指向**本技能**远端URL 不含 skill-template 仓库名
- [ ] `git log` 首条提交属于本技能(非模板历史)
## 17. 常见错误 ## 17. 常见错误
@@ -751,11 +802,65 @@ uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-relea
- 用户说明在根 `README.md`Agent 资料在 `references/`,开发规范在 `development/` - 用户说明在根 `README.md`Agent 资料在 `references/`,开发规范在 `development/`
- 统一走 `scripts/main.py` 作为 CLI 入口 - 统一走 `scripts/main.py` 作为 CLI 入口
### 错误 5平台放 slug 最前
表现:
- slug 形如 `erp-download-settlement-report`(平台段错误地放在最前)
- 语义校验或 `test_slug_naming` 失败
要改:
- 动词放首段,平台放末段,例如 `download-settlement-report-erp`
- 见 [`NAMING.md`](NAMING.md)
### 错误 6noun 段过长导致超 48 字符
表现:
- slug 超过 48 字符
- 中间名词段堆砌过多
要改:
- 精简 noun-phrase保留核心业务含义
- 见 [`NAMING.md`](NAMING.md) §3 硬约束
### 错误 7只改目录名未改 slug
表现:
- 与「错误 1」相同但目录已改名而 `SKILL.md` / `constants.py` 仍为占位
要检查:
- `SKILL.md` 的 `metadata.openclaw.slug`
- `scripts/util/constants.py` 的 `SKILL_SLUG`
### 错误 8复制模板时带走了 `.git`
表现:
- `git remote -v` 仍显示 skill-template 的 origin
- push 到错误仓库,或 log 里全是模板历史
- 新技能目录里仍有 `.openclaw-skill-template`
修复:
```powershell
Remove-Item -Recurse -Force .git
Remove-Item -Force .openclaw-skill-template -ErrorAction SilentlyContinue
git init
git remote add origin <新技能仓库 URL>
```
以后复制请优先用 `tools/scaffold_skill.ps1`(见 [`tools/README.md`](../tools/README.md))。
## 18. 推荐开发顺序总结 ## 18. 推荐开发顺序总结
如果让一个新人照着做,我建议他按这个顺序: 如果让一个新人照着做,我建议他按这个顺序:
1. 复制模板并改目录名 1. 按 [`NAMING.md`](NAMING.md) 确定 slug复制模板并改目录名
2. 改 `SKILL.md` 与根 `README.md` 2. 改 `SKILL.md` 与根 `README.md`
3. 改 `scripts/util/constants.py` 3. 改 `scripts/util/constants.py`
4. 改 `references/` 与 `development/` 4. 改 `references/` 与 `development/`

237
development/NAMING.md Normal file
View File

@@ -0,0 +1,237 @@
# 技能 slug 命名规范verb-noun-platform
> 本文是匠厂**新业务 skill** 的 slug / Git 仓库名 / 目录名命名权威说明。复制 `skill-template` 出去开发新技能前**必读**Agent 在开发者询问命名或 slug 时也应先读本文。
---
## 1. 适用范围
| 范围 | 说明 |
|------|------|
| ✅ **适用** | 从 `skill-template` **复制出去的新技能** |
| ❌ **不追溯** | 已上线历史技能(如 `account-manager``reconcile-finance``scrape-contacts-lianjia` 等)保持不动 |
| ❌ **不修改** | 其他仓库(尤其 `account-manager`)不在本文约束范围内 |
| ✅ **模板占位** | 模板本体的 `your-skill-slug` 占位必须继续通过全部测试 |
**新技能勿仿**历史 2 段无平台形态(如 `reconcile-finance`)或不符合 verb-noun-platform 的旧 repo 名;校验器对历史 repo **不追溯**,但规范中明确标注为反模式。
---
## 2. 命名格式
标准形态(约 95% 新技能):
```text
{verb}-{noun-phrase}-{platform}
```
| 部分 | 规则 |
|------|------|
| **verb** | 1 个词,来自 [动词白名单](#5-动词白名单) |
| **noun-phrase** | 13 个词(按 `-` 分段),描述业务对象 |
| **platform** | 1 个词,来自 [平台白名单](#6-平台白名单);可含连字符(如 `single-window` |
**展示层分工不变:**
| 字段 | 语言 | 示例 |
|------|------|------|
| slug / repo / 目录 / `SKILL_SLUG` / DB 文件名 | **英文** kebab-case | `download-settlement-report-amazon` |
| `SKILL.md``name``description`、根 `README.md` | **中文** | `亚马逊结算报表下载` |
---
## 3. 硬约束(新技能强制校验)
| 约束 | 值 |
|------|-----|
| 格式 | `{verb}-{noun-phrase}-{platform}`(标准型,默认) |
| 总段数(按 `-` 分割) | **35 段**标准型scope 型见 [§4](#4-三类合法形态) |
| 总长度 | **≤ 48 字符** |
| 字符集 | 小写字母、数字、连字符kebab-case |
| 禁止 | 中文 slug、下划线、**平台放最前**(如 `amazon-download-xxx` |
实现校验见 `scripts/util/slug_naming.py``scripts/db/display_metadata_validator.py``SKILL.md` frontmatter 自检时调用。
---
## 4. 三类合法形态
### 4.1 标准型95%
`verb-noun-platform``verb-noun-noun-platform`35 段)
- 第 1 段 ∈ 动词白名单
- 最后 1 段(或 hyphenated 平台的最后 2 段)∈ 平台白名单
- 中间为 noun-phrase13 段)
示例:
- `download-settlement-report-amazon`
- `add-product-shopee`
- `enter-sales-receipt-kingdee`
- `fill-export-declaration-single-window`
### 4.2 scope 型(少见)
`verb-noun-scope`
- 最后 1 段 ∈ scope 白名单:`all` | `batch` | `multi` | `default`
- 第 1 段 ∈ 动词白名单
- 总段数 35
示例:`generate-article-all`
> 说明:`geo-generate-article-all` 若作为新技能且不在豁免表、首段 `geo` 不在动词表,应写为 `generate-article-all`。
### 4.3 历史豁免(不校验语义)
以下 slug 跳过 verb/platform 语义校验(见 `LEGACY_EXEMPT_SLUGS`
- `account-manager`
- `skill-template`
- `your-skill-slug` / `your_skill_slug`(模板占位)
---
## 5. 动词白名单
词表文件:[`assets/naming/verbs.txt`](../assets/naming/verbs.txt)(每行一词,便于后续扩展)
| 动词 | 动词 | 动词 | 动词 |
|------|------|------|------|
| scrape | receive | download | export |
| add | fill | enter | bind |
| withdraw | reconcile | verify | submit |
| query | process | reply | generate |
| draft | review | confirm | track |
| upload | match | triage | ship |
| reprice | | | |
---
## 6. 平台白名单
词表文件:[`assets/naming/platforms.txt`](../assets/naming/platforms.txt)
| 平台 | 平台 | 平台 |
|------|------|------|
| amazon | shopee | xinghang |
| alibaba | kingdee | icbc |
| pingpong | worldfirst | paypal |
| lianlian | single-window | export-rebate |
| etax | | |
scope 词表:[`assets/naming/scopes.txt`](../assets/naming/scopes.txt) — `all``batch``multi``default`
---
## 7. slug 与展示字段分工
```yaml
---
name: 亚马逊结算报表下载 # 中文,用户可见
description: "从卖家中心下载结算报表并入库" # 中文说明
metadata:
openclaw:
slug: download-settlement-report-amazon # 英文机器标识
---
```
同步修改:
- 目录名
- `metadata.openclaw.slug`
- `scripts/util/constants.py` 中的 `SKILL_SLUG`
- 默认数据库路径 `{slug}/{slug}.db`
三者必须一致。
---
## 8. 与 account-manager 的关系
slug **末段 platform**`account-manager` 中的 `platform_key` **语义对应**,但**不要求字符串完全相等**。
| slug 末段 (platform) | account-manager platform_key示例 |
|----------------------|--------------------------------------|
| amazon | `amazon_sim` / `amazon` |
| shopee | `shopee` |
| kingdee | `kingdee` |
新技能若需账号下发,在 `REQUIREMENTS.md` §0 填写拟用的 `platform_key`,并与兄弟技能调用参数对齐。
---
## 9. 复制新技能流程
1. **先定 slug** — 按本文规范拟定 `{verb}-{noun-phrase}-{platform}`,确认在白名单内且 ≤ 48 字符
2. **复制目录****推荐** [`tools/scaffold_skill.ps1`](../tools/scaffold_skill.ps1)`-Slug` 与目录名一致);勿整包复制带走 `.git`
2.5. **清理 Git若未用 scaffold** — 删除 `.git``.openclaw-skill-template``.pytest_cache`;再 `git init``git remote add origin`
3. **全局替换**`your-skill-slug` → 新 slug更新 `SKILL.md``constants.py`、文档中的占位
4. **填写 REQUIREMENTS** — 完成 [`REQUIREMENTS.md`](REQUIREMENTS.md) §0 技能标识
5. **开发 & 自检**`python tests/run_tests.py -v`,含 slug 语义测试;`git remote -v` 确认非 template
---
## 10. 正例 / 反例
### ✅ 正例
| slug | 说明 |
|------|------|
| `download-settlement-report-amazon` | 标准型4 段 |
| `add-product-shopee` | 标准型3 段 |
| `enter-sales-receipt-kingdee` | 标准型 |
| `fill-export-declaration-single-window` | 标准型,平台含连字符 |
| `generate-article-all` | scope 型 |
### ❌ 反例
| slug | 原因 |
|------|------|
| `amazon-download-settlement-report` | 平台在最前 |
| `enter-cross-border-sales-receipt-voucher-kingdee` | 过长、noun 臃肿 |
| `reconcile-finance` | 新技能禁止 2 段无平台 |
| `download_settlement_amazon` | 非 kebab-case下划线 |
---
## 11. 反模式
- **平台前置**`amazon-download-...``shopee-add-...`
- **英文占位 name**:复制后仍保留 `My Skill` 或纯英文 slug 风格名称
- **只改目录名**`SKILL.md` slug 与 `constants.SKILL_SLUG` 未同步
- **noun 堆砌**:中间段过多导致超 48 字符或难以阅读
- **仿历史 2 段 repo**`reconcile-finance``content-manager`
---
## 12. 演进说明(当前不启用)
当技能库规模 **> 200** 时,可再评估引入 `{domain}-verb-noun-platform` 前缀形态。**当前不启用**,新技能一律使用本文 verb-noun-platform 标准型或 scope 型。
---
## 13. 历史形态说明
以下为代表性**历史 repo**,新技能**勿仿**
| 历史 slug | 说明 |
|-----------|------|
| `account-manager` | 跨平台账号管理,非 verb-noun-platform |
| `reconcile-finance` | 2 段无 platform |
| `scrape-contacts-lianjia` | 平台名不在当前白名单(链家) |
| `skill-template` | 模板仓库本身 |
校验器通过 `LEGACY_EXEMPT_SLUGS` 对模板占位与少数豁免 slug 跳过语义检查;**新复制出去的业务 skill 不在豁免表内**,必须满足 §3 硬约束。
---
## 14. 相关文档
| 文档 | 内容 |
|------|------|
| [`DEVELOPMENT.md`](DEVELOPMENT.md) | 复制模板步骤、发布前检查清单 |
| [`REQUIREMENTS.md`](REQUIREMENTS.md) | §0 技能标识必填项 |
| [`../references/SCHEMA.md`](../references/SCHEMA.md) | kebab-case 与数据库路径 |
| [`../SKILL.md`](../SKILL.md) | frontmatter 命名摘要 |

View File

@@ -3,12 +3,15 @@
本目录面向**人类开发者**与 **AI 编程代理**。开始定制 skill 前,建议按以下顺序阅读: 本目录面向**人类开发者**与 **AI 编程代理**。开始定制 skill 前,建议按以下顺序阅读:
1. [`REQUIREMENTS.md`](REQUIREMENTS.md) — 需求文档模板与验收标准 1. [`REQUIREMENTS.md`](REQUIREMENTS.md) — 需求文档模板与验收标准
2. [`DEVELOPMENT.md`](DEVELOPMENT.md) — 完整开发步骤与目录规范 2. [`NAMING.md`](NAMING.md) — slug / 仓库名命名规范(复制模板前必读)
3. [`TESTING.md`](TESTING.md) — 测试分层、隔离数据根与档位开关 3. [`DEVELOPMENT.md`](DEVELOPMENT.md) — 完整开发步骤与目录规范
4. [`ADAPTER.md`](ADAPTER.md) — 涉及外部系统对接时 4. [`TESTING.md`](TESTING.md) — 测试分层、隔离数据根与档位开关
5. [`RPA.md`](RPA.md) — 涉及浏览器 / 桌面 / 手机自动化 5. [`ADAPTER.md`](ADAPTER.md) — 涉及外部系统对接
6. [`CONFIG.md`](CONFIG.md) — `.env` 规范与 bootstrap 机制 6. [`RPA.md`](RPA.md) — 涉及浏览器 / 桌面 / 手机自动化时
7. [`RUNTIME.md`](RUNTIME.md) — 共享 runtime、数据路径、发布打包与编码约定 7. [`CONFIG.md`](CONFIG.md) — `.env` 规范与 bootstrap 机制
8. [`RUNTIME.md`](RUNTIME.md) — 共享 runtime、数据路径、发布打包与编码约定
脚手架与 Git 防串库:[`../tools/README.md`](../tools/README.md)`scaffold_skill.ps1`)。
Agent 调用契约见 [`../references/CLI.md`](../references/CLI.md)、[`../references/SCHEMA.md`](../references/SCHEMA.md)。 Agent 调用契约见 [`../references/CLI.md`](../references/CLI.md)、[`../references/SCHEMA.md`](../references/SCHEMA.md)。
用户市场说明见根目录 [`README.md`](../README.md),不要写进本目录。 用户市场说明见根目录 [`README.md`](../README.md),不要写进本目录。

View File

@@ -15,6 +15,14 @@
# REQUIREMENTS # REQUIREMENTS
## 0. 技能标识(复制 REQUIREMENTS 时必填)
- 拟定 slug英文符合 `development/NAMING.md`
- 中文 name
- 中文 description一句话
- account-manager platform_key如有
- 命名形态:标准型 / scope 型
## 1. 文档目标 ## 1. 文档目标
说明这份需求文档的作用,以及它解决什么问题。 说明这份需求文档的作用,以及它解决什么问题。

View File

@@ -138,7 +138,7 @@ from jiangchang_skill_core.rpa.stealth import stealth_enabled, STEALTH_INIT_SCRI
- `examples/real_browser_rpa/scripts/service/task_rpa.py` — RPA 主流程 - `examples/real_browser_rpa/scripts/service/task_rpa.py` — RPA 主流程
- `examples/real_browser_rpa/scripts/service/account_client.py` — account-manager CLI 封装 - `examples/real_browser_rpa/scripts/service/account_client.py` — account-manager CLI 封装
仿真浏览器 RPA 的 adapter 分层权威参考见 `examples/simulator_browser_rpa/scripts/service/adapter/`**不能**替代真实浏览器 RPA 规范。 仿真浏览器 RPA 的 adapter 分层权威参考见 `examples/simulator_browser_rpa/`(含 async `simulator_playwright.py`、薄 `simulator_rpa` adapter、`account_client.py` subprocess**不能**替代真实浏览器 RPA 规范。
### 1.6 仿真浏览器 RPA 示例(必读) ### 1.6 仿真浏览器 RPA 示例(必读)
@@ -146,12 +146,32 @@ from jiangchang_skill_core.rpa.stealth import stealth_enabled, STEALTH_INIT_SCRI
1. **必须先阅读** `examples/simulator_browser_rpa/README.md` 1. **必须先阅读** `examples/simulator_browser_rpa/README.md`
2. 再参考: 2. 再参考:
- `examples/simulator_browser_rpa/scripts/service/browser_session.py` - `examples/simulator_browser_rpa/scripts/service/browser_session.py` — async persistent context**不含 goto**
- `examples/simulator_browser_rpa/scripts/service/adapter/`base / mock / simulator_rpa / dispatch - `examples/simulator_browser_rpa/scripts/service/simulator_playwright.py` — RPA 主流程(厚
- `examples/simulator_browser_rpa/scripts/service/task_service.py` - `examples/simulator_browser_rpa/scripts/service/adapter/`(薄 `simulator_rpa` + mock + dispatch
- `examples/simulator_browser_rpa/scripts/service/account_client.py` — account-manager subprocess
- `examples/simulator_browser_rpa/scripts/service/task_service.py` — async 编排
- `examples/simulator_browser_rpa/sandbox/demo_app.html` - `examples/simulator_browser_rpa/sandbox/demo_app.html`
> **先选类型再写代码**:真实第三方网站 → `real_browser_rpa`;自有 sandbox → `simulator_browser_rpa`。不要跨类型照抄 selector 或业务流程。 > **先选类型再写代码**:真实第三方网站 → `real_browser_rpa`;自有 sandbox / 行业仿真 → `simulator_browser_rpa`。不要跨类型照抄 selector 或业务流程。
### 1.7 行业仿真平台浏览器 RPAjc2009 类)
适用:**自有或共享 `sandbox.jc2009.com` 等行业仿真平台**,不是真实高风控第三方站。
工程范式与 §1.5 相同:
| 项 | 标准 |
|----|------|
| Playwright | **async** 贯穿;禁止 sync Playwright 用于完整 RPA 主路径 |
| 分层 | **薄 adapter** + `{domain}_playwright.py`(示例:`simulator_playwright.py`+ `account_client.py` subprocess |
| 登录 | **双层**:门户 HITL`#portal-user` / `#portal-pass` 类泛化 DOM+ 业务系统登录(技能内自写) |
| 账号 | `url` 用行业根,不用 `/login``auth_strategy=per_session_manual``pick_web_account` + `release_lease` |
| 禁止 | **不要** `import account-manager``rpa_helpers` 等内部模块 |
| Selector | 用户可见文案 / `get_by_role` / `name` 优先;`data-testid` 有则用、无则 fallback共享 sandbox **不要求**为技能加 testid |
| Profile | Chrome persistent profile 可能缓存旧 SPA → 联调排障:手工 `--user-data-dir` 清站点数据 |
权威 example**修正后的** `examples/simulator_browser_rpa/`(不是任何生产技能仓库)。
--- ---

View File

@@ -32,6 +32,7 @@
- 标准时间字段:`created_at` / `updated_at` 默认值、trigger 维护与 `datetime_unix_seconds` 元数据(见 `test_timestamp_columns.py` - 标准时间字段:`created_at` / `updated_at` 默认值、trigger 维护与 `datetime_unix_seconds` 元数据(见 `test_timestamp_columns.py`
- **adapter profile**[`tests/test_adapter_profile_policy.py`](../tests/test_adapter_profile_policy.py) + [`adapter_test_utils`](../tests/adapter_test_utils.py) ——验证在未授权情况下绝不误判开启真实网络/RPA。 - **adapter profile**[`tests/test_adapter_profile_policy.py`](../tests/test_adapter_profile_policy.py) + [`adapter_test_utils`](../tests/adapter_test_utils.py) ——验证在未授权情况下绝不误判开启真实网络/RPA。
- **发布打包守护**[`tests/test_release_packaging_constraints.py`](../tests/test_release_packaging_constraints.py) ——`scripts/**/*.py` 单文件 < 1000 行、文本文件 UTF-8 without BOM。 - **发布打包守护**[`tests/test_release_packaging_constraints.py`](../tests/test_release_packaging_constraints.py) ——`scripts/**/*.py` 单文件 < 1000 行、文本文件 UTF-8 without BOM。
- **slug 语义**[`tests/test_slug_naming.py`](../tests/test_slug_naming.py) ——verb-noun-platform 命名规范校验。
**原则:`tests/test_*.py` 不允许隐形访问外网、不允许拉起真实浏览器、不允许读写开发者机器的真实数据根**。如果需要仿真服务器,也应仅在 integration并由档位变量放行 **原则:`tests/test_*.py` 不允许隐形访问外网、不允许拉起真实浏览器、不允许读写开发者机器的真实数据根**。如果需要仿真服务器,也应仅在 integration并由档位变量放行
@@ -129,6 +130,20 @@ Golden fixture 流程同理([`tests/samples/test_golden_cases.py.sample`](../t
--- ---
## 7.1 simulator_rpa 联调检查表
浏览器 RPA 走 `simulator_rpa` 档位联调前,逐项确认:
- [ ] 数据目录 `.env``OPENCLAW_TEST_TARGET=simulator_rpa`(非 `mock` / `unit`
- [ ] account-manager`platform ensure` + 账号 `active` + 有 `profile_dir`
- [ ] 目标 sandbox UI 已部署(跨团队;本地可用 `sandbox/demo_app.html`
- [ ] 失败先查 `rpa-artifacts/` 截图,再查 Chrome profile 缓存(`--user-data-dir` 手工打开清站点数据)
- [ ] 默认 `python tests/run_tests.py -v` 仍全部通过mock 离线example 内 `pytest` 不启真实浏览器
参考实现:`examples/simulator_browser_rpa/README.md`
---
## 8. 新 skill 的最小测试清单 ## 8. 新 skill 的最小测试清单
以下清单 **原文摘自 [`tests/README.md`](../tests/README.md) 「新技能最小测试清单」**(复制新仓库后逐项勾选): 以下清单 **原文摘自 [`tests/README.md`](../tests/README.md) 「新技能最小测试清单」**(复制新仓库后逐项勾选):

View File

@@ -19,13 +19,23 @@
## 两类浏览器 RPA 的区别 ## 两类浏览器 RPA 的区别
| | real_browser_rpa | simulator_browser_rpa | | | real_browser_rpa | simulator_browser_rpa(修正后) |
|---|---|---| |---|---|---|
| 目标 | 真实第三方网页 | 自有仿真页面 / sandbox | | 目标 | 真实第三方网页 | 自有仿真页面 / sandbox / jc2009 类 |
| 风控 | 高(验证码、登录态、反爬) | 低DOM 可控) | | 风控 | 高(验证码、登录态、反爬) | 低DOM 可控) |
| 账号 | 通常需 account-manager + profile | 示例可用 env / demo 账号 | | 账号 | account-manager + profile | account-manager + profile`simulator_rpa` 档) |
| 用途 | 生产级真实采集/操作参考 | adapter 分层 + 表单 RPA 教学 | | 用途 | 生产级真实采集/操作参考 | adapter 分层 + 表单 RPA 教学 |
### 工程范式对比
| 工程范式 | real_browser_rpa | simulator_browser_rpa修正后 |
|---|---|---|
| async Playwright | ✅ | ✅ |
| 薄 adapter + `*_playwright.py` | `task_rpa.py` | `simulator_playwright.py` |
| account-manager subprocess | ✅ | ✅(`simulator_rpa` 档位) |
两者工程范式一致,差异在目标系统(真实站 vs 仿真站/本地 demo
## 如何选择示例 ## 如何选择示例
| 你的场景 | 参考目录 | | 你的场景 | 参考目录 |

View File

@@ -1,20 +1,20 @@
# Simulator Browser RPA Example # Simulator Browser RPA Example
这是一个**仿真浏览器 RPA** 的可复制参考实现:在可控 sandbox 页面上,通过 Playwright 完成登录、表单填写、弹窗确认、结果解析。 这是一个**仿真浏览器 RPA** 的可复制参考实现:在可控 sandbox 页面上,通过 **async Playwright** 完成门户门闩、两步登录、表单填写、PIN 弹窗、结果解析。
本示例抽象自已跑通的「批量提交类」浏览器自动化 skill但已去业务化不包含真实银行/真实资金语义。 本示例抽象自已跑通的「工行批量提交类」浏览器自动化 skill 的工程范式,但已**去业务化**——不包含真实银行/真实资金语义,也**不是** Amazon 或任何生产技能的复制
## 适合什么场景 ## 适合什么场景
- 自有 **仿真系统 / sandbox / training 页面** - 自有 **行业仿真系统** / `sandbox.jc2009.com` / 本地 `sandbox/demo_app.html`
- 需要 **浏览器 RPA** 做端到端演示或回归 - 需要 **浏览器 RPA** 做端到端演示或回归
- 表单较复杂多步登录、Tab 切换、动态增行、PIN 弹窗、成功页解析 - 表单较复杂:门户 HITL、多步登录、Tab 切换、动态增行、PIN 弹窗、成功页解析
- 需要 **adapter 分层**mockCI与 simulator_rpa半集成可切换 - 需要 **adapter 分层**`mock`CI`simulator_rpa`(半集成)可切换
## 不适合什么场景 ## 不适合什么场景
- **真实第三方网站 + 高风控 + 人工验证码** → 优先看 [`../real_browser_rpa/README.md`](../real_browser_rpa/README.md) - **真实第三方网站 + 高风控 + 人工验证码** → 优先看 [`../real_browser_rpa/README.md`](../real_browser_rpa/README.md)
- 目标系统有稳定 **官方 API** → 不要硬写浏览器 RPA,后续参考 `real_api` - 目标系统有稳定 **官方 API** → 不要硬写浏览器 RPA
- **不要把本示例的业务字段、selector、sandbox URL 原样复制**到你的 skill - **不要把本示例的业务字段、selector、sandbox URL 原样复制**到你的 skill
## 目录结构 ## 目录结构
@@ -23,22 +23,24 @@
simulator_browser_rpa/ simulator_browser_rpa/
├── README.md ├── README.md
├── sandbox/ ├── sandbox/
│ └── demo_app.html # 本地仿真页面(稳定 data-testid / name │ └── demo_app.html # 本地仿真页面(门户门闩 + 批量提交教学 DOM
├── scripts/ ├── scripts/
│ ├── service/ │ ├── service/
│ │ ├── browser_session.py # 浏览器启动URL 仅 page.goto │ │ ├── browser_session.py # async 启动 persistent context不含 goto
│ │ ├── account_client.py # 账号来源集中封装 │ │ ├── account_client.py # account-manager subprocess对齐 real_browser_rpa
│ │ ├── task_service.py # 编排层 │ │ ├── simulator_playwright.py # RPA 主流程(厚)
│ │ ├── task_service.py # async 编排层
│ │ └── adapter/ │ │ └── adapter/
│ │ ├── base.py # BatchItem / BatchSubmitResult 契约 │ │ ├── base.py # BatchItem / BatchSubmitResult 契约
│ │ ├── mock.py # mock/unit 档 │ │ ├── mock.py # mock/unit 档
│ │ ├── simulator_rpa.py │ │ ├── simulator_rpa.py # 薄 adapterpick / lease / 委托 RPA
│ │ └── __init__.py # select_adapter │ │ └── __init__.py # select_adapter
│ └── util/ │ └── util/
│ ├── constants.py │ ├── constants.py
│ └── logging.py │ └── logging.py
└── tests/ └── tests/
├── test_adapter_dispatch.py ├── test_adapter_dispatch.py
├── test_no_rpa_helpers_import.py
├── test_simulator_rpa_unit.py ├── test_simulator_rpa_unit.py
└── test_stop_reason.py └── test_stop_reason.py
``` ```
@@ -47,91 +49,95 @@ simulator_browser_rpa/
| 文件 | 职责 | | 文件 | 职责 |
|---|---| |---|---|
| `browser_session.py` | 系统 Chrome/Edge + launch args 规范;支持 profile / 无 profile | | `browser_session.py` | async 系统 Chrome/Edge + launch args**不在此 goto** |
| `account_client.py` | 示例从环境变量读账号;真实 skill 可在此集中对接 account-manager | | `account_client.py` | **唯一** account-manager subprocess 封装(`pick_web_account` / `release_lease` |
| `adapter/base.py` | 数据契约与 `BatchAdapterBase` | | `simulator_playwright.py` | 门户轮询、登录、批量表单、PIN、解析 `batch_id`、截图 |
| `adapter/simulator_rpa.py` | **薄** adapterpick 账号、lease、委托 RPA、`finally release_lease` |
| `adapter/mock.py` | 不启浏览器,供 unit/mock/CI | | `adapter/mock.py` | 不启浏览器,供 unit/mock/CI |
| `adapter/simulator_rpa.py` | 仿真 RPA 主流程 | | `task_service.py` | async 校验输入 → 选 adapter → `await submit_batch` |
| `task_service.py` | 校验输入 → 选 adapter → 提交 → 返回 summary | | `sandbox/demo_app.html` | 可控 DOM含可选门户门闩 + 原批量提交流程 |
| `sandbox/demo_app.html` | 可控 DOM配合 selector 稳定 |
| `tests/` | dispatch、URL 不进 launch args、失败转结果 |
## 核心流程 ## 核心流程
1. `task_service.run_batch_submit(target, items)` 校验参数 1. `await run_batch_submit(target, items)` 校验参数
2. `select_adapter()``OPENCLAW_TEST_TARGET` 选择 mock 或 simulator RPA 2. `select_adapter()``OPENCLAW_TEST_TARGET` 选择 mock 或 simulator RPA
3. simulator 档:`pick_simulator_account()` `set_credentials()``submit_batch()` 3. `simulator_rpa` 档:`SimulatorBrowserRpaAdapter.submit_batch` `pick_web_account(sim_batch)``submit_batch_rpa``release_lease`
4. RPA登录两步 → 批量页 → 填表 → 提交 → PIN → 解析 `batch_id` 4. RPA门户门闩(若可见则 HITL 轮询)→ 登录两步 → 批量页 → 填表 → 提交 → PIN → 解析 `batch_id`
5. 失败截图(有 `artifacts_dir` 时)→ 返回 `BatchSubmitResult` 5. 失败截图(有 `artifacts_dir` 时)→ 返回 `BatchSubmitResult`
## 复制到新 skill 时怎么做 ## 两档仿真目标
按 copy map **选择性复制**到 `scripts/`**不要**整包照搬本目录,**不要**把本 README 当作新技能根 README。 | 档位 | 目标 | 说明 |
|---|---|---|
| 本地教学 | `sandbox/demo_app.html``file://` | 结构教学 + mock 单测默认 |
| 线上联调 | `https://sandbox.jc2009.com/...` | 设 `SIMULATOR_BASE_URL` + account-manager `sim_batch` 账号 |
联调时账号 `url` 用行业根路径,**不要**写 `/login` 子路径;`auth_strategy=per_session_manual`
## account-manager 注册示例(泛化)
```bash
python .../account-manager/scripts/main.py account add-web \
--platform sim_batch \
--login-id demo001 \
--url file:///.../demo_app.html \
--environment simulator \
--auth-strategy per_session_manual
```
线上 sandbox 时将 `--url` 换为 `https://sandbox.jc2009.com/...` 占位路径(按团队部署填写)。
**PIN/密码演示值**RPA 从 account 的 `login_id` 取账号;`password` / `token_pin` 允许从 account 字段或环境变量 `SIMULATOR_PASSWORD` / `SIMULATOR_TOKEN_PIN` 回退;验证码固定演示值 `0000`(仅 sandbox 用)。
## Copy map复制到新技能 `scripts/` ## Copy map复制到新技能 `scripts/`
| 来源examples | 目标(新技能) | 说明 | | 来源examples | 目标(新技能) | 说明 |
|---|---|---| |---|---|---|
| `examples/simulator_browser_rpa/scripts/service/browser_session.py` | `scripts/service/browser_session.py` | URL 仅 `page.goto`,不进 launch args | | `browser_session.py` | `scripts/service/browser_session.py` | asyncURL 仅 `page.goto`,不进 launch args |
| `examples/simulator_browser_rpa/scripts/service/account_client.py` | `scripts/service/account_client.py` | 账号来源集中封装 | | `account_client.py` | `scripts/service/account_client.py` | subprocess对齐 `real_browser_rpa` |
| `examples/simulator_browser_rpa/scripts/service/task_service.py` | `scripts/service/task_service.py` | 编排层,合并业务逻辑 | | `simulator_playwright.py` | `scripts/service/<platform>_playwright.py` | RPA 主流程(厚) |
| `examples/simulator_browser_rpa/scripts/service/adapter/` | `scripts/service/<domain>_adapter/` | **adapter 分层权威参考** | | `adapter/` | `scripts/service/<domain>_adapter/` | `simulator_rpa` + mock |
| `examples/simulator_browser_rpa/scripts/util/constants.py` | `scripts/util/constants.py` | 须替换业务常量 | | `task_service.py` | `scripts/service/task_service.py` | async 编排 |
| `examples/simulator_browser_rpa/scripts/util/logging.py` | `scripts/util/logging.py` | 可选 | | `util/constants.py` | `scripts/util/constants.py` | 须替换 `TARGET_PLATFORM` 等业务常量 |
| `examples/simulator_browser_rpa/tests/` | `tests/` | 可按需复制 dispatch / unit 测试 | | `tests/` | `tests/` | 可按需复制 dispatch / unit / 禁止 import 守护 |
| `examples/simulator_browser_rpa/sandbox/demo_app.html` | — | **仅仿真测试**,不进入生产 skill | | `sandbox/demo_app.html` | — | **仅仿真测试**,不进入生产 skill |
| `examples/simulator_browser_rpa/README.md` | — | **只读参考**,不复制为新技能根 README |
**边界说明:**
- `sandbox/` 只用于仿真测试与本地回归,**不进入**真实生产 skill
- `adapter/` 分层base / mock / simulator_rpa是仿真浏览器 RPA 的权威参考
- `mock` / `simulator_rpa` 只用于测试与仿真,**不要**混入 `real_browser_rpa` 的真实站点逻辑
- `demo_app.html` 是可运行沙箱页面,**不是**真实业务系统
**保留:**
- adapter 分层base / mock / simulator_rpa / `select_adapter`
- `browser_session.py` 启动规范(**URL 不进 launch args**
- `account_client.py` 集中封装模式
- 关键节点结构化日志 + 失败截图
- 测试dispatch、launch args、错误转 `BatchSubmitResult`
**替换:**
- `BatchItem` 字段与校验规则
- `sandbox/demo_app.html``SIMULATOR_BASE_URL` 指向的页面
- 全部 DOM selector必须 F12 实测)
- `task_service.py` 输入输出与 DB/任务日志
- mock 返回字段、成功页 `batch_id` 解析逻辑
## 明确禁止 ## 明确禁止
- 不要复制 `__pycache__``.pytest_cache``artifacts/``recordings/` 等生成物 - **不要** `import account-manager` 内部模块(`rpa_helpers``inject_account_manager_scripts_path``get_account_credential`
- **不要在技能侧**执行 `playwright install` / `pip install playwright`Python 包由宿主共享 runtime 提供) - **不要** sync Playwright 用于完整 RPA 主路径
- **不要把 URL 放进** `launch_persistent_context(args=[...])`
- **不要在** `task_service.py` / RPA 主流程里散落 account-manager subprocess - **不要在** `task_service.py` / RPA 主流程里散落 account-manager subprocess
- **不要把真实生产系统**配置到 `simulator_rpa` 档位 - **不要在技能侧**执行 `playwright install` / `pip install playwright`
- 不要照抄 slug、logger 名、演示账号到生产 skill - **不要把 URL 放进** `launch_persistent_context(args=[...])`
- **不要** vendor `scripts/jiangchang_skill_core/`
- **不要**照抄 slug、logger 名、演示账号到生产 skill
> 旧版 `pick_simulator_account()` / `SimulatorAccount` / `set_credentials()` 已废弃;账号统一由 account-manager 下发。
## 联调检查表
- [ ] 数据目录 `.env``OPENCLAW_TEST_TARGET=simulator_rpa`(非 mock
- [ ] account-manager`platform ensure` + 账号 `active` + 有 `profile_dir`
- [ ] 目标 sandbox UI 已部署(跨团队)
- [ ] 失败先查 `rpa-artifacts/` 截图,再查 Chrome profile 缓存(可用 `--user-data-dir` 手工打开清站点数据)
- [ ] 默认 `tests/run_tests.py` / 本 example `pytest` 仍全部通过mock 离线)
## 依赖说明 ## 依赖说明
- Python 3.10+ - Python 3.10+
- Python 包 `playwright`:由**宿主共享 runtime** 提供;技能侧不要写入 `requirements.txt`,不要 `pip install playwright` - Python 包 `playwright`:由**宿主共享 runtime** 提供;技能侧不要 `pip install playwright`
- 默认使用**系统 Chrome/Edge**,不使用 Playwright 内置 Chromium - 默认使用**系统 Chrome/Edge**;技能侧不要 `playwright install`
- 技能侧不要执行 `playwright install` / `playwright install chromium` - `jiangchang_skill_core.runtime_env.find_chrome_executable`(可选)
- 本地独立开发若缺 Playwright Python 包,仅作开发机临时补装;生产由宿主解决
- `jiangchang_skill_core.runtime_env.find_chrome_executable`(可选,用于查找系统浏览器)
## 环境变量(示例) ## 环境变量(示例)
| 变量 | 说明 | | 变量 | 说明 |
|---|---| |---|---|
| `OPENCLAW_TEST_TARGET` | `mock`/`unit` → Mock默认 → Simulator RPA | | `OPENCLAW_TEST_TARGET` | `mock`/`unit` → Mock`simulator_rpa` → 仿真 RPA |
| `SIMULATOR_BASE_URL` | 仿真页 URL默认本地 `sandbox/demo_app.html` 的 file URI | | `SIMULATOR_BASE_URL` | 仿真页 URL默认本地 `demo_app.html` 的 file URI |
| `SIMULATOR_LOGIN_ID` / `PASSWORD` / `TOKEN_PIN` | 演示账号 | | `SIMULATOR_PASSWORD` / `SIMULATOR_TOKEN_PIN` | 演示密码/PIN 回退(账号字段优先) |
| `SIMULATOR_PROFILE_DIR` | 可选 persistent profile | | `PORTAL_LOGIN_WAIT_SEC` | 门户 HITL 超时(默认 120 |
| `OPENCLAW_BROWSER_HEADLESS` | `0` 默认有头;`1` 无头模式CI 可用) | | `OPENCLAW_BROWSER_HEADLESS` | `0` 默认有头;`1` 无头CI 可用) |
## 运行测试 ## 运行测试
@@ -140,9 +146,9 @@ cd examples/simulator_browser_rpa
python -m pytest tests/ -v python -m pytest tests/ -v
``` ```
测试为 mock/纯函数,**不需要启动真实浏览器**。 测试为 mock/纯函数,**不启动真实浏览器**。
## 当前状态 ## 当前状态
- 示例代码与 sandbox 页面已就绪,可作为仿真浏览器 RPA 复制起点 - async Playwright + 薄 adapter + `simulator_playwright.py` 工程范式已对齐 `real_browser_rpa`
- 未接入完整 CLI、数据库与 entitlement。 - 未接入完整 CLI、数据库与 entitlement。

View File

@@ -15,7 +15,16 @@
<body> <body>
<h1>仿真业务系统 · 批量提交(示例)</h1> <h1>仿真业务系统 · 批量提交(示例)</h1>
<section id="login-step1-section"> <section id="portal-section">
<h2>平台门户登录(示例)</h2>
<form name="portalGate">
<label>门户账号 <input id="portal-user" name="portalUser" /></label>
<label>门户密码 <input id="portal-pass" name="portalPass" type="password" /></label>
<button type="button" id="portal-enter">进入业务系统</button>
</form>
</section>
<section id="login-step1-section" class="hidden">
<h2>登录 · 第一步</h2> <h2>登录 · 第一步</h2>
<form name="simLoginStep1" data-testid="login-step1"> <form name="simLoginStep1" data-testid="login-step1">
<label>账号 <input name="loginId" data-testid="login-id" /></label> <label>账号 <input name="loginId" data-testid="login-id" /></label>
@@ -69,6 +78,11 @@
const DEMO_CAPTCHA = "0000"; const DEMO_CAPTCHA = "0000";
let rowCount = 0; let rowCount = 0;
document.getElementById('portal-enter').addEventListener('click', function () {
document.getElementById('portal-section').classList.add('hidden');
document.getElementById('login-step1-section').classList.remove('hidden');
});
document.querySelector('form[name="simLoginStep1"]').addEventListener('submit', function (e) { document.querySelector('form[name="simLoginStep1"]').addEventListener('submit', function (e) {
e.preventDefault(); e.preventDefault();
document.getElementById('login-step1-section').classList.add('hidden'); document.getElementById('login-step1-section').classList.add('hidden');

View File

@@ -1,40 +1,199 @@
"""仿真账号来源封装(示例级;真实 skill 可在此集中对接 account-manager)。""" """account-manager CLI 集成(仅 subprocess不 import 兄弟 skill 内部模块)。"""
from __future__ import annotations from __future__ import annotations
import json
import logging
import os import os
from dataclasses import dataclass import subprocess
from typing import Optional import sys
from typing import Any, Dict, List, Optional
from util.constants import ( from jiangchang_skill_core.runtime_env import get_sibling_skills_root
DEFAULT_DEMO_LOGIN_ID,
DEFAULT_DEMO_PASSWORD, from util.constants import LEASE_HOLDER, LEASE_TTL_SEC, TARGET_PLATFORM
DEFAULT_DEMO_TOKEN_PIN, from util.logging import mask_text
ENV_SIMULATOR_LOGIN_ID,
ENV_SIMULATOR_PASSWORD, logger = logging.getLogger(__name__)
ENV_SIMULATOR_PROFILE_DIR,
ENV_SIMULATOR_TOKEN_PIN, PLACEHOLDER_PLATFORM = TARGET_PLATFORM
ACCOUNT_SETUP_MESSAGE = (
f"未找到可用的 {PLACEHOLDER_PLATFORM} 账号,所以还没有打开浏览器。"
f"请先在 account-manager 中添加 platform={PLACEHOLDER_PLATFORM}"
"status=active、带 profile_dir 的账号后重新运行。"
) )
LEASE_BUSY_MESSAGE = "目标平台账号当前被其他任务占用,请等待释放租约后重试。"
@dataclass
class SimulatorAccount:
login_id: str
password: str
token_pin: str
profile_dir: str = ""
def pick_simulator_account() -> SimulatorAccount: class AccountManagerError(Exception):
def __init__(self, code: str, message: str) -> None:
super().__init__(message)
self.code = code
self.message = message
def mask_login_id(login_id: str) -> str:
return mask_text(login_id)
def _resolve_account_manager_main() -> str:
"""解析 account-manager CLI 入口路径。
优先级ACCOUNT_MANAGER_ROOT → get_sibling_skills_root → 开发机兜底。
""" """
示例账号来源:优先读环境变量,否则返回内置 demo 账号。 env_root = (os.getenv("ACCOUNT_MANAGER_ROOT") or "").strip()
if env_root and os.path.isfile(os.path.join(env_root, "scripts", "main.py")):
return os.path.join(os.path.abspath(env_root), "scripts", "main.py")
真实 skill 若依赖 account-manager应把 subprocess / sibling_bridge 调用 scripts_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
集中在本模块,不要散落到 task_service 或 RPA 主流程中。 skills_root = get_sibling_skills_root(scripts_dir)
""" candidate = os.path.join(skills_root, "account-manager", "scripts", "main.py")
return SimulatorAccount( if os.path.isfile(candidate):
login_id=(os.getenv(ENV_SIMULATOR_LOGIN_ID) or DEFAULT_DEMO_LOGIN_ID).strip(), return candidate
password=(os.getenv(ENV_SIMULATOR_PASSWORD) or DEFAULT_DEMO_PASSWORD).strip(),
token_pin=(os.getenv(ENV_SIMULATOR_TOKEN_PIN) or DEFAULT_DEMO_TOKEN_PIN).strip(), dev = r"D:\OpenClaw\client-commons\account-manager\scripts\main.py"
profile_dir=(os.getenv(ENV_SIMULATOR_PROFILE_DIR) or "").strip(), if os.path.isfile(dev):
return dev
raise AccountManagerError(
"ACCOUNT_NOT_FOUND",
"未找到 account-manager请配置 ACCOUNT_MANAGER_ROOT 或安装 account-manager 技能。",
) )
def _parse_last_json(stdout: str) -> Any:
lines = [ln.strip() for ln in (stdout or "").splitlines() if ln.strip()]
for raw in reversed(lines):
if raw.startswith("ERROR:"):
continue
try:
return json.loads(raw)
except json.JSONDecodeError:
continue
raise AccountManagerError("UNKNOWN_ERROR", "stdout 中没有可解析的 JSON。")
def _run_argv(argv_suffix: List[str]) -> subprocess.CompletedProcess[str]:
main_py = _resolve_account_manager_main()
env = os.environ.copy()
return subprocess.run(
[sys.executable, main_py, *argv_suffix],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
env=env,
)
def _normalize_error_code(raw: str) -> str:
code = (raw or "").strip()
if code.startswith("ERROR:"):
code = code[6:]
return code
def _validate_pick_payload(payload: dict[str, Any]) -> dict[str, Any]:
if payload.get("success") is False:
err = payload.get("error") if isinstance(payload.get("error"), dict) else {}
code = _normalize_error_code(str(err.get("code") or ""))
message = str(err.get("message") or "")
if code == "LEASE_CONFLICT" or "LEASE_CONFLICT" in code:
raise AccountManagerError("LEASE_CONFLICT", LEASE_BUSY_MESSAGE)
if code in ("NO_ACCOUNT", "ACCOUNT_NOT_FOUND") or "NO_ACCOUNT" in code:
raise AccountManagerError("NO_ACCOUNT", message or "没有可用账号。")
raise AccountManagerError(code or "PICK_WEB_FAILED", message or "pick-web 失败。")
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "pick-web 返回格式异常。")
if not payload.get("profile_dir"):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "pick-web 返回的账号缺少 profile_dir。")
return payload
def _pick_web_with_lease(platform: str) -> Dict[str, Any]:
proc = _run_argv(
[
"account",
"pick-web",
"--platform",
platform,
"--lease",
"--holder",
LEASE_HOLDER,
"--ttl-sec",
LEASE_TTL_SEC,
]
)
out = proc.stdout or ""
if proc.returncode != 0 and not out.strip():
raise AccountManagerError(
"PICK_WEB_FAILED",
(proc.stderr or "").strip() or "pick-web 子进程失败",
)
payload = _parse_last_json(out)
if isinstance(payload, dict) and payload.get("success") is False:
return _validate_pick_payload(payload)
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "pick-web 返回格式异常。")
return payload
def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
proc = _run_argv(["account", "get", str(account_id)])
out = proc.stdout or ""
if proc.returncode != 0 or out.startswith("ERROR:"):
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"指定账号 {account_id} 不存在或获取失败。")
raw = _parse_last_json(out)
if isinstance(raw, dict) and "data" in raw and isinstance(raw["data"], dict):
data = raw["data"]
elif isinstance(raw, dict):
data = raw
else:
raise AccountManagerError("ACCOUNT_NOT_FOUND", "account get 返回非对象 JSON。")
platform_key = str(data.get("platform_key") or "").lower()
if platform_key != platform.lower():
raise AccountManagerError(
"ACCOUNT_NOT_FOUND",
f"账号 {account_id} 不是 {platform} 平台账号。",
)
profile_dir = str(data.get("profile_dir") or "").strip()
if not profile_dir:
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 {account_id} 缺少 profile_dir。")
data["profile_dir"] = profile_dir
data["lease_token"] = ""
return data
def pick_web_account(platform: str, account_id: Optional[str] = None) -> dict:
"""获取网页账号profile_dir + lease_token"""
platform_key = (platform or PLACEHOLDER_PLATFORM).strip() or PLACEHOLDER_PLATFORM
if account_id:
try:
aid = int(account_id)
except (TypeError, ValueError):
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 ID 无效:{account_id}")
return _pick_by_id(platform_key, aid)
try:
return _pick_web_with_lease(platform_key)
except AccountManagerError as exc:
if exc.code in ("NO_ACCOUNT", "ACCOUNT_NOT_FOUND") or "NO_ACCOUNT" in exc.code:
raise AccountManagerError("ACCOUNT_SETUP_REQUIRED", ACCOUNT_SETUP_MESSAGE) from exc
raise
def release_lease(lease_token: Optional[str]) -> None:
token = (lease_token or "").strip()
if not token:
return
try:
proc = _run_argv(["lease", "release", token])
if proc.returncode != 0:
logger.warning("lease_release_failed stdout=%s", (proc.stdout or "").strip())
except Exception:
logger.warning("lease_release_exception", exc_info=True)

View File

@@ -28,5 +28,5 @@ class BatchSubmitResult:
class BatchAdapterBase: class BatchAdapterBase:
name: str = "base" name: str = "base"
def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult: async def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult:
raise NotImplementedError raise NotImplementedError

View File

@@ -11,7 +11,7 @@ from service.adapter.base import BatchAdapterBase, BatchItem, BatchSubmitResult
class MockBatchAdapter(BatchAdapterBase): class MockBatchAdapter(BatchAdapterBase):
name = "mock" name = "mock"
def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult: async def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult:
del target del target
if not items: if not items:
return BatchSubmitResult( return BatchSubmitResult(

View File

@@ -1,78 +1,45 @@
"""仿真浏览器 RPA:操作 sandbox/demo_app.html 完成批量提交""" """仿真浏览器 RPA 薄 adapterpick 账号、lease、委托 RPA、release lease"""
from __future__ import annotations from __future__ import annotations
import logging import logging
import os
import random
import re
import time
from typing import List, Optional from typing import List, Optional
from jiangchang_skill_core import config from jiangchang_skill_core import config
from service.account_client import AccountManagerError, pick_web_account, release_lease
from service.adapter.base import BatchAdapterBase, BatchItem, BatchSubmitResult from service.adapter.base import BatchAdapterBase, BatchItem, BatchSubmitResult
from service.browser_session import browser_session, find_chrome_executable from service.browser_session import find_chrome_executable
from util.constants import ( from service.simulator_playwright import submit_batch_rpa
DEFAULT_DEMO_CAPTCHA, from util.constants import TARGET_PLATFORM, resolve_simulator_base_url
DEFAULT_DEMO_LOGIN_ID,
DEFAULT_DEMO_PASSWORD,
DEFAULT_DEMO_TOKEN_PIN,
DEFAULT_TIMEOUT_MS,
LOGIN_NAVIGATE_TIMEOUT_MS,
SUBMIT_NAVIGATE_TIMEOUT_MS,
resolve_simulator_base_url,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_PAUSE_MIN_MS = 500
_PAUSE_MAX_MS = 2000
def _resolve_entry_url(account: dict) -> str:
class RpaError(Exception): url = str(account.get("url") or "").strip()
def __init__(self, message: str, screenshot_path: Optional[str] = None): if url:
super().__init__(message) return url.rstrip("/")
self.screenshot_path = screenshot_path return resolve_simulator_base_url()
class SimulatorBrowserRpaAdapter(BatchAdapterBase): class SimulatorBrowserRpaAdapter(BatchAdapterBase):
name = "simulator_browser_rpa" name = "simulator_rpa"
def __init__( def __init__(
self, self,
base_url: Optional[str] = None,
headless: Optional[bool] = None,
artifacts_dir: Optional[str] = None, artifacts_dir: Optional[str] = None,
headless: Optional[bool] = None,
) -> None: ) -> None:
self.base_url = (base_url or resolve_simulator_base_url()).rstrip("/") self.artifacts_dir = artifacts_dir
if headless is None: if headless is None:
self.headless = config.get_bool("OPENCLAW_BROWSER_HEADLESS", default=False) self.headless = config.get_bool("OPENCLAW_BROWSER_HEADLESS", default=False)
else: else:
self.headless = headless self.headless = headless
self.artifacts_dir = artifacts_dir
self._login_id: str = DEFAULT_DEMO_LOGIN_ID async def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult:
self._password: str = DEFAULT_DEMO_PASSWORD
self._token_pin: str = DEFAULT_DEMO_TOKEN_PIN
self._profile_dir: Optional[str] = None
def set_credentials(
self,
login_id: str,
password: str,
token_pin: str,
profile_dir: str,
) -> None:
self._login_id = login_id
self._password = password
self._token_pin = token_pin
stripped = (profile_dir or "").strip()
self._profile_dir = stripped or None
def submit_batch(self, target: str, items: List[BatchItem]) -> BatchSubmitResult:
try: try:
from playwright.sync_api import sync_playwright # noqa: F401 from playwright.async_api import async_playwright # noqa: F401
except ImportError: except ImportError:
return BatchSubmitResult( return BatchSubmitResult(
ok=False, ok=False,
@@ -106,236 +73,37 @@ class SimulatorBrowserRpaAdapter(BatchAdapterBase):
artifacts={"adapter": self.name}, artifacts={"adapter": self.name},
) )
start_url = self.base_url lease_token: Optional[str] = None
try: try:
with browser_session( account = pick_web_account(TARGET_PLATFORM)
start_url, lease_token = account.get("lease_token")
profile_dir=self._profile_dir, base_url = _resolve_entry_url(account)
return await submit_batch_rpa(
account,
target=target,
items=items,
base_url=base_url,
artifacts_dir=self.artifacts_dir,
headless=self.headless, headless=self.headless,
) as page: )
try: except AccountManagerError as exc:
logger.info("rpa_login_start url=%s", start_url)
self._login(page)
logger.info("rpa_login_done")
self._goto_batch_page(page)
logger.info("rpa_batch_page_ready")
self._fill_batch_form(page, target, items)
logger.info("rpa_batch_form_filled rows=%s", len(items))
batch_id = self._submit_and_get_batch_id(page)
logger.info("rpa_submit_success batch_id=%s", batch_id)
total = sum(it.amount for it in items)
return BatchSubmitResult(
ok=True,
batch_id=batch_id,
submitted_count=len(items),
submitted_amount=total,
error_msg=None,
artifacts={"adapter": self.name},
)
except RpaError as exc:
logger.error("rpa_failed: %s", exc)
arts = {"adapter": self.name}
if exc.screenshot_path:
arts["screenshot"] = exc.screenshot_path
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg=str(exc),
artifacts=arts,
)
except Exception as exc:
sp = self._safe_screenshot(page, "unexpected_error")
logger.exception("rpa_unexpected: %s", exc)
arts = {"adapter": self.name}
if sp:
arts["screenshot"] = sp
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg=f"未预期异常:{type(exc).__name__}: {exc}",
artifacts=arts,
)
except Exception as outer:
return BatchSubmitResult( return BatchSubmitResult(
ok=False, ok=False,
batch_id=None, batch_id=None,
submitted_count=0, submitted_count=0,
submitted_amount=0.0, submitted_amount=0.0,
error_msg=f"启动浏览器失败:{outer}", error_msg=f"ERROR:{exc.code} {exc.message}",
artifacts={"adapter": self.name}, artifacts={"adapter": self.name},
) )
def _pause(self, min_ms: int = _PAUSE_MIN_MS, max_ms: int = _PAUSE_MAX_MS) -> None:
time.sleep(random.uniform(min_ms, max_ms) / 1000.0)
def _login(self, page) -> None:
try:
page.wait_for_selector('form[name="simLoginStep1"]', timeout=LOGIN_NAVIGATE_TIMEOUT_MS)
self._pause()
page.fill('form[name="simLoginStep1"] input[name="loginId"]', self._login_id)
self._pause()
page.click('form[name="simLoginStep1"] button[type="submit"]')
except Exception as exc: except Exception as exc:
sp = self._safe_screenshot(page, "login_step1_failed") logger.exception("adapter_unexpected: %s", exc)
raise RpaError(f"登录第一步失败:{exc}", screenshot_path=sp) from exc return BatchSubmitResult(
ok=False,
try: batch_id=None,
page.wait_for_selector('form[name="simLoginStep2"]', timeout=DEFAULT_TIMEOUT_MS) submitted_count=0,
self._pause() submitted_amount=0.0,
page.fill('form[name="simLoginStep2"] input[name="password"]', self._password) error_msg=f"未预期异常:{type(exc).__name__}: {exc}",
page.fill('form[name="simLoginStep2"] input[name="captcha"]', DEFAULT_DEMO_CAPTCHA) artifacts={"adapter": self.name},
self._pause() )
page.click('form[name="simLoginStep2"] button[type="submit"]') finally:
except Exception as exc: release_lease(lease_token)
sp = self._safe_screenshot(page, "login_step2_failed")
raise RpaError(f"登录第二步失败:{exc}", screenshot_path=sp) from exc
try:
page.wait_for_selector('form[name="simBatchForm"]', timeout=LOGIN_NAVIGATE_TIMEOUT_MS)
except Exception as exc:
sp = self._safe_screenshot(page, "login_navigate_failed")
raise RpaError(f"登录后未进入批量提交页:{exc}", screenshot_path=sp) from exc
def _goto_batch_page(self, page) -> None:
try:
page.wait_for_selector('form[name="simBatchForm"]', timeout=DEFAULT_TIMEOUT_MS)
self._pause()
except Exception as exc:
sp = self._safe_screenshot(page, "goto_batch_failed")
raise RpaError(f"进入批量提交页失败:{exc}", screenshot_path=sp) from exc
def _select_target_account(self, page, target: str) -> None:
sel = page.locator('select[name="fromAccountNo"]')
sel.wait_for(timeout=DEFAULT_TIMEOUT_MS)
n = sel.locator("option").count()
matched_value: Optional[str] = None
first_value: Optional[str] = None
for i in range(n):
opt = sel.locator("option").nth(i)
val = opt.get_attribute("value") or ""
if not val.strip():
continue
label_txt = opt.inner_text()
if first_value is None:
first_value = val
if target.strip() and target.strip() in label_txt:
matched_value = val
break
pick = matched_value or first_value
if not pick:
raise RpaError("来源账户下拉无可选项")
if not matched_value and target.strip():
logger.warning("target '%s' 未匹配选项,回退 value=%s", target, pick)
sel.select_option(value=pick)
self._pause()
def _fill_batch_form(self, page, target: str, items: List[BatchItem]) -> None:
try:
page.get_by_role("button", name=re.compile(r"页面填写")).click()
except Exception as exc:
sp = self._safe_screenshot(page, "tab_switch_failed")
raise RpaError(f"切换到页面填写失败:{exc}", screenshot_path=sp) from exc
try:
self._select_target_account(page, target)
except RpaError:
raise
except Exception as exc:
sp = self._safe_screenshot(page, "select_account_failed")
raise RpaError(f"选择来源账户失败:{exc}", screenshot_path=sp) from exc
try:
existing = page.locator("tr[data-row-index]").count()
need_more = max(0, len(items) - existing)
add_btn = page.get_by_role("button", name=re.compile(r"新增一行"))
for _ in range(need_more):
self._pause(400, 900)
add_btn.click()
except Exception as exc:
sp = self._safe_screenshot(page, "add_rows_failed")
raise RpaError(f"新增明细行失败:{exc}", screenshot_path=sp) from exc
for it in items:
try:
row = page.locator(f'tr[data-row-index="{it.row_index}"]')
row.wait_for(timeout=DEFAULT_TIMEOUT_MS)
self._pause(300, 800)
row.locator('input[name="name"]').fill(it.name)
row.locator('input[name="account"]').fill(it.account)
row.locator('input[name="note"]').fill(it.note or "")
row.locator('input[name="amount"]').fill(f"{it.amount}")
except Exception as exc:
sp = self._safe_screenshot(page, f"fill_row_{it.row_index}_failed")
raise RpaError(f"填第 {it.row_index} 行失败:{exc}", screenshot_path=sp) from exc
try:
page.fill('input[name="purpose"]', "仿真批量提交示例")
except Exception as exc:
logger.warning("填写用途失败(非致命):%s", exc)
def _submit_and_get_batch_id(self, page) -> str:
try:
self._pause(800, 1500)
page.locator('form[name="simBatchForm"] button[type="submit"]').click()
except Exception as exc:
sp = self._safe_screenshot(page, "click_submit_failed")
raise RpaError(f"点击提交失败:{exc}", screenshot_path=sp) from exc
dlg_sel = 'div[role="dialog"][aria-modal="true"]'
try:
page.wait_for_selector(dlg_sel, timeout=DEFAULT_TIMEOUT_MS)
except Exception as exc:
sp = self._safe_screenshot(page, "pin_dialog_not_open")
raise RpaError(f"PIN 弹窗未出现:{exc}", screenshot_path=sp) from exc
try:
pin_input = page.locator(f'{dlg_sel} input[type="password"][inputmode="numeric"]')
self._pause(500, 1200)
pin_input.fill(self._token_pin)
except Exception as exc:
sp = self._safe_screenshot(page, "fill_pin_failed")
raise RpaError(f"填写 PIN 失败:{exc}", screenshot_path=sp) from exc
try:
page.locator(dlg_sel).get_by_role("button", name=re.compile(r"确认提交")).click()
except Exception as exc:
sp = self._safe_screenshot(page, "click_confirm_failed")
raise RpaError(f"确认提交失败:{exc}", screenshot_path=sp) from exc
try:
page.wait_for_selector('[data-testid="batch-id"]', timeout=SUBMIT_NAVIGATE_TIMEOUT_MS)
except Exception as exc:
sp = self._safe_screenshot(page, "success_area_not_visible")
raise RpaError(f"提交后未出现成功区域:{exc}", screenshot_path=sp) from exc
batch_id = page.locator('[data-testid="batch-id"]').get_attribute("data-batch-id") or ""
if not batch_id:
batch_id = page.locator('[data-testid="batch-id"]').inner_text().strip()
url = page.url
if not batch_id and "#/batch/" in url:
batch_id = url.split("#/batch/", 1)[1].split("?")[0].split("#")[0].strip("/")
if not batch_id:
sp = self._safe_screenshot(page, "batch_id_empty")
raise RpaError(f"无法解析 batch_idurl={url}", screenshot_path=sp)
return batch_id
def _safe_screenshot(self, page, tag: str) -> Optional[str]:
if not self.artifacts_dir:
return None
try:
os.makedirs(self.artifacts_dir, exist_ok=True)
path = os.path.join(self.artifacts_dir, f"{tag}_{int(time.time())}.png")
page.screenshot(path=path, full_page=True)
logger.info("screenshot_saved path=%s", path)
return path
except Exception as exc:
logger.warning("screenshot_failed tag=%s err=%s", tag, exc)
return None

View File

@@ -1,16 +1,14 @@
"""浏览器会话启动(系统 Chrome/EdgeURL 仅通过 page.goto 打开)。""" """浏览器会话启动(系统 Chrome/Edge + async Playwright persistent context)。"""
from __future__ import annotations from __future__ import annotations
import logging import logging
import os import os
from contextlib import contextmanager from typing import Any, List, Optional, Tuple
from typing import Any, Generator, Iterator, List, Optional, Tuple
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
DEFAULT_PAGE_TIMEOUT_MS = 15_000 DEFAULT_PAGE_TIMEOUT_MS = 15_000
GOTO_TIMEOUT_MS = 60_000
CHROME_LAUNCH_ARGS: List[str] = [ CHROME_LAUNCH_ARGS: List[str] = [
"--start-maximized", "--start-maximized",
@@ -33,11 +31,11 @@ def _headless_from_env() -> bool:
return v in ("1", "true", "yes", "on") return v in ("1", "true", "yes", "on")
def _find_chrome_executable() -> str | None: def find_chrome_executable() -> str | None:
try: try:
from jiangchang_skill_core.runtime_env import find_chrome_executable from jiangchang_skill_core.runtime_env import find_chrome_executable as _find
chrome = find_chrome_executable() chrome = _find()
if chrome and os.path.isfile(chrome): if chrome and os.path.isfile(chrome):
return chrome return chrome
except ImportError: except ImportError:
@@ -59,80 +57,70 @@ def _clear_node_options() -> None:
os.environ.pop("NODE_OPTIONS", None) os.environ.pop("NODE_OPTIONS", None)
@contextmanager async def start_browser_session(
def browser_session( profile_dir: str,
start_url: str,
*, *,
profile_dir: Optional[str] = None,
headless: Optional[bool] = None, headless: Optional[bool] = None,
) -> Generator[Any, None, None]: ) -> Tuple[Any, Any, Any]:
""" """
同步 Playwright 浏览器会话 启动持久化 Chrome profile返回 (playwright, context, page)
- Playwright Python 包由宿主共享 runtime 提供;技能侧不要 playwright install URL 不在此文件打开;由 simulator_playwright.py 负责 page.goto
- launch args 仅放 Chrome 参数,不含 URL。
- 流程launch → new_page → goto(start_url)
""" """
from playwright.sync_api import sync_playwright from playwright.async_api import async_playwright
_clear_node_options() _clear_node_options()
hl = headless if headless is not None else _headless_from_env() hl = headless if headless is not None else _headless_from_env()
chrome = _find_chrome_executable() chrome = find_chrome_executable()
if not chrome: if not chrome:
raise RuntimeError( raise RuntimeError(
"未检测到 Chrome 或 Edge 浏览器,请安装系统浏览器后重试。" "ERROR:MISSING_BROWSER 未检测到 Chrome 或 Edge 浏览器,请安装系统浏览器后重试。"
) )
args = chrome_launch_args() args = chrome_launch_args()
logger.info( logger.info(
"browser_launch chrome=%s profile_dir=%s start_url=%s args=%s headless=%s", "browser_launch chrome=%s profile_dir=%s args=%s headless=%s",
chrome, chrome,
profile_dir or "", profile_dir,
start_url,
args, args,
hl, hl,
) )
browser_holder = None pw = await async_playwright().start()
context = None context = None
with sync_playwright() as pw: try:
try: context = await pw.chromium.launch_persistent_context(
if profile_dir: user_data_dir=profile_dir,
context = pw.chromium.launch_persistent_context( headless=hl,
user_data_dir=profile_dir, executable_path=chrome,
headless=hl, locale="zh-CN",
executable_path=chrome, no_viewport=True,
locale="zh-CN", args=args,
no_viewport=True, ignore_default_args=["--enable-automation"],
args=args, )
ignore_default_args=["--enable-automation"], await context.add_init_script(STEALTH_INIT_SCRIPT)
)
else:
browser_holder = pw.chromium.launch(
headless=hl,
executable_path=chrome,
args=args,
ignore_default_args=["--enable-automation"],
)
context = browser_holder.new_context(no_viewport=True, locale="zh-CN")
context.add_init_script(STEALTH_INIT_SCRIPT) page = context.pages[0] if context.pages else await context.new_page()
page = context.new_page() page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS)
page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS) return pw, context, page
page.goto(start_url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT_MS) except Exception:
yield page if context is not None:
finally:
try: try:
if context is not None: await context.close()
context.close() except Exception:
except Exception as exc: pass
logger.warning("browser_context_close_failed: %s", exc) await pw.stop()
if browser_holder is not None: raise
try:
browser_holder.close()
except Exception as exc:
logger.warning("browser_close_failed: %s", exc)
def find_chrome_executable() -> str | None: async def close_browser_session(playwright: Any, context: Any) -> None:
return _find_chrome_executable() try:
if context is not None:
await context.close()
except Exception:
pass
try:
if playwright is not None:
await playwright.stop()
except Exception:
pass

View File

@@ -0,0 +1,400 @@
"""仿真批量提交 RPA 主流程async Playwright"""
from __future__ import annotations
import asyncio
import logging
import os
import random
import re
import time
from dataclasses import dataclass
from typing import Any, List, Optional
from jiangchang_skill_core import config
from service.adapter.base import BatchItem, BatchSubmitResult
from service.browser_session import (
close_browser_session,
find_chrome_executable,
start_browser_session,
)
from util.constants import (
DEFAULT_DEMO_CAPTCHA,
DEFAULT_DEMO_LOGIN_ID,
DEFAULT_DEMO_PASSWORD,
DEFAULT_DEMO_TOKEN_PIN,
DEFAULT_TIMEOUT_MS,
ENV_SIMULATOR_PASSWORD,
ENV_SIMULATOR_TOKEN_PIN,
LOGIN_NAVIGATE_TIMEOUT_MS,
SUBMIT_NAVIGATE_TIMEOUT_MS,
portal_login_wait_sec,
)
logger = logging.getLogger(__name__)
GOTO_TIMEOUT_MS = 60_000
_PAUSE_MIN_MS = 500
_PAUSE_MAX_MS = 2000
class RpaError(Exception):
def __init__(self, message: str, screenshot_path: Optional[str] = None):
super().__init__(message)
self.screenshot_path = screenshot_path
@dataclass
class _Credentials:
login_id: str
password: str
token_pin: str
def _resolve_credentials(account: dict[str, Any]) -> _Credentials:
login_id = str(
account.get("login_id") or account.get("username") or DEFAULT_DEMO_LOGIN_ID
).strip()
password = str(
account.get("password")
or os.getenv(ENV_SIMULATOR_PASSWORD)
or config.get(ENV_SIMULATOR_PASSWORD)
or DEFAULT_DEMO_PASSWORD
).strip()
token_pin = str(
account.get("token_pin")
or account.get("pin")
or os.getenv(ENV_SIMULATOR_TOKEN_PIN)
or config.get(ENV_SIMULATOR_TOKEN_PIN)
or DEFAULT_DEMO_TOKEN_PIN
).strip()
return _Credentials(login_id=login_id, password=password, token_pin=token_pin)
def _headless() -> bool:
return config.get_bool("OPENCLAW_BROWSER_HEADLESS", default=False)
async def _pause(min_ms: int = _PAUSE_MIN_MS, max_ms: int = _PAUSE_MAX_MS) -> None:
await asyncio.sleep(random.uniform(min_ms, max_ms) / 1000.0)
async def _visible(locator) -> bool:
try:
return await locator.count() > 0 and await locator.is_visible(timeout=500)
except Exception:
return False
async def _wait_portal_gate(page, wait_sec: int) -> None:
portal_user = page.locator("#portal-user")
try:
if not await _visible(portal_user):
return
except Exception:
return
print(f"[门户] 检测到平台门户登录门闩,请手工完成门户登录,最多等待 {wait_sec} 秒...")
deadline = time.monotonic() + wait_sec
while time.monotonic() < deadline:
try:
portal_section = page.locator("#portal-section")
login_step1 = page.locator('form[name="simLoginStep1"]')
if not await _visible(portal_section) or await _visible(login_step1):
print("[门户] 门户门闩已通过,进入业务系统登录")
return
except Exception:
pass
await asyncio.sleep(2.0)
raise RpaError("ERROR:LOGIN_TIMEOUT 门户登录超时,请重新运行并及时完成门户登录。")
async def _login(page, creds: _Credentials, artifacts_dir: str | None) -> None:
try:
await page.wait_for_selector(
'form[name="simLoginStep1"]', timeout=LOGIN_NAVIGATE_TIMEOUT_MS
)
await _pause()
await page.fill('form[name="simLoginStep1"] input[name="loginId"]', creds.login_id)
await _pause()
await page.click('form[name="simLoginStep1"] button[type="submit"]')
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "login_step1_failed")
raise RpaError(f"登录第一步失败:{exc}", screenshot_path=sp) from exc
try:
await page.wait_for_selector('form[name="simLoginStep2"]', timeout=DEFAULT_TIMEOUT_MS)
await _pause()
await page.fill('form[name="simLoginStep2"] input[name="password"]', creds.password)
await page.fill('form[name="simLoginStep2"] input[name="captcha"]', DEFAULT_DEMO_CAPTCHA)
await _pause()
await page.click('form[name="simLoginStep2"] button[type="submit"]')
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "login_step2_failed")
raise RpaError(f"登录第二步失败:{exc}", screenshot_path=sp) from exc
try:
await page.wait_for_selector('form[name="simBatchForm"]', timeout=LOGIN_NAVIGATE_TIMEOUT_MS)
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "login_navigate_failed")
raise RpaError(f"登录后未进入批量提交页:{exc}", screenshot_path=sp) from exc
async def _goto_batch_page(page, artifacts_dir: str | None) -> None:
try:
await page.wait_for_selector('form[name="simBatchForm"]', timeout=DEFAULT_TIMEOUT_MS)
await _pause()
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "goto_batch_failed")
raise RpaError(f"进入批量提交页失败:{exc}", screenshot_path=sp) from exc
async def _select_target_account(page, target: str) -> None:
sel = page.locator('select[name="fromAccountNo"]')
await sel.wait_for(timeout=DEFAULT_TIMEOUT_MS)
n = await sel.locator("option").count()
matched_value: Optional[str] = None
first_value: Optional[str] = None
for i in range(n):
opt = sel.locator("option").nth(i)
val = (await opt.get_attribute("value")) or ""
if not val.strip():
continue
label_txt = await opt.inner_text()
if first_value is None:
first_value = val
if target.strip() and target.strip() in label_txt:
matched_value = val
break
pick = matched_value or first_value
if not pick:
raise RpaError("来源账户下拉无可选项")
if not matched_value and target.strip():
logger.warning("target '%s' 未匹配选项,回退 value=%s", target, pick)
await sel.select_option(value=pick)
await _pause()
async def _fill_batch_form(
page, target: str, items: List[BatchItem], artifacts_dir: str | None
) -> None:
try:
await page.get_by_role("button", name=re.compile(r"页面填写")).click()
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "tab_switch_failed")
raise RpaError(f"切换到页面填写失败:{exc}", screenshot_path=sp) from exc
try:
await _select_target_account(page, target)
except RpaError:
raise
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "select_account_failed")
raise RpaError(f"选择来源账户失败:{exc}", screenshot_path=sp) from exc
try:
existing = await page.locator("tr[data-row-index]").count()
need_more = max(0, len(items) - existing)
add_btn = page.get_by_role("button", name=re.compile(r"新增一行"))
for _ in range(need_more):
await _pause(400, 900)
await add_btn.click()
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "add_rows_failed")
raise RpaError(f"新增明细行失败:{exc}", screenshot_path=sp) from exc
for it in items:
try:
row = page.locator(f'tr[data-row-index="{it.row_index}"]')
await row.wait_for(timeout=DEFAULT_TIMEOUT_MS)
await _pause(300, 800)
await row.locator('input[name="name"]').fill(it.name)
await row.locator('input[name="account"]').fill(it.account)
await row.locator('input[name="note"]').fill(it.note or "")
await row.locator('input[name="amount"]').fill(f"{it.amount}")
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, f"fill_row_{it.row_index}_failed")
raise RpaError(f"填第 {it.row_index} 行失败:{exc}", screenshot_path=sp) from exc
try:
await page.fill('input[name="purpose"]', "仿真批量提交示例")
except Exception as exc:
logger.warning("填写用途失败(非致命):%s", exc)
async def _submit_and_get_batch_id(page, token_pin: str, artifacts_dir: str | None) -> str:
try:
await _pause(800, 1500)
await page.locator('form[name="simBatchForm"] button[type="submit"]').click()
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "click_submit_failed")
raise RpaError(f"点击提交失败:{exc}", screenshot_path=sp) from exc
dlg_sel = 'div[role="dialog"][aria-modal="true"]'
try:
await page.wait_for_selector(dlg_sel, timeout=DEFAULT_TIMEOUT_MS)
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "pin_dialog_not_open")
raise RpaError(f"PIN 弹窗未出现:{exc}", screenshot_path=sp) from exc
try:
pin_input = page.locator(f'{dlg_sel} input[type="password"][inputmode="numeric"]')
await _pause(500, 1200)
await pin_input.fill(token_pin)
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "fill_pin_failed")
raise RpaError(f"填写 PIN 失败:{exc}", screenshot_path=sp) from exc
try:
await page.locator(dlg_sel).get_by_role("button", name=re.compile(r"确认提交")).click()
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "click_confirm_failed")
raise RpaError(f"确认提交失败:{exc}", screenshot_path=sp) from exc
try:
await page.wait_for_selector('[data-testid="batch-id"]', timeout=SUBMIT_NAVIGATE_TIMEOUT_MS)
except Exception as exc:
sp = await _safe_screenshot(page, artifacts_dir, "success_area_not_visible")
raise RpaError(f"提交后未出现成功区域:{exc}", screenshot_path=sp) from exc
batch_id = await page.locator('[data-testid="batch-id"]').get_attribute("data-batch-id") or ""
if not batch_id:
batch_id = (await page.locator('[data-testid="batch-id"]').inner_text()).strip()
url = page.url
if not batch_id and "#/batch/" in url:
batch_id = url.split("#/batch/", 1)[1].split("?")[0].split("#")[0].strip("/")
if not batch_id:
sp = await _safe_screenshot(page, artifacts_dir, "batch_id_empty")
raise RpaError(f"无法解析 batch_idurl={url}", screenshot_path=sp)
return batch_id
async def _safe_screenshot(page, artifacts_dir: str | None, tag: str) -> Optional[str]:
if not artifacts_dir:
return None
try:
os.makedirs(artifacts_dir, exist_ok=True)
path = os.path.join(artifacts_dir, f"{tag}_{int(time.time())}.png")
await page.screenshot(path=path, full_page=True)
logger.info("screenshot_saved path=%s", path)
return path
except Exception as exc:
logger.warning("screenshot_failed tag=%s err=%s", tag, exc)
return None
async def submit_batch_rpa(
account: dict[str, Any],
*,
target: str,
items: list[BatchItem],
base_url: str,
artifacts_dir: str | None,
video_session: Any = None,
headless: bool | None = None,
) -> BatchSubmitResult:
del video_session # 示例占位;真实 skill 可接入 RpaVideoSession
profile_dir = str(account.get("profile_dir") or "").strip()
if not profile_dir:
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg="账号缺少 profile_dir",
artifacts={"adapter": "simulator_rpa"},
)
if not items:
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg="items 为空",
artifacts={"adapter": "simulator_rpa"},
)
if not find_chrome_executable():
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg="未检测到 Chrome 或 Edge 浏览器,请安装系统浏览器后重试。",
artifacts={"adapter": "simulator_rpa"},
)
creds = _resolve_credentials(account)
portal_wait = portal_login_wait_sec()
hl = headless if headless is not None else _headless()
pw = None
context = None
page = None
try:
pw, context, page = await start_browser_session(profile_dir, headless=hl)
logger.info("rpa_goto url=%s", base_url)
await page.goto(base_url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT_MS)
await _wait_portal_gate(page, portal_wait)
logger.info("rpa_login_start")
await _login(page, creds, artifacts_dir)
logger.info("rpa_login_done")
await _goto_batch_page(page, artifacts_dir)
logger.info("rpa_batch_page_ready")
await _fill_batch_form(page, target, items, artifacts_dir)
logger.info("rpa_batch_form_filled rows=%s", len(items))
batch_id = await _submit_and_get_batch_id(page, creds.token_pin, artifacts_dir)
logger.info("rpa_submit_success batch_id=%s", batch_id)
total = sum(it.amount for it in items)
return BatchSubmitResult(
ok=True,
batch_id=batch_id,
submitted_count=len(items),
submitted_amount=total,
error_msg=None,
artifacts={"adapter": "simulator_rpa"},
)
except RpaError as exc:
logger.error("rpa_failed: %s", exc)
arts = {"adapter": "simulator_rpa"}
if exc.screenshot_path:
arts["screenshot"] = exc.screenshot_path
elif page is not None:
sp = await _safe_screenshot(page, artifacts_dir, "rpa_error")
if sp:
arts["screenshot"] = sp
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg=str(exc),
artifacts=arts,
)
except Exception as exc:
logger.exception("rpa_unexpected: %s", exc)
arts = {"adapter": "simulator_rpa"}
if page is not None:
sp = await _safe_screenshot(page, artifacts_dir, "unexpected_error")
if sp:
arts["screenshot"] = sp
return BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg=f"未预期异常:{type(exc).__name__}: {exc}",
artifacts=arts,
)
finally:
await close_browser_session(pw, context)

View File

@@ -7,8 +7,7 @@ import tempfile
import uuid import uuid
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from service.account_client import pick_simulator_account from service.adapter import BatchItem, select_adapter
from service.adapter import BatchItem, SimulatorBrowserRpaAdapter, select_adapter
from util.logging import get_logger from util.logging import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -42,7 +41,7 @@ def _validate_items(items: List[Dict[str, Any]]) -> tuple[List[BatchItem], Optio
return parsed, None return parsed, None
def run_batch_submit( async def run_batch_submit(
target: str, target: str,
items: List[Dict[str, Any]], items: List[Dict[str, Any]],
*, *,
@@ -53,6 +52,7 @@ def run_batch_submit(
最小编排示例。 最小编排示例。
真实 skill 中还应:写 task_logs、entitlement、RpaVideoSession 等。 真实 skill 中还应:写 task_logs、entitlement、RpaVideoSession 等。
账号 pick / lease 在 adapter 内完成,编排层不直接碰 account-manager。
""" """
del force # 示例占位;真实 skill 可用于跳过确认 del force # 示例占位;真实 skill 可用于跳过确认
@@ -70,19 +70,15 @@ def run_batch_submit(
uuid.uuid4().hex[:8], uuid.uuid4().hex[:8],
) )
os.makedirs(art_dir, exist_ok=True) os.makedirs(art_dir, exist_ok=True)
logger.info("batch_submit_start target=%s item_count=%s artifacts_dir=%s", target_key, len(batch_items), art_dir) logger.info(
"batch_submit_start target=%s item_count=%s artifacts_dir=%s",
target_key,
len(batch_items),
art_dir,
)
adapter = select_adapter(artifacts_dir=art_dir) adapter = select_adapter(artifacts_dir=art_dir)
if isinstance(adapter, SimulatorBrowserRpaAdapter): result = await adapter.submit_batch(target_key, batch_items)
account = pick_simulator_account()
adapter.set_credentials(
account.login_id,
account.password,
account.token_pin,
account.profile_dir,
)
result = adapter.submit_batch(target_key, batch_items)
summary: Dict[str, Any] = { summary: Dict[str, Any] = {
"success": result.ok, "success": result.ok,
"target": target_key, "target": target_key,

View File

@@ -5,12 +5,19 @@ from __future__ import annotations
import os import os
from pathlib import Path from pathlib import Path
from jiangchang_skill_core import config
SKILL_SLUG = "example-simulator-browser-rpa" SKILL_SLUG = "example-simulator-browser-rpa"
LOG_LOGGER_NAME = "openclaw.skill.example_simulator_browser_rpa" LOG_LOGGER_NAME = "openclaw.skill.example_simulator_browser_rpa"
TARGET_PLATFORM = "sim_batch"
LEASE_HOLDER = "example-simulator-browser-rpa"
LEASE_TTL_SEC = "1800"
DEFAULT_TIMEOUT_MS = 15_000 DEFAULT_TIMEOUT_MS = 15_000
LOGIN_NAVIGATE_TIMEOUT_MS = 30_000 LOGIN_NAVIGATE_TIMEOUT_MS = 30_000
SUBMIT_NAVIGATE_TIMEOUT_MS = 60_000 SUBMIT_NAVIGATE_TIMEOUT_MS = 60_000
DEFAULT_PORTAL_LOGIN_WAIT_SEC = 120
DEFAULT_DEMO_LOGIN_ID = "demo001" DEFAULT_DEMO_LOGIN_ID = "demo001"
DEFAULT_DEMO_PASSWORD = "demo-password" DEFAULT_DEMO_PASSWORD = "demo-password"
@@ -18,15 +25,26 @@ DEFAULT_DEMO_CAPTCHA = "0000"
DEFAULT_DEMO_TOKEN_PIN = "123456" DEFAULT_DEMO_TOKEN_PIN = "123456"
ENV_SIMULATOR_BASE_URL = "SIMULATOR_BASE_URL" ENV_SIMULATOR_BASE_URL = "SIMULATOR_BASE_URL"
ENV_SIMULATOR_LOGIN_ID = "SIMULATOR_LOGIN_ID"
ENV_SIMULATOR_PASSWORD = "SIMULATOR_PASSWORD" ENV_SIMULATOR_PASSWORD = "SIMULATOR_PASSWORD"
ENV_SIMULATOR_TOKEN_PIN = "SIMULATOR_TOKEN_PIN" ENV_SIMULATOR_TOKEN_PIN = "SIMULATOR_TOKEN_PIN"
ENV_SIMULATOR_PROFILE_DIR = "SIMULATOR_PROFILE_DIR" ENV_PORTAL_LOGIN_WAIT_SEC = "PORTAL_LOGIN_WAIT_SEC"
_EXAMPLE_ROOT = Path(__file__).resolve().parents[2] _EXAMPLE_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_DEMO_APP_PATH = _EXAMPLE_ROOT / "sandbox" / "demo_app.html" DEFAULT_DEMO_APP_PATH = _EXAMPLE_ROOT / "sandbox" / "demo_app.html"
def portal_login_wait_sec() -> int:
raw = (
os.getenv(ENV_PORTAL_LOGIN_WAIT_SEC)
or config.get(ENV_PORTAL_LOGIN_WAIT_SEC)
or str(DEFAULT_PORTAL_LOGIN_WAIT_SEC)
)
try:
return max(1, int(str(raw).strip()))
except (TypeError, ValueError):
return DEFAULT_PORTAL_LOGIN_WAIT_SEC
def resolve_simulator_base_url() -> str: def resolve_simulator_base_url() -> str:
env = (os.getenv(ENV_SIMULATOR_BASE_URL) or "").strip() env = (os.getenv(ENV_SIMULATOR_BASE_URL) or "").strip()
if env: if env:

View File

@@ -3,6 +3,8 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
import re
from typing import Any
from util.constants import LOG_LOGGER_NAME from util.constants import LOG_LOGGER_NAME
@@ -13,3 +15,24 @@ def get_logger(name: str | None = None) -> logging.Logger:
def configure_logging(level: int = logging.INFO) -> None: def configure_logging(level: int = logging.INFO) -> None:
logging.basicConfig(level=level, format="%(levelname)s %(name)s %(message)s") logging.basicConfig(level=level, format="%(levelname)s %(name)s %(message)s")
def mask_text(value: str, *, keep_start: int = 2, keep_end: int = 2) -> str:
s = (value or "").strip()
if len(s) >= 11 and s.isdigit():
return s[:3] + "****" + s[-4:]
if len(s) > keep_start + keep_end:
return s[:keep_start] + "****" + s[-keep_end:]
return "****"
def safe_log_value(value: Any) -> str:
if value is None:
return ""
text = str(value)
if re.fullmatch(r"\d{7,}", text):
return mask_text(text)
if "@" in text and len(text) > 6:
local, _, domain = text.partition("@")
return f"{mask_text(local, keep_start=1, keep_end=1)}@{domain}"
return text

View File

@@ -42,20 +42,20 @@ class TestAdapterDispatch(unittest.TestCase):
self.assertIsInstance(select_adapter(), SimulatorBrowserRpaAdapter) self.assertIsInstance(select_adapter(), SimulatorBrowserRpaAdapter)
class TestMockAdapter(unittest.TestCase): class TestMockAdapter(unittest.IsolatedAsyncioTestCase):
def test_mock_success(self) -> None: async def test_mock_success(self) -> None:
items = [ items = [
BatchItem(row_index=1, name="Alice", account="1001", amount=10.5), BatchItem(row_index=1, name="Alice", account="1001", amount=10.5),
BatchItem(row_index=2, name="Bob", account="1002", amount=20.0), BatchItem(row_index=2, name="Bob", account="1002", amount=20.0),
] ]
result = MockBatchAdapter().submit_batch("acct-demo-001", items) result = await MockBatchAdapter().submit_batch("acct-demo-001", items)
self.assertTrue(result.ok) self.assertTrue(result.ok)
self.assertEqual(result.submitted_count, 2) self.assertEqual(result.submitted_count, 2)
self.assertEqual(result.submitted_amount, 30.5) self.assertEqual(result.submitted_amount, 30.5)
self.assertTrue((result.batch_id or "").startswith("MOCK-")) self.assertTrue((result.batch_id or "").startswith("MOCK-"))
def test_mock_empty_items(self) -> None: async def test_mock_empty_items(self) -> None:
result = MockBatchAdapter().submit_batch("acct-demo-001", []) result = await MockBatchAdapter().submit_batch("acct-demo-001", [])
self.assertFalse(result.ok) self.assertFalse(result.ok)
self.assertEqual(result.error_msg, "items 为空") self.assertEqual(result.error_msg, "items 为空")

View File

@@ -0,0 +1,39 @@
# -*- coding: utf-8 -*-
"""禁止 import account-manager 内部 RPA 辅助模块。"""
from __future__ import annotations
import os
import re
import unittest
from pathlib import Path
FORBIDDEN_PATTERNS = (
re.compile(r"\brpa_helpers\b"),
re.compile(r"\binject_account_manager_scripts_path\b"),
re.compile(r"\bget_account_credential\b"),
)
SCRIPTS_ROOT = Path(__file__).resolve().parents[1] / "scripts"
class TestNoRpaHelpersImport(unittest.TestCase):
def test_scripts_tree_has_no_forbidden_account_manager_imports(self) -> None:
violations: list[str] = []
for root, _dirs, files in os.walk(SCRIPTS_ROOT):
for name in files:
if not name.endswith(".py"):
continue
path = Path(root) / name
text = path.read_text(encoding="utf-8")
for pattern in FORBIDDEN_PATTERNS:
if pattern.search(text):
violations.append(f"{path.relative_to(SCRIPTS_ROOT)}: {pattern.pattern}")
self.assertEqual(
violations,
[],
msg="Forbidden account-manager internal imports found:\n" + "\n".join(violations),
)
if __name__ == "__main__":
unittest.main()

View File

@@ -2,31 +2,27 @@
"""SimulatorBrowserRpaAdapter 单元测试(不启动真实浏览器)。""" """SimulatorBrowserRpaAdapter 单元测试(不启动真实浏览器)。"""
from __future__ import annotations from __future__ import annotations
import asyncio
import unittest import unittest
from unittest.mock import MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from service.adapter import BatchItem, SimulatorBrowserRpaAdapter from service.adapter import BatchItem, SimulatorBrowserRpaAdapter
from service.browser_session import CHROME_LAUNCH_ARGS from service.browser_session import CHROME_LAUNCH_ARGS
from service.simulator_playwright import RpaError
class TestSimulatorRpaUnit(unittest.TestCase): class TestSimulatorRpaUnit(unittest.IsolatedAsyncioTestCase):
def test_set_credentials_and_empty_profile_dir(self) -> None: async def test_submit_batch_empty_items(self) -> None:
adapter = SimulatorBrowserRpaAdapter() adapter = SimulatorBrowserRpaAdapter()
adapter.set_credentials("demo001", "pwd", "pin", " ") result = await adapter.submit_batch("acct-demo-001", [])
self.assertEqual(adapter._login_id, "demo001")
self.assertIsNone(adapter._profile_dir)
def test_submit_batch_empty_items(self) -> None:
adapter = SimulatorBrowserRpaAdapter()
result = adapter.submit_batch("acct-demo-001", [])
self.assertFalse(result.ok) self.assertFalse(result.ok)
self.assertIn("items", result.error_msg or "") self.assertIn("items", result.error_msg or "")
@patch("service.adapter.simulator_rpa.find_chrome_executable", return_value=None) @patch("service.adapter.simulator_rpa.find_chrome_executable", return_value=None)
def test_no_chrome_returns_error_without_launch(self, _chrome: MagicMock) -> None: async def test_no_chrome_returns_error_without_launch(self, _chrome: MagicMock) -> None:
adapter = SimulatorBrowserRpaAdapter() adapter = SimulatorBrowserRpaAdapter()
items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)] items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)]
result = adapter.submit_batch("acct-demo-001", items) result = await adapter.submit_batch("acct-demo-001", items)
self.assertFalse(result.ok) self.assertFalse(result.ok)
msg = result.error_msg or "" msg = result.error_msg or ""
self.assertTrue("Chrome" in msg or "Edge" in msg or "浏览器" in msg) self.assertTrue("Chrome" in msg or "Edge" in msg or "浏览器" in msg)
@@ -38,38 +34,100 @@ class TestSimulatorRpaUnit(unittest.TestCase):
self.assertFalse(lowered.startswith("https://")) self.assertFalse(lowered.startswith("https://"))
self.assertFalse(lowered.startswith("file://")) self.assertFalse(lowered.startswith("file://"))
@patch("service.adapter.simulator_rpa.browser_session") @patch("service.adapter.simulator_rpa.submit_batch_rpa", new_callable=AsyncMock)
@patch("service.adapter.simulator_rpa.pick_web_account")
@patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe") @patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe")
def test_start_url_opened_via_goto_not_launch_args( async def test_adapter_delegates_to_playwright_with_account_url(
self, self,
_chrome: MagicMock, _chrome: MagicMock,
mock_session: MagicMock, mock_pick: MagicMock,
mock_rpa: AsyncMock,
) -> None: ) -> None:
from service.adapter.simulator_rpa import RpaError from service.adapter.base import BatchSubmitResult
adapter = SimulatorBrowserRpaAdapter(base_url="file:///demo/demo_app.html") mock_pick.return_value = {
"login_id": "demo001",
"profile_dir": "/tmp/profile",
"url": "file:///demo/demo_app.html",
"lease_token": "tok-1",
}
mock_rpa.return_value = BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg="login fail",
artifacts={"adapter": "simulator_rpa"},
)
adapter = SimulatorBrowserRpaAdapter(artifacts_dir="/tmp/art")
items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)] items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)]
result = await adapter.submit_batch("acct-demo-001", items)
page = MagicMock()
mock_cm = MagicMock()
mock_cm.__enter__.return_value = page
mock_cm.__exit__.return_value = False
mock_session.return_value = mock_cm
with patch.object(adapter, "_login", side_effect=RpaError("login fail")):
result = adapter.submit_batch("acct-demo-001", items)
self.assertFalse(result.ok) self.assertFalse(result.ok)
mock_session.assert_called_once() mock_rpa.assert_awaited_once()
called_start_url = mock_session.call_args[0][0] call_kwargs = mock_rpa.call_args.kwargs
self.assertEqual(called_start_url, "file:///demo/demo_app.html") self.assertEqual(call_kwargs["base_url"], "file:///demo/demo_app.html")
for arg in CHROME_LAUNCH_ARGS: for arg in CHROME_LAUNCH_ARGS:
self.assertNotIn(arg, called_start_url) self.assertNotIn(arg, call_kwargs["base_url"])
def test_safe_screenshot_without_artifacts_dir(self) -> None: @patch("service.simulator_playwright.start_browser_session", new_callable=AsyncMock)
adapter = SimulatorBrowserRpaAdapter(artifacts_dir=None) async def test_playwright_opens_url_via_goto_not_launch_args(self, mock_start: AsyncMock) -> None:
page = MagicMock() from service.simulator_playwright import submit_batch_rpa
self.assertIsNone(adapter._safe_screenshot(page, "tag"))
page = AsyncMock()
page.goto = AsyncMock()
mock_start.return_value = (MagicMock(), MagicMock(), page)
account = {"profile_dir": "/tmp/p", "login_id": "demo001"}
items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)]
with patch(
"service.simulator_playwright._wait_portal_gate",
new_callable=AsyncMock,
return_value=None,
):
with patch(
"service.simulator_playwright._login",
new_callable=AsyncMock,
side_effect=RpaError("login fail"),
):
with patch(
"service.simulator_playwright.find_chrome_executable",
return_value="C:/Chrome/chrome.exe",
):
with patch(
"service.simulator_playwright.close_browser_session",
new_callable=AsyncMock,
):
result = await submit_batch_rpa(
account,
target="acct-demo-001",
items=items,
base_url="file:///demo/demo_app.html",
artifacts_dir=None,
)
self.assertFalse(result.ok)
page.goto.assert_awaited_once()
goto_url = page.goto.call_args[0][0]
self.assertEqual(goto_url, "file:///demo/demo_app.html")
for arg in CHROME_LAUNCH_ARGS:
self.assertNotIn(arg, goto_url)
async def test_safe_screenshot_without_artifacts_dir(self) -> None:
from service.simulator_playwright import _safe_screenshot
page = AsyncMock()
self.assertIsNone(await _safe_screenshot(page, None, "tag"))
class TestSyncHelpers(unittest.TestCase):
def test_asyncio_run_smoke(self) -> None:
async def _noop() -> bool:
return True
self.assertTrue(asyncio.run(_noop()))
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -3,51 +3,70 @@
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from unittest.mock import MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from service.adapter import BatchItem, SimulatorBrowserRpaAdapter from service.adapter import BatchItem, SimulatorBrowserRpaAdapter
from service.adapter.simulator_rpa import RpaError from service.adapter.base import BatchSubmitResult
class TestBatchSubmitFailureResult(unittest.TestCase): class TestBatchSubmitFailureResult(unittest.IsolatedAsyncioTestCase):
@patch("service.adapter.simulator_rpa.release_lease")
@patch("service.adapter.simulator_rpa.submit_batch_rpa", new_callable=AsyncMock)
@patch("service.adapter.simulator_rpa.pick_web_account")
@patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe") @patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe")
@patch("service.adapter.simulator_rpa.browser_session") async def test_rpa_error_converts_to_failed_result(
def test_rpa_error_converts_to_failed_result(self, mock_session: MagicMock, _chrome: MagicMock) -> None: self,
page = MagicMock() _chrome: MagicMock,
mock_cm = MagicMock() mock_pick: MagicMock,
mock_cm.__enter__.return_value = page mock_rpa: AsyncMock,
mock_cm.__exit__.return_value = False _release: MagicMock,
mock_session.return_value = mock_cm ) -> None:
mock_pick.return_value = {
"profile_dir": "/tmp/profile",
"lease_token": "tok",
}
mock_rpa.return_value = BatchSubmitResult(
ok=False,
batch_id=None,
submitted_count=0,
submitted_amount=0.0,
error_msg="登录第一步失败timeout",
artifacts={"adapter": "simulator_rpa", "screenshot": "/tmp/art/x.png"},
)
adapter = SimulatorBrowserRpaAdapter(artifacts_dir="/tmp/art") adapter = SimulatorBrowserRpaAdapter(artifacts_dir="/tmp/art")
items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)] items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)]
result = await adapter.submit_batch("acct-demo-001", items)
with patch.object(adapter, "_login", side_effect=RpaError("登录第一步失败timeout", screenshot_path="/tmp/art/x.png")):
result = adapter.submit_batch("acct-demo-001", items)
self.assertFalse(result.ok) self.assertFalse(result.ok)
self.assertIn("登录第一步失败", result.error_msg or "") self.assertIn("登录第一步失败", result.error_msg or "")
self.assertEqual(result.artifacts.get("adapter"), "simulator_browser_rpa") self.assertEqual(result.artifacts.get("adapter"), "simulator_rpa")
self.assertEqual(result.artifacts.get("screenshot"), "/tmp/art/x.png") self.assertEqual(result.artifacts.get("screenshot"), "/tmp/art/x.png")
@patch("service.adapter.simulator_rpa.release_lease")
@patch("service.adapter.simulator_rpa.submit_batch_rpa", new_callable=AsyncMock)
@patch("service.adapter.simulator_rpa.pick_web_account")
@patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe") @patch("service.adapter.simulator_rpa.find_chrome_executable", return_value="C:/Chrome/chrome.exe")
@patch("service.adapter.simulator_rpa.browser_session") async def test_unexpected_error_includes_adapter_name(
def test_unexpected_error_includes_adapter_name(self, mock_session: MagicMock, _chrome: MagicMock) -> None: self,
page = MagicMock() _chrome: MagicMock,
mock_cm = MagicMock() mock_pick: MagicMock,
mock_cm.__enter__.return_value = page mock_rpa: AsyncMock,
mock_cm.__exit__.return_value = False _release: MagicMock,
mock_session.return_value = mock_cm ) -> None:
mock_pick.return_value = {
"profile_dir": "/tmp/profile",
"lease_token": "tok",
}
mock_rpa.side_effect = RuntimeError("boom")
adapter = SimulatorBrowserRpaAdapter(artifacts_dir=None) adapter = SimulatorBrowserRpaAdapter(artifacts_dir=None)
items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)] items = [BatchItem(row_index=1, name="A", account="1", amount=1.0)]
result = await adapter.submit_batch("acct-demo-001", items)
with patch.object(adapter, "_login", side_effect=RuntimeError("boom")):
result = adapter.submit_batch("acct-demo-001", items)
self.assertFalse(result.ok) self.assertFalse(result.ok)
self.assertIn("未预期异常", result.error_msg or "") self.assertIn("未预期异常", result.error_msg or "")
self.assertEqual(result.artifacts.get("adapter"), "simulator_browser_rpa") self.assertEqual(result.artifacts.get("adapter"), "simulator_rpa")
self.assertNotIn("screenshot", result.artifacts) self.assertNotIn("screenshot", result.artifacts)

View File

@@ -29,7 +29,7 @@ metadata:
- `name`:用户可见的中文业务名称(可含品牌,如 `GEO 文章生成``1688 联系人采集`)。 - `name`:用户可见的中文业务名称(可含品牌,如 `GEO 文章生成``1688 联系人采集`)。
- `description`:能力说明,不是技能名称。 - `description`:能力说明,不是技能名称。
- `metadata.openclaw.slug`机器标识kebab-case 英文;目录名与默认数据库文件名使用 slug。 - `metadata.openclaw.slug`机器标识kebab-case 英文;目录名与默认数据库文件名使用 slug。slug 语义规范见 [`../development/NAMING.md`](../development/NAMING.md);本文件仅描述数据库与 kebab-case 格式。
- 复制模板后不得将 `your-skill-slug``My Skill` 等占位内容作为正式技能发布名称。 - 复制模板后不得将 `your-skill-slug``My Skill` 等占位内容作为正式技能发布名称。
## 数据库路径与命名 ## 数据库路径与命名

View File

@@ -18,6 +18,7 @@ from db.display_metadata import (
) )
from db.timestamp_columns import has_updated_at_trigger from db.timestamp_columns import has_updated_at_trigger
from util.constants import SKILL_SLUG from util.constants import SKILL_SLUG
from util.slug_naming import LEGACY_EXEMPT_SLUGS, validate_slug_semantics
_TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"}) _TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"})
_ENGLISH_PLACEHOLDER_NAME_PATTERNS = ( _ENGLISH_PLACEHOLDER_NAME_PATTERNS = (
@@ -127,6 +128,14 @@ def validate_skill_frontmatter(skill_root: str, *, allow_template_placeholders:
f"SKILL.md slug ({slug!r}) 与 util.constants.SKILL_SLUG ({constants_slug!r}) 不一致" f"SKILL.md slug ({slug!r}) 与 util.constants.SKILL_SLUG ({constants_slug!r}) 不一致"
) )
if slug and slug not in LEGACY_EXEMPT_SLUGS and _KEBAB_SLUG.fullmatch(slug):
naming_errors, naming_warnings = validate_slug_semantics(
slug,
strict=not allow_template_placeholders,
)
report.errors.extend(naming_errors)
report.warnings.extend(naming_warnings)
return report return report

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.14" SKILL_VERSION = "1.0.32"
LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug" LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug"
PLATFORM_KIT_MIN_VERSION = "1.0.17" PLATFORM_KIT_MIN_VERSION = "1.0.17"

227
scripts/util/slug_naming.py Normal file
View File

@@ -0,0 +1,227 @@
"""技能 slug 语义校验verb-noun-platform 规范,见 development/NAMING.md"""
from __future__ import annotations
import os
import re
_KEBAB_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
_MAX_SLUG_LENGTH = 48
_MIN_SEGMENTS = 3
_MAX_SEGMENTS = 5
_MAX_NOUN_SEGMENTS = 3
_SKILL_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
_NAMING_DIR = os.path.join(_SKILL_ROOT, "assets", "naming")
_BUILTIN_VERBS = frozenset(
{
"scrape",
"receive",
"download",
"export",
"add",
"fill",
"enter",
"bind",
"withdraw",
"reconcile",
"verify",
"submit",
"query",
"process",
"reply",
"generate",
"draft",
"review",
"confirm",
"track",
"upload",
"match",
"triage",
"ship",
"reprice",
}
)
_BUILTIN_PLATFORMS = frozenset(
{
"amazon",
"shopee",
"xinghang",
"alibaba",
"kingdee",
"icbc",
"pingpong",
"worldfirst",
"paypal",
"lianlian",
"single-window",
"export-rebate",
"etax",
}
)
_BUILTIN_SCOPES = frozenset({"all", "batch", "multi", "default"})
LEGACY_EXEMPT_SLUGS: frozenset[str] = frozenset(
{
"account-manager",
"skill-template",
"your-skill-slug",
"your_skill_slug",
}
)
def load_wordlist(filename: str) -> frozenset[str]:
"""从 assets/naming/{filename} 加载词表;文件不存在时 fallback 内置集合。"""
fallbacks = {
"verbs.txt": _BUILTIN_VERBS,
"platforms.txt": _BUILTIN_PLATFORMS,
"scopes.txt": _BUILTIN_SCOPES,
}
fallback = fallbacks.get(filename, frozenset())
path = os.path.join(_NAMING_DIR, filename)
if not os.path.isfile(path):
return fallback
words: set[str] = set()
with open(path, encoding="utf-8") as f:
for line in f:
word = line.strip()
if word and not word.startswith("#"):
words.add(word)
return frozenset(words) if words else fallback
def parse_slug_segments(slug: str) -> list[str]:
"""'-' 分割;非法字符由上层 kebab 校验处理。"""
slug = slug.strip()
if not slug:
return []
return slug.split("-")
def _match_platform_suffix(
segments: list[str],
platforms: frozenset[str],
) -> tuple[str, int] | None:
"""返回 (platform_name, platform_part_count) 或 None。"""
for platform in sorted(platforms, key=lambda item: (-item.count("-"), item)):
parts = platform.split("-")
part_count = len(parts)
if len(segments) < part_count + 2:
continue
if segments[-part_count:] == parts:
return platform, part_count
return None
def classify_slug_form(segments: list[str], *, slug: str = "") -> str:
"""
返回: 'legacy_exempt' | 'standard' | 'scope' | 'invalid'
- legacy_exempt: slug in LEGACY_EXEMPT_SLUGS
- standard: 末段匹配 platforms, 首段 in verbs, 3<=len<=5
- scope: 末段 in scopes, 首段 in verbs, 3<=len<=5
- invalid: 其他
"""
if slug in LEGACY_EXEMPT_SLUGS:
return "legacy_exempt"
if not segments or not (_MIN_SEGMENTS <= len(segments) <= _MAX_SEGMENTS):
return "invalid"
verbs = load_wordlist("verbs.txt")
platforms = load_wordlist("platforms.txt")
scopes = load_wordlist("scopes.txt")
if segments[0] not in verbs:
return "invalid"
if segments[-1] in scopes:
return "scope"
if _match_platform_suffix(segments, platforms) is not None:
return "standard"
return "invalid"
def validate_slug_semantics(
slug: str,
*,
strict: bool = True,
) -> tuple[list[str], list[str]]:
"""返回 (errors, warnings)。"""
slug = slug.strip()
if slug in LEGACY_EXEMPT_SLUGS:
return [], []
errors: list[str] = []
warnings: list[str] = []
if not _KEBAB_SLUG.fullmatch(slug):
msg = f"slug 必须为 kebab-case小写字母、数字、连字符{slug!r}"
(errors if strict else warnings).append(msg)
return errors, warnings
if len(slug) > _MAX_SLUG_LENGTH:
msg = f"slug 长度不得超过 {_MAX_SLUG_LENGTH} 字符(当前 {len(slug)}{slug!r}"
(errors if strict else warnings).append(msg)
segments = parse_slug_segments(slug)
segment_count = len(segments)
if segment_count < _MIN_SEGMENTS:
msg = (
f"slug 段数应为 {_MIN_SEGMENTS}{_MAX_SEGMENTS}verb-noun-platform"
f"当前为 {segment_count} 段:{slug!r};新技能禁止 2 段无平台形态(如 reconcile-finance"
)
(errors if strict else warnings).append(msg)
return errors, warnings
if segment_count > _MAX_SEGMENTS:
msg = f"slug 段数不得超过 {_MAX_SEGMENTS}(当前 {segment_count}{slug!r}"
(errors if strict else warnings).append(msg)
verbs = load_wordlist("verbs.txt")
platforms = load_wordlist("platforms.txt")
scopes = load_wordlist("scopes.txt")
if segments[0] in platforms and segments[0] not in verbs:
msg = f"平台名不应放在 slug 最前:{slug!r}(应为 verb-noun-platform"
(errors if strict else warnings).append(msg)
if segments[0] not in verbs:
msg = f"slug 首段应为动词白名单中的词:{segments[0]!r}slug={slug!r}"
(errors if strict else warnings).append(msg)
form = classify_slug_form(segments, slug=slug)
if form == "scope":
if segments[-1] not in scopes:
msg = f"slug 末段应为 scope 白名单中的词:{segments[-1]!r}slug={slug!r}"
(errors if strict else warnings).append(msg)
noun_count = segment_count - 2
elif form == "standard":
platform_match = _match_platform_suffix(segments, platforms)
if platform_match is None:
msg = f"slug 末段应匹配平台白名单:{slug!r}"
(errors if strict else warnings).append(msg)
noun_count = 0
else:
_platform_name, platform_part_count = platform_match
noun_count = segment_count - 1 - platform_part_count
else:
if segment_count >= _MIN_SEGMENTS:
if segments[-1] not in scopes:
platform_match = _match_platform_suffix(segments, platforms)
if platform_match is None:
msg = f"slug 末段应匹配平台或 scope 白名单:{slug!r}"
(errors if strict else warnings).append(msg)
noun_count = 0
if noun_count > _MAX_NOUN_SEGMENTS:
warnings.append(
f"slug 中间名词段建议不超过 {_MAX_NOUN_SEGMENTS} 个词(当前 {noun_count}{slug!r}"
)
return errors, warnings

View File

@@ -324,7 +324,7 @@ name: My Skill
description: demo description: demo
metadata: metadata:
openclaw: openclaw:
slug: demo-skill slug: query-demo-info
--- ---
""" """
with open(os.path.join(tmp, "SKILL.md"), "w", encoding="utf-8") as f: with open(os.path.join(tmp, "SKILL.md"), "w", encoding="utf-8") as f:

View File

@@ -141,6 +141,15 @@ class TestDocsStandards(unittest.TestCase):
msg="SKILL.md must state README.md is user-facing documentation", msg="SKILL.md must state README.md is user-facing documentation",
) )
def test_naming_md_exists_and_covers_standards(self) -> None:
text = self._read("development/NAMING.md")
self.assertIn("verb-noun-platform", text)
self.assertIn("48", text)
self.assertTrue("LEGACY" in text or "历史" in text)
self.assertTrue("verbs.txt" in text or "动词白名单" in text)
self.assertTrue("platforms.txt" in text or "平台白名单" in text)
self.assertTrue("仅适用于新创建" in text or "新技能" in text)
def test_release_workflow_uses_main_channel(self) -> None: def test_release_workflow_uses_main_channel(self) -> None:
text = self._read(".github/workflows/release_skill.yaml") text = self._read(".github/workflows/release_skill.yaml")
uses_line = _extract_workflow_uses_line(text) uses_line = _extract_workflow_uses_line(text)
@@ -165,6 +174,84 @@ class TestDocsStandards(unittest.TestCase):
self.assertIn("RpaVideoSession", text) self.assertIn("RpaVideoSession", text)
self.assertIn("1.0.17", text) self.assertIn("1.0.17", text)
def test_rpa_md_forbids_rpa_helpers_import(self) -> None:
text = self._read("development/RPA.md")
self.assertIn("rpa_helpers", text)
self.assertTrue(
"不要 import" in text or "不要" in text and "import" in text,
msg="RPA.md must document forbidding rpa_helpers import",
)
def test_rpa_md_covers_simulator_playwright_layering(self) -> None:
text = self._read("development/RPA.md")
self.assertTrue(
"simulator_playwright" in text
or ("薄 adapter" in text and "playwright.py" in text),
msg="RPA.md must describe thin adapter + *_playwright.py layering",
)
def test_rpa_md_covers_industry_simulator_portal(self) -> None:
text = self._read("development/RPA.md")
self.assertTrue(
any(marker in text for marker in ("jc2009", "行业仿真", "门户")),
msg="RPA.md must mention jc2009 / industry simulator / portal gate",
)
def test_simulator_example_readme_covers_async_and_account_manager(self) -> None:
text = self._read("examples/simulator_browser_rpa/README.md")
self.assertIn("async", text.lower())
self.assertTrue(
"account-manager" in text.lower() or "pick-web" in text or "pick_web_account" in text,
msg="simulator_browser_rpa README must mention async and account-manager",
)
def test_development_md_documents_scaffold_and_git_cleanup(self) -> None:
text = self._read("development/DEVELOPMENT.md")
self.assertTrue(
"scaffold_skill" in text or "scaffold_skill.ps1" in text,
msg="DEVELOPMENT.md must document scaffold_skill",
)
self.assertTrue(
("Remove-Item" in text and ".git" in text)
or ("删除" in text and ".git" in text),
msg="DEVELOPMENT.md must document deleting .git",
)
def test_naming_md_copy_flow_documents_scaffold_or_git_cleanup(self) -> None:
text = self._read("development/NAMING.md")
self.assertTrue(
"scaffold" in text.lower() or ("清理" in text and ".git" in text),
msg="NAMING.md §9 must mention scaffold or Git cleanup",
)
def test_scaffold_ps1_exists(self) -> None:
path = os.path.join(get_skill_root(), "tools", "scaffold_skill.ps1")
self.assertTrue(os.path.isfile(path), msg="tools/scaffold_skill.ps1 must exist")
def test_guarded_docs_do_not_reference_amazon_implementation(self) -> None:
forbidden = (
"download-settlement-report-amazon",
"amazon_sim",
"#amz-email",
)
guarded_docs = (
"development/RPA.md",
"development/ADAPTER.md",
"development/DEVELOPMENT.md",
"development/CONFIG.md",
"development/TESTING.md",
"examples/README.md",
"examples/simulator_browser_rpa/README.md",
)
for rel in guarded_docs:
text = self._read(rel)
for fragment in forbidden:
self.assertNotIn(
fragment,
text,
msg=f"{rel} must not reference Amazon implementation {fragment!r}",
)
def test_adapter_md_covers_four_tiers_and_config_get(self) -> None: def test_adapter_md_covers_four_tiers_and_config_get(self) -> None:
text = self._read("development/ADAPTER.md") text = self._read("development/ADAPTER.md")
for marker in ( for marker in (

View File

@@ -0,0 +1,39 @@
# -*- coding: utf-8 -*-
"""禁止在模板主骨架 scripts/ 中 import account-manager 内部 RPA 辅助模块。"""
from __future__ import annotations
import os
import re
import unittest
from _support import get_skill_root
FORBIDDEN_PATTERNS = (
re.compile(r"\brpa_helpers\b"),
re.compile(r"\binject_account_manager_scripts_path\b"),
re.compile(r"\bget_account_credential\b"),
)
class TestNoRpaHelpersImport(unittest.TestCase):
def test_scripts_tree_has_no_forbidden_account_manager_imports(self) -> None:
scripts_dir = os.path.join(get_skill_root(), "scripts")
violations: list[str] = []
for root, _dirs, files in os.walk(scripts_dir):
for name in files:
if not name.endswith(".py"):
continue
rel = os.path.relpath(os.path.join(root, name), scripts_dir)
text = open(os.path.join(root, name), encoding="utf-8").read()
for pattern in FORBIDDEN_PATTERNS:
if pattern.search(text):
violations.append(f"{rel}: {pattern.pattern}")
self.assertEqual(
violations,
[],
msg="Forbidden account-manager internal imports in scripts/:\n" + "\n".join(violations),
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,64 @@
# -*- coding: utf-8 -*-
"""skill-template 脚手架与 Git 防串库守护。"""
from __future__ import annotations
import os
import unittest
from _support import get_skill_root
TEMPLATE_MARKER = ".openclaw-skill-template"
TEMPLATE_SLUG = "your-skill-slug"
def _read_skill_slug() -> str:
constants_path = os.path.join(get_skill_root(), "scripts", "util", "constants.py")
with open(constants_path, encoding="utf-8") as f:
for line in f:
if line.strip().startswith("SKILL_SLUG"):
_, _, value = line.partition("=")
return value.strip().strip('"').strip("'")
return ""
class TestScaffoldGuard(unittest.TestCase):
def test_template_repo_has_marker(self) -> None:
slug = _read_skill_slug()
if slug != TEMPLATE_SLUG:
self.skipTest(f"当前 SKILL_SLUG={slug!r},非模板占位,跳过 template marker 断言")
marker = os.path.join(get_skill_root(), TEMPLATE_MARKER)
self.assertTrue(
os.path.isfile(marker),
msg=f"skill-template 源仓库根目录必须存在 {TEMPLATE_MARKER}",
)
def test_business_skill_must_not_keep_template_marker(self) -> None:
slug = _read_skill_slug()
if slug == TEMPLATE_SLUG:
self.skipTest("当前仍为模板占位 slug由 test_template_repo_has_marker 守护")
marker = os.path.join(get_skill_root(), TEMPLATE_MARKER)
self.assertFalse(
os.path.isfile(marker),
msg=(
"业务技能不得保留 .openclaw-skill-template"
"请使用 tools/scaffold_skill 或删除标记文件并检查是否误复制了模板 .git"
),
)
def test_development_md_documents_git_cleanup(self) -> None:
path = os.path.join(get_skill_root(), "development", "DEVELOPMENT.md")
with open(path, encoding="utf-8") as f:
text = f.read()
self.assertIn(".git", text)
self.assertTrue(
"scaffold_skill" in text or "scaffold" in text.lower(),
msg="DEVELOPMENT.md must document scaffold_skill",
)
self.assertTrue(
"remote" in text.lower() or "git remote" in text.lower(),
msg="DEVELOPMENT.md must mention git remote self-check",
)
if __name__ == "__main__":
unittest.main()

87
tests/test_slug_naming.py Normal file
View File

@@ -0,0 +1,87 @@
# -*- coding: utf-8 -*-
"""slug 语义校验verb-noun-platform"""
from __future__ import annotations
import unittest
from util.slug_naming import (
LEGACY_EXEMPT_SLUGS,
classify_slug_form,
parse_slug_segments,
validate_slug_semantics,
)
class TestSlugNaming(unittest.TestCase):
def test_parse_slug_segments(self) -> None:
self.assertEqual(parse_slug_segments("download-settlement-report-amazon"), [
"download",
"settlement",
"report",
"amazon",
])
def test_classify_standard(self) -> None:
segments = parse_slug_segments("download-settlement-report-amazon")
self.assertEqual(classify_slug_form(segments, slug="download-settlement-report-amazon"), "standard")
def test_classify_scope(self) -> None:
segments = parse_slug_segments("generate-article-all")
self.assertEqual(classify_slug_form(segments, slug="generate-article-all"), "scope")
def test_classify_legacy_exempt(self) -> None:
segments = parse_slug_segments("your-skill-slug")
self.assertEqual(classify_slug_form(segments, slug="your-skill-slug"), "legacy_exempt")
def test_valid_hyphenated_platform_slug(self) -> None:
slug = "fill-export-declaration-single-window"
segments = parse_slug_segments(slug)
self.assertEqual(classify_slug_form(segments, slug=slug), "standard")
errors, warnings = validate_slug_semantics(slug)
self.assertEqual(errors, [], msg=f"errors={errors!r} warnings={warnings!r}")
def test_valid_standard_slugs(self) -> None:
for slug in (
"download-settlement-report-amazon",
"add-product-shopee",
"generate-article-all",
"your-skill-slug",
"account-manager",
):
with self.subTest(slug=slug):
errors, warnings = validate_slug_semantics(slug)
self.assertEqual(errors, [], msg=f"errors={errors!r} warnings={warnings!r}")
def test_invalid_slugs(self) -> None:
cases = (
"amazon-download-settlement-report",
"enter-cross-border-sales-receipt-voucher-kingdee",
"reconcile-finance",
"download-settlement-report-unknownplat",
"foobar-settlement-report-amazon",
)
for slug in cases:
with self.subTest(slug=slug):
errors, _warnings = validate_slug_semantics(slug)
self.assertTrue(errors, msg=f"expected errors for {slug!r}")
def test_length_boundary_48_passes(self) -> None:
slug = "query-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-amazon"
self.assertEqual(len(slug), 48)
errors, _warnings = validate_slug_semantics(slug)
self.assertEqual(errors, [])
def test_length_boundary_49_fails(self) -> None:
slug = "query-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-amazon"
self.assertEqual(len(slug), 49)
errors, _warnings = validate_slug_semantics(slug)
self.assertTrue(errors)
self.assertTrue(any("48" in err for err in errors))
def test_legacy_exempt_set(self) -> None:
self.assertIn("your-skill-slug", LEGACY_EXEMPT_SLUGS)
self.assertIn("account-manager", LEGACY_EXEMPT_SLUGS)
if __name__ == "__main__":
unittest.main()

47
tools/README.md Normal file
View File

@@ -0,0 +1,47 @@
# skill-template 工具
## 推荐:脚手架创建新技能
**不要**用资源管理器整文件夹复制 `skill-template`(会带走隐藏 `.git`,导致 push 串库)。
Windows 首选:
```powershell
cd D:\OpenClaw\client-commons\skill-template
.\tools\scaffold_skill.ps1 -Slug my-new-skill -Destination D:\OpenClaw\client-gdcm\my-new-skill
```
跨平台Python 3.10+
```bash
python tools/scaffold_skill.py --slug my-new-skill --destination /path/to/my-new-skill
```
脚本行为:
- 复制模板到 `-Destination`(目录末段须与 `-Slug` 一致)
- **排除**`.git``.pytest_cache``__pycache__``.env``.env.local``*.pyc`
- 删除目标中的 `.openclaw-skill-template` 标记
- **不**自动 `git init`;打印后续 Git 与测试命令
目标目录若已存在且非空,需加 `-Force` / `--force`(且目标**不得**含 `.git`)。
## 手工复制后的补救
若已整目录复制,必须先清理再初始化:
```powershell
cd <新技能目录>
Remove-Item -Recurse -Force .git
Remove-Item -Recurse -Force .pytest_cache -ErrorAction SilentlyContinue
Remove-Item -Force .openclaw-skill-template -ErrorAction SilentlyContinue
git init
git remote add origin <新技能仓库 URL>
git remote -v
```
确认 `origin` **不是** skill-template 的远端 URL。
## 模板标记
源仓库根目录有 `.openclaw-skill-template`(仅 template 保留)。业务技能不得保留该文件;`tests/test_scaffold_guard.py` 会守护。

107
tools/scaffold_skill.ps1 Normal file
View File

@@ -0,0 +1,107 @@
#Requires -Version 5.1
<#
.SYNOPSIS
Scaffold a new skill directory from skill-template (excludes .git).
.EXAMPLE
.\tools\scaffold_skill.ps1 -Slug my-new-skill -Destination D:\OpenClaw\client-gdcm\my-new-skill
#>
[CmdletBinding()]
param(
[Parameter(Mandatory = $true)]
[string]$Slug,
[Parameter(Mandatory = $true)]
[string]$Destination,
[switch]$Force
)
$ErrorActionPreference = "Stop"
function Test-KebabSlug([string]$Value) {
return $Value -match '^[a-z0-9]+(-[a-z0-9]+)*$'
}
function Write-ScaffoldNextSteps([string]$TargetPath) {
Write-Host ""
Write-Host "=== 脚手架复制完成 ===" -ForegroundColor Green
Write-Host ("目标目录: " + $TargetPath)
Write-Host ""
Write-Host "请在本机继续执行(未自动 git init避免误操作" -ForegroundColor Yellow
Write-Host (' cd "' + $TargetPath + '"')
Write-Host " Remove-Item -Recurse -Force .git -ErrorAction SilentlyContinue"
Write-Host " git init"
Write-Host " git remote add origin <你的新技能仓库 URL>"
Write-Host " git remote -v"
Write-Host " python tests/run_tests.py -v"
Write-Host ""
}
$TemplateRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path
$TargetPath = $Destination
if (-not (Test-KebabSlug $Slug)) {
throw "Slug must be kebab-case. Got: $Slug"
}
$leaf = Split-Path -Leaf ($TargetPath.TrimEnd('\', '/'))
if ($leaf -ne $Slug) {
throw "Destination leaf '$leaf' must match -Slug '$Slug'."
}
$parent = Split-Path -Parent $TargetPath
if (-not $parent -or -not (Test-Path -LiteralPath $parent)) {
throw "Parent directory does not exist: $parent"
}
$gitPath = Join-Path $TargetPath ".git"
if (Test-Path -LiteralPath $gitPath) {
throw "Target already has .git at $gitPath . Remove it first or use an empty destination."
}
$targetExists = Test-Path -LiteralPath $TargetPath
if ($targetExists) {
$entries = @(Get-ChildItem -LiteralPath $TargetPath -Force -ErrorAction SilentlyContinue)
if ($entries.Count -gt 0 -and -not $Force) {
throw "Target exists and is not empty: $TargetPath . Use -Force or pick an empty directory."
}
}
$excludeDirs = @(".git", ".pytest_cache", "__pycache__")
$excludeFiles = @(".env", ".env.local")
function Copy-TemplateTree {
param(
[string]$Source,
[string]$Dest
)
if (-not (Test-Path -LiteralPath $Dest)) {
New-Item -ItemType Directory -Path $Dest -Force | Out-Null
}
Get-ChildItem -LiteralPath $Source -Force | ForEach-Object {
$name = $_.Name
if ($excludeDirs -contains $name) { return }
if ($excludeFiles -contains $name) { return }
if ($name -like "*.pyc") { return }
$srcPath = $_.FullName
$dstPath = Join-Path $Dest $name
if ($_.PSIsContainer) {
Copy-TemplateTree -Source $srcPath -Dest $dstPath
}
else {
Copy-Item -LiteralPath $srcPath -Destination $dstPath -Force
}
}
}
Write-Host "Copy template: $TemplateRoot -> $TargetPath" -ForegroundColor Cyan
Copy-TemplateTree -Source $TemplateRoot -Dest $TargetPath
$marker = Join-Path $TargetPath ".openclaw-skill-template"
if (Test-Path -LiteralPath $marker) {
Remove-Item -LiteralPath $marker -Force
}
Write-ScaffoldNextSteps -TargetPath (Resolve-Path -LiteralPath $TargetPath).Path

106
tools/scaffold_skill.py Normal file
View File

@@ -0,0 +1,106 @@
#!/usr/bin/env python3
"""跨平台 skill-template 脚手架(与 scaffold_skill.ps1 行为一致)。"""
from __future__ import annotations
import argparse
import re
import shutil
import sys
from pathlib import Path
EXCLUDE_DIR_NAMES = {".git", ".pytest_cache", "__pycache__"}
EXCLUDE_FILE_NAMES = {".env", ".env.local"}
KEBAB_SLUG = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
def _ignore_copy(_dir: str, names: list[str]) -> set[str]:
ignored: set[str] = set()
for name in names:
if name in EXCLUDE_DIR_NAMES or name in EXCLUDE_FILE_NAMES:
ignored.add(name)
elif name.endswith(".pyc"):
ignored.add(name)
return ignored
def _print_next_steps(target: Path) -> None:
print()
print("=== 脚手架复制完成 ===")
print(f"目标目录: {target}")
print()
print("请在本机继续执行(未自动 git init避免误操作")
print(f' cd "{target}"')
print(" Remove-Item -Recurse -Force .git -ErrorAction SilentlyContinue # 若曾手工复制漏删")
print(" git init")
print(" git remote add origin <你的新技能仓库 URL>")
print(" git remote -v # 确认 remote 不是 skill-template")
print(" python tests/run_tests.py -v")
print()
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="从 skill-template 脚手架复制新技能目录")
parser.add_argument("--slug", required=True, help="kebab-case slug须与目标目录名一致")
parser.add_argument("--destination", required=True, help="新技能完整目标路径")
parser.add_argument("--force", action="store_true", help="目标非空时允许覆盖(目标不得含 .git")
args = parser.parse_args(argv)
slug = args.slug.strip()
if not KEBAB_SLUG.match(slug):
print(f"ERROR: Slug 必须是 kebab-case当前: {slug}", file=sys.stderr)
return 1
target = Path(args.destination).resolve()
if target.name != slug:
print(
f"ERROR: 目标目录末段 '{target.name}' 与 slug '{slug}' 不一致。",
file=sys.stderr,
)
return 1
parent = target.parent
if not parent.is_dir():
print(f"ERROR: 父目录不存在: {parent}", file=sys.stderr)
return 1
if (target / ".git").exists():
print(
f"ERROR: 目标已存在 .git请先删除: {target / '.git'}",
file=sys.stderr,
)
return 1
template_root = Path(__file__).resolve().parent.parent
if target.exists():
entries = list(target.iterdir())
if entries and not args.force:
print(
f"ERROR: 目标目录已存在且非空: {target}\n请加 --force 或换空目录。",
file=sys.stderr,
)
return 1
target.mkdir(parents=True, exist_ok=True)
print(f"复制模板: {template_root} -> {target}")
for item in template_root.iterdir():
name = item.name
if name in EXCLUDE_DIR_NAMES or name in EXCLUDE_FILE_NAMES:
continue
if name.endswith(".pyc"):
continue
dest = target / name
if item.is_dir():
shutil.copytree(item, dest, ignore=_ignore_copy, dirs_exist_ok=True)
else:
shutil.copy2(item, dest)
marker = target / ".openclaw-skill-template"
if marker.is_file():
marker.unlink()
_print_next_steps(target)
return 0
if __name__ == "__main__":
raise SystemExit(main())