From 1bbcc65cf3a3f87ad667f0f1991e34a606469b58 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 7 Jun 2026 11:28:45 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=96=B0=E8=B0=83=E6=95=B4=E6=A8=A1?= =?UTF-8?q?=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 19 +- SKILL.md | 10 +- references/CLI.md | 14 + references/CONFIG.md | 2 +- references/DEVELOPMENT.md | 29 +- references/RPA.md | 6 +- references/RUNTIME.md | 57 ++- references/TESTING.md | 3 +- requirements.txt | 8 +- scripts/jiangchang_skill_core/__init__.py | 16 - scripts/jiangchang_skill_core/config.py | 126 ----- scripts/jiangchang_skill_core/rpa/__init__.py | 66 --- .../jiangchang_skill_core/rpa/anti_detect.py | 216 -------- .../jiangchang_skill_core/rpa/artifacts.py | 29 -- scripts/jiangchang_skill_core/rpa/browser.py | 58 --- scripts/jiangchang_skill_core/rpa/errors.py | 20 - .../jiangchang_skill_core/rpa/human_login.py | 97 ---- scripts/jiangchang_skill_core/rpa/stealth.py | 67 --- .../rpa/video_session.py | 461 ------------------ scripts/jiangchang_skill_core/runtime_env.py | 109 ----- .../jiangchang_skill_core/unified_logging.py | 213 -------- scripts/main.py | 3 + scripts/service/sibling_bridge.py | 16 +- scripts/service/task_service.py | 17 +- scripts/util/constants.py | 3 +- tests/README.md | 9 +- tests/_support.py | 10 + tests/test_cli_smoke.py | 23 +- tests/test_platform_import.py | 117 +++++ tests/test_runtime_diagnostics_integration.py | 72 +++ tests/test_template_runtime_standard.py | 77 +++ 31 files changed, 448 insertions(+), 1525 deletions(-) delete mode 100644 scripts/jiangchang_skill_core/__init__.py delete mode 100644 scripts/jiangchang_skill_core/config.py delete mode 100644 scripts/jiangchang_skill_core/rpa/__init__.py delete mode 100644 scripts/jiangchang_skill_core/rpa/anti_detect.py delete mode 100644 scripts/jiangchang_skill_core/rpa/artifacts.py delete mode 100644 scripts/jiangchang_skill_core/rpa/browser.py delete mode 100644 scripts/jiangchang_skill_core/rpa/errors.py delete mode 100644 scripts/jiangchang_skill_core/rpa/human_login.py delete mode 100644 scripts/jiangchang_skill_core/rpa/stealth.py delete mode 100644 scripts/jiangchang_skill_core/rpa/video_session.py delete mode 100644 scripts/jiangchang_skill_core/runtime_env.py delete mode 100644 scripts/jiangchang_skill_core/unified_logging.py create mode 100644 tests/test_platform_import.py create mode 100644 tests/test_runtime_diagnostics_integration.py create mode 100644 tests/test_template_runtime_standard.py diff --git a/README.md b/README.md index 887b381..280f214 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ ## 模板目标 - 对齐当前规范 skill 的目录结构：`assets/`、`references/`、`scripts/`、`tests/`、`evals/` -- 对齐当前规范脚手架分层：`scripts/cli`、`scripts/db`、`scripts/service`、`scripts/util`、`scripts/jiangchang_skill_core` +- 对齐当前规范脚手架分层：`scripts/cli`、`scripts/db`、`scripts/service`、`scripts/util`；公共能力从共享 runtime 的 `jiangchang-platform-kit` 引用 - 提供最小可运行入口：`python scripts/main.py health` / `version`，以及通用任务命令骨架 `run` / `logs` / `log-get` - 内置 **隔离测试体系**：`IsolatedDataRoot`、双键环境、`OPENCLAW_TEST_TARGET` 档位与 adapter profile 策略（详见 [`references/TESTING.md`](references/TESTING.md)） - 让新技能从一开始就按规范落地，不再沿用旧模板的 `docs/`、`optional/`、`skill_main.py` 结构 @@ -56,9 +56,22 @@ python scripts/main.py version ## Python 依赖（requirements.txt） -- 在技能**根目录**维护 `requirements.txt`，声明本技能特有的 Python 三方包。 +- 模板默认声明 `jiangchang-platform-kit>=1.0.11` 作为公共 SDK 依赖；`jiangchang_skill_core` 来自共享 runtime，**不在技能仓库内复制**。 +- 新技能**不得 vendored** `scripts/jiangchang_skill_core/`；复制模板后若发现该目录，应删除并改为依赖共享 venv。 +- 在技能**根目录**维护 `requirements.txt`，声明本技能特有的 Python 三方包（在 platform-kit 之外按需追加）。 - 匠厂宿主在**安装或更新**技能后，会将该文件中的依赖安装到共享 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 中提示用户自行安装。 + +## 公共能力边界 + +以下能力均由 `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 诊断 diff --git a/SKILL.md b/SKILL.md index b6d5fd1..bdfb1d7 100644 --- a/SKILL.md +++ b/SKILL.md @@ -6,6 +6,7 @@ author: 深圳匠厂科技有限公司 metadata: openclaw: slug: your-skill-slug + platform_kit_min_version: "1.0.11" emoji: "📦" category: "通用" developer_ids: @@ -29,7 +30,7 @@ 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` 引用。 ## 最小命令 @@ -40,9 +41,12 @@ python {baseDir}/scripts/main.py version ## 运行依赖 -- 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.11`（`jiangchang_skill_core` 包）；**新技能不得**在仓库内保留 `scripts/jiangchang_skill_core/` 副本。 +- config、logging、runtime_env、rpa、media-assets、video_session、runtime_diagnostics 等均从共享 venv 的 `jiangchang_skill_core` import，而非技能目录副本。 +- 本技能**特有**的 Python 三方依赖请声明在根目录 `requirements.txt`（默认已含 `jiangchang-platform-kit>=1.0.11`）；宿主安装/更新技能后会自动安装到共享 venv。 - Skill 代码**不要**自行 `pip install`；系统级依赖（如 VC++ Runtime）仅在 `health` / preflight 中提示用户安装。 +- `health` 使用 `collect_runtime_diagnostics` 输出统一 runtime 诊断（只读，不下载/修复 media-assets）。 ## 重要说明 diff --git a/references/CLI.md b/references/CLI.md index db8ab48..3a0fb68 100644 --- a/references/CLI.md +++ b/references/CLI.md @@ -9,6 +9,20 @@ python {baseDir}/scripts/main.py health python {baseDir}/scripts/main.py version ``` +## health 标准输出（runtime diagnostics） + +`health` 委托 `jiangchang_skill_core.collect_runtime_diagnostics`，输出统一 runtime 诊断。典型字段： + +- `python_executable` — 当前 Python 解释器路径 +- `platform_kit_version` / `platform_kit_min_version` — 已安装与要求的 platform-kit 版本 +- `jiangchang_skill_core_file` — 公共库加载来源（若来自技能目录副本会报 warning） +- `media_assets_root` — 共享媒体资源根目录 +- `ffmpeg_path` / `ffmpeg_available` — ffmpeg 探测结果 +- `background_music_mp3_count` — 可用背景音乐数量 +- `runtime_issue[warning|error]` — 非致命/致命问题列表 + +`health` **只读诊断**，不下载、不修复 media-assets，不执行业务 DB / 邮箱 / 订单等检查。 + ## 通用业务命令骨架 模板默认提供以下命令（复制后按需保留或扩展）： diff --git a/references/CONFIG.md b/references/CONFIG.md index 2b9e0f2..b864cf1 100644 --- a/references/CONFIG.md +++ b/references/CONFIG.md @@ -69,7 +69,7 @@ config.ensure_env_file() # 首次把 .env.example copy 到数 ### 引用方式 -实现位于 `jiangchang-platform-kit` 的 `jiangchang_skill_core.config`；模板 vendor 拷贝在 `scripts/jiangchang_skill_core/config.py`： +`jiangchang_skill_core.config` 来自共享 runtime 安装的 `jiangchang-platform-kit`；业务代码统一 `from jiangchang_skill_core import config`，**不要在技能内复制 config.py**： ```python from jiangchang_skill_core import config diff --git a/references/DEVELOPMENT.md b/references/DEVELOPMENT.md index 662d536..f489d42 100644 --- a/references/DEVELOPMENT.md +++ b/references/DEVELOPMENT.md @@ -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.11` 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,8 +172,9 @@ scripts/ ### 3.1 `requirements.txt` 依赖规范 -技能根目录的 `requirements.txt` 是**标准文件**，用于声明本技能特有的 Python 三方依赖。 +技能根目录的 `requirements.txt` 是**标准文件**，用于声明本技能 Python 三方依赖。 +- 模板默认包含 `jiangchang-platform-kit>=1.0.11`（公共 SDK）；复制后保留该行，再按需追加技能特有依赖。 - 匠厂宿主安装/更新技能后，会将其安装到共享 venv：`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。 - **不要**在业务代码中 `subprocess` / `pip install`；缺依赖由 `health` 报错，由宿主负责安装。 - **版本约束尽量收窄**，降低多技能共享 venv 时的冲突风险。推荐范围写法： @@ -186,7 +183,23 @@ scripts/ - 不推荐：`chromadb>=0.5.0` - 推荐：`chromadb>=0.5.23,<0.6` - **不要**把系统组件（VC++ Runtime、浏览器安装包等）写进 `requirements.txt`；这类前置条件写在 `health` / preflight 文档与错误提示中。 -- 若技能无额外 Python 依赖，可保留仅含注释的 `requirements.txt`（宿主会识别为 empty 并跳过安装）。 +- 若技能无额外 Python 依赖，可保留仅含 `jiangchang-platform-kit>=1.0.11` 的最小 `requirements.txt`。 + +### 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 的标准步骤 diff --git a/references/RPA.md b/references/RPA.md index 964f9bb..a7e33d3 100644 --- a/references/RPA.md +++ b/references/RPA.md @@ -62,10 +62,9 @@ pip install playwright # 仅装包；用系统 Chrome，无需 playwright inst ### 引用方式 -共享实现位于 `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 +75,9 @@ from jiangchang_skill_core.rpa import ( from jiangchang_skill_core.rpa.stealth import stealth_enabled, STEALTH_INIT_SCRIPT ``` +- `RpaVideoSession` 来自 platform-kit；ffmpeg、背景音乐、media-assets 由 platform-kit 统一解析。 +- `health` 对上述资源做只读诊断，不下载、不修复。 + 完整示例见 `scripts/service/example_browser_rpa.py` 与 `scripts/service/example_adapter/sim_rpa.py`。 --- diff --git a/references/RUNTIME.md b/references/RUNTIME.md index 5bcfc27..196d2de 100644 --- a/references/RUNTIME.md +++ b/references/RUNTIME.md @@ -1,5 +1,51 @@ # 运行时约定 +## 共享 Python Runtime + +**skill-template** 及复制出的新技能，公共能力均来自宿主匠厂安装的共享 Python Runtime（`jiangchang-platform-kit>=1.0.11` 及其传递依赖）。`jiangchang_skill_core` **不得**在技能仓库内 vendored，应由共享 venv 的 site-packages 提供。 + +| 场景 | 推荐做法 | +|------|----------| +| 日常运行 | 由宿主匠厂触发技能 | +| 手工排查 / 测试机 | 使用共享 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 + +通用: + {baseDir}/scripts/main.py health +``` + +`` 通常位于 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。数据根由宿主注入；开发模式下也可能通过 `CLAW_DATA_ROOT` 解析（见 `jiangchang_skill_core.runtime_env`）。 + +## Runtime 诊断（platform-kit 1.0.11+） + +`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 下载或修复。 + +## media-assets / ffmpeg / 背景音乐 + +背景音乐、ffmpeg 等共享资源由 platform-kit 统一解析，默认路径： + +```text +{JIANGCHANG_DATA_ROOT}/shared/media-assets +``` + +（与 `CLAW_DATA_ROOT` 解析规则一致，由宿主注入。） + +RPA 录屏成片（`RpaVideoSession`）、ffmpeg 路径、背景音乐探测均由 platform-kit 提供；技能只 import，不重复实现。 + ## 目录结构 新技能建议采用以下根目录结构： @@ -10,14 +56,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 +78,12 @@ {...}/{skill_slug}.db ``` +## 明确禁止 + +- **不得** vendored `scripts/jiangchang_skill_core/` +- **不得**在技能内本地重复实现 `RuntimeDiagnostics` / `collect_runtime_diagnostics` +- **不得**写死开发机绝对路径作为运行时约定 + ## 测试时的运行时隔离 本模板的 `tests/_support.IsolatedDataRoot` 会在测试期间同时设置： diff --git a/references/TESTING.md b/references/TESTING.md index aa457ab..5c8347c 100644 --- a/references/TESTING.md +++ b/references/TESTING.md @@ -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.11` 导入来源、文档/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)； diff --git a/requirements.txt b/requirements.txt index 4d8e491..b410b02 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,9 @@ # Skill-specific Python dependencies. # Keep this file minimal. -# Platform-provided shared dependencies are installed by Jiangchang runtime. -# Add only dependencies required by this skill. +# Platform shared APIs are provided by jiangchang-platform-kit from Jiangchang shared runtime. # -# Browser RPA: uses system Chrome/Edge; no `playwright install chromium` needed. +# Required platform SDK. Do not vendor jiangchang_skill_core inside a skill. +jiangchang-platform-kit>=1.0.11 + +# Browser RPA example: uses system Chrome/Edge; no `playwright install chromium` needed. playwright>=1.42.0,<2 diff --git a/scripts/jiangchang_skill_core/__init__.py b/scripts/jiangchang_skill_core/__init__.py deleted file mode 100644 index 1867304..0000000 --- a/scripts/jiangchang_skill_core/__init__.py +++ /dev/null @@ -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", -] diff --git a/scripts/jiangchang_skill_core/config.py b/scripts/jiangchang_skill_core/config.py deleted file mode 100644 index 621a6b1..0000000 --- a/scripts/jiangchang_skill_core/config.py +++ /dev/null @@ -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 diff --git a/scripts/jiangchang_skill_core/rpa/__init__.py b/scripts/jiangchang_skill_core/rpa/__init__.py deleted file mode 100644 index 20ec47c..0000000 --- a/scripts/jiangchang_skill_core/rpa/__init__.py +++ /dev/null @@ -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, -) diff --git a/scripts/jiangchang_skill_core/rpa/anti_detect.py b/scripts/jiangchang_skill_core/rpa/anti_detect.py deleted file mode 100644 index d86c6e5..0000000 --- a/scripts/jiangchang_skill_core/rpa/anti_detect.py +++ /dev/null @@ -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)) diff --git a/scripts/jiangchang_skill_core/rpa/artifacts.py b/scripts/jiangchang_skill_core/rpa/artifacts.py deleted file mode 100644 index d08668d..0000000 --- a/scripts/jiangchang_skill_core/rpa/artifacts.py +++ /dev/null @@ -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 diff --git a/scripts/jiangchang_skill_core/rpa/browser.py b/scripts/jiangchang_skill_core/rpa/browser.py deleted file mode 100644 index 8d4ec77..0000000 --- a/scripts/jiangchang_skill_core/rpa/browser.py +++ /dev/null @@ -1,58 +0,0 @@ -"""统一 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, -) -> "BrowserContext": - """用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。 - - headless=None 时读 OPENCLAW_BROWSER_HEADLESS。 - 录屏由 RpaVideoSession（ffmpeg）负责，本函数不向 Playwright 传递任何录屏参数。 - """ - 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 - - context = await playwright.chromium.launch_persistent_context(**launch_kwargs) - if stealth_enabled(): - await context.add_init_script(STEALTH_INIT_SCRIPT) - return context diff --git a/scripts/jiangchang_skill_core/rpa/errors.py b/scripts/jiangchang_skill_core/rpa/errors.py deleted file mode 100644 index ce17562..0000000 --- a/scripts/jiangchang_skill_core/rpa/errors.py +++ /dev/null @@ -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", -] diff --git a/scripts/jiangchang_skill_core/rpa/human_login.py b/scripts/jiangchang_skill_core/rpa/human_login.py deleted file mode 100644 index 267b2ca..0000000 --- a/scripts/jiangchang_skill_core/rpa/human_login.py +++ /dev/null @@ -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 diff --git a/scripts/jiangchang_skill_core/rpa/stealth.py b/scripts/jiangchang_skill_core/rpa/stealth.py deleted file mode 100644 index ee5eff8..0000000 --- a/scripts/jiangchang_skill_core/rpa/stealth.py +++ /dev/null @@ -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"] diff --git a/scripts/jiangchang_skill_core/rpa/video_session.py b/scripts/jiangchang_skill_core/rpa/video_session.py deleted file mode 100644 index 072d258..0000000 --- a/scripts/jiangchang_skill_core/rpa/video_session.py +++ /dev/null @@ -1,461 +0,0 @@ -"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。""" - -from __future__ import annotations - -import asyncio -import os -import random -import shutil -import subprocess -import sys -import time -from dataclasses import dataclass -from datetime import datetime -from pathlib import Path -from typing import Any, Dict, IO, List, Optional - -from .. import config - -_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", -) - -_CAPTURE_FRAMERATE = 15 - - -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: - 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 _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 _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]: - """Windows gdigrab 全桌面录制；非 Windows 返回 None。""" - if sys.platform != "win32": - return None - return [ - "ffmpeg", - "-y", - "-f", - "gdigrab", - "-framerate", - str(_CAPTURE_FRAMERATE), - "-i", - "desktop", - "-c:v", - "libx264", - "-preset", - "ultrafast", - "-pix_fmt", - "yuv420p", - capture_path, - ] - - -def _compose_capture_mp4( - capture_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(capture_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(capture_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 层统一 ffmpeg 桌面录屏；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.capture_path: 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._subtitles_dir = os.path.join(self._artifact_root, "subtitles") - self._logs_dir = os.path.join(self._artifact_root, "logs") - self._ffmpeg_record_log_path = os.path.join(self._logs_dir, "ffmpeg-record.log") - self._capture_path = os.path.join(self._artifact_root, "capture.mp4") - 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 - self._ffmpeg_proc: Optional[subprocess.Popen] = None - self._ffmpeg_record_log_file: Optional[IO[str]] = None - - if self.enabled: - os.makedirs(self._videos_dir, exist_ok=True) - os.makedirs(self._subtitles_dir, exist_ok=True) - os.makedirs(self._logs_dir, exist_ok=True) - os.makedirs(self._artifact_root, exist_ok=True) - - def _record_log_hint(self) -> str: - if self.enabled and self._ffmpeg_record_log_path: - return f"; record_log={self._ffmpeg_record_log_path}" - return "" - - def _append_warning(self, code: str, detail: str = "") -> None: - msg = code if not detail else f"{code}: {detail}" - if code in ( - "ffmpeg_record_start_failed", - "ffmpeg_capture_missing", - "ffmpeg_compose_failed", - ): - msg += self._record_log_hint() - self.warnings.append(msg) - - def step(self, text: str, *, duration: float = 4.0) -> None: - self.add_step(text, duration=duration) - - def add_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)) - - def _close_ffmpeg_record_log(self) -> None: - fh = self._ffmpeg_record_log_file - self._ffmpeg_record_log_file = None - if fh is None: - return - try: - fh.close() - except Exception: - pass - - def _start_ffmpeg_capture(self) -> None: - if not self.enabled: - return - if not shutil.which("ffmpeg"): - self.warnings.append("ffmpeg_not_found") - return - - cmd = _build_desktop_record_cmd(self._capture_path) - if cmd is None: - self._append_warning( - "ffmpeg_record_start_failed", - "desktop capture only supported on Windows", - ) - return - - stderr_target: Any = subprocess.DEVNULL - try: - os.makedirs(self._logs_dir, exist_ok=True) - self._ffmpeg_record_log_file = open( - self._ffmpeg_record_log_path, "a", encoding="utf-8" - ) - stderr_target = self._ffmpeg_record_log_file - except Exception as exc: - self.warnings.append(f"ffmpeg_record_log_open_failed: {exc}") - - try: - self._ffmpeg_proc = subprocess.Popen( - cmd, - stdin=subprocess.PIPE, - stdout=subprocess.DEVNULL, - stderr=stderr_target, - ) - self.capture_path = self._capture_path - except Exception as exc: - self._append_warning("ffmpeg_record_start_failed", str(exc)) - self._ffmpeg_proc = None - self._close_ffmpeg_record_log() - - def _stop_ffmpeg_capture(self) -> None: - proc = self._ffmpeg_proc - if proc is None: - self._close_ffmpeg_record_log() - return - if proc.poll() is None: - try: - if proc.stdin: - proc.stdin.write(b"q") - proc.stdin.flush() - except Exception: - try: - proc.terminate() - except Exception: - pass - try: - proc.wait(timeout=20) - except subprocess.TimeoutExpired: - try: - proc.kill() - except Exception: - pass - try: - proc.wait(timeout=5) - except Exception: - pass - self._ffmpeg_proc = None - self._close_ffmpeg_record_log() - time.sleep(0.3) - - async def __aenter__(self) -> "RpaVideoSession": - if self.enabled: - self._t0 = time.monotonic() - if self.title: - self.add_step(self.title, duration=3.0) - self._start_ffmpeg_capture() - return self - - async def __aexit__(self, exc_type, exc, tb) -> None: - if self.enabled: - self._stop_ffmpeg_capture() - await self.finalize() - - async def finalize(self) -> None: - """停止录制后合成最终 MP4；失败只记 warning，不抛异常。""" - if not self.enabled: - self.artifacts = self.summary() - return - - if not self._steps and self.title: - self.add_step(self.title) - - try: - capture = Path(self._capture_path) - if not capture.is_file() or capture.stat().st_size <= 0: - self._append_warning("ffmpeg_capture_missing") - self.artifacts = self.summary() - return - - self.capture_path = str(capture.resolve()) - - entries = list(self._steps) - if entries: - entries[-1].duration = max(entries[-1].duration, 5.0) - try: - _generate_srt(entries, self._srt_path) - except Exception as exc: - self.warnings.append(f"subtitle_write_failed: {exc}") - - music = _resolve_music_file() - if music is None: - self.warnings.append("background_music_missing") - - srt_file = Path(self._srt_path) - use_srt = srt_file.is_file() and srt_file.stat().st_size > 0 - - ok, err = _compose_capture_mp4( - capture, - srt_file if use_srt else 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_capture_mp4( - capture, - srt_file if use_srt else Path(self._srt_path), - Path(self._planned_output), - music_file=None, - ) - if not ok and not use_srt: - ok, err = _run_ffmpeg( - [ - "ffmpeg", - "-y", - "-i", - str(capture), - "-c:v", - "libx264", - "-preset", - "fast", - "-crf", - "23", - "-pix_fmt", - "yuv420p", - str(self._planned_output), - ] - ) - if not ok: - self._append_warning("ffmpeg_compose_failed", err) - self.artifacts = self.summary() - return - - self.output_video_path = str(Path(self._planned_output).resolve()) - except Exception as exc: - self.warnings.append(f"video_finalize_error: {exc}") - - self.artifacts = self.summary() - - def summary(self) -> Dict[str, Any]: - record_log = ( - os.path.abspath(self._ffmpeg_record_log_path) - if self.enabled and self._ffmpeg_record_log_path - else None - ) - return { - "enabled": self.enabled, - "path": self.output_video_path, - "capture_path": self.capture_path, - "record_log_path": record_log, - "warnings": list(self.warnings), - } - - def to_artifact_dict(self) -> Dict[str, Any]: - data = self.summary() - data["batch_id"] = self.batch_id - return data diff --git a/scripts/jiangchang_skill_core/runtime_env.py b/scripts/jiangchang_skill_core/runtime_env.py deleted file mode 100644 index 6d93477..0000000 --- a/scripts/jiangchang_skill_core/runtime_env.py +++ /dev/null @@ -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() diff --git a/scripts/jiangchang_skill_core/unified_logging.py b/scripts/jiangchang_skill_core/unified_logging.py deleted file mode 100644 index 0e3c788..0000000 --- a/scripts/jiangchang_skill_core/unified_logging.py +++ /dev/null @@ -1,213 +0,0 @@ -""" -统一文件日志：{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_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, Tuple - -from .runtime_env import get_data_root, get_user_id - -_skill_slug: str = "" -_logger_name: str = "" - -# 避免 logging 内部错误再向 stderr 打印 "--- Logging error ---" -logging.raiseExceptions = False - - -def get_unified_logs_dir() -> str: - path = os.path.join(get_data_root(), get_user_id(), "logs") - try: - os.makedirs(path, exist_ok=True) - except OSError: - pass - return path - - -def get_skill_log_file_path() -> str: - """与 setup 写入的主日志文件一致（供终端提示等）。""" - override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip() - if override: - parent = os.path.dirname(os.path.abspath(override)) - if parent: - try: - os.makedirs(parent, exist_ok=True) - except OSError: - pass - return os.path.abspath(override) - return os.path.join(get_unified_logs_dir(), "jiangchang.log") - - -def _can_open_log_file(log_path: str) -> Tuple[bool, str]: - """探测日志文件是否可追加写入（真实 open，非 os.access）。""" - try: - parent = os.path.dirname(os.path.abspath(log_path)) - if parent: - os.makedirs(parent, exist_ok=True) - with open(log_path, "a", encoding="utf-8"): - pass - return True, "" - except Exception as exc: - return False, str(exc) - - -def ensure_trace_for_process() -> str: - """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。""" - 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: - """subprocess.run(..., env=...) 时使用，保证带上当前 trace。""" - ensure_trace_for_process() - tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() - if not tid: - tid = 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 _clear_logger_handlers(log: logging.Logger) -> None: - for h in list(log.handlers): - log.removeHandler(h) - try: - h.close() - except Exception: - pass - - -def _attach_fallback_handler(log: logging.Logger) -> None: - """文件日志不可用时仅挂 NullHandler，不向 stderr 打印 logging 异常。""" - log.addHandler(logging.NullHandler()) - - -def setup_skill_logging(skill_slug: str, logger_name: str) -> None: - """ - 幂等：为指定 logger 挂载统一文件日志与可选 stderr。 - 须在进程早期调用（如 CLI main 在业务日志之前）。 - 文件日志不可写时降级为 NullHandler，不影响 CLI 返回码。 - """ - global _skill_slug, _logger_name - logging.raiseExceptions = False - ensure_trace_for_process() - _skill_slug = skill_slug - _logger_name = logger_name - - log = logging.getLogger(logger_name) - _clear_logger_handlers(log) - log.setLevel(_log_level_from_env()) - fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT) - filt = _SkillContextFilter() - - log_path = get_skill_log_file_path() - ok, _err = _can_open_log_file(log_path) - if ok: - try: - fh = TimedRotatingFileHandler( - log_path, - when="midnight", - interval=1, - backupCount=_backup_count(), - encoding="utf-8", - delay=False, - ) - fh.setFormatter(fmt) - fh.addFilter(filt) - log.addHandler(fh) - except Exception: - _attach_fallback_handler(log) - else: - _attach_fallback_handler(log) - - if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in ( - "1", - "true", - "yes", - "on", - ): - try: - sh = logging.StreamHandler(sys.stderr) - sh.setLevel(logging.WARNING) - sh.setFormatter(fmt) - sh.addFilter(filt) - log.addHandler(sh) - except Exception: - pass - 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: - """ - 独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。 - """ - global _skill_slug - logging.raiseExceptions = False - ensure_trace_for_process() - _skill_slug = skill_slug - lg = logging.getLogger(logger_name) - _clear_logger_handlers(lg) - lg.setLevel(level) - fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT) - filt = _SkillContextFilter() - - ok, _err = _can_open_log_file(log_path) - if ok: - try: - fh = logging.FileHandler(log_path, encoding="utf-8", delay=False) - fh.setFormatter(fmt) - fh.addFilter(filt) - lg.addHandler(fh) - except Exception: - _attach_fallback_handler(lg) - else: - _attach_fallback_handler(lg) - lg.propagate = False - return lg diff --git a/scripts/main.py b/scripts/main.py index 8813480..4b0eed6 100644 --- a/scripts/main.py +++ b/scripts/main.py @@ -5,6 +5,9 @@ skill-template CLI 入口。 复制为新技能后，请修改注释与常量，但保留当前分层结构： cli（argv）→ service（业务编排）→ db（持久化）→ util（通用工具）。 + +jiangchang_skill_core 来自宿主共享 Python runtime 安装的 jiangchang-platform-kit， +不在技能仓库内 vendored。 """ from __future__ import annotations diff --git a/scripts/service/sibling_bridge.py b/scripts/service/sibling_bridge.py index 657b548..521daae 100644 --- a/scripts/service/sibling_bridge.py +++ b/scripts/service/sibling_bridge.py @@ -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）， -# 那是发布类技能的特定需求，不属于通用模板。 -# ============================================================ diff --git a/scripts/service/task_service.py b/scripts/service/task_service.py index b31d775..baab75d 100644 --- a/scripts/service/task_service.py +++ b/scripts/service/task_service.py @@ -7,12 +7,14 @@ from __future__ import annotations import json -import sys from typing import Optional +from jiangchang_skill_core import collect_runtime_diagnostics, format_runtime_health_lines + from db import task_logs_repository as tlr from service.entitlement_service import check_entitlement -from util.constants import SKILL_SLUG, SKILL_VERSION +from util.constants import PLATFORM_KIT_MIN_VERSION, SKILL_SLUG, SKILL_VERSION +from util.runtime_paths import get_skill_root from util.timeutil import unix_to_iso @@ -89,7 +91,16 @@ def cmd_log_get(log_id: str) -> int: 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(), + ) + status = "failed" if runtime.has_fatal_issues else "ok" + print(f"{SKILL_SLUG} health: {status}") + for line in format_runtime_health_lines(runtime): + print(line) + return 1 if runtime.has_fatal_issues else 0 def cmd_version() -> int: diff --git a/scripts/util/constants.py b/scripts/util/constants.py index d44c0a5..471c5af 100644 --- a/scripts/util/constants.py +++ b/scripts/util/constants.py @@ -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.11" diff --git a/tests/README.md b/tests/README.md index 1f0d489..6bc2efb 100644 --- a/tests/README.md +++ b/tests/README.md @@ -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 diagnostics；warning 不导致非零退出）。 - [ ] `python scripts/main.py version` 输出 JSON，且 `skill` 与目录名 / `SKILL.md` / `constants.SKILL_SLUG` 一致。 - [ ] 所有 DB / 文件写入都在 `IsolatedDataRoot`（或等价隔离）下测试，不写真实数据目录。 - [ ] 外部系统默认使用 mock / `FakeAdapter`，不访问真实 API，不打开真实 RPA。 @@ -74,6 +74,11 @@ 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`` | + +复制为新技能后**不得删除**上述架构守护测试，除非同步更新模板标准并理解影响。 ### 1.3 数据隔离（``_support.IsolatedDataRoot``） diff --git a/tests/_support.py b/tests/_support.py index 62a551b..c674470 100644 --- a/tests/_support.py +++ b/tests/_support.py @@ -39,6 +39,16 @@ def get_skill_root() -> str: return _SKILL_ROOT +def platform_kit_version_patch(version: str = "1.0.11"): + """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: """ 在独立临时目录下模拟数据根，避免污染本机默认数据目录。 diff --git a/tests/test_cli_smoke.py b/tests/test_cli_smoke.py index 33a0909..25f212c 100644 --- a/tests/test_cli_smoke.py +++ b/tests/test_cli_smoke.py @@ -8,7 +8,7 @@ import os import unittest from contextlib import redirect_stderr, redirect_stdout -from _support import IsolatedDataRoot +from _support import IsolatedDataRoot, platform_kit_version_patch # scripts/ 已由 _support 注入 sys.path from cli.app import main @@ -27,10 +27,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() diff --git a/tests/test_platform_import.py b/tests/test_platform_import.py new file mode 100644 index 0000000..939a023 --- /dev/null +++ b/tests/test_platform_import.py @@ -0,0 +1,117 @@ +# -*- 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() + + +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_11(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.11") + + 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.11") + + req_path = os.path.join(get_skill_root(), "requirements.txt") + with open(req_path, encoding="utf-8") as f: + req = f.read() + self.assertIn("jiangchang-platform-kit>=1.0.11", req) + + installed = metadata.version("jiangchang-platform-kit") + 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() diff --git a/tests/test_runtime_diagnostics_integration.py b/tests/test_runtime_diagnostics_integration.py new file mode 100644 index 0000000..1804760 --- /dev/null +++ b/tests/test_runtime_diagnostics_integration.py @@ -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() diff --git a/tests/test_template_runtime_standard.py b/tests/test_template_runtime_standard.py new file mode 100644 index 0000000..e063e72 --- /dev/null +++ b/tests/test_template_runtime_standard.py @@ -0,0 +1,77 @@ +# -*- 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", +) + +POSITIVE_MARKERS = ( + "jiangchang-platform-kit", + "1.0.11", + "共享 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) + + +if __name__ == "__main__": + unittest.main()