Compare commits

..

9 Commits

Author SHA1 Message Date
7827cc75b2 chore: auto release commit (2026-06-13 14:59:54)
All checks were successful
技能自动化发布 / release (push) Successful in 10s
2026-06-13 14:59:54 +08:00
a24f6809da chore: auto release commit (2026-06-08 14:19:45)
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-06-08 14:19:46 +08:00
5b555e4313 去掉依赖 2026-06-07 14:50:44 +08:00
4c5bc31b0e 继续优化代码 2026-06-07 12:02:17 +08:00
45cf1741e1 继续完善代码 2026-06-07 11:51:23 +08:00
1bbcc65cf3 重新调整模板 2026-06-07 11:28:45 +08:00
ae5466854f fix: sync unified logging probe before file handler attach
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-06-02 15:18:59 +08:00
b7163f12a5 chore: sync ffmpeg video session and logging fallback
All checks were successful
技能自动化发布 / release (push) Successful in 4s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-02 14:29:02 +08:00
f52536bdcc chore: sync ffmpeg RpaVideoSession, remove Playwright recording
All checks were successful
技能自动化发布 / release (push) Successful in 5s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-02 14:07:02 +08:00
46 changed files with 1616 additions and 1496 deletions

View File

@@ -1,5 +1,5 @@
# ── 运行模式(对应 adapter 四档,见 references/ADAPTER.md──
OPENCLAW_TEST_TARGET=simulator_rpa # mock | simulator_rpa | real_api | real_rpa
# ── 运行模式 / adapter 档位(见 references/ADAPTER.md──
OPENCLAW_TEST_TARGET=mock # mock | simulator_rpa | real_api | real_rpa
# ── 目标系统 ──
TARGET_BASE_URL=https://sandbox.jc2009.com # 仿真/生产地址,可被进程环境变量覆盖

View File

@@ -5,8 +5,9 @@
## 模板目标
- 对齐当前规范 skill 的目录结构:`assets/``references/``scripts/``tests/``evals/`
- 对齐当前规范脚手架分层:`scripts/cli``scripts/db``scripts/service``scripts/util``scripts/jiangchang_skill_core`
- 提供最小可运行入口:`python scripts/main.py health` / `version`,以及通用任务命令骨架 `run` / `logs` / `log-get`
- 对齐当前规范脚手架分层:`scripts/cli``scripts/db``scripts/service``scripts/util`;公共能力从共享 runtime 的 `jiangchang-platform-kit` 引用
- 提供最小可运行入口:`python scripts/main.py health` / `config-path` / `version`,以及通用任务命令骨架 `run` / `logs` / `log-get`
- 内置 **配置 bootstrap**`.env.example` 首次落盘到 `{DATA_ROOT}/{USER_ID}/{slug}/.env`,升级后合并缺失 key`scripts/util/config_bootstrap.py`
- 内置 **隔离测试体系**`IsolatedDataRoot`、双键环境、`OPENCLAW_TEST_TARGET` 档位与 adapter profile 策略(详见 [`references/TESTING.md`](references/TESTING.md)
- 让新技能从一开始就按规范落地,不再沿用旧模板的 `docs/``optional/``skill_main.py` 结构
@@ -56,9 +57,24 @@ python scripts/main.py version
## Python 依赖requirements.txt
- 在技能**根目录**维护 `requirements.txt`,声明本技能特有的 Python 三方包
- 匠厂宿主在**安装或更新**技能后,会将该文件中的依赖安装到共享 Python runtime
- `jiangchang-platform-kit``playwright`、config、runtime diagnostics、RPA 公共能力等由**宿主共享 runtime** 提供;`jiangchang_skill_core` **不在技能仓库内复制**
- 新技能**不得 vendored** `scripts/jiangchang_skill_core/`;复制模板后若发现该目录,应删除并改为依赖共享 venv。
- 在技能**根目录**维护 `requirements.txt`**只声明技能特有** Python 三方包;**不要**重复声明 `jiangchang-platform-kit``playwright`
- `SKILL.md``metadata.openclaw.platform_kit_min_version`(当前 `1.0.13`)是运行契约/兼容性声明,供宿主安装与启用时校验,**不是** pip 依赖声明。`jiangchang-platform-kit>=1.0.13` 是当前 RPA 视频/旁白/背景音乐/前后缓冲等标准能力的最低版本。
- 匠厂宿主在**安装或更新**技能后,会将技能 `requirements.txt` 中的依赖安装到共享 Python runtime
`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`Windows 示例:`D:\jiangchang-data\python-runtime\.venv`)。
- 平台通用依赖由宿主 `resources/jiangchang-python-runtime/pyproject.toml` 维护;**不要**把业务依赖写进宿主 pyproject。
- **Skill 业务代码不得自动 `pip install`**;缺依赖时应在 `health` / preflight 中给出清晰错误。
- **系统级组件**(如 Microsoft Visual C++ Redistributable**不要**写进 `requirements.txt`;仅在 `health` / preflight 中提示用户自行安装。
- 独立本地开发环境若缺少公共包,可手动安装;**生产/宿主运行**由共享 runtime 负责,无需在技能内 `pip install playwright`
## 公共能力边界
以下能力均由 `jiangchang-platform-kit`(共享 runtime 的 `jiangchang_skill_core`)提供,技能只写业务逻辑与薄封装,**不要**在技能内复制实现:
- `config``.env` 三级优先级读取与首次落盘
- `runtime_env` — 数据根、用户 ID、兄弟技能根解析
- `unified_logging` — 统一日志与 trace
- `rpa` — 浏览器/桌面/手机 RPA 共享库
- `media_assets` — 共享媒体资源探测与路径解析
- `video_session` — RPA 录屏成片ffmpeg、字幕、背景音乐
- `runtime_diagnostics``health` 统一 runtime 诊断

View File

@@ -6,6 +6,7 @@ author: 深圳匠厂科技有限公司
metadata:
openclaw:
slug: your-skill-slug
platform_kit_min_version: "1.0.13"
emoji: "📦"
category: "通用"
developer_ids:
@@ -29,20 +30,27 @@ allowed-tools:
- 根目录结构参考现有规范技能:`assets/``references/``scripts/``tests/``evals/`
- CLI 入口固定为 `scripts/main.py`
- 业务逻辑按 `cli / db / service / util / jiangchang_skill_core` 分层
- 业务逻辑按 `cli / db / service / util` 分层;公共能力从 `jiangchang-platform-kit` 提供的 `jiangchang_skill_core` 引用
## 最小命令
```bash
python {baseDir}/scripts/main.py health
python {baseDir}/scripts/main.py config-path
python {baseDir}/scripts/main.py version
```
配置:仓库 `.env.example` 为模板;用户 `.env``{CLAW_DATA_ROOT}/{CLAW_USER_ID}/{slug}/.env`,启动时自动 bootstrap。优先级进程环境变量 > 用户 `.env` > `.env.example`
## 运行依赖
- Python 运行环境由匠厂宿主注入共享 runtime`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`
- 本技能**特有**的 Python 三方依赖请声明在根目录 `requirements.txt`;宿主安装/更新技能后会自动安装到共享 venv
- Python 运行环境由匠厂宿主注入**共享 runtime**`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`
- 公共能力来自共享 runtime 安装的 `jiangchang-platform-kit>=1.0.13``jiangchang_skill_core` 包);**新技能不得**在仓库内保留 `scripts/jiangchang_skill_core/` 副本
- config、logging、runtime_env、rpa、media-assets、video_session、runtime_diagnostics 等均从共享 venv 的 `jiangchang_skill_core` import而非技能目录副本。
- 根目录 `requirements.txt` **只声明技能特有** Python 三方依赖;`jiangchang-platform-kit``playwright` 等公共能力由宿主共享 runtime 提供,**不要**写入技能 requirements。宿主安装/更新技能后会将技能 requirements 安装到共享 venv。
- `metadata.openclaw.platform_kit_min_version`(当前 `1.0.13`)是运行契约/兼容性声明,供宿主安装与启用时校验,**不是**技能 pip 依赖声明。`jiangchang-platform-kit>=1.0.13` 是当前 RPA 视频/旁白/背景音乐/前后缓冲等标准能力的最低版本。
- Skill 代码**不要**自行 `pip install`;系统级依赖(如 VC++ Runtime仅在 `health` / preflight 中提示用户安装。
- `health` 使用 `collect_runtime_diagnostics` 输出统一 runtime 诊断(只读,不下载/修复 media-assets
## 重要说明

8
pytest.ini Normal file
View File

@@ -0,0 +1,8 @@
[pytest]
testpaths = tests
python_files = test_*.py *_test.py
python_classes = Test*
python_functions = test_*
norecursedirs = .git __pycache__ .pytest_cache integration desktop samples fixtures artifacts diagnostics
filterwarnings =
ignore::DeprecationWarning

View File

@@ -1,6 +1,6 @@
# 适配器标准:真实/仿真 × API/RPA 四档模式
> 凡是"连接三方系统"ERP、CRM、SaaS、银行等的 skill都应采用本文的 **adapter 四档模式**。这样同一套业务逻辑可以在不同档位间切换:开发用仿真、CI 用 mock、上线用真实,互不影响。
> 凡是"连接三方系统"ERP、CRM、SaaS、银行等的 skill都应采用本文的 **adapter 四档模式**。这样同一套业务逻辑可以在不同档位间切换:开发用 mock、半集成用 simulator、上线用真实,互不影响。
## 为什么要分档
@@ -9,104 +9,97 @@
| | API有接口 | RPA操作界面 |
|---|---|---|
| **真实系统** | `real_api` 真实 API | `real_rpa` 真实界面 |
| **仿真系统** | `mock` 离线仿真 | `sim_rpa` 仿真平台界面 |
| **仿真/离线** | `mock` 纯内存/fixture | `simulator_rpa` 仿真站点/桌面仿真 |
### 四档定义
| 档位 | 含义 | 默认策略 |
|------|------|----------|
| **`mock`** | 纯内存或 fixture**默认单测/CI** | 模板 `.env.example` 默认 `OPENCLAW_TEST_TARGET=mock` |
| **`simulator_rpa`** | 仿真站点或桌面仿真,可半集成 | 开发联调可选 |
| **`real_api`** | 真实 API | **必须** `ALLOW_REAL_API=1` |
| **`real_rpa`** | 真实浏览器/真实系统 | **必须** `ALLOW_REAL_RPA=1` |
- **mock**:纯离线、不联网,给单测 / CI / 开发自测,**保证可重复**。
- **sim_rpa**:操作仿真平台(如 `sandbox.jc2009.com`),跑真实端到端流程但不碰生产**默认开发档位**
- **simulator_rpa**:操作仿真平台(如 `sandbox.jc2009.com`),跑端到端流程但不碰生产。
- **real_api**:有官方接口时**首选**(最稳、最快、最易维护)。
- **real_rpa**:没有 API 只能操作生产界面,**风险最高、放最后**。
> 推荐优先级:**real_api > sim_rpa > real_rpa**mock 永远保留做 CI。
> 推荐优先级:**real_api > simulator_rpa > real_rpa**mock 永远保留做 CI。
## 权限开关
未显式授权时**不得**落到真实 API/RPA
| 变量 | 作用 |
|------|------|
| `ALLOW_REAL_API=1` | 允许 `real_api` 访问真实 HTTP |
| `ALLOW_REAL_RPA=1` | 允许 `real_rpa` 驱动真实浏览器/RPA |
| `ALLOW_WRITE_ACTIONS=1` | 在 `real_*` 下允许**写**操作(提交表单、下单等) |
进程环境变量优先于用户 `.env`(见 `CONFIG.md` 三层优先级)。
## 目录骨架
业务逻辑只依赖 `base` 里定义的接口与数据契约,具体档位实现各自一个文件:
```
scripts/service/<domain>_adapter/
__init__.py # dispatch按档位返回对应 adapter 实例
base.py # 数据契约dataclass+ AdapterBase 接口NotImplementedError
base.py # 数据契约dataclass+ AdapterBase 接口
mock.py # 离线仿真,给 CI/单测
real_api.py # 真实系统 API
sim_rpa.py # 仿真平台 RPAPlaywright/pywinauto/uiautomator2
sim_rpa.py # 仿真平台 RPA
real_rpa.py # 真实系统 RPA占位谨慎实现
```
> 参考实现:`disburse-payroll-icbc/scripts/service/disburse_adapter/`(已有 `base.py` / `mock.py` / `icbc_simulator_rpa.py`)。
> 模板示例:`scripts/service/example_adapter/`(复制改名即用)。
### 引用方式
```python
from service.example_adapter import get_adapter
from service.example_adapter.base import ExampleItem
adapter = get_adapter() # 由 OPENCLAW_TEST_TARGET 决定档位
result = adapter.do_batch(
target="https://sandbox.jc2009.com",
items=[ExampleItem(id="1", label="demo")],
)
```
mock 档位纯离线验证:
```bash
set OPENCLAW_TEST_TARGET=mock
python -c "import sys; sys.path.insert(0,'scripts'); from service.example_adapter import get_adapter; from service.example_adapter.base import ExampleItem; r=get_adapter().do_batch('x',[ExampleItem('1','a')]); print(r.ok, r.serial_no)"
```
## base.py 写法(要点)
1. **数据契约用 `@dataclass`**:输入项(如 `PayrollItem`)、批量结果(如 `BatchResult{ok, serial_no, error_msg, artifacts}`)。
2. **接口方法抛 `NotImplementedError`**:所有档位实现同一签名。
3. **以"批"为单位返回**:真实 RPA 往往是整批提交,结果按批次返回,不要假设逐条。
4. **结果里带 `artifacts`**:截图/录屏/日志路径,便于排查。
```python
@dataclass
class XxxResult:
ok: bool
serial_no: Optional[str]
error_msg: Optional[str]
artifacts: Dict[str, Any] = field(default_factory=dict)
class AdapterBase:
name = "base"
def do_batch(self, target: str, items: list) -> XxxResult:
raise NotImplementedError
```
模板示例:`scripts/service/example_adapter/`(复制改名即用)。
## 档位 dispatch
`OPENCLAW_TEST_TARGET` 统一决定用哪个 adapter(与现有测试体系一致)
`OPENCLAW_TEST_TARGET` 统一决定用哪个 adapter
| `OPENCLAW_TEST_TARGET` | adapter | 用途 |
|---|---|---|
| `unit` / `mock` | `MockAdapter` | 单测 / CI离线 |
| `simulator_rpa`(默认) | `SimRpaAdapter` | 开发/演示,操作仿真平台 |
| `simulator_rpa` | `SimRpaAdapter` | 开发/演示,操作仿真平台 |
| `real_api` | `RealApiAdapter` | 生产,走官方接口 |
| `real_rpa` | `RealRpaAdapter` | 生产,操作真实界面(最后手段) |
| `real_rpa` | `RealRpaAdapter` | 生产,操作真实界面 |
```python
# __init__.py
def get_adapter():
target = (os.environ.get("OPENCLAW_TEST_TARGET") or "simulator_rpa").lower()
target = (os.environ.get("OPENCLAW_TEST_TARGET") or "mock").lower()
if target in ("unit", "mock"):
return MockAdapter()
if target == "real_api":
return RealApiAdapter()
if target == "real_rpa":
return RealRpaAdapter()
return SimRpaAdapter() # 默认
return SimRpaAdapter()
```
> 实际可改为读 `.env` 里的运行模式(见 `CONFIG.md``OPENCLAW_TEST_TARGET` 作为进程级覆盖优先
`get_adapter()` 实现应配合 `tests/adapter_test_utils.py` 的 profile 策略:**未授权时不 silently 开启真实网络/RPA**
## 测试要求
## contract tests
- **mock 档必须能纯离线跑通**,进 CI。
- **每个 adapter 单测同一组用例**(契约测试),保证档位间行为一致。
- RPA 档的真实联网测试单独标记,不进默认 CI参考 `tests/desktop/`)。
每个 adapter **必须覆盖同一套契约测试**(复制 `tests/samples/test_service_contract.py.sample`
- 至少覆盖:**timeout**、**unauthorized**、**invalid response**、**empty result**
- **mock 不允许真实网络**
- 使用 `FakeAdapter` 或等价 stub 模拟异常路径
## 兄弟依赖
依赖 account-manager 或其他兄弟技能时:
1.`SKILL.md``metadata.openclaw.dependencies.required` 声明。
2. 调用兄弟技能走统一 **`service.sibling_bridge`**`call_sibling_json`**不要**在 service 层散落 `subprocess.run`
3. **lease 必须 release**account-manager 租约);进程被 kill 后可能残留 lease需在运维文档说明排查方式查 account-manager lease 列表 / 手动释放)。
```python
from service.sibling_bridge import call_sibling_json
result = call_sibling_json("account-manager", ["list", "--limit", "10"])
```
## 相关文档

View File

@@ -1,4 +1,4 @@
# 通用业务技能模板 CLI 模板
# 通用业务技能模板 CLI 标准
`{baseDir}` 替换为技能根目录(含 `SKILL.md``scripts/` 的目录)。所有命令通过 `python {baseDir}/scripts/main.py` 调用。
@@ -6,41 +6,82 @@
```bash
python {baseDir}/scripts/main.py health
python {baseDir}/scripts/main.py config-path
python {baseDir}/scripts/main.py version
```
## 通用业务命令骨架
## 标准行为
模板默认提供以下命令(复制后按需保留或扩展):
- **任意 CLI 启动前**执行 `bootstrap_skill_config()``main.py``cli.app.main()` 均触发)。
- **`health`**:只读 runtime 诊断,**不下载、不修复 media-assets不执行业务动作**;不输出敏感值。
- **`config-path`**:输出 JSON包含 `skill``env_path``example_path`
- **`version`**:输出 JSON`version``skill`)。
- **`run`**:长时间无 stdout **不代表卡死**;应通过 `logs` / `log-get` 排查。
- **任务完成后**若有 video artifactCLI 应打印录屏路径、录屏日志、视频/音频诊断(见 `task_run_support._print_video_summary`)。
## config-path配置路径
```json
{
"skill": "your-skill-slug",
"env_path": "{DATA_ROOT}/{USER_ID}/your-skill-slug/.env",
"example_path": "{skill_root}/.env.example"
}
```
## health 标准输出runtime diagnostics
`health` 委托 `jiangchang_skill_core.collect_runtime_diagnostics`,典型字段:
- `python_executable` — 当前 Python 解释器路径
- `platform_kit_version` / `platform_kit_min_version` / `platform_kit_version_ok`
- `jiangchang_skill_core_file` — 公共库加载来源(若来自技能目录副本会报 warning
- `media_assets_root``ffmpeg_path``ffmpeg_available``background_music_mp3_count`
- `runtime_issue[warning|error]` — 非致命/致命问题列表
- `env_path` / `env_exists` / `example_path` — 用户 `.env` 与仓库模板路径
## 通用业务命令骨架
```bash
python {baseDir}/scripts/main.py run
python {baseDir}/scripts/main.py run -t <target> -i <input_id>
python {baseDir}/scripts/main.py logs
python {baseDir}/scripts/main.py logs --task-type disburse --status failed
python {baseDir}/scripts/main.py logs --task-type demo --status failed
python {baseDir}/scripts/main.py log-get <log_id>
```
含义:
- `run`:执行一次任务(具体语义由你的 skill 决定)
- `logs`:列出最近 N 条任务日志,可按 task_type / status / target 过滤
- `log-get`:按 log_id 查看单条任务日志JSON
| 命令 | 用途 |
|------|------|
| `run` | 执行一次任务 |
| `logs` | 列出最近 N 条任务日志 |
| `log-get` | 按 log_id 查看单条任务日志JSON |
## 若你的技能是发布类
## 查询/导出类(可选)
可以把 `run` 命令对外别名为 `publish`,参数 `--target` 别名为 `--account-id`,参数 `--input-id` 保持不变
复制后按需增加,例如:`contacts``contact-get``export` 或业务等价命令
## 若你的技能依赖兄弟技能
使用 `service.sibling_bridge.call_sibling_json(skill_slug, args)`,不要硬编码兄弟技能名到 sibling_bridge.py 自身。
使用 `service.sibling_bridge.call_sibling_json(skill_slug, args)`,不要硬编码兄弟技能名到 `sibling_bridge.py` 自身;并在 `SKILL.md` 声明 `dependencies.required`
```bash
# 假设 base 与兄弟技能同级
python {baseDir}/../<sibling-skill-slug>/scripts/main.py ...
## 手工排查命令(推荐)
**建议使用宿主共享 python-runtime**,避免技能目录内临时 venv 加载不到公共库:
```text
Windows:
{JIANGCHANG_DATA_ROOT}\python-runtime\.venv\Scripts\python.exe {baseDir}\scripts\main.py health
通用:
<shared-python> {baseDir}/scripts/main.py health
```
`<shared-python>` 通常位于 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`
**不建议**在生产/测试机使用技能目录内 `uv run python`,以免加载不到共享 runtime 中的 `jiangchang-platform-kit` / `playwright`
## 模板约定
- 最小模板至少保留 `health` / `version`
- 最小模板至少保留 `health` / `config-path` / `version`
- 通用业务技能建议使用 `run` / `logs` / `log-get`
- 不要再用旧模板的 `publish` / `--account-id` / `--article-id` 命名

View File

@@ -4,22 +4,23 @@
## 核心原则
1. **`.env.example` 进仓库**:带注释的全量默认配置,是配置项的"单一事实来源"
2. **首次运行自动 copy**`setup` / `doctor` / 首次 `run` 时,把 `.env.example` 复制到
`{CLAW_DATA_ROOT}/{user}/{slug}/.env`**已存在则不覆盖**,保护用户改过的值)。
3. **三级优先级读取**`进程环境变量` > `数据目录/.env` > `.env.example 默认值`
4. **敏感信息绝不进 `.env`**:密码/密钥/口令走 account-manager 的 `secret-ref`(见下"红线")。
1. **三层优先级**`进程环境变量` > `{数据目录}/.env` > 仓库 `.env.example` 默认值
2. **`.env.example` 进仓库**:带注释的全量默认配置,是配置项的"单一事实来源"。
3. **首次运行自动 copybootstrap**`scripts/main.py` 启动与 `cli.app.main()` 均会调用 `util.config_bootstrap.bootstrap_skill_config()`,把仓库根目录 `.env.example` 复制到
`{CLAW_DATA_ROOT}/{user}/{slug}/.env`**已存在则不整体覆盖**,保护用户改过的值)
4. **技能升级合并缺失项**`merge_missing_env_keys` 会把 `.env.example` 中新增、但用户 `.env` 尚未包含的 key **追加**进去(带注释块)。
5. **敏感信息绝不进 `.env`**:密码/密钥/口令走 account-manager 的 `secret-ref`(见下"红线")。
## 标准配置项
## 标准配置项`.env` 分组)
每个 skill 的 `.env.example` 至少包含这些(按需增减):
每个 skill 的 `.env.example` 至少包含这些通用项(按需增减):
```ini
# ── 运行模式(对应 adapter 四档,见 ADAPTER.md──
OPENCLAW_TEST_TARGET=simulator_rpa # mock | simulator_rpa | real_api | real_rpa
# ── 运行模式 / adapter 档位(见 ADAPTER.md──
OPENCLAW_TEST_TARGET=mock # mock | simulator_rpa | real_api | real_rpa
# ── 目标系统 ──
TARGET_BASE_URL=https://sandbox.jc2009.com # 仿真/生产地址,可被进程环境变量覆盖
# ── 目标系统(示例占位,复制后改为业务地址)──
TARGET_BASE_URL=https://sandbox.jc2009.com
# ── 默认账号(仅非敏感标识;密码走 account-manager──
DEFAULT_LOGIN_ID=04110001
@@ -28,23 +29,24 @@ DEFAULT_LOGIN_ID=04110001
OPENCLAW_BROWSER_HEADLESS=0 # 0=有头(默认) 1=无头(CI)
OPENCLAW_PLAYWRIGHT_STEALTH=1 # 1=开启反检测指纹(默认)
# ── 存证 ──
OPENCLAW_RECORD_VIDEO=1 # 1=录屏+字幕+背景音+MP4成片RPA技能默认开,见 RPA.md
# ── 录屏与失败存证 ──
OPENCLAW_RECORD_VIDEO=1 # 1=录屏+字幕+旁白+背景音+MP4RPA 默认开,见 RPA.md
OPENCLAW_ARTIFACTS_ON_FAILURE=1 # 1=失败截图(默认)
# ── 节流 / 超时 ──
STEP_DELAY_MIN=1.0 # 步骤间随机等待下限(秒)
STEP_DELAY_MAX=5.0 # 步骤间随机等待上限(秒)
STEP_DELAY_MIN=1.0
STEP_DELAY_MAX=5.0
HUMAN_WAIT_TIMEOUT=180 # 滑块/验证码/2FA 等人工超时(秒)
```
**业务专属配置必须带技能前缀**(如 `DEMO_XXX``MY_SKILL_XXX`**不要污染全局命名空间**(禁止 `SCRAPE_1688_*``RECEIVE_ORDER_*` 等跨技能前缀写入模板)。
## 🚫 红线:敏感信息不进 `.env`
`.env` 只放**非敏感运行参数**模式、地址、开关、超时。密码、密钥、动态口令、token——
`.env` 只放**非敏感运行参数**(模式、地址、开关、超时、账号标识。密码、密钥、动态口令、token——
-**account-manager** 注册凭据,用 `--secret-storage env --secret-ref XXX` 引用;
- 真实值由宿主/用户通过环境变量注入,**不落任何文件、不进 git**。
- 参考 `disburse-payroll-icbc/scripts/cli/cmd_setup.py`:账号写入 account-manager密码只登记 `secret-ref`(如 `ICBC_SIM_PASSWORD`)。
- 真实值由宿主/用户通过进程环境变量注入,**不落任何文件、不进 git**。
`.gitignore` 必须包含:
@@ -53,43 +55,41 @@ HUMAN_WAIT_TIMEOUT=180 # 滑块/验证码/2FA 等人工超时(
*.env.local
```
## 配置读取层(模板提供)
## 配置 bootstrap(模板提供)
模板会提供一个统一 `config.py`,约定行为
模板`scripts/util/config_bootstrap.py` 提供 `bootstrap_skill_config()`
```python
config.get("TARGET_BASE_URL") # 三级优先级解析
from util.config_bootstrap import bootstrap_skill_config
bootstrap_skill_config() # main.py 与 cli.app.main() 启动时调用
```
内部委托共享 runtime 的 `jiangchang_skill_core.config`
```python
config.ensure_env_file(SKILL_SLUG, example_path) # 首次落盘
config.merge_missing_env_keys(example_path, env_path) # 升级后追加缺失 key
```
业务代码**只通过 `config.get*` 读配置**,不直接 `os.environ`
```python
config.get("TARGET_BASE_URL")
config.get_bool("OPENCLAW_BROWSER_HEADLESS")
config.get_float("STEP_DELAY_MIN", 1.0)
config.ensure_env_file() # 首次把 .env.example copy 到数据目录
```
- 启动时调一次 `ensure_env_file()` 完成落盘。
- 所有业务代码**只通过 `config.get*` 读配置**,不直接 `os.environ`,便于统一管理与测试覆盖。
## health / config-path
### 引用方式
- **`health`**:输出 `collect_runtime_diagnostics` 字段(`platform_kit_version_ok``ffmpeg_path` 等),**不打印敏感值**;补充 `env_path` / `env_exists` / `example_path`
- **`config-path`**:输出 JSON包含 `skill``env_path`(用户数据目录 `.env`)、`example_path`(仓库 `.env.example` 绝对路径),便于排查落盘位置。
实现位于 `jiangchang-platform-kit``jiangchang_skill_core.config`;模板 vendor 拷贝在 `scripts/jiangchang_skill_core/config.py`
## doctor / setup 命令(可选)
```python
from jiangchang_skill_core import config
from util.constants import SKILL_SLUG
from util.runtime_paths import get_skill_root
import os
每个 skill 可提供自检命令:
example = os.path.join(get_skill_root(), ".env.example")
config.ensure_env_file(SKILL_SLUG, example)
headless = config.get_bool("OPENCLAW_BROWSER_HEADLESS")
target = config.get("TARGET_BASE_URL")
timeout = config.get_int("HUMAN_WAIT_TIMEOUT", 180)
```
## doctor / setup 命令
每个 skill 应提供自检命令(参考 `disburse``test_doctor`
- `python scripts/main.py doctor`检查浏览器是否安装、Profile/账号是否就绪、`.env` 是否已落盘、设备(手机)是否连接。
- `python scripts/main.py doctor`检查浏览器、Profile/账号、`.env` 落盘、设备等。
- `python scripts/main.py setup`:初始化演示账号、首次落盘 `.env`
## 相关文档

View File

@@ -97,7 +97,6 @@ your-skill/
├─ scripts/
│ ├─ cli/
│ ├─ db/
│ ├─ jiangchang_skill_core/
│ ├─ service/
│ └─ util/
├─ tests/
@@ -125,7 +124,6 @@ scripts/
├─ main.py
├─ cli/
├─ db/
├─ jiangchang_skill_core/
├─ service/
└─ util/
```
@@ -151,9 +149,7 @@ scripts/
- `util/`
作用:常量、日志、路径、时间工具、通用帮助函数
- `jiangchang_skill_core/`
作用运行时环境、统一日志、配置读取、RPA 共享库vendor 自 platform-kit
一般按现有规范技能复制,不要自己乱改结构
公共能力config、logging、runtime_env、rpa、media_assets、video_session、runtime_diagnostics从共享 runtime 的 `jiangchang-platform-kit>=1.0.13` import**不得**在 `scripts/` 下保留 `jiangchang_skill_core/` 副本。
## 3.2 开发 RPA 类 skill
@@ -168,7 +164,7 @@ scripts/
from jiangchang_skill_core import config
from jiangchang_skill_core.rpa import launch_persistent_browser, anti_detect, wait_for_captcha_pass
```
vendor 源码在 `scripts/jiangchang_skill_core/rpa/`(与 `jiangchang-platform-kit` 保持同步拷贝)
上述 import 来自宿主共享 runtime 安装的 `jiangchang-platform-kit`,不是技能目录副本
4. **mock 档必须离线可跑**`OPENCLAW_TEST_TARGET=mock`sim_rpa / real_* 按需单独测。
5. **桌面/手机**:本期标准见 RPA.md 第 2/3 节,复用 `jiangchang_desktop_sdk` / `screencast`**不要在新 skill 里重复造包**(尚待实战验证)。
@@ -176,9 +172,11 @@ scripts/
### 3.1 `requirements.txt` 依赖规范
技能根目录的 `requirements.txt` 是**标准文件**,用于声明本技能特有的 Python 三方依赖。
技能根目录的 `requirements.txt` 是**标准文件**,用于声明本技能**特有** Python 三方依赖。
- 匠厂宿主安装/更新技能后,会将其安装到共享 venv`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。
- **公共依赖**`jiangchang-platform-kit`、`playwright`、config、runtime diagnostics、RPA 公共能力等)由**宿主共享 runtime** 提供,**不要**写入技能 `requirements.txt`。
- `SKILL.md` 的 `metadata.openclaw.platform_kit_min_version`(当前 `1.0.13`)是运行契约/兼容性声明,供宿主安装与启用时校验,**不是** pip 依赖声明。
- 匠厂宿主安装/更新技能后,会将技能 `requirements.txt` 安装到共享 venv`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。
- **不要**在业务代码中 `subprocess` / `pip install`;缺依赖由 `health` 报错,由宿主负责安装。
- **版本约束尽量收窄**,降低多技能共享 venv 时的冲突风险。推荐范围写法:
- `pkg>=1.2.0,<2`
@@ -186,7 +184,24 @@ scripts/
- 不推荐:`chromadb>=0.5.0`
- 推荐:`chromadb>=0.5.23,<0.6`
- **不要**把系统组件VC++ Runtime、浏览器安装包等写进 `requirements.txt`;这类前置条件写在 `health` / preflight 文档与错误提示中。
- 若技能无额外 Python 依赖,可保留仅含注释的 `requirements.txt`(宿主会识别为 empty 并跳过安装)
- 若技能无额外 Python 依赖,可保留空文件或仅含注释说明;**不要**为 platform-kit 或 playwright 保留占位行
- 独立本地开发环境若缺少公共包,可手动安装;**生产/宿主运行**由共享 runtime 负责。
### 3.3 公共能力不要复制
以下模块由 platform-kit 提供,技能内**不要**重复实现或 vendored 拷贝:
| 能力 | 导入方式 |
|------|----------|
| config | `from jiangchang_skill_core import config` |
| logging | `from jiangchang_skill_core.unified_logging import ...` 或 `util.logging_config` 薄封装 |
| runtime_env | `from jiangchang_skill_core.runtime_env import ...` 或 `util.runtime_paths` 薄封装 |
| rpa | `from jiangchang_skill_core.rpa import ...` |
| media_assets | `from jiangchang_skill_core.media_assets import ...` |
| video_session | `from jiangchang_skill_core.rpa.video_session import RpaVideoSession` |
| runtime_diagnostics | `from jiangchang_skill_core import collect_runtime_diagnostics, format_runtime_health_lines` |
`health` 应委托 `collect_runtime_diagnostics`,不要在 `scripts/util/` 自建 `runtime_diagnostics.py`。
## 4. 开发一个新 skill 的标准步骤

View File

@@ -1,64 +1,83 @@
---
description: "规范化的新技能模板说明复制后替换名称、CLI 示例与 task_service 等业务实现。"
description: "用用户能理解的方式说明这个技能能做什么、适合什么场景,以及如何开始使用。"
---
# 技能模板说明
# 技能名称】
这个仓库是**给开发者复制的新技能模板**,不是终端用户直接调用的业务 skill。
用一两句话说明这个技能帮助用户完成什么工作。
这里不要解释技术实现,要从用户的业务目标出发描述价值。
## 它提供什么
## 这个技能能帮你做什么
- 标准目录结构
- 最小 CLI 入口(`run` / `logs` / `log-get` / `health` / `version`
- 通用任务日志表骨架(`task_logs`
- `service` 层占位模块(`task_service.py`、通用 `sibling_bridge.py`
- 成熟的隔离测试体系IsolatedDataRoot、FakeAdapter、6 档测试 target、profile 策略)
- 与现有规范 skill 一致的发布脚本与 GitHub workflow
用 3 到 6 条业务能力说明技能价值,例如:
## 复制后你需要改什么
- 【自动整理/查询/生成/同步/核对/提交】某类业务数据
- 帮你减少重复操作,降低人工录入或核对成本
- 在任务完成后给出清晰的处理结果、失败原因或后续建议
- `SKILL.md` 中的名称、描述、slug、触发说明
- `SKILL.md``metadata.openclaw.developer_ids`(如需让非公开技能默认授权给开发者查看)
- `references/CLI.md` 里的命令示例
- `scripts/util/constants.py` 中的 slug / 版本 / logger 名
- `scripts/service/task_service.py` 中的 `cmd_run` 真实业务逻辑
- 如有特定 task_type`references/SCHEMA.md` 中补充字段映射说明
## 适合什么场景
## `developer_ids` 是做什么的
列出典型使用场景。请使用用户熟悉的业务语言,例如:
`metadata.openclaw.developer_ids` 是平台发布元数据,不是终端用户文案。
- 每天需要重复处理同一类订单、单据、报表或客户资料
- 需要从多个系统中查询信息并汇总结果
- 需要按固定规则检查数据是否完整、是否异常
- 需要把处理结果保存为文件、记录或任务报告
它用于声明:当技能发布后,如果平台侧将该技能设置为 `access_scope = 0`(不公开),哪些开发者用户仍应自动获得可见权限。
## 使用前需要准备什么
约定如下
说明用户在使用前要准备的东西,例如
- 字段位置:`SKILL.md` -> `metadata.openclaw.developer_ids`
- 推荐格式:正整数数组,例如 `[1032, 12428]`
- 发布效果:发布接口会把这些用户补写到 `skill_user_access`
- 第一个 ID 会作为主开发者同步到 `skills.developer_id`
- 相关账号已登录,且具备查询、导出或提交权限
- 需要处理的文件、时间范围、订单号、客户名称、项目名称等信息
- 如果涉及外部系统,请确认网络、账号、验证码或审批流程可正常使用
如果你的技能本来就是公开技能,或暂时不需要开发者专属可见性,这个字段可以留空数组。
## 你可以这样告诉 Agent
## 不建议再保留的旧结构
用自然语言示例展示用户可以怎么发起任务,例如:
- 旧模板里的 `docs/`
- 旧模板里的 `optional/`
- 旧入口 `scripts/skill_main.py`
- “帮我查询今天的【业务对象】,整理成表格。”
- “帮我核对这些【数据/单据】是否有异常,并告诉我问题在哪里。”
- “帮我把【时间范围】内的【业务数据】导出并生成处理报告。”
- “帮我根据这个文件里的内容,逐条完成【业务动作】。”
新模板统一使用 `scripts/main.py` 作为入口
注意:示例应是自然语言,不要写命令行命令
## 参考文档索引
## 执行后你会得到什么
| 文档 | 内容 |
|------|------|
| `DEVELOPMENT.md` | 从复制模板到开发出新 skill 的完整教程 |
| `RUNTIME.md` | 运行时目录结构与 `CLAW_*` 环境变量约定 |
| `CLI.md` | CLI 命令与参数示例 |
| `SCHEMA.md` | task_logs 与字段映射说明 |
| `TESTING.md` | 测试 target 与隔离体系 |
| **`RPA.md`** | **三端 RPA 标准(浏览器/桌面/手机):选型、拟人操作、反反爬、人工兜底、存证** |
| **`ADAPTER.md`** | **真实/仿真 × API/RPA 四档适配器模式** |
| **`CONFIG.md`** | **`.env` 配置规范与首次自动落盘机制** |
| `REQUIREMENTS.md` | 依赖与运行要求 |
说明用户会看到什么结果,例如:
> 开发"操作浏览器/桌面/手机"或"对接 ERP/CRM/SaaS/银行"类 skill 前,**务必先读 `RPA.md` + `ADAPTER.md` + `CONFIG.md`** 这三份标准,避免重复踩坑。
- 任务是否成功完成
- 处理了多少条数据
- 哪些内容成功、哪些失败
- 失败原因和建议处理方式
- 生成的文件、表格、报告或记录
## 注意事项
用用户能理解的方式说明边界和风险,例如:
- 请确认输入范围准确,避免处理错数据
- 涉及提交、付款、删除、审批等高风险动作时,技能应在执行前请求确认
- 如果外部系统登录失效、权限不足或页面变化,任务可能需要用户介入
- 不建议在未核对数据的情况下直接执行批量操作
## 常见问题
### 为什么任务没有查到数据?
可能是时间范围、关键词、账号权限或系统筛选条件不正确。请先确认输入信息是否准确。
### 为什么任务执行到一半需要我操作?
可能遇到了登录、验证码、权限确认、二次审批或系统异常。请根据 Agent 的提示完成必要操作。
### 结果文件在哪里查看?
请根据 Agent 返回的结果说明查看生成文件、任务记录或处理报告。
## 给技能开发者的填写提醒
复制这个模板创建新技能后,请把本文中的【占位内容】替换成真实业务说明。
请始终从用户视角描述技能价值,不要把实现方式、代码文件、测试方法、发布流程写在这里。
技术实现细节应写入开发文档,而不是用户说明页。

View File

@@ -166,8 +166,10 @@
### Python 包依赖requirements.txt
- 本技能若需额外 Python 三方包,写入根目录 `requirements.txt`
- 匠厂宿主安装/更新技能后,会安装到 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`
- 本技能若需**特有** Python 三方包,写入根目录 `requirements.txt`
- `jiangchang-platform-kit``playwright` 等公共能力由宿主共享 runtime 提供,**不要**写入技能 requirements
- `SKILL.md``platform_kit_min_version` 是运行契约/兼容性声明,供宿主校验,**不是** pip 依赖声明。
- 匠厂宿主安装/更新技能后,会将技能 requirements 安装到 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`
- 版本须尽量收窄(如 `requests>=2.31.0,<3`),避免共享 venv 冲突。
- **不要**把系统运行时组件(如 Microsoft Visual C++ Redistributable写进 requirements.txt。

View File

@@ -28,44 +28,73 @@
## 1. 浏览器(标准已成熟)
**选型Playwright + 系统 Chrome/Edge。** 这是团队验证最充分的一条线(参考 `1688-scrape-contacts``disburse-payroll-icbc`)。
**选型Playwright + 系统 Chrome/Edge。** 这是团队验证最充分的一条线(标杆技能 `1688-scrape-contacts``receive-order` 已落地)。
### 1.0 生产路径(优先)
| 项 | 标准 |
|----|------|
| 驱动 | PlaywrightPython |
| 浏览器 | 系统已装 Chrome/Edge`channel="chrome"`**不用内置 Chromium无需 `playwright install`** |
| 登录态 | `launch_persistent_context` + 用户 Profile 目录(由 account-manager 下发) |
| 模式 | 默认有头;`OPENCLAW_BROWSER_HEADLESS=1` 转无头 |
| 反反爬 | stealth 指纹注入 + 拟人输入/鼠标/延迟 + 滑块停下等人工(见下) |
| 浏览器 | **优先系统 Chrome/Edge** + `launch_persistent_context``channel="chrome"` 或 Edge不用内置 Chromium**技能内不要 `playwright install`** |
| 登录态 | 持久 Profile 目录账号、profile、lease **统一走 account-manager**(或对应兄弟技能),不硬编码密码 |
| CDP | **仅作诊断 / 桌面宿主类场景****不要**作为强风控站点的默认生产路径 |
| 行为 | 模拟真实用户:真实点击、键盘、鼠标、地址栏输入;**不要**拼接搜索结果 URL、DOM 注入、`el.value=`、JS 跳转 |
| 模式 | 默认有头 `OPENCLAW_BROWSER_HEADLESS=0`;无头仅 CI |
| 反检测 | stealth 默认开 `OPENCLAW_PLAYWRIGHT_STEALTH=1`(见 1.1 |
### 1.1 防反爬三件套(必做)
### 1.1 Playwright 启动标准
1. **指纹淡化stealth**:启动时 `add_init_script` 注入,抹平自动化特征——
`navigator.webdriver=undefined``chrome.runtime``permissions.query``plugins``languages``hardwareConcurrency``deviceMemory`、WebGL vendor
启动参数 `--disable-blink-features=AutomationControlled` + `ignore_default_args=["--enable-automation"]`
开关:`OPENCLAW_PLAYWRIGHT_STEALTH`(默认开)。
2. **拟人操作**
- 输入:逐字符 `keyboard.type(delay=90~240ms)`**先真实点击聚焦再输入,绝不 `el.value=`**
- 鼠标:贝塞尔曲线轨迹移动 + 微抖动;进场随机晃动 2~4 次
- 导航:用真实点击触发,**不要 `window.location.href=` / `el.click()` 这类 JS 跳转/点击**。
- 翻页:真实点击翻页箭头,注意排除禁用态选择器。
- 延迟:每步之间 `random_delay`
3. **风控页 = 停下等人工**:检测到滑块/拦截URL 含 `punish/x5sec/baxia/tmd`,或 DOM 命中 `#nocaptcha/.nc-container/punish-component` 等)→ 抛 `ERROR:CAPTCHA_NEED_HUMAN`,轮询等人工通过或超时,**不自动操作滑块**。
1. **默认有头**`OPENCLAW_BROWSER_HEADLESS=0``.env.example` 默认值)。
2. **stealth 默认开**`OPENCLAW_PLAYWRIGHT_STEALTH=1`;通过 `add_init_script` 注入指纹淡化脚本
3. **不要在技能里自行安装 playwright**;由宿主共享 runtime 提供
4. **不要默认传 `--no-sandbox`**(除非特定容器环境且已评估风险)。
5. **不要默认传 `--disable-blink-features=AutomationControlled`**platform-kit stealth 已覆盖,额外 flag 可能适得其反。
6. **可以** `ignore_default_args=["--enable-automation"]`platform-kit `launch_persistent_browser` 已处理)
7. **强风控平台**:优先真实点击、键盘、鼠标、地址栏、持久 profile**不要**直接拼接搜索结果 URL 或 DOM 注入
> 参考实现:`1688-scrape-contacts/scripts/service/anti_detect.py`、`util/playwright_stealth.py`、`service/login_service.py`(验证码等待)。已提升为共享库 `jiangchang_skill_core.rpa`platform-kit
指纹淡化stealth典型项`navigator.webdriver=undefined``chrome.runtime``permissions.query``plugins``languages` 等。共享实现见 `jiangchang_skill_core.rpa`platform-kit **>= 1.0.13**)。
### 1.2 安装
**拟人操作**(必做):
- 输入:逐字符 `keyboard.type(delay=90~240ms)`**先真实点击聚焦再输入,绝不 `el.value=`**。
- 鼠标:贝塞尔曲线轨迹 + 微抖动;进场随机晃动。
- 导航:用真实点击触发,**不要 `window.location.href=` / JS 点击跳转**。
- 翻页:真实点击翻页控件,注意排除禁用态。
- 延迟:每步之间 `random_delay``.env``STEP_DELAY_MIN/MAX`)。
### 1.2 页面启动标准
优先级(高 → 低):
1. **launch args `start_url`**:浏览器启动即打开目标首页(推荐生产路径)。
2. **profile preferences**:作为兜底(例如 homepage / startup URLs
3. **程序化导航 `page.goto`**:仅最后兜底;强风控站点慎用裸 `goto` 深链。
**生产路径不要依赖 CDP 接管现有页面**CDP 仅用于诊断或桌面宿主已打开浏览器的场景。
### 1.3 HITL / 验证码
- 自动处理失败时**允许等待人工**滑块、短信、人脸、U盾、动态口令等。
- 等待人工验证时必须有:**字幕 step**(用户可见动作说明)、**结构化日志**、**超时**`HUMAN_WAIT_TIMEOUT`)。
- 检测到风控页URL/DOM 特征)→ 抛 `ERROR:CAPTCHA_NEED_HUMAN`,轮询等待或超时,**不要强行绕过平台安全机制**。
- **不要自动操作滑块**。
> 共享库:`jiangchang_skill_core.rpa.wait_for_captcha_pass`;参考 `scripts/service/example_browser_rpa.py`。
### 1.4 安装
生产/宿主运行:`playwright` 由共享 runtime 提供,无需在技能内 `pip install`
独立本地开发环境若缺少 Python 包 `playwright` 时,可手动执行:
```bash
pip install playwright # 仅装;用系统 Chrome无需 playwright install chromium
pip install playwright # 仅独立开发补装;用系统 Chrome无需 playwright install chromium
```
### 引用方式
共享实现位于 `jiangchang-platform-kit``jiangchang_skill_core.rpa`;模板通过 vendor 拷贝引用
共享实现位于宿主共享 runtime 安装的 `jiangchang-platform-kit``jiangchang_skill_core.rpa`)。复制后的业务技能直接 import**技能仓库不得保留 rpa 公共代码副本**
```python
# scripts/jiangchang_skill_core/rpa/ — vendored from platform-kit
from jiangchang_skill_core.rpa import (
launch_persistent_browser,
anti_detect,
@@ -76,6 +105,9 @@ from jiangchang_skill_core.rpa import (
from jiangchang_skill_core.rpa.stealth import stealth_enabled, STEALTH_INIT_SCRIPT
```
- `RpaVideoSession` 来自 platform-kit **>= 1.0.13**ffmpeg、背景音乐、media-assets 由 platform-kit 统一解析;已提供前置/后置缓冲、字幕、TTS 旁白、背景音乐循环、结尾淡出。
- `health` 对上述资源做只读诊断,不下载、不修复。
完整示例见 `scripts/service/example_browser_rpa.py``scripts/service/example_adapter/sim_rpa.py`
---
@@ -155,16 +187,49 @@ skill 退出/抛错统一用 `ERROR:` 前缀 + 稳定码,方便宿主与上层
---
## 5. 存证规范
## 5. 存证与录屏规范
### 5.1 截图存证
- **路径**`{CLAW_DATA_ROOT}/{user}/{slug}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.png`
- **失败必截图**:受 `OPENCLAW_ARTIFACTS_ON_FAILURE`(默认开)控制。
- **运行录屏成片**RPA 型技能默认 `OPENCLAW_RECORD_VIDEO=1`,表示录屏、中文字幕、背景音乐、最终 MP4 **一体开启**。由 service 核心中的 `RpaVideoSession` 统一触发(见 `jiangchang_skill_core.rpa.video_session`),不在 CLI/E2E 入口单独实现
- **最终视频**`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
- **中间产物**`{skill_data_dir}/rpa-artifacts/{batch_id}/`raw-video、字幕、失败截图等**不放最终 MP4**。
- Playwright 使用 `record_video_dir` 指向 `rpa-artifacts/{batch_id}/raw-video/`
- **Playwright 不负责录屏**,仅浏览器自动化
- **常见 tag**`before_submit` / `after_submit` / `captcha` / `login_fail` / `error`
### 5.2 RPA 视频 step 标准
字幕是**用户可见动作说明**不是技术日志。step 要贴近真实动作,不要只在大流程入口打点。
推荐关键动作示例(按业务裁剪):
- 启动浏览器
- 打开首页
- 检查登录状态
- 定位输入框
- 输入关键词xxx
- 点击搜索
- 等待结果
- 打开详情页
- 提取信息
- 写入结果
- 任务完成
技术诊断、重复跳过、DB 写入可以显示但**通常不需要旁白**。
`title` / `closing_title` 必须由 skill 传入**中文业务文案**(如「开始执行示例任务」「示例任务执行完成」)。
### 5.3 录屏成片标准
- RPA skill 默认 `OPENCLAW_RECORD_VIDEO=1`
- 使用 platform-kit 的 **`RpaVideoSession`****skill 不要自行合成视频**(不要自己调 ffmpeg 拼 MP4
- `OPENCLAW_RECORD_VIDEO=0` 时 session 无副作用(不启 ffmpeg、不写字幕文件
- **ffmpeg 是唯一录屏器**Windows`gdigrab` + `desktop`)。
- **最终视频**`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
- **中间产物**`rpa-artifacts/{batch_id}/capture.mp4``subtitles/``logs/` 等。
- 任务完成后 CLI / `result_summary` 应包含:`video_path``raw_video``video_log``video_warnings``music_path``voiceover_path``audio_warnings`(见 `scripts/service/task_run_support.py`)。
模板最小示范见 `scripts/service/task_service.py``_run_template_demo()`
---
## 相关文档

View File

@@ -1,5 +1,61 @@
# 运行时约定
## 共享 Python Runtime
**skill-template** 及复制出的新技能,公共能力均来自宿主匠厂安装的共享 Python Runtime`jiangchang-platform-kit>=1.0.13` 及其传递依赖,含 `playwright`)。`jiangchang_skill_core` **不得**在技能仓库内 vendored应由共享 venv 的 site-packages 提供。
技能根目录 `requirements.txt` **只声明技能特有依赖****不要**重复声明 `jiangchang-platform-kit``playwright``SKILL.md``platform_kit_min_version` 是运行契约,**不是** pip 依赖声明。
| 场景 | 推荐做法 |
|------|----------|
| 日常运行 | 由宿主匠厂触发技能 |
| 手工排查 / 测试机 | 使用共享 runtime 的 `python.exe` 执行 `scripts/main.py` |
| **不推荐** | 在技能目录内 `uv run python …` — 可能创建临时 venv加载不到宿主共享 runtime |
占位命令(路径因环境而异,勿写死本机目录):
```text
Windows:
{JIANGCHANG_DATA_ROOT}\python-runtime\.venv\Scripts\python.exe {baseDir}\scripts\main.py health
通用:
<shared-python> {baseDir}/scripts/main.py health
```
`<shared-python>` 通常位于 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。数据根由宿主注入;开发模式下也可能通过 `CLAW_DATA_ROOT` 解析(见 `jiangchang_skill_core.runtime_env`)。
## Runtime 诊断platform-kit 1.0.13+
`health` 命令通过 **`jiangchang_skill_core.collect_runtime_diagnostics`** 输出共享 runtime 诊断,**不在技能内重复实现**。典型字段:
- `skill_slug``python_executable``platform_kit_version``platform_kit_min_version`
- `jiangchang_skill_core_file` — 若从技能目录加载会输出 `runtime_issue[warning]`
- `CLAW_DATA_ROOT` / `JIANGCHANG_DATA_ROOT` / `resolved_data_root`
- `media_assets_root``ffmpeg_path``background_music_*`
- `record_video_enabled``runtime_issue[warning|error]`
`health` 是**只读诊断****不会**触发 media-assets 下载或修复。并补充 `env_path` / `env_exists` / `example_path` 配置路径字段。
## 配置 bootstrap
- 仓库内 `.env.example` 是配置模板(单一事实来源)。
- 用户实际 `.env``{CLAW_DATA_ROOT}/{CLAW_USER_ID}/{skill_slug}/.env`
- `scripts/main.py``cli.app.main()` 启动时调用 `util.config_bootstrap.bootstrap_skill_config()`
- 配置优先级:**进程环境变量** > **用户 `.env`** > **`.env.example` 默认值**。
- 公共 `config` / `merge_missing_env_keys` 来自共享 runtime 的 `jiangchang-platform-kit>=1.0.13`**不得** vendored `scripts/jiangchang_skill_core/`
## media-assets / ffmpeg / 背景音乐
背景音乐、ffmpeg 等共享资源由 platform-kit 统一解析,默认路径:
```text
{JIANGCHANG_DATA_ROOT}/shared/media-assets
```
(与 `CLAW_DATA_ROOT` 解析规则一致,由宿主注入。)
RPA 录屏成片(`RpaVideoSession`、ffmpeg 路径、背景音乐探测均由 platform-kit 提供;技能只 import不重复实现。
## 目录结构
新技能建议采用以下根目录结构:
@@ -10,14 +66,13 @@
- `tests/`
- `evals/`
## `scripts/` 分层
**标准 `scripts/` 分层(不含 `jiangchang_skill_core/` 副本):**
- `scripts/main.py`:唯一 CLI 入口
- `scripts/cli/`:参数解析与命令分发
- `scripts/db/`SQLite 或本地持久化层
- `scripts/service/`:业务用例与外部交互
- `scripts/util/`:通用工具、常量、日志、路径
- `scripts/jiangchang_skill_core/`:运行时与统一日志副本
- `scripts/util/`:通用工具、常量、日志、路径薄封装
## 数据路径
@@ -33,6 +88,12 @@
{...}/{skill_slug}.db
```
## 明确禁止
- **不得** vendored `scripts/jiangchang_skill_core/`
- **不得**在技能内本地重复实现 `RuntimeDiagnostics` / `collect_runtime_diagnostics`
- **不得**写死开发机绝对路径作为运行时约定
## 测试时的运行时隔离
本模板的 `tests/_support.IsolatedDataRoot` 会在测试期间同时设置:

View File

@@ -22,7 +22,8 @@
必跑套件要像一个紧张的守门员:**快、确定、离线**。典型覆盖:
- CLI导入 [`cli.app`](../scripts/cli/app.py) 走解析链路、`health` / `version` / `logs` / `log-get` 冒烟;
- CLI导入 [`cli.app`](../scripts/cli/app.py) 走解析链路、`health`runtime diagnostics/ `version` / `logs` / `log-get` 冒烟;
- 架构守护:无 `scripts/jiangchang_skill_core/``platform-kit>=1.0.13` 导入来源、文档/runtime 标准(见 `test_platform_import.py` 等);
- **真实 subprocess**[`tests/test_entrypoint_subprocess.py`](../tests/test_entrypoint_subprocess.py) 再调用一遍 `python scripts/main.py`,防路径漂移;
- 运行时:`runtime_paths`**`CLAW_*` / `JIANGCHANG_*` 并发兜底;
- `SKILL.md` YAML slug vs [`constants.SKILL_SLUG`](../scripts/util/constants.py)
@@ -178,7 +179,53 @@ Golden fixture 流程同理([`tests/samples/test_golden_cases.py.sample`](../t
---
## 延伸阅读
## 11. pytest 收集卫生标准
模板根目录提供 `pytest.ini`(或等价配置),约束:
- **只收集** `test_*.py` / `*_test.py`
- **不让** `.txt`、结果文件、日志文件被 pytest 误收集
- `norecursedirs` 排除 `integration/``desktop/``samples/``fixtures/``artifacts/``diagnostics/`
测试结果文件**不要**放在 `tests/` 根目录;应放 `tests/artifacts/``tests/diagnostics/` 并加入 `.gitignore`
默认 `python tests/run_tests.py` 仍只发现 `tests/` 根目录一层 `test_*.py`(不递归子目录),与 pytest 策略一致。
---
## 12. RPA / video 测试标准
- `RpaVideoSession` 调用**不跑真实 ffmpeg**:单测中使用 `unittest.mock` patch session 或设 `OPENCLAW_RECORD_VIDEO=0`
- 断言 `title` / `closing_title` 是**中文**业务文案。
- 断言 video artifact 会进入 `result_summary``video_path``raw_video``video_log` 等)。
- 断言 step 文案贴近用户动作不是技术日志如「准备执行示例任务」而非「enter cmd_run」
参考 `tests/test_video_service.py`
---
## 13. 宿主 E2E 标准
- 使用 `jiangchang_desktop_sdk.e2e_helpers`(见 `tests/desktop/``.sample`)。
- **不伪造**用户数据目录;通过宿主 IPC 获取真实 `skill_data_dir`
- E2E **不自动生成**真实密钥或生产凭证。
---
## 14. 测试前检查清单(模板守护)
复制新 skill 或修改模板后,确认:
- [ ] `requirements.txt` **不含** `jiangchang-platform-kit` / `playwright`
- [ ]`scripts/jiangchang_skill_core/` vendored 副本
- [ ] `platform_kit_min_version` **>= 1.0.13**`SKILL.md` + `constants.py`
- [ ] `health` 能输出 `platform_kit_version_ok`(或等价诊断行)
- [ ] `config-path` 可输出用户 `.env` 路径 JSON
- [ ] `pytest.ini` 存在且 `python_files` 只收集 `test_*.py` / `*_test.py`
- [ ] pytest **不会**误收集 `tests/` 下的 `.txt` / 日志 / 结果文件
- [ ] `python tests/run_tests.py -v` 全部通过
---
- [`tests/README.md`](../tests/README.md) — 表格、变量与目录细则
- [`references/DEVELOPMENT.md`](DEVELOPMENT.md) §14.5 — 测试驱动的开发顺序

View File

@@ -12,7 +12,7 @@ Set-StrictMode -Version Latest
$ErrorActionPreference = "Stop"
$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$sharedScript = Join-Path $scriptDir "..\jiangchang-platform-kit\tools\release.ps1"
$sharedScript = Join-Path $scriptDir "..\..\jiangchang-platform-kit\tools\release.ps1"
$sharedScript = [System.IO.Path]::GetFullPath($sharedScript)
if (-not (Test-Path $sharedScript)) {

View File

@@ -1,7 +1 @@
# Skill-specific Python dependencies.
# Keep this file minimal.
# Platform-provided shared dependencies are installed by Jiangchang runtime.
# Add only dependencies required by this skill.
#
# Browser RPA: uses system Chrome/Edge; no `playwright install chromium` needed.
playwright>=1.42.0,<2
# 公共依赖由宿主共享 runtime 提供;这里只声明技能特有 Python 依赖。

View File

@@ -7,12 +7,14 @@ import sys
from typing import List, Optional
from service.task_service import (
cmd_config_path,
cmd_health,
cmd_log_get,
cmd_logs,
cmd_run,
cmd_version,
)
from util.config_bootstrap import bootstrap_skill_config
from util.argparse_zh import ZhArgumentParser
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.logging_config import get_skill_logger, setup_skill_logging
@@ -55,6 +57,7 @@ def _print_full_usage() -> None:
print(" python main.py logs [--limit N] [--status s] [--task-type t] [--target-id tid]")
print(" python main.py log-get <log_id>")
print(" python main.py health")
print(" python main.py config-path")
print(" python main.py version")
@@ -88,6 +91,9 @@ def build_parser() -> ZhArgumentParser:
sp = sub.add_parser("health", help="健康检查")
sp.set_defaults(handler=lambda _a: cmd_health())
sp = sub.add_parser("config-path", help="输出用户 .env 与模板路径")
sp.set_defaults(handler=lambda _a: cmd_config_path())
sp = sub.add_parser("version", help="版本信息(JSON)")
sp.set_defaults(handler=lambda _a: cmd_version())
return p
@@ -95,12 +101,15 @@ def build_parser() -> ZhArgumentParser:
def main(argv: Optional[List[str]] = None) -> int:
argv = argv if argv is not None else sys.argv[1:]
bootstrap_skill_config()
setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
get_skill_logger().info("cli_start argv=%s", sys.argv)
if not argv:
_print_full_usage()
return 1
if len(argv) == 2 and argv[0] not in {"run", "logs", "log-get", "health", "version", "-h", "--help"}:
if len(argv) == 2 and argv[0] not in {
"run", "logs", "log-get", "health", "config-path", "version", "-h", "--help"
}:
return cmd_run(target=argv[0], input_id=argv[1])
parser = build_parser()
args = parser.parse_args(argv)

View File

@@ -1,16 +0,0 @@
# Vendored from jiangchang-platform-kit/sdk/jiangchang_skill_core/ — keep runtime_env + unified_logging + config + rpa in sync.
from .config import ensure_env_file, get, get_bool, get_float, get_int
try:
from . import rpa
except ImportError:
rpa = None # type: ignore[assignment,misc]
__all__ = [
"ensure_env_file",
"get",
"get_bool",
"get_float",
"get_int",
"rpa",
]

View File

@@ -1,126 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""三级优先级配置读取 + 首次 .env 落盘(进程 env > 数据目录 .env > .env.example"""
from __future__ import annotations
import os
import shutil
from typing import Any
from .runtime_env import get_data_root, get_user_id
_skill_slug: str | None = None
_example_path: str | None = None
_env_file_path: str | None = None
_user_env_cache: dict[str, str] | None = None
_example_defaults_cache: dict[str, str] | None = None
def _parse_env_file(path: str) -> dict[str, str]:
"""标准库手写 .env 解析(不引 python-dotenv"""
result: dict[str, str] = {}
if not path or not os.path.isfile(path):
return result
with open(path, encoding="utf-8") as f:
for raw_line in f:
line = raw_line.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
value = value.strip()
if not key:
continue
if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
value = value[1:-1]
elif " #" in value:
value = value.split(" #", 1)[0].strip()
result[key] = value
return result
def _get_user_env() -> dict[str, str]:
global _user_env_cache
if _user_env_cache is not None:
return _user_env_cache
if _env_file_path and os.path.isfile(_env_file_path):
_user_env_cache = _parse_env_file(_env_file_path)
else:
_user_env_cache = {}
return _user_env_cache
def _get_example_defaults() -> dict[str, str]:
global _example_defaults_cache
if _example_defaults_cache is not None:
return _example_defaults_cache
if _example_path and os.path.isfile(_example_path):
_example_defaults_cache = _parse_env_file(_example_path)
else:
_example_defaults_cache = {}
return _example_defaults_cache
def reset_cache() -> None:
"""测试用:清空解析缓存。"""
global _user_env_cache, _example_defaults_cache
_user_env_cache = None
_example_defaults_cache = None
def ensure_env_file(skill_slug: str, example_path: str) -> str:
"""首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env已存在不覆盖返回路径。"""
global _skill_slug, _example_path, _env_file_path
_skill_slug = skill_slug
_example_path = os.path.abspath(example_path)
dest_dir = os.path.join(get_data_root(), get_user_id(), skill_slug)
os.makedirs(dest_dir, exist_ok=True)
dest = os.path.join(dest_dir, ".env")
_env_file_path = dest
if not os.path.isfile(dest) and os.path.isfile(_example_path):
shutil.copy2(_example_path, dest)
reset_cache()
return dest
def get(key: str, default: Any = None) -> str | None:
"""进程环境变量 > 数据目录 .env > .env.example 默认值。"""
env_val = os.environ.get(key)
if env_val is not None and env_val != "":
return env_val
user_val = _get_user_env().get(key)
if user_val is not None and user_val != "":
return user_val
example_val = _get_example_defaults().get(key)
if example_val is not None and example_val != "":
return example_val
return default
def get_bool(key: str, default: bool = False) -> bool:
val = get(key)
if val is None:
return default
return str(val).strip().lower() in ("1", "true", "yes", "on")
def get_float(key: str, default: float) -> float:
val = get(key)
if val is None:
return default
try:
return float(val)
except (TypeError, ValueError):
return default
def get_int(key: str, default: int) -> int:
val = get(key)
if val is None:
return default
try:
return int(val)
except (TypeError, ValueError):
return default

View File

@@ -1,66 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""浏览器 RPA 共享库stealth、拟人操作、启动封装、HITL 等待、失败存证。"""
from __future__ import annotations
from . import anti_detect, artifacts, errors, human_login, stealth
try:
from .browser import launch_persistent_browser
except ImportError:
launch_persistent_browser = None # type: ignore[misc, assignment]
try:
from .video_session import RpaVideoSession
except ImportError:
RpaVideoSession = None # type: ignore[misc, assignment]
__all__ = [
"anti_detect",
"artifacts",
"errors",
"human_login",
"stealth",
"launch_persistent_browser",
"RpaVideoSession",
# re-export common anti_detect helpers
"random_delay",
"human_delay_short",
"human_delay_page",
"human_delay_batch",
"human_mouse_move",
"human_mouse_wiggle",
"human_click",
"human_type",
"human_scroll",
# human_login
"wait_for_captcha_pass",
"wait_for_login",
"DEFAULT_CAPTCHA_MARKERS",
# artifacts
"artifact_path",
"capture_failure",
# stealth
"STEALTH_INIT_SCRIPT",
"stealth_enabled",
"persistent_context_launch_parts",
]
from .anti_detect import (
human_click,
human_delay_batch,
human_delay_page,
human_delay_short,
human_mouse_move,
human_mouse_wiggle,
human_scroll,
human_type,
random_delay,
)
from .artifacts import artifact_path, capture_failure
from .human_login import DEFAULT_CAPTCHA_MARKERS, wait_for_captcha_pass, wait_for_login
from .stealth import (
STEALTH_INIT_SCRIPT,
persistent_context_launch_parts,
stealth_enabled,
)

View File

@@ -1,216 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""防反爬:随机延迟、人工鼠标轨迹模拟、拟人滚动。
所有延迟使用 random.uniform() + asyncio.sleep()。
鼠标移动使用三次贝塞尔曲线 + 随机抖动,避免机械直线轨迹。
"""
from __future__ import annotations
import asyncio
import math
import random
from typing import Optional
# ── 随机延迟 ────────────────────────────────────────────────
async def random_delay(min_s: float, max_s: float) -> float:
"""随机延迟 min_s ~ max_s 秒,返回实际延迟秒数。"""
delay = random.uniform(min_s, max_s)
await asyncio.sleep(delay)
return delay
async def human_delay_short() -> float:
"""页面内操作间短延迟 0.5~2 秒(点击/滚动)。"""
return await random_delay(0.5, 2.0)
async def human_delay_page(min_s: float = 2, max_s: float = 8) -> float:
"""页面间/店铺间切换延迟(可配置范围)。"""
return await random_delay(min_s, max_s)
async def human_delay_batch(min_s: float = 5, max_s: float = 15) -> float:
"""批次间/翻页间长延迟。"""
return await random_delay(min_s, max_s)
# ── 鼠标轨迹 ────────────────────────────────────────────────
async def human_mouse_move(
page,
target_x: float,
target_y: float,
steps: Optional[int] = None,
) -> None:
"""模拟人工鼠标移动到目标坐标。
使用三次贝塞尔曲线生成 S/C 形路径,中间控制点随机偏移,
每步添加微小随机抖动,避免完美直线移动。
Args:
page: Playwright Page 对象。
target_x: 目标 X 坐标。
target_y: 目标 Y 坐标。
steps: 移动步数None 时按距离自动计算)。
"""
if steps is None:
dist = math.hypot(target_x, target_y)
steps = max(8, int(dist / 15))
# 随机起始点(模拟鼠标从页面其他位置移动过来)
start_x = target_x + random.uniform(-300, 300)
start_y = target_y + random.uniform(-200, 200)
# 两个贝塞尔控制点(制造弯曲路径)
dx = target_x - start_x
dy = target_y - start_y
cp1_x = start_x + dx * random.uniform(0.2, 0.5) + random.uniform(-120, 120)
cp1_y = start_y + dy * random.uniform(0.1, 0.4) + random.uniform(-100, 100)
cp2_x = start_x + dx * random.uniform(0.5, 0.8) + random.uniform(-100, 100)
cp2_y = start_y + dy * random.uniform(0.4, 0.7) + random.uniform(-80, 80)
for i in range(steps + 1):
t = i / steps
# 三次贝塞尔: B(t) = (1-t)³·P0 + 3(1-t)²·t·P1 + 3(1-t)·t²·P2 + t³·P3
u = 1 - t
x = (
(u ** 3) * start_x
+ 3 * (u ** 2) * t * cp1_x
+ 3 * u * (t ** 2) * cp2_x
+ (t ** 3) * target_x
)
y = (
(u ** 3) * start_y
+ 3 * (u ** 2) * t * cp1_y
+ 3 * u * (t ** 2) * cp2_y
+ (t ** 3) * target_y
)
# 微小随机抖动
x += random.uniform(-1.5, 1.5)
y += random.uniform(-1.5, 1.5)
await page.mouse.move(x, y)
await asyncio.sleep(random.uniform(0.003, 0.015))
# 最终精确到位
await page.mouse.move(target_x, target_y)
async def human_click(
page,
selector: Optional[str] = None,
*,
x: Optional[float] = None,
y: Optional[float] = None,
) -> None:
"""拟人点击:贝塞尔轨迹移动 → 微延迟 → 点击。
支持两种定位方式:
- selector: CSS 选择器,自动取元素中心附近随机偏移
- x, y: 直接指定坐标(自动加微小随机偏移)
Args:
page: Playwright Page 对象。
selector: CSS 选择器(与 x/y 二选一)。
x: 目标 X 坐标。
y: 目标 Y 坐标。
"""
if selector:
locator = page.locator(selector).first
box = await locator.bounding_box()
if box is None:
raise ValueError(f"无法获取元素边界: {selector}")
target_x = box["x"] + box["width"] * random.uniform(0.3, 0.7)
target_y = box["y"] + box["height"] * random.uniform(0.3, 0.7)
elif x is not None and y is not None:
target_x = x + random.uniform(-3, 3)
target_y = y + random.uniform(-3, 3)
else:
raise ValueError("必须提供 selector 或 (x, y) 坐标")
await human_mouse_move(page, target_x, target_y)
await human_delay_short()
await page.mouse.click(target_x, target_y)
# ── 滚动模拟 ────────────────────────────────────────────────
async def human_scroll(
page, direction: str = "down", amount: Optional[int] = None
) -> None:
"""模拟人工滚动(分段 + 变速 + 微延迟)。
Args:
page: Playwright Page 对象。
direction: "up""down"
amount: 滚动像素总量,默认随机 300~800。
"""
if amount is None:
amount = random.randint(300, 800)
sign = -1 if direction == "up" else 1
remaining = amount
step_count = random.randint(3, 7)
for i in range(step_count):
step = remaining // (step_count - i) + random.randint(-30, 30)
step = max(10, min(step, remaining))
remaining -= step
await page.mouse.wheel(0, sign * step)
await asyncio.sleep(random.uniform(0.05, 0.2))
if remaining > 0:
await page.mouse.wheel(0, sign * remaining)
await human_delay_short()
# ── 进场鼠标随机晃动 ────────────────────────────────────────
async def human_mouse_wiggle(page) -> None:
"""页面进入后随机移动鼠标 2~4 次(贝塞尔轨迹),模拟真人到场环顾。"""
try:
vp = page.viewport_size or {"width": 1280, "height": 800}
except Exception:
vp = {"width": 1280, "height": 800}
w, h = vp.get("width", 1280), vp.get("height", 800)
for _ in range(random.randint(2, 4)):
tx = random.uniform(w * 0.2, w * 0.8)
ty = random.uniform(h * 0.2, h * 0.7)
await human_mouse_move(page, tx, ty)
await asyncio.sleep(random.uniform(0.2, 0.8))
# ── 拟人逐字输入 ────────────────────────────────────────────
async def human_type(page, locator, text: str) -> None:
"""真人式输入:贝塞尔移动到输入框 → 真实点击聚焦 → 逐字符 keyboard.type。
全程使用 isTrusted=true 的可信事件,绝不用 JS 设置 value。
"""
box = await locator.bounding_box()
if box is None:
raise ValueError("human_type: 无法获取输入框边界")
tx = box["x"] + box["width"] * random.uniform(0.3, 0.7)
ty = box["y"] + box["height"] * random.uniform(0.3, 0.7)
await human_mouse_move(page, tx, ty)
await asyncio.sleep(random.uniform(0.15, 0.5))
await page.mouse.click(tx, ty)
await asyncio.sleep(random.uniform(0.2, 0.6))
# 清空已有内容(用键盘,不用 JS
await page.keyboard.press("Control+A")
await page.keyboard.press("Delete")
await asyncio.sleep(random.uniform(0.2, 0.5))
for ch in text:
await page.keyboard.type(ch, delay=random.uniform(90, 240))

View File

@@ -1,29 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""RPA 失败存证路径与截图。"""
from __future__ import annotations
import os
from datetime import datetime
def _artifacts_enabled() -> bool:
v = (os.environ.get("OPENCLAW_ARTIFACTS_ON_FAILURE") or "1").strip().lower()
return v not in ("0", "false", "no", "off")
def artifact_path(data_dir: str, batch_id: str, tag: str, ext: str = "png") -> str:
"""返回 {data_dir}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.{ext}"""
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
directory = os.path.join(data_dir, "rpa-artifacts", batch_id)
os.makedirs(directory, exist_ok=True)
return os.path.join(directory, f"{tag}_{ts}.{ext}")
async def capture_failure(page, data_dir: str, batch_id: str, tag: str) -> str:
"""受 OPENCLAW_ARTIFACTS_ON_FAILURE 控制,截图并返回路径;禁用时返回空字符串。"""
if not _artifacts_enabled():
return ""
path = artifact_path(data_dir, batch_id, tag, ext="png")
await page.screenshot(path=path, full_page=True)
return path

View File

@@ -1,62 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""统一 persistent context 启动封装。"""
from __future__ import annotations
import os
from typing import TYPE_CHECKING
from ..runtime_env import find_chrome_executable
from .stealth import (
STEALTH_INIT_SCRIPT,
persistent_context_launch_parts,
stealth_enabled,
)
if TYPE_CHECKING:
from playwright.async_api import BrowserContext
async def launch_persistent_browser(
playwright,
*,
profile_dir: str,
channel: str = "chrome",
executable_path: str | None = None,
headless: bool | None = None,
extra_args: list[str] | None = None,
record_video_dir: str | None = None,
) -> "BrowserContext":
"""用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。
headless=None 时读 OPENCLAW_BROWSER_HEADLESS。
record_video_dir 非空则开录屏。
"""
if headless is None:
v = (os.environ.get("OPENCLAW_BROWSER_HEADLESS") or "0").strip().lower()
headless = v in ("1", "true", "yes", "on")
chrome = executable_path or find_chrome_executable()
if not chrome:
raise RuntimeError(
"ERROR:MISSING_BROWSER 未检测到 Chrome 或 Edge 浏览器,请安装后重试。"
)
args, ignore = persistent_context_launch_parts(extra_args=extra_args)
launch_kwargs: dict = dict(
user_data_dir=profile_dir,
headless=headless,
executable_path=chrome,
locale="zh-CN",
no_viewport=True,
args=args,
)
if ignore is not None:
launch_kwargs["ignore_default_args"] = ignore
if record_video_dir:
launch_kwargs["record_video_dir"] = record_video_dir
context = await playwright.chromium.launch_persistent_context(**launch_kwargs)
if stealth_enabled():
await context.add_init_script(STEALTH_INIT_SCRIPT)
return context

View File

@@ -1,20 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""RPA 场景统一错误码stdout / 异常消息前缀)。"""
ERR_REQUIRE_LOGIN = "ERROR:REQUIRE_LOGIN"
ERR_LOGIN_TIMEOUT = "ERROR:LOGIN_TIMEOUT"
ERR_CAPTCHA_NEED_HUMAN = "ERROR:CAPTCHA_NEED_HUMAN"
ERR_RATE_LIMITED = "ERROR:RATE_LIMITED"
ERR_MISSING_BROWSER = "ERROR:MISSING_BROWSER"
ERR_DEVICE_NOT_READY = "ERROR:DEVICE_NOT_READY"
ERR_WINDOW_NOT_FOUND = "ERROR:WINDOW_NOT_FOUND"
__all__ = [
"ERR_REQUIRE_LOGIN",
"ERR_LOGIN_TIMEOUT",
"ERR_CAPTCHA_NEED_HUMAN",
"ERR_RATE_LIMITED",
"ERR_MISSING_BROWSER",
"ERR_DEVICE_NOT_READY",
"ERR_WINDOW_NOT_FOUND",
]

View File

@@ -1,97 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""验证码/登录 HITL 等待(站点无关通用部分)。"""
from __future__ import annotations
import asyncio
DEFAULT_CAPTCHA_MARKERS = (
"punish",
"baxia",
"nc_login",
"tmd",
"x5sec",
"安全验证",
"请按住滑块",
"拖动到最右边",
)
async def wait_for_captcha_pass(
page,
captcha_markers: tuple[str, ...] = DEFAULT_CAPTCHA_MARKERS,
timeout_sec: int = 120,
) -> bool:
"""
检测到验证码/拦截页时,暂停等待用户手工完成验证。
每 3 秒轮询一次页面状态,直到页面不再是拦截状态或超时。
Returns:
True — 验证码已通过,可以继续
False — 超时,用户未在规定时间内完成验证
"""
print(
f"[验证码] 检测到安全拦截,请在浏览器中手工完成验证。"
f"等待最多 {timeout_sec} 秒..."
)
elapsed = 0
while elapsed < timeout_sec:
await asyncio.sleep(3)
elapsed += 3
try:
url = page.url or ""
text = ""
try:
text = await page.locator("body").inner_text(timeout=2_000)
except Exception:
pass
haystack = (url + " " + text[:1000]).lower()
still_blocked = any(m.lower() in haystack for m in captcha_markers)
if not still_blocked:
print(f"[验证码] 已通过!({elapsed}s 后检测到正常页面)")
return True
except Exception:
pass
print(f"[验证码] 超时 {timeout_sec}s用户未完成验证")
return False
async def wait_for_login(
page,
success_selectors: list[str],
timeout_sec: int = 180,
) -> bool:
"""
轮询等待登录成功标志出现(站点专属选择器由调用方传入)。
Args:
page: Playwright Page 对象。
success_selectors: 登录成功后应出现的 CSS 选择器列表。
timeout_sec: 最长等待秒数。
Returns:
True — 检测到登录成功标志
False — 超时
"""
print(
f"[登录] 请在浏览器中完成登录。等待最多 {timeout_sec} 秒,"
"期间程序不会操作页面..."
)
elapsed = 0
while elapsed < timeout_sec:
await asyncio.sleep(2)
elapsed += 2
for selector in success_selectors:
try:
el = await page.query_selector(selector)
if el is not None:
print("[登录] 检测到登录成功!")
return True
except Exception:
pass
print(f"[登录] 超时 {timeout_sec}s未完成登录")
return False

View File

@@ -1,67 +0,0 @@
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
"""淡化 Playwright 自动化指纹,降低 baxia/x5sec 风控命中率。
关闭:环境变量 OPENCLAW_PLAYWRIGHT_STEALTH=0或 false/off/no
"""
from __future__ import annotations
import os
STEALTH_INIT_SCRIPT = """
(() => {
try { Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); } catch (e) {}
try {
window.chrome = window.chrome || {};
window.chrome.runtime = window.chrome.runtime || {};
} catch (e) {}
try {
const orig = navigator.permissions && navigator.permissions.query;
if (orig) {
navigator.permissions.query = (p) =>
p && p.name === 'notifications'
? Promise.resolve({ state: Notification.permission })
: orig(p);
}
} catch (e) {}
try {
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
} catch (e) {}
try {
Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] });
} catch (e) {}
try {
Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 });
} catch (e) {}
try {
Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 });
} catch (e) {}
try {
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function (p) {
if (p === 37445) return 'Intel Inc.';
if (p === 37446) return 'Intel Iris OpenGL Engine';
return getParameter.call(this, p);
};
} catch (e) {}
})();
"""
def stealth_enabled() -> bool:
v = (os.environ.get("OPENCLAW_PLAYWRIGHT_STEALTH") or "1").strip().lower()
return v not in ("0", "false", "no", "off")
def persistent_context_launch_parts(
*, extra_args: list[str] | None = None
) -> tuple[list[str], list[str] | None]:
"""返回 (args, ignore_default_args)。ignore_default_args 为 None 时不要传给 launch。"""
args = ["--start-maximized"]
if extra_args:
args = args + list(extra_args)
if not stealth_enabled():
return args, None
if "--disable-blink-features=AutomationControlled" not in args:
args.append("--disable-blink-features=AutomationControlled")
return args, ["--enable-automation"]

View File

@@ -1,289 +0,0 @@
"""RPA 运行录屏会话Playwright raw video → 字幕 + 背景音乐 → 最终 MP4。"""
from __future__ import annotations
import asyncio
import os
import random
import subprocess
import sys
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from .. import config
# 与 screencast/composer.py 保持一致的字幕样式
_JIANGCHANG_SUBTITLE_STYLE = (
"FontName=Arial,FontSize=14,"
"PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
"Outline=2,Shadow=1,Alignment=2"
)
_DEFAULT_MUSIC_ROOT = os.environ.get(
"MEDIA_ASSETS_ROOT",
r"D:\OpenClaw\client-commons\media-assets",
)
def _record_video_enabled() -> bool:
return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
def _resolve_music_file() -> Optional[Path]:
music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music"
if not music_dir.is_dir():
return None
files = list(music_dir.rglob("*.mp3"))
if not files:
return None
return random.choice(files)
@dataclass
class _StepEntry:
start: float
text: str
duration: float = 4.0
def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
"""根据步骤时间线生成 SRT防重叠"""
MIN_DURATION = 1.0
GAP = 0.1
steps = sorted(entries, key=lambda e: e.start)
if not steps:
steps = [_StepEntry(0.0, "任务执行中", 4.0)]
for i in range(len(steps) - 1):
cur, nxt = steps[i], steps[i + 1]
max_end = nxt.start - GAP
new_dur = max_end - cur.start
if new_dur < MIN_DURATION:
cur.duration = MIN_DURATION
nxt.start = cur.start + MIN_DURATION + GAP
else:
cur.duration = min(cur.duration, new_dur)
def _fmt(s: float) -> str:
h = int(s // 3600)
m = int((s % 3600) // 60)
sec = int(s % 60)
ms = int((s % 1) * 1000)
return f"{h:02d}:{m:02d}:{sec:02d},{ms:03d}"
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
for i, e in enumerate(steps, 1):
f.write(f"{i}\n")
f.write(f"{_fmt(e.start)} --> {_fmt(e.start + e.duration)}\n")
f.write(f"{e.text}\n\n")
def _escape_srt_path(path: str) -> str:
return path.replace("\\", "/").replace(":", "\\:")
def _find_raw_webm(raw_dir: str) -> Optional[Path]:
root = Path(raw_dir)
if not root.is_dir():
return None
webms = list(root.rglob("*.webm"))
if not webms:
return None
return max(webms, key=lambda p: p.stat().st_size)
def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)
except FileNotFoundError:
return False, "ffmpeg not found"
if result.returncode != 0:
err = (result.stderr or result.stdout or "").strip()
return False, err[:500] if err else f"exit {result.returncode}"
return True, ""
def _compose_webm_mp4(
webm_path: Path,
srt_path: Path,
output_path: Path,
*,
music_file: Optional[Path] = None,
music_volume: float = 0.15,
) -> tuple[bool, str]:
output_path.parent.mkdir(parents=True, exist_ok=True)
srt_esc = _escape_srt_path(str(srt_path.resolve()))
subtitle_filter = (
f"subtitles='{srt_esc}'"
f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
)
if music_file and music_file.is_file():
filter_complex = (
f"[0:v]{subtitle_filter}[vout];"
f"[1:a]volume={music_volume},apad[aout]"
)
cmd = [
"ffmpeg", "-y",
"-i", str(webm_path),
"-i", str(music_file),
"-filter_complex", filter_complex,
"-map", "[vout]",
"-map", "[aout]",
"-c:v", "libx264",
"-preset", "fast",
"-crf", "23",
"-c:a", "aac",
"-b:a", "128k",
"-shortest",
"-pix_fmt", "yuv420p",
str(output_path),
]
else:
cmd = [
"ffmpeg", "-y",
"-i", str(webm_path),
"-vf", subtitle_filter,
"-c:v", "libx264",
"-preset", "fast",
"-crf", "23",
"-c:a", "aac",
"-b:a", "128k",
"-pix_fmt", "yuv420p",
str(output_path),
]
return _run_ffmpeg(cmd)
class RpaVideoSession:
"""Service/RPA 层统一录屏会话OPENCLAW_RECORD_VIDEO=1 时启用完整成片流程。"""
def __init__(
self,
*,
skill_slug: str,
skill_data_dir: str,
batch_id: str,
title: str = "",
) -> None:
self.skill_slug = skill_slug
self.skill_data_dir = os.path.abspath(skill_data_dir)
self.batch_id = batch_id
self.title = title
self.enabled = _record_video_enabled()
self.warnings: List[str] = []
self.output_video_path: Optional[str] = None
self.record_video_dir: Optional[str] = None
self.artifacts: Dict[str, Any] = {}
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_batch = "".join(c if c.isalnum() or c in "-_" else "_" for c in batch_id)
self._video_stem = f"{skill_slug}_{ts}_{safe_batch}"
self._videos_dir = os.path.join(self.skill_data_dir, "videos")
self._artifact_root = os.path.join(
self.skill_data_dir, "rpa-artifacts", batch_id
)
self._raw_video_dir = os.path.join(self._artifact_root, "raw-video")
self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
self._logs_dir = os.path.join(self._artifact_root, "logs")
self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt")
self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
self._steps: List[_StepEntry] = []
self._t0: Optional[float] = None
if self.enabled:
os.makedirs(self._videos_dir, exist_ok=True)
os.makedirs(self._raw_video_dir, exist_ok=True)
os.makedirs(self._subtitles_dir, exist_ok=True)
os.makedirs(self._logs_dir, exist_ok=True)
self.record_video_dir = self._raw_video_dir
def step(self, text: str, *, duration: float = 4.0) -> None:
"""记录一步骤字幕(中文说明)。"""
if not self.enabled:
return
if self._t0 is None:
self._t0 = time.monotonic()
elapsed = time.monotonic() - self._t0
self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration))
async def __aenter__(self) -> "RpaVideoSession":
if self.enabled:
self._t0 = time.monotonic()
if self.title:
self.step(self.title, duration=3.0)
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
if self.enabled:
await self.finalize()
async def finalize(self) -> None:
"""在 Playwright context 已 close 后合成最终 MP4失败只记 warning。"""
if not self.enabled:
return
if not self._steps and self.title:
self.step(self.title)
try:
await asyncio.sleep(0.5)
webm = _find_raw_webm(self._raw_video_dir)
if webm is None:
self.warnings.append("no_playwright_raw_video")
self.artifacts = self.to_artifact_dict()
return
entries = list(self._steps)
if entries:
entries[-1].duration = max(entries[-1].duration, 5.0)
_generate_srt(entries, self._srt_path)
music = _resolve_music_file()
if music is None:
self.warnings.append("background_music_missing")
ok, err = _compose_webm_mp4(
webm,
Path(self._srt_path),
Path(self._planned_output),
music_file=music,
)
if not ok and music is not None:
self.warnings.append(f"ffmpeg_with_music_failed: {err}")
ok, err = _compose_webm_mp4(
webm,
Path(self._srt_path),
Path(self._planned_output),
music_file=None,
)
if not ok:
self.warnings.append(f"ffmpeg_compose_failed: {err}")
self.artifacts = self.to_artifact_dict()
return
self.output_video_path = str(Path(self._planned_output).resolve())
self.artifacts = self.to_artifact_dict()
except Exception as exc:
self.warnings.append(f"video_finalize_error: {exc}")
self.artifacts = self.to_artifact_dict()
def to_artifact_dict(self) -> Dict[str, Any]:
return {
"enabled": self.enabled,
"path": self.output_video_path,
"warnings": list(self.warnings),
"batch_id": self.batch_id,
}

View File

@@ -1,109 +0,0 @@
"""
JIANGCHANG_* 数据根与用户目录:解析规则 + 可选本地 CLI 默认值。
模板复制后通常无需大改;如组织环境不同,再按项目实际调整。
"""
from __future__ import annotations
import os
import sys
CLI_LOCAL_DEV_ENABLED = True
DEFAULT_LOCAL_USER_ID = "10032"
WIN_DEFAULT_DATA_ROOT = r"D:\jiangchang-data"
WIN_DEFAULT_JIANGCHANG_APP_ROOT = r"D:\AI\jiangchang"
def platform_default_data_root() -> str:
if sys.platform == "win32":
return WIN_DEFAULT_DATA_ROOT
return os.path.join(os.path.expanduser("~"), ".jiangchang-data")
def get_data_root() -> str:
env = (
os.getenv("CLAW_DATA_ROOT")
or os.getenv("JIANGCHANG_DATA_ROOT")
or ""
).strip()
if env:
return env
return platform_default_data_root()
def get_user_id() -> str:
return (
os.getenv("CLAW_USER_ID")
or os.getenv("JIANGCHANG_USER_ID")
or ""
).strip() or "_anon"
def _looks_like_skills_root(path: str) -> bool:
if not path or not os.path.isdir(path):
return False
# 通过常见基础技能存在与否,判断这是不是 skills 根目录
# 这里列举的是平台基础设施类技能,不包含具体业务技能
for marker in (
"llm-manager",
"skill-template",
):
if os.path.isdir(os.path.join(path, marker)):
return True
return False
def get_skills_root() -> str:
for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
v = (os.getenv(key) or "").strip()
if v:
return os.path.normpath(v)
app = (os.getenv("JIANGCHANG_APP_ROOT") or "").strip()
if sys.platform == "win32" and not app:
app = WIN_DEFAULT_JIANGCHANG_APP_ROOT
if app:
nested = os.path.join(app, "skills")
if _looks_like_skills_root(nested):
return os.path.normpath(nested)
if _looks_like_skills_root(app):
return os.path.normpath(app)
if sys.platform == "win32":
nested = os.path.join(WIN_DEFAULT_JIANGCHANG_APP_ROOT, "skills")
if _looks_like_skills_root(nested):
return os.path.normpath(nested)
if _looks_like_skills_root(WIN_DEFAULT_JIANGCHANG_APP_ROOT):
return os.path.normpath(WIN_DEFAULT_JIANGCHANG_APP_ROOT)
return os.path.normpath(os.path.join(os.path.expanduser("~"), ".openclaw", "skills"))
def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str:
if skill_scripts_dir:
scripts = os.path.abspath(skill_scripts_dir)
skill_root = os.path.dirname(scripts)
inferred = os.path.dirname(skill_root)
if _looks_like_skills_root(inferred):
return os.path.normpath(inferred)
for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
v = (os.getenv(key) or "").strip()
if v:
return os.path.normpath(v)
return get_skills_root()
def apply_cli_local_defaults() -> None:
enabled = CLI_LOCAL_DEV_ENABLED
if not enabled:
v = (os.getenv("JIANGCHANG_CLI_LOCAL_DEV") or "").strip().lower()
enabled = v in ("1", "true", "yes", "on")
if not enabled:
return
if not (os.getenv("CLAW_DATA_ROOT") or os.getenv("JIANGCHANG_DATA_ROOT") or "").strip():
os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
if not (os.getenv("CLAW_USER_ID") or os.getenv("JIANGCHANG_USER_ID") or "").strip():
os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()

View File

@@ -1,137 +0,0 @@
"""
统一文件日志:{DATA_ROOT}/{USER_ID}/logs/jiangchang.log
按日轮转;行内带 trace_id 与 skill_slug便于跨技能排查。
"""
from __future__ import annotations
import logging
import os
import sys
import uuid
from logging.handlers import TimedRotatingFileHandler
from typing import Optional
from .runtime_env import get_data_root, get_user_id
_skill_slug: str = ""
_logger_name: str = ""
def get_unified_logs_dir() -> str:
path = os.path.join(get_data_root(), get_user_id(), "logs")
os.makedirs(path, exist_ok=True)
return path
def get_skill_log_file_path() -> str:
override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip()
if override:
parent = os.path.dirname(os.path.abspath(override))
if parent:
os.makedirs(parent, exist_ok=True)
return os.path.abspath(override)
return os.path.join(get_unified_logs_dir(), "jiangchang.log")
def ensure_trace_for_process() -> str:
existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
if existing:
return existing
tid = uuid.uuid4().hex[:12]
os.environ["JIANGCHANG_TRACE_ID"] = tid
return tid
def subprocess_env_with_trace(environ: Optional[dict] = None) -> dict:
ensure_trace_for_process()
tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or ensure_trace_for_process()
base = os.environ if environ is None else environ
return {**base, "JIANGCHANG_TRACE_ID": tid}
class _SkillContextFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or "-"
record.skill_slug = _skill_slug or "-"
return True
_FORMAT = "%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
_DATEFMT = "%Y-%m-%dT%H:%M:%S"
def _log_level_from_env() -> int:
v = (os.getenv("JIANGCHANG_LOG_LEVEL") or "INFO").strip().upper()
return getattr(logging, v, None) or logging.INFO
def _backup_count() -> int:
try:
n = int((os.getenv("JIANGCHANG_LOG_BACKUP_COUNT") or "30").strip())
return max(1, min(n, 365))
except ValueError:
return 30
def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
global _skill_slug, _logger_name
ensure_trace_for_process()
_skill_slug = skill_slug
_logger_name = logger_name
log = logging.getLogger(logger_name)
if log.handlers:
return
log.setLevel(_log_level_from_env())
path = get_skill_log_file_path()
fh = TimedRotatingFileHandler(
path,
when="midnight",
interval=1,
backupCount=_backup_count(),
encoding="utf-8",
delay=True,
)
fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
fh.setFormatter(fmt)
fh.addFilter(_SkillContextFilter())
log.addHandler(fh)
if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in ("1", "true", "yes", "on"):
sh = logging.StreamHandler(sys.stderr)
sh.setLevel(logging.WARNING)
sh.setFormatter(fmt)
sh.addFilter(_SkillContextFilter())
log.addHandler(sh)
log.propagate = False
def get_skill_logger() -> logging.Logger:
if not _logger_name:
raise RuntimeError("get_skill_logger: call setup_skill_logging first")
return logging.getLogger(_logger_name)
def attach_unified_file_handler(
log_path: str,
*,
skill_slug: str,
logger_name: str,
level: int = logging.DEBUG,
) -> logging.Logger:
global _skill_slug
ensure_trace_for_process()
_skill_slug = skill_slug
lg = logging.getLogger(logger_name)
lg.handlers.clear()
lg.setLevel(level)
parent = os.path.dirname(os.path.abspath(log_path))
if parent:
os.makedirs(parent, exist_ok=True)
fh = logging.FileHandler(log_path, encoding="utf-8")
fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
fh.setFormatter(fmt)
fh.addFilter(_SkillContextFilter())
lg.addHandler(fh)
lg.propagate = False
return lg

View File

@@ -5,6 +5,9 @@ skill-template CLI 入口。
复制为新技能后,请修改注释与常量,但保留当前分层结构:
cliargv→ service业务编排→ db持久化→ util通用工具
jiangchang_skill_core 来自宿主共享 Python runtime 安装的 jiangchang-platform-kit
不在技能仓库内 vendored。
"""
from __future__ import annotations
@@ -25,6 +28,10 @@ from jiangchang_skill_core.runtime_env import apply_cli_local_defaults
apply_cli_local_defaults()
from util.config_bootstrap import bootstrap_skill_config
bootstrap_skill_config()
from cli.app import main
if __name__ == "__main__":

View File

@@ -50,14 +50,13 @@ class SimRpaAdapter(AdapterBase):
batch_id=batch_id,
title="技能仿真演示",
) as video:
video.step("正在打开目标页面")
video.add_step("正在打开目标页面")
try:
async with async_playwright() as pw:
context = await launch_persistent_browser(
pw,
profile_dir=profile_dir,
headless=config.get_bool("OPENCLAW_BROWSER_HEADLESS"),
record_video_dir=video.record_video_dir,
)
page = context.pages[0] if context.pages else await context.new_page()
await page.goto(url, wait_until="domcontentloaded", timeout=60_000)

View File

@@ -1,7 +1,7 @@
"""兄弟技能 CLI 调用模板
"""兄弟技能 CLI 调用工具
用工具:通过子进程调用同级其他 skill 的 main.py并解析 JSON 输出。
具体调用哪些兄弟技能,由具体业务 skill 决定(在 task_service.py 中按需 import 调用)
通过子进程调用同级其他 skill 的 main.py并解析 JSON 输出。
具体调用哪些兄弟技能,由复制后的业务 skill 在 service 层按需决定
"""
from __future__ import annotations
@@ -50,16 +50,8 @@ def call_sibling_json(skill_slug: str, args: List[str]) -> Optional[Dict[str, An
def get_sibling_main_path(skill_slug: str) -> str:
"""获取兄弟技能的 main.py 绝对路径。
使用示例(在你的 task_service.py 中):
使用示例(在 task_service.py 中):
from service.sibling_bridge import get_sibling_main_path, call_sibling_json
# 调用名为 account-manager 的兄弟技能:
result = call_sibling_json("account-manager", ["list", "--limit", "10"])
"""
return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
# ============================================================
# 复制本模板后,按需在你自己的 task_service.py 中调用上面的工具。
# 不要在本文件中硬编码具体兄弟技能函数(如 get_account_manager_main_path
# 那是发布类技能的特定需求,不属于通用模板。
# ============================================================

View File

@@ -0,0 +1,84 @@
"""任务运行辅助:录屏信息组装与 CLI 输出(通用模板,无业务逻辑)。"""
from __future__ import annotations
from typing import Any, Dict, Optional
def build_video_info(
video_summary: dict[str, Any],
compose_context_before: Optional[dict[str, Any]] = None,
compose_context_after: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
"""将 RpaVideoSession.summary() 转为 task log / CLI 可用的 video 块。"""
return {
"enabled": video_summary.get("enabled", False),
"path": video_summary.get("path"),
"capture_path": video_summary.get("capture_path"),
"record_log_path": video_summary.get("record_log_path"),
"warnings": list(video_summary.get("warnings") or []),
"voiceover_path": video_summary.get("voiceover_path"),
"music_path": video_summary.get("music_path"),
"audio_warnings": list(video_summary.get("audio_warnings") or []),
"compose_context": {
**(compose_context_before or {}),
"after_compose": compose_context_after or {},
},
}
def merge_video_into_result_summary(
payload: dict[str, Any],
video_info: Optional[dict[str, Any]],
) -> dict[str, Any]:
"""把 video artifact 字段写入 result_summary JSON 对象。"""
if not video_info:
return payload
payload = dict(payload)
payload["video"] = video_info
payload["video_path"] = video_info.get("path")
payload["raw_video"] = video_info.get("capture_path")
payload["video_log"] = video_info.get("record_log_path")
payload["video_warnings"] = list(video_info.get("warnings") or [])
payload["voiceover_path"] = video_info.get("voiceover_path")
payload["music_path"] = video_info.get("music_path")
payload["audio_warnings"] = list(video_info.get("audio_warnings") or [])
return payload
def _print_video_summary(video_block: Optional[dict[str, Any]]) -> None:
"""任务完成后在 CLI 打印录屏路径、日志与诊断信息。"""
if not isinstance(video_block, dict) or not video_block.get("enabled"):
return
record_log = video_block.get("record_log_path")
if record_log:
print(f"录屏日志:{record_log}")
path = video_block.get("path")
if path:
print(f"录屏路径:{path}")
else:
print("录屏路径:未生成")
warnings = video_block.get("warnings") or []
if warnings:
print(f"视频诊断:{'; '.join(str(w) for w in warnings)}")
capture = video_block.get("capture_path")
if capture:
print(f"原始录屏路径:{capture}")
voiceover = video_block.get("voiceover_path")
if voiceover:
print(f"旁白路径:{voiceover}")
music = video_block.get("music_path")
if music:
print(f"背景音乐:{music}")
audio_warnings = video_block.get("audio_warnings") or []
if audio_warnings:
print(f"音频诊断:{'; '.join(str(w) for w in audio_warnings)}")
compose_ctx = video_block.get("compose_context")
if isinstance(compose_ctx, dict):
print(
"视频合成上下文:"
f"media_assets_root={compose_ctx.get('media_assets_root', '')}; "
f"ffmpeg_path={compose_ctx.get('ffmpeg_path', '')}; "
f"background_music_issue={compose_ctx.get('background_music_issue') or ''}; "
f"background_music_sample_path={compose_ctx.get('background_music_sample_path', '')}"
)

View File

@@ -6,25 +6,81 @@
from __future__ import annotations
import asyncio
import json
import sys
from typing import Optional
import os
import uuid
from typing import Any, Dict, Optional
from jiangchang_skill_core import collect_runtime_diagnostics, config, format_runtime_health_lines
from jiangchang_skill_core.rpa.video_session import RpaVideoSession
from db import task_logs_repository as tlr
from service.entitlement_service import check_entitlement
from util.constants import SKILL_SLUG, SKILL_VERSION
from service.task_run_support import (
_print_video_summary,
build_video_info,
merge_video_into_result_summary,
)
from util.constants import PLATFORM_KIT_MIN_VERSION, SKILL_SLUG, SKILL_VERSION
from util.runtime_paths import get_skill_data_dir, get_skill_root
from util.timeutil import unix_to_iso
async def _run_template_demo(target: Optional[str], input_id: Optional[str]) -> tuple[int, dict[str, Any]]:
"""最小 RpaVideoSession 示范:不启动浏览器、不执行业务,仅演示录屏包裹范式。"""
batch_id = uuid.uuid4().hex[:12]
data_dir = get_skill_data_dir()
async with RpaVideoSession(
skill_slug=SKILL_SLUG,
skill_data_dir=data_dir,
batch_id=batch_id,
title="开始执行示例任务",
closing_title="示例任务执行完成",
) as video:
video.add_step("准备执行示例任务")
if target:
video.add_step(f"接收目标参数:{target}")
if input_id:
video.add_step(f"接收输入标识:{input_id}")
video.add_step("示例任务执行完成")
video_summary = video.summary()
video_info = build_video_info(video_summary, {}, {})
return 1, video_info
def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int:
"""通用任务执行入口模板。复制后请实现真实业务逻辑。"""
_ = (target, input_id)
ok, reason = check_entitlement(SKILL_SLUG)
if not ok:
print(f"{reason}")
return 1
rc, video_info = asyncio.run(_run_template_demo(target, input_id))
_print_video_summary(video_info)
summary_payload = merge_video_into_result_summary(
{
"template_demo": True,
"target": target,
"input_id": input_id,
},
video_info,
)
tlr.save_task_log(
task_type="demo",
target_id=target,
input_id=input_id,
input_title="模板示例任务",
status="failed",
error_msg="模板仓库未实现真实业务",
result_summary=json.dumps(summary_payload, ensure_ascii=False),
)
print("❌ 这是模板仓库,请复制后在 scripts/service/task_service.py 中实现 cmd_run 的真实业务逻辑。")
return 1
return rc
def cmd_logs(
@@ -88,8 +144,43 @@ def cmd_log_get(log_id: str) -> int:
return 0
def cmd_config_path() -> int:
example_path = os.path.join(get_skill_root(), ".env.example")
env_path = config.get_env_file_path() or ""
print(
json.dumps(
{
"skill": SKILL_SLUG,
"env_path": env_path,
"example_path": os.path.abspath(example_path),
},
ensure_ascii=False,
)
)
return 0
def cmd_health() -> int:
return 0 if sys.version_info >= (3, 10) else 1
runtime = collect_runtime_diagnostics(
skill_slug=SKILL_SLUG,
platform_kit_min_version=PLATFORM_KIT_MIN_VERSION,
skill_root=get_skill_root(),
)
example_path = os.path.join(get_skill_root(), ".env.example")
env_path = config.get_env_file_path() or ""
env_exists = bool(env_path and os.path.isfile(env_path))
health_status = "failed" if runtime.has_fatal_issues else "ok"
lines = [
f"{SKILL_SLUG} health: {health_status}",
*format_runtime_health_lines(runtime),
f"env_path: {env_path}",
f"env_exists: {env_exists}",
f"example_path: {os.path.abspath(example_path)}",
]
for line in lines:
print(line)
return 1 if runtime.has_fatal_issues else 0
def cmd_version() -> int:

View File

@@ -0,0 +1,19 @@
"""技能配置初始化:.env.example 落盘与用户 .env 缺失项合并。"""
from __future__ import annotations
import os
from jiangchang_skill_core import config
from util.constants import SKILL_SLUG, SKILL_VERSION
from util.runtime_paths import get_skill_root
def bootstrap_skill_config() -> str:
"""确保用户数据目录 .env 存在,并追加 .env.example 中的新配置项。"""
example_path = os.path.join(get_skill_root(), ".env.example")
env_path = config.ensure_env_file(SKILL_SLUG, example_path)
comment = f"{SKILL_SLUG} v{SKILL_VERSION}" if SKILL_VERSION else None
config.merge_missing_env_keys(example_path, env_path, comment_skill=comment)
return env_path

View File

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

View File

@@ -30,8 +30,8 @@
从本模板复制出新技能仓库后,建议至少逐项确认:
- [ ] `python tests/run_tests.py -v` 能通过。
- [ ] `python scripts/main.py health` 能通过。
- [ ] `python tests/run_tests.py -v` 能通过(含 platform-kit 导入与无 vendored core 守护)
- [ ] `python scripts/main.py health` 能通过(输出 runtime diagnosticswarning 不导致非零退出)
- [ ] `python scripts/main.py version` 输出 JSON且 `skill` 与目录名 / `SKILL.md` / `constants.SKILL_SLUG` 一致。
- [ ] 所有 DB / 文件写入都在 `IsolatedDataRoot`(或等价隔离)下测试,不写真实数据目录。
- [ ] 外部系统默认使用 mock / `FakeAdapter`,不访问真实 API不打开真实 RPA。
@@ -39,7 +39,7 @@
- [ ] 至少有 1 个缺必填字段 / 非法输入测试。
- [ ] 如果有 adapter至少覆盖 timeout / unauthorized / invalid response可用 fake 模拟)。
- [ ] 如果有解析 / 计算 / 校验类业务,至少保留 1 组 golden fixture脱敏
- [ ] 如果技能有 Python 三方依赖,`requirements.txt` 已声明且版本约束合理(尽量收窄范围)。
- [ ] 如果技能有**特有** Python 三方依赖,`requirements.txt` 已声明且版本约束合理(尽量收窄范围)公共依赖platform-kit、playwright由宿主共享 runtime 提供,**不要**写入技能 requirements
- [ ] `health` 能在依赖缺失或原生运行库未就绪时给出清晰、可操作的错误(非裸 traceback
- [ ] 真实 API / 真实 RPA 测试只放 `tests/integration/`,并且默认 `.sample`,不默认运行。
@@ -74,6 +74,13 @@ python tests/run_tests.py test_cli_smoke
| ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` |
| DB / ``task_logs`` 骨架冒烟 | ``test_db_smoke.py`` |
| **adapter profile / 外呼授权策略** | ``test_adapter_profile_policy.py`` + ``adapter_test_utils.py``(非 ``test_`` 前缀,不自动当用例收集) |
| **无 vendored ``jiangchang_skill_core``** | ``test_platform_import.py`` |
| **runtime diagnostics 架构守护** | ``test_runtime_diagnostics_integration.py`` |
| **文档/runtime 标准守护** | ``test_template_runtime_standard.py`` |
| **配置 bootstrap / config-path** | ``test_config_bootstrap.py`` |
| **health runtime diagnostics** | ``test_health_runtime.py`` |
复制为新技能后**不得删除**上述架构守护测试,除非同步更新模板标准并理解影响。
### 1.3 数据隔离(``_support.IsolatedDataRoot``

View File

@@ -39,6 +39,16 @@ def get_skill_root() -> str:
return _SKILL_ROOT
def platform_kit_version_patch(version: str = "1.0.13"):
"""Mock installed jiangchang-platform-kit version for health/diagnostics tests."""
from unittest.mock import patch
return patch(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
return_value=version,
)
class IsolatedDataRoot:
"""
在独立临时目录下模拟数据根,避免污染本机默认数据目录。

View File

@@ -8,7 +8,9 @@ import os
import unittest
from contextlib import redirect_stderr, redirect_stdout
from _support import IsolatedDataRoot
from _support import IsolatedDataRoot, platform_kit_version_patch
from jiangchang_skill_core import config
# scripts/ 已由 _support 注入 sys.path
from cli.app import main
@@ -27,10 +29,23 @@ class TestCliSmoke(unittest.TestCase):
self.assertIn("health", out)
def test_health_zero(self) -> None:
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["health"])
self.assertEqual(rc, 0)
old_record = os.environ.get("OPENCLAW_RECORD_VIDEO")
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
try:
buf = io.StringIO()
with platform_kit_version_patch(), redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["health"])
self.assertEqual(rc, 0)
out = buf.getvalue()
self.assertIn("health:", out)
self.assertIn("python_executable:", out)
self.assertIn("platform_kit_version:", out)
self.assertIn("jiangchang_skill_core_file:", out)
finally:
if old_record is None:
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
else:
os.environ["OPENCLAW_RECORD_VIDEO"] = old_record
def test_version_json_and_matches_constants_slug(self) -> None:
buf = io.StringIO()
@@ -58,17 +73,25 @@ class TestCliSmoke(unittest.TestCase):
self.assertIn("数字", buf.getvalue())
def test_run_placeholder_returns_nonzero_without_network(self) -> None:
"""占位 run不验证业务成功,仅确认模板提示且退出码非 0无 AUTH_BASE 时不发 HTTP"""
"""占位 runRpaVideoSession 示范 + 模板提示OPENCLAW_RECORD_VIDEO=0 时不启 ffmpeg"""
old_auth = os.environ.pop("JIANGCHANG_AUTH_BASE_URL", None)
old_record = os.environ.get("OPENCLAW_RECORD_VIDEO")
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
try:
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["run"])
self.assertNotEqual(rc, 0)
self.assertIn("模板", buf.getvalue())
with IsolatedDataRoot(user_id="_cli_run"):
config.reset_cache()
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["run"])
self.assertNotEqual(rc, 0)
self.assertIn("模板", buf.getvalue())
finally:
if old_auth is not None:
os.environ["JIANGCHANG_AUTH_BASE_URL"] = old_auth
if old_record is None:
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
else:
os.environ["OPENCLAW_RECORD_VIDEO"] = old_record
if __name__ == "__main__":

View File

@@ -0,0 +1,116 @@
# -*- coding: utf-8 -*-
"""配置落盘、合并与读取优先级。"""
from __future__ import annotations
import io
import json
import os
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
from _support import IsolatedDataRoot, get_skill_root, platform_kit_version_patch
from jiangchang_skill_core import config
from cli.app import main
from util.config_bootstrap import bootstrap_skill_config
from util.constants import SKILL_SLUG
class TestConfigBootstrap(unittest.TestCase):
def test_first_run_creates_user_env(self) -> None:
with IsolatedDataRoot(user_id="_cfg_test"):
config.reset_cache()
path = bootstrap_skill_config()
self.assertTrue(os.path.isfile(path))
self.assertTrue(path.replace("\\", "/").endswith(f"/{SKILL_SLUG}/.env"))
def test_existing_env_not_overwritten(self) -> None:
with IsolatedDataRoot(user_id="_cfg_keep"):
config.reset_cache()
path = bootstrap_skill_config()
with open(path, "w", encoding="utf-8") as f:
f.write("OPENCLAW_TEST_TARGET=simulator_rpa\n")
with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False, suffix=".example") as tmp:
tmp.write("OPENCLAW_TEST_TARGET=mock\nHUMAN_WAIT_TIMEOUT=60\n")
tmp_path = tmp.name
try:
config.merge_missing_env_keys(tmp_path, path, comment_skill="test")
with open(path, encoding="utf-8") as f:
content = f.read()
self.assertIn("OPENCLAW_TEST_TARGET=simulator_rpa", content)
self.assertNotIn("OPENCLAW_TEST_TARGET=mock", content)
finally:
os.unlink(tmp_path)
config.reset_cache()
def test_merge_appends_missing_keys(self) -> None:
with IsolatedDataRoot(user_id="_cfg_merge"):
config.reset_cache()
path = bootstrap_skill_config()
with open(path, "w", encoding="utf-8") as f:
f.write("OPENCLAW_TEST_TARGET=simulator_rpa\n")
example = os.path.join(get_skill_root(), ".env.example")
added = config.merge_missing_env_keys(example, path, comment_skill="test")
self.assertIn("HUMAN_WAIT_TIMEOUT", added)
with open(path, encoding="utf-8") as f:
content = f.read()
self.assertIn("HUMAN_WAIT_TIMEOUT=180", content)
def test_config_priority_process_user_example(self) -> None:
with IsolatedDataRoot(user_id="_cfg_prio"):
config.reset_cache()
example = os.path.join(get_skill_root(), ".env.example")
path = config.ensure_env_file(SKILL_SLUG, example)
with open(path, "w", encoding="utf-8") as f:
f.write("PRIORITY_TEST_KEY=from_user\n")
config.reset_cache()
os.environ["PRIORITY_TEST_KEY"] = "from_process"
self.assertEqual(config.get("PRIORITY_TEST_KEY"), "from_process")
del os.environ["PRIORITY_TEST_KEY"]
config.reset_cache()
self.assertEqual(config.get("PRIORITY_TEST_KEY"), "from_user")
with open(path, "w", encoding="utf-8") as f:
f.write("# empty user\n")
config.reset_cache()
with open(example, encoding="utf-8") as f:
example_text = f.read()
self.assertIn("OPENCLAW_TEST_TARGET=mock", example_text)
self.assertEqual(config.get("OPENCLAW_TEST_TARGET"), "mock")
def test_config_path_outputs_json(self) -> None:
with IsolatedDataRoot(user_id="_cfg_path"):
config.reset_cache()
bootstrap_skill_config()
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["config-path"])
self.assertEqual(rc, 0)
payload = json.loads(buf.getvalue().strip())
self.assertEqual(payload["skill"], SKILL_SLUG)
self.assertTrue(payload["env_path"])
self.assertTrue(payload["example_path"].endswith(".env.example"))
def test_health_does_not_print_sensitive_plaintext(self) -> None:
with IsolatedDataRoot(user_id="_cfg_health"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
config.reset_cache()
path = bootstrap_skill_config()
with open(path, "a", encoding="utf-8") as f:
f.write("FAKE_SECRET_TOKEN=supersecret12345\n")
config.reset_cache()
buf = io.StringIO()
with platform_kit_version_patch(), redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["health"])
self.assertEqual(rc, 0)
out = buf.getvalue()
self.assertNotIn("supersecret12345", out)
self.assertIn("env_path:", out)
self.assertIn("env_exists: True", out)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
"""文档与 pytest 配置守护:防止模板标准退化。"""
from __future__ import annotations
import configparser
import os
import unittest
from _support import get_skill_root
class TestDocsStandards(unittest.TestCase):
def _read(self, rel_path: str) -> str:
path = os.path.join(get_skill_root(), rel_path)
with open(path, encoding="utf-8") as f:
return f.read()
def test_rpa_md_covers_video_and_playwright_standards(self) -> None:
text = self._read("references/RPA.md")
self.assertIn("title", text)
self.assertIn("closing_title", text)
self.assertIn("中文", text)
self.assertIn("--no-sandbox", text)
self.assertIn("--disable-blink-features=AutomationControlled", text)
self.assertIn("RpaVideoSession", text)
self.assertIn("1.0.13", text)
def test_adapter_md_covers_four_tiers_and_allow_flags(self) -> None:
text = self._read("references/ADAPTER.md")
for marker in (
"mock",
"simulator_rpa",
"real_api",
"real_rpa",
"ALLOW_REAL_API",
"ALLOW_REAL_RPA",
"ALLOW_WRITE_ACTIONS",
"sibling_bridge",
):
self.assertIn(marker, text, msg=f"ADAPTER.md missing {marker!r}")
def test_cli_md_mentions_shared_python_runtime(self) -> None:
text = self._read("references/CLI.md")
self.assertIn("python-runtime", text)
self.assertIn("uv run python", text)
def test_testing_md_mentions_pytest_txt_collection_guard(self) -> None:
text = self._read("references/TESTING.md")
self.assertIn(".txt", text)
self.assertIn("pytest.ini", text)
def test_docs_do_not_claim_1_0_11_as_current_standard(self) -> None:
for rel in (
"README.md",
"SKILL.md",
"references/RUNTIME.md",
"references/RPA.md",
"references/CONFIG.md",
"references/TESTING.md",
):
text = self._read(rel)
self.assertNotIn(
">=1.0.11",
text.replace(" ", ""),
msg=f"{rel} still claims 1.0.11 as current minimum",
)
class TestPytestIniCollection(unittest.TestCase):
def test_pytest_ini_exists_and_limits_python_files(self) -> None:
ini_path = os.path.join(get_skill_root(), "pytest.ini")
self.assertTrue(os.path.isfile(ini_path), "pytest.ini must exist at repo root")
parser = configparser.ConfigParser()
parser.read(ini_path, encoding="utf-8")
python_files = parser.get("pytest", "python_files", fallback="")
self.assertIn("test_*.py", python_files)
self.assertIn("*_test.py", python_files)
if __name__ == "__main__":
unittest.main()

View File

@@ -10,6 +10,19 @@ import unittest
from _support import IsolatedDataRoot, get_skill_root
import importlib.metadata as metadata
from jiangchang_skill_core import version_ge
from util.constants import PLATFORM_KIT_MIN_VERSION
def _platform_kit_meets_template_minimum() -> bool:
try:
installed = metadata.version("jiangchang-platform-kit")
except metadata.PackageNotFoundError:
return False
return version_ge(installed, PLATFORM_KIT_MIN_VERSION)
def _run_main(args: list[str], timeout: int = 15) -> subprocess.CompletedProcess[str]:
root = get_skill_root()
@@ -30,6 +43,10 @@ def _run_main(args: list[str], timeout: int = 15) -> subprocess.CompletedProcess
class TestEntrypointSubprocess(unittest.TestCase):
@unittest.skipUnless(
_platform_kit_meets_template_minimum(),
f"installed jiangchang-platform-kit below template minimum {PLATFORM_KIT_MIN_VERSION}",
)
def test_health_exit_zero(self) -> None:
with IsolatedDataRoot():
cp = _run_main(["health"])

View File

@@ -0,0 +1,178 @@
# -*- coding: utf-8 -*-
"""health / runtime 诊断:验证委托 platform-kit 公共 API。"""
from __future__ import annotations
import io
import os
import unittest
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from unittest.mock import patch
from _support import IsolatedDataRoot, get_skill_root, platform_kit_version_patch
import jiangchang_skill_core
from cli.app import main
from jiangchang_skill_core import (
collect_runtime_diagnostics,
format_runtime_health_lines,
is_jiangchang_skill_core_from_skill_tree,
runtime_diagnostics_dict,
)
from util.config_bootstrap import bootstrap_skill_config
from util.constants import PLATFORM_KIT_MIN_VERSION, SKILL_SLUG
from util.runtime_paths import get_skill_root as skill_root_path
def _collect_diag():
return collect_runtime_diagnostics(
skill_slug=SKILL_SLUG,
platform_kit_min_version=PLATFORM_KIT_MIN_VERSION,
skill_root=skill_root_path(),
)
class TestHealthRuntimeDiagnostics(unittest.TestCase):
def test_collect_runtime_diagnostics_fields(self) -> None:
with IsolatedDataRoot(user_id="_health_rt"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
diag = _collect_diag()
payload = runtime_diagnostics_dict(diag)
for key in (
"skill_slug",
"python_executable",
"platform_kit_version",
"jiangchang_skill_core_file",
"resolved_data_root",
"media_assets_root",
"ffmpeg_available",
"background_music_mp3_count",
"runtime_issues",
):
self.assertIn(key, payload)
self.assertEqual(diag.skill_slug, SKILL_SLUG)
self.assertEqual(diag.platform_kit_min_version, PLATFORM_KIT_MIN_VERSION)
def test_format_runtime_health_lines_contains_core_fields(self) -> None:
diag = _collect_diag()
text = "\n".join(format_runtime_health_lines(diag))
self.assertIn("skill_slug:", text)
self.assertIn("python_executable:", text)
self.assertIn("platform_kit_version:", text)
self.assertIn("jiangchang_skill_core_file:", text)
self.assertIn("media_assets_root:", text)
self.assertIn("background_music_mp3_count:", text)
def test_health_cli_prints_runtime_diagnostics(self) -> None:
with IsolatedDataRoot(user_id="_health_cli"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
bootstrap_skill_config()
buf = io.StringIO()
with platform_kit_version_patch(), redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["health"])
out = buf.getvalue()
self.assertEqual(rc, 0, msg=out)
self.assertIn("python_executable:", out)
self.assertIn("platform_kit_version:", out)
self.assertIn("jiangchang_skill_core_file:", out)
self.assertIn("media_assets_root:", out)
self.assertIn("background_music_mp3_count:", out)
self.assertIn("env_path:", out)
self.assertIn("example_path:", out)
def test_health_fails_when_platform_kit_missing(self) -> None:
with IsolatedDataRoot(user_id="_health_no_pkg"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
bootstrap_skill_config()
with patch(
"jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
return_value=None,
):
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["health"])
self.assertEqual(rc, 1)
self.assertIn("platform_kit_not_installed", buf.getvalue())
def test_background_music_issue_is_warning_not_fatal(self) -> None:
from jiangchang_skill_core.media_assets import MediaAssetsStatus
with IsolatedDataRoot(user_id="_health_music_warn"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
media_root = Path(os.environ["CLAW_DATA_ROOT"]) / "shared" / "media-assets"
ffmpeg_path = media_root / "tools" / "ffmpeg.exe"
ffmpeg_path.parent.mkdir(parents=True, exist_ok=True)
ffmpeg_path.write_bytes(b"x" * 256)
status = MediaAssetsStatus(
root=media_root,
exists=True,
ready=False,
source="local",
warnings=["music_dir_missing"],
ffmpeg_path=ffmpeg_path,
music_dir=None,
)
music_probe = {
"music_root": None,
"mp3_count": 0,
"usable_count": 0,
"sample_path": None,
"issue": "music_dir_missing",
}
with patch(
"jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
return_value=status,
), patch(
"jiangchang_skill_core.runtime_diagnostics.probe_background_music",
return_value=music_probe,
), platform_kit_version_patch():
diag = _collect_diag()
music_issues = [i for i in diag.issues if i.code == "background_music_unavailable"]
self.assertTrue(music_issues)
self.assertEqual(music_issues[0].severity, "warning")
self.assertFalse(diag.has_fatal_issues)
def test_skill_tree_core_load_detected_as_issue(self) -> None:
skill_root = get_skill_root()
fake_core = os.path.join(
skill_root, "scripts", "fake_pkg", "jiangchang_skill_core", "__init__.py"
)
self.assertTrue(
is_jiangchang_skill_core_from_skill_tree(
skill_root=skill_root, core_file=fake_core
)
)
self.assertFalse(
is_jiangchang_skill_core_from_skill_tree(
skill_root=skill_root,
core_file=jiangchang_skill_core.__file__,
)
)
def test_skill_tree_core_load_issue_in_diagnostics(self) -> None:
skill_root = get_skill_root()
fake_core = os.path.join(skill_root, "scripts", "jiangchang_skill_core", "__init__.py")
with patch.object(jiangchang_skill_core, "__file__", fake_core):
diag = _collect_diag()
codes = diag.issue_codes()
self.assertIn("jiangchang_skill_core_loaded_from_skill_tree", codes)
messages = [issue.message for issue in diag.issues]
self.assertTrue(any("skill tree" in msg.lower() for msg in messages))
class TestHealthRuntimeWithFakeMedia(unittest.TestCase):
def test_probe_counts_mp3_without_download(self) -> None:
with IsolatedDataRoot(user_id="_health_media"):
media_root = Path(os.environ["CLAW_DATA_ROOT"]) / "shared" / "media-assets"
music_dir = media_root / "music"
music_dir.mkdir(parents=True)
(music_dir / "demo.mp3").write_bytes(b"\x00" * 1024)
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
diag = _collect_diag()
self.assertGreaterEqual(diag.background_music_mp3_count, 1)
self.assertGreaterEqual(diag.background_music_usable_count, 1)
self.assertIsNone(diag.background_music_issue)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,136 @@
# -*- coding: utf-8 -*-
"""防回归:公共能力必须从共享 platform-kit 导入,而非技能 vendored 副本。"""
from __future__ import annotations
import importlib.metadata as metadata
import os
import re
import sys
import unittest
from _support import get_scripts_dir, get_skill_root
def _parse_platform_kit_min_version(skill_md_text: str) -> str:
parts = skill_md_text.split("---", 2)
if len(parts) < 2:
raise ValueError("missing frontmatter")
m = re.search(
r"platform_kit_min_version:\s*[\"']?([^\"'\s]+)[\"']?",
parts[1],
)
if not m:
raise ValueError("platform_kit_min_version not found")
return m.group(1).strip()
def _requirement_dependency_lines(req_text: str) -> list[str]:
return [
line.strip()
for line in req_text.splitlines()
if line.strip() and not line.strip().startswith("#")
]
class TestPlatformImportSource(unittest.TestCase):
def test_jiangchang_skill_core_not_from_skill_scripts(self) -> None:
scripts_dir = os.path.abspath(get_scripts_dir())
skill_root = os.path.abspath(get_skill_root())
vendored = os.path.join(scripts_dir, "jiangchang_skill_core")
self.assertFalse(
os.path.isdir(vendored),
f"vendored package must not exist: {vendored}",
)
import jiangchang_skill_core
import jiangchang_skill_core.rpa.video_session as video_session
core_file = os.path.abspath(jiangchang_skill_core.__file__)
video_file = os.path.abspath(video_session.__file__)
self.assertNotIn(
scripts_dir.lower(),
core_file.lower(),
msg=f"jiangchang_skill_core loaded from skill scripts: {core_file}",
)
self.assertNotIn(
skill_root.lower(),
core_file.lower(),
msg=f"jiangchang_skill_core loaded from skill tree: {core_file}",
)
self.assertNotIn(
scripts_dir.lower(),
video_file.lower(),
msg=f"video_session loaded from skill scripts: {video_file}",
)
from jiangchang_skill_core.rpa.video_session import RpaVideoSession
self.assertTrue(callable(RpaVideoSession))
def test_skill_tree_load_issue_helper(self) -> None:
from jiangchang_skill_core import is_jiangchang_skill_core_from_skill_tree
skill_root = os.path.abspath(get_skill_root())
fake = os.path.join(skill_root, "scripts", "jiangchang_skill_core", "__init__.py")
self.assertTrue(
is_jiangchang_skill_core_from_skill_tree(skill_root=skill_root, core_file=fake)
)
self.assertFalse(
is_jiangchang_skill_core_from_skill_tree(
skill_root=skill_root, core_file=sys.executable
)
)
def test_platform_kit_min_version_is_1_0_13(self) -> None:
from jiangchang_skill_core import version_ge
from util.constants import PLATFORM_KIT_MIN_VERSION
self.assertEqual(PLATFORM_KIT_MIN_VERSION, "1.0.13")
md_path = os.path.join(get_skill_root(), "SKILL.md")
with open(md_path, encoding="utf-8") as f:
md = f.read()
self.assertEqual(_parse_platform_kit_min_version(md), "1.0.13")
req_path = os.path.join(get_skill_root(), "requirements.txt")
with open(req_path, encoding="utf-8") as f:
req = f.read()
dep_lines = _requirement_dependency_lines(req)
dep_body = "\n".join(dep_lines).lower()
self.assertNotIn("jiangchang-platform-kit", dep_body)
self.assertNotIn("playwright", dep_body)
try:
installed = metadata.version("jiangchang-platform-kit")
except metadata.PackageNotFoundError:
self.skipTest("jiangchang-platform-kit not installed in current interpreter")
if not version_ge(installed, PLATFORM_KIT_MIN_VERSION):
self.skipTest(
f"installed jiangchang-platform-kit {installed!r} < required "
f"{PLATFORM_KIT_MIN_VERSION!r}; upgrade shared runtime to enforce this check"
)
self.assertTrue(
version_ge(installed, PLATFORM_KIT_MIN_VERSION),
msg=f"installed {installed!r} < required {PLATFORM_KIT_MIN_VERSION!r}",
)
def test_collect_runtime_diagnostics_importable(self) -> None:
from jiangchang_skill_core import collect_runtime_diagnostics
self.assertTrue(callable(collect_runtime_diagnostics))
def test_main_import_smoke(self) -> None:
scripts_dir = get_scripts_dir()
if scripts_dir not in sys.path:
sys.path.insert(0, scripts_dir)
from cli.app import build_parser
parser = build_parser()
self.assertIsNotNone(parser)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,72 @@
# -*- coding: utf-8 -*-
"""Architecture guard: runtime diagnostics must come from platform-kit."""
from __future__ import annotations
import os
import unittest
from _support import get_scripts_dir, get_skill_root
_FORBIDDEN_DEFINITIONS = (
"class RuntimeDiagnostics",
"class RuntimeIssue",
"def collect_runtime_diagnostics",
"def format_runtime_health_lines",
"def runtime_diagnostics_dict",
"def _is_git_lfs_pointer",
"def _is_usable_audio_file",
"def _probe_background_music",
)
class TestRuntimeDiagnosticsIntegration(unittest.TestCase):
def test_no_vendored_jiangchang_skill_core(self) -> None:
vendored = os.path.join(get_scripts_dir(), "jiangchang_skill_core")
self.assertFalse(os.path.isdir(vendored), vendored)
def test_runtime_diagnostics_module_removed(self) -> None:
legacy = os.path.join(get_scripts_dir(), "util", "runtime_diagnostics.py")
self.assertFalse(os.path.isfile(legacy), legacy)
def test_util_py_files_do_not_reimplement_runtime_diagnostics(self) -> None:
util_dir = os.path.join(get_scripts_dir(), "util")
for name in os.listdir(util_dir):
if not name.endswith(".py"):
continue
path = os.path.join(util_dir, name)
with open(path, encoding="utf-8") as f:
text = f.read()
for forbidden in _FORBIDDEN_DEFINITIONS:
self.assertNotIn(
forbidden,
text,
msg=f"{name} must not define {forbidden}",
)
def test_task_service_imports_platform_kit_diagnostics(self) -> None:
path = os.path.join(get_scripts_dir(), "service", "task_service.py")
with open(path, encoding="utf-8") as f:
text = f.read()
self.assertIn("from jiangchang_skill_core import", text)
self.assertIn("collect_runtime_diagnostics", text)
self.assertIn("format_runtime_health_lines", text)
self.assertIn("PLATFORM_KIT_MIN_VERSION", text)
self.assertNotIn("from util.runtime_diagnostics import", text)
def test_collect_runtime_diagnostics_callable_from_platform_kit(self) -> None:
from jiangchang_skill_core import collect_runtime_diagnostics
from util.constants import PLATFORM_KIT_MIN_VERSION, SKILL_SLUG
from util.runtime_paths import get_skill_root
diag = collect_runtime_diagnostics(
skill_slug=SKILL_SLUG,
platform_kit_min_version=PLATFORM_KIT_MIN_VERSION,
skill_root=get_skill_root(),
)
self.assertEqual(diag.skill_slug, SKILL_SLUG)
self.assertEqual(diag.platform_kit_min_version, PLATFORM_KIT_MIN_VERSION)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,82 @@
# -*- coding: utf-8 -*-
"""Docs guard: template must promote shared runtime, not vendored jiangchang_skill_core."""
from __future__ import annotations
import os
import unittest
from _support import get_skill_root
DOC_PATHS = (
"README.md",
"SKILL.md",
"references/RUNTIME.md",
"references/CLI.md",
"references/DEVELOPMENT.md",
"references/RPA.md",
"references/CONFIG.md",
)
FORBIDDEN_PHRASES = (
"scripts/jiangchang_skill_core/:运行时",
"运行时与统一日志副本",
"vendor 源码在 scripts/jiangchang_skill_core",
"模板通过 vendor 拷贝引用",
"vendor 自 platform-kit",
"模板 vendor 拷贝在 scripts/jiangchang_skill_core",
"模板默认声明 `jiangchang-platform-kit",
"模板默认包含 `jiangchang-platform-kit",
"默认已含 `jiangchang-platform-kit",
"保留仅含 `jiangchang-platform-kit",
)
POSITIVE_MARKERS = (
"jiangchang-platform-kit",
"1.0.13",
"共享 runtime",
)
class TestTemplateRuntimeStandard(unittest.TestCase):
def _read(self, rel_path: str) -> str:
path = os.path.join(get_skill_root(), rel_path)
with open(path, encoding="utf-8") as f:
return f.read()
def test_guarded_docs_avoid_vendored_core_guidance(self) -> None:
for rel_path in DOC_PATHS:
text = self._read(rel_path)
for phrase in FORBIDDEN_PHRASES:
self.assertNotIn(
phrase,
text,
msg=f"{rel_path} must not contain {phrase!r}",
)
def test_guarded_docs_promote_shared_runtime_standard(self) -> None:
combined = "\n".join(self._read(p) for p in DOC_PATHS)
for marker in POSITIVE_MARKERS:
self.assertIn(
marker,
combined,
msg=f"docs should mention {marker!r}",
)
self.assertTrue(
"不得 vendored" in combined or "不要 vendor" in combined,
msg="docs should state skills must not vendor jiangchang_skill_core",
)
def test_runtime_md_mentions_collect_runtime_diagnostics(self) -> None:
text = self._read("references/RUNTIME.md")
self.assertIn("collect_runtime_diagnostics", text)
def test_cli_md_mentions_health_runtime_diagnostics(self) -> None:
text = self._read("references/CLI.md")
self.assertIn("python_executable", text)
self.assertIn("platform_kit_version", text)
self.assertIn("jiangchang_skill_core_file", text)
self.assertIn("python-runtime", text)
if __name__ == "__main__":
unittest.main()

161
tests/test_video_service.py Normal file
View File

@@ -0,0 +1,161 @@
# -*- coding: utf-8 -*-
"""RpaVideoSession 模板示范与 result_summary / CLI 输出测试。"""
from __future__ import annotations
import io
import json
import os
import re
import unittest
from contextlib import redirect_stderr, redirect_stdout
from unittest.mock import AsyncMock, MagicMock, patch
from _support import IsolatedDataRoot, get_skill_root
from jiangchang_skill_core import config
class TestEnvExampleVideoDefaults(unittest.TestCase):
def test_env_example_contains_required_keys(self) -> None:
path = os.path.join(get_skill_root(), ".env.example")
with open(path, encoding="utf-8") as f:
text = f.read()
for key in (
"OPENCLAW_TEST_TARGET=mock",
"OPENCLAW_RECORD_VIDEO=1",
"OPENCLAW_ARTIFACTS_ON_FAILURE=1",
"OPENCLAW_BROWSER_HEADLESS=0",
"OPENCLAW_PLAYWRIGHT_STEALTH=1",
):
self.assertIn(key, text, msg=f"missing {key!r} in .env.example")
class TestPrintVideoSummary(unittest.TestCase):
def test_cli_prints_video_path_when_enabled(self) -> None:
from service.task_run_support import _print_video_summary
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
_print_video_summary(
{
"enabled": True,
"path": r"D:\data\videos\your-skill-slug_demo.mp4",
"record_log_path": r"D:\data\logs\ffmpeg-record.log",
"warnings": [],
}
)
out = buf.getvalue()
self.assertIn("录屏路径", out)
self.assertIn("your-skill-slug_demo.mp4", out)
self.assertIn("录屏日志", out)
def test_disabled_video_prints_nothing(self) -> None:
from service.task_run_support import _print_video_summary
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
_print_video_summary({"enabled": False, "path": r"D:\tmp\x.mp4"})
self.assertEqual(buf.getvalue(), "")
class TestMergeVideoIntoResultSummary(unittest.TestCase):
def test_result_summary_includes_video_fields(self) -> None:
from service.task_run_support import merge_video_into_result_summary
video_info = {
"enabled": True,
"path": r"D:\data\videos\demo.mp4",
"capture_path": r"D:\data\capture.mp4",
"record_log_path": r"D:\data\logs\ffmpeg-record.log",
"warnings": ["test_warning"],
"voiceover_path": r"D:\data\voiceover.wav",
"music_path": r"D:\data\music.mp3",
"audio_warnings": ["tts_skipped"],
}
payload = merge_video_into_result_summary({"demo": True}, video_info)
self.assertIn("video", payload)
self.assertEqual(payload["video_path"], video_info["path"])
self.assertEqual(payload["raw_video"], video_info["capture_path"])
self.assertEqual(payload["video_log"], video_info["record_log_path"])
self.assertEqual(payload["video_warnings"], ["test_warning"])
self.assertEqual(payload["voiceover_path"], video_info["voiceover_path"])
self.assertEqual(payload["music_path"], video_info["music_path"])
self.assertEqual(payload["audio_warnings"], ["tts_skipped"])
class TestTemplateRunVideoSession(unittest.TestCase):
def _fake_video_session(self, mock_video: MagicMock):
class _FakeVideoSession:
def __init__(self, **kwargs):
self.kwargs = kwargs
async def __aenter__(self):
return mock_video
async def __aexit__(self, *args):
return None
return _FakeVideoSession
def test_cmd_run_creates_video_session_with_chinese_titles(self) -> None:
with IsolatedDataRoot(user_id="_video_run"):
os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
config.reset_cache()
from service import task_service
mock_video = MagicMock()
mock_video.add_step = MagicMock()
mock_video.summary.return_value = {
"enabled": False,
"path": None,
"capture_path": None,
"record_log_path": None,
"warnings": [],
"voiceover_path": None,
"music_path": None,
"audio_warnings": [],
}
created: list[dict] = []
class _CapturingFakeSession:
def __init__(self, **kwargs):
created.append(kwargs)
async def __aenter__(self):
return mock_video
async def __aexit__(self, *args):
return None
with patch.object(task_service, "RpaVideoSession", _CapturingFakeSession):
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = task_service.cmd_run(target="demo-target", input_id="42")
self.assertEqual(rc, 1)
self.assertTrue(created, "RpaVideoSession should be constructed")
kwargs = created[0]
self.assertEqual(kwargs.get("title"), "开始执行示例任务")
self.assertEqual(kwargs.get("closing_title"), "示例任务执行完成")
self.assertTrue(re.search(r"[\u4e00-\u9fff]", kwargs.get("title", "")))
self.assertTrue(re.search(r"[\u4e00-\u9fff]", kwargs.get("closing_title", "")))
mock_video.add_step.assert_any_call("准备执行示例任务")
mock_video.add_step.assert_any_call("示例任务执行完成")
mock_video.summary.assert_called()
out = buf.getvalue()
self.assertIn("模板", out)
from db import task_logs_repository as tlr
rows = tlr.list_task_logs(1)
self.assertTrue(rows)
summary = json.loads(rows[0][7])
self.assertIn("video", summary)
self.assertIn("video_path", summary)
if __name__ == "__main__":
unittest.main()