chore: package env example in skill releases

Co-authored-by: Cursor <cursoragent@cursor.com>

.github/workflows/publish.yml

@@ -2,8 +2,10 @@ name: Publish Python Package to Gitea
 
 on:
   push:
+    branches:
+      - main
     tags:
-      - 'v*'
+      - "v*"
   workflow_dispatch:
 
 jobs:
@@ -23,6 +25,13 @@ jobs:
     steps:
       - uses: https://git.jc2009.com/admin/actions-checkout@v4
 
+      - name: Resolve and patch package version
+        env:
+          GITHUB_REF: ${{ github.ref }}
+          GITHUB_RUN_NUMBER: ${{ github.run_number }}
+          GITHUB_RUN_ID: ${{ github.run_id }}
+        run: python3.12 tools/ci_set_package_version.py
+
       - name: Install build tools
         run: python3.12 -m pip install --upgrade build twine --break-system-packages
 
@@ -36,4 +45,4 @@ jobs:
         run: |
           python3.12 -m twine upload \
             --repository-url https://git.jc2009.com/api/packages/client-jiangchang/pypi \
-            dist/*
+            dist/*

.github/workflows/reusable-release-skill.yaml

@@ -113,6 +113,20 @@ jobs:
             "copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \
             "copy_dir('evals', os.path.join(PACKAGE, 'evals'), common_ignore)" \
             '' \
+            "req_src = os.path.abspath('requirements.txt')" \
+            "req_dst = os.path.join(PACKAGE, 'requirements.txt')" \
+            'if os.path.isfile(req_src):' \
+            "    shutil.copy2(req_src, req_dst)" \
+            "    print('Copied requirements.txt')" \
+            '' \
+            "env_example_src = os.path.abspath('.env.example')" \
+            "env_example_dst = os.path.join(PACKAGE, '.env.example')" \
+            'if os.path.isfile(env_example_src):' \
+            "    shutil.copy2(env_example_src, env_example_dst)" \
+            "    print('Copied .env.example')" \
+            'if os.path.isfile(env_example_src) and not os.path.isfile(env_example_dst):' \
+            "    raise RuntimeError('.env.example exists in skill root but was not packaged')" \
+            '' \
             "print('Package top-level entries:')" \
             'for name in sorted(os.listdir(PACKAGE)):' \
             "    print(' -', name)" \

README.md

@@ -54,15 +54,66 @@ finally:
     client.disconnect()
 ```
 
-## 5. 发版流程（维护者）
+### 桌面 E2E（`jiangchang_desktop_sdk.e2e_helpers`）
 
-```bash
-git tag v0.1.0
-git push origin v0.1.0
-# CI 自动打包并发布到 Gitea Package Registry（见 .github/workflows/publish.yml）
+Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录，不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` / `CLAW_USER_ID` 来伪造用户（否则会回退到 `_anon`）。
+
+推荐 preflight 流程：
+
+1. `require_logged_in_skill_data_dir(page, skill_slug)` — 经 `jiangchang:resolve-skill-data-dir` 解析路径，并断言已登录（非 `_anon` / `anon` / `anonymous`）。
+2. `assert_skill_env_file(ctx, ["API_KEY", ...])` — 检查 `{skill_data}/config/.env` 中必填项已配置且非占位符；E2E 不会自动生成真实密钥。
+3. `send_prompt_via_composer(page, prompt)` — 普通文本逐字键入；内容中的 `\n` / `\r\n` / `\r` 用 **Shift+Enter** 输入换行；最后用 **Enter** 发送（禁止 DOM 注入、剪贴板、JS 伪造 input/change）。
+
+诊断时可调用 `get_runtime_env(page)` 查看主进程解析的环境，**不要**用它拼用户数据路径。
+
+```python
+from jiangchang_desktop_sdk.e2e_helpers import (
+    assert_skill_env_file,
+    require_logged_in_skill_data_dir,
+    send_prompt_via_composer,
+)
+
+ctx = require_logged_in_skill_data_dir(page, "demo-skill")
+assert_skill_env_file(ctx, ["API_KEY"])
+send_prompt_via_composer(page, "第一行\n第二行")
 ```
 
-仓库内还保留 `examples/`、`tools/`、`python-runtime/` 目录及既有 Skill / 前端相关的 GitHub Actions 工作流，与本 PyPI 包的发布彼此独立。
+## 5. 发版流程（维护者）
+
+### main 分支自动预发布
+
+每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **唯一** 的 post 版本（基于 `pyproject.toml` 中的基础版本，例如 `0.1.0.post42`）。版本号不会写回仓库，仅在 CI 构建前临时改写 `pyproject.toml` 与 `jiangchang_desktop_sdk.__version__`。
+
+```bash
+git push origin main
+```
+
+测试人员升级安装：
+
+```bash
+python -m pip install --upgrade \
+  --index-url https://git.jc2009.com/api/packages/client-jiangchang/pypi/simple/ \
+  --extra-index-url https://pypi.org/simple \
+  jiangchang-platform-kit
+```
+
+### 正式版本（tag）
+
+打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v0.1.1` → PyPI 包 `0.1.1`）：
+
+```bash
+git tag v0.1.1
+git push origin v0.1.1
+```
+
+### 安装后验证
+
+```bash
+python -c "from jiangchang_desktop_sdk.e2e_helpers import send_prompt_via_composer; print('e2e_helpers OK')"
+python -c "from screencast import run_screencast; print('screencast OK')"
+```
+
+仓库内还保留 `examples/`、`tools/`、`python-runtime/` 目录及既有 Skill / 前端相关的 GitHub Actions 工作流，与本 PyPI 包的发布彼此独立。`tools/screencast/` 为兼容薄壳，pip 包内实现位于 `screencast` 包（`src/screencast/`）。
 
 ## v0.2.0 — data-jcid migration (internal refactor)
 

pyproject.toml

@@ -13,6 +13,7 @@ authors = [{ name = "client-jiangchang" }]
 dependencies = [
     "requests>=2.31.0",
     "playwright>=1.42.0",
+    "mss>=9.0.1",
 ]
 
 [project.urls]

sanity_check.py

@@ -0,0 +1,68 @@
+"""
+Stage 2 集成验证脚本——一次性 sanity check。
+
+跑前准备:
+  1. 启动匠厂客户端(确认能正常打开聊天页 + 至少配好一个可用模型)
+  2. 在本仓库激活 Python venv,确认 playwright 已装
+  3. 直接 python sanity_check.py 即可
+"""
+import sys
+import time
+import traceback
+
+print("=" * 60)
+print("Stage 2 SDK Sanity Check")
+print("=" * 60)
+
+try:
+    from jiangchang_desktop_sdk import JiangchangDesktopClient
+    print("[1/6] import OK")
+except Exception as e:
+    print(f"[FAIL] import failed: {e}")
+    sys.exit(1)
+
+try:
+    with JiangchangDesktopClient() as c:
+        print("[2/6] context entered")
+
+        c.ensure_app_running()
+        print("[3/6] ensure_app_running OK")
+
+        c.wait_gateway_ready()
+        print("[4/6] wait_gateway_ready OK (note: returns even if gw not actually ready)")
+
+        c.new_task()
+        print("[5/6] new_task OK")
+
+        t0 = time.time()
+        answer = c.ask("你好,请用一句话回复,确认 SDK 可用")
+        elapsed = time.time() - t0
+        print(f"[6/6] ask() returned in {elapsed:.1f}s")
+        print(f"      answer length: {len(answer)} chars")
+        print(f"      answer preview: {answer[:200]!r}")
+        assert answer.strip(), "Empty answer!"
+
+        print()
+        print("[BONUS] verifying read() works")
+        msgs = c.read()
+        print(f"      read() returned {len(msgs)} messages")
+        for i, m in enumerate(msgs[-4:]):
+            print(f"      [-{len(msgs)-i}] role={m.role!r} id={m.id!r} "
+                  f"content_len={len(m.content) if m.content else 0}")
+
+        print()
+        print("[BONUS] verifying send_file raises NotImplementedError")
+        try:
+            c.send_file("/tmp/fake.txt")
+            print("      [FAIL] send_file did NOT raise!")
+        except NotImplementedError as e:
+            print(f"      [OK] send_file raised: {str(e)[:80]}...")
+
+    print()
+    print("=" * 60)
+    print("ALL CHECKS PASSED ✓")
+    print("=" * 60)
+except Exception as e:
+    print(f"\n[FAIL] sanity check crashed:")
+    traceback.print_exc()
+    sys.exit(2)

src/jiangchang_desktop_sdk/client.py

@@ -219,63 +219,13 @@ class JiangchangDesktopClient:
         return False
 
     def _activate_and_maximize_windows(self) -> bool:
-        """Windows 专属：用 Win32 API 把主窗口拉到桌面最前并最大化。
+        """Windows 专属：把匠厂主窗口拉到桌面最前并最大化。
 
-        EnumWindows 遍历所有顶层窗口，按标题模糊匹配（含"匠厂/Jiangchang/ClawX/OpenClaw"），
-        找到后 ShowWindow(SW_MAXIMIZE) + SetForegroundWindow，让录屏看到铺满桌面的匠厂，
-        避免任务栏边角和其他应用露出来。
-
-        为什么改用 ctypes（替代上一版 PowerShell AppActivate）：
-        1) AppActivate 只激活不最大化，依然能录到桌面边缘其他窗口；
-        2) ctypes 不 fork powershell，启动更快（~0ms vs ~300-800ms）；
-        3) 直接拿到 hwnd 后续可扩展（如监控窗口尺寸）。
+        实现已抽到 jiangchang_desktop_sdk.window_win32 模块（SDK 子模块），
+        给 screencast 工具也复用。本方法保留为类内入口（兼容旧调用）。
         """
-        if os.name != "nt":
-            return False
-        try:
-            import ctypes
-            from ctypes import wintypes
-        except Exception as exc:
-            _logger.debug("[bring_to_front] ctypes 不可用：%s", exc)
-            return False
-
-        user32 = ctypes.windll.user32
-        SW_MAXIMIZE = 3
-        keywords = ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
-        matching: list = []
-
-        @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
-        def _enum_proc(hwnd, _lparam):
-            if not user32.IsWindowVisible(hwnd):
-                return True
-            length = user32.GetWindowTextLengthW(hwnd)
-            if length == 0:
-                return True
-            buf = ctypes.create_unicode_buffer(length + 1)
-            user32.GetWindowTextW(hwnd, buf, length + 1)
-            if any(kw in buf.value for kw in keywords):
-                matching.append((hwnd, buf.value))
-            return True
-
-        try:
-            user32.EnumWindows(_enum_proc, 0)
-        except Exception as exc:
-            _logger.debug("[bring_to_front] EnumWindows 失败：%s", exc)
-            return False
-
-        if not matching:
-            _logger.debug("[bring_to_front] 未找到匠厂主窗口")
-            return False
-
-        hwnd, title = matching[0]
-        try:
-            user32.ShowWindow(hwnd, SW_MAXIMIZE)
-            user32.SetForegroundWindow(hwnd)
-            _logger.debug("[bring_to_front] 已激活+最大化 hwnd=%s title=%s", hwnd, title)
-            return True
-        except Exception as exc:
-            _logger.debug("[bring_to_front] ShowWindow/SetForegroundWindow 失败：%s", exc)
-            return False
+        from .window_win32 import activate_and_maximize_jiangchang_window
+        return activate_and_maximize_jiangchang_window()
 
     def _launch_via_executable(self, cdp_port: int) -> bool:
         """回退：直接 spawn 可执行文件，带 --remote-debugging-port。"""

src/jiangchang_desktop_sdk/e2e_helpers.py

@@ -0,0 +1,470 @@
+# -*- coding: utf-8 -*-
+"""
+桌面 E2E 通用 helper：所有 skill 的 desktop E2E 测试共用。
+
+为什么独立出来：
+- disburse-payroll-icbc 实战验证：SDK 的 ask() 在 user_idx<0 时有死循环 bug，
+  且 send_file() 是 NotImplementedError。需要绕开它们自写完整测试流程
+- 这些 helper（拖拽附件、等 user 节点、多信号合流等待 agent 完成）是
+  **所有 skill 都通用的**，不应该每个 skill 都内嵌一份
+
+设计原则：
+- 所有 helper 只接受 Playwright Page 作为入口（不依赖 SDK client 实例）
+- 时间常量默认值留出宽容（user message 60s / agent complete 1200s）
+- 失败一律抛 TimeoutError 或 AssertionError，由调用方决定怎么截屏/上报
+
+依赖匠厂 DOM 锚点（docs/JIANGCHANG_CUSTOM_ANCHORS.md 守护）：
+- [data-testid="chat-composer-input"]
+- [data-testid="chat-page"][data-sending]
+- [data-role="user"] / [data-role="assistant"]
+- [data-testid="chat-execution-graph"][data-collapsed]
+- window.__jc_sending__
+"""
+from __future__ import annotations
+
+import base64
+import re
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterator, Optional, Tuple
+
+from playwright.sync_api import Page
+
+
+__all__ = [
+    "SkillDataDir",
+    "resolve_skill_data_dir",
+    "require_logged_in_skill_data_dir",
+    "get_runtime_env",
+    "assert_skill_env_file",
+    "drop_file_into_composer",
+    "wait_for_attachment_ready",
+    "wait_for_composer_enabled",
+    "send_prompt_via_composer",
+    "wait_for_user_message",
+    "wait_for_agent_complete",
+]
+
+_NEWLINE = object()
+
+_ANONYMOUS_USER_IDS = frozenset({"", "_anon", "anon", "anonymous"})
+
+_DEFAULT_PLACEHOLDER_VALUES = frozenset({
+    "",
+    "xxx",
+    "sk-xxx",
+    "your-key",
+    "your_api_key",
+    "todo",
+    "changeme",
+    "replace-me",
+    "<your-key>",
+})
+
+@dataclass(frozen=True)
+class SkillDataDir:
+    """匠厂宿主解析出的 skill 用户数据目录上下文。"""
+
+    root: Path
+    path: Path
+    user_id: str
+    skill_slug: str
+
+
+def _normalize_ws(text: str) -> str:
+    return re.sub(r"\s+", "", text)
+
+
+def _invoke_ipc(page: Page, channel: str, payload: dict | None = None) -> dict:
+    """通过匠厂 preload 暴露的 IPC 调用主进程 handler。"""
+    result = page.evaluate(
+        """
+        async ({ channel, payload }) => {
+          const electron = window.electron;
+          if (!electron || !electron.ipcRenderer
+              || typeof electron.ipcRenderer.invoke !== 'function') {
+            throw new Error(
+              'window.electron.ipcRenderer.invoke 不可用：请在匠厂桌面客户端页面内运行 E2E，'
+              + '且 preload 已暴露 jiangchang IPC'
+            );
+          }
+          try {
+            if (payload === null || payload === undefined) {
+              return await electron.ipcRenderer.invoke(channel);
+            }
+            return await electron.ipcRenderer.invoke(channel, payload);
+          } catch (err) {
+            const msg = err && err.message ? err.message : String(err);
+            throw new Error(channel + ' IPC 调用失败: ' + msg);
+          }
+        }
+        """,
+        {"channel": channel, "payload": payload},
+    )
+    if not isinstance(result, dict):
+        raise RuntimeError(
+            f"{channel} 返回非对象: {result!r}"
+        )
+    return result
+
+
+def resolve_skill_data_dir(
+    page: Page,
+    skill_slug: str,
+    *,
+    create: bool = True,
+) -> SkillDataDir:
+    """通过宿主 IPC 解析当前登录用户下的 skill 数据目录（不拼路径、不用 _anon 默认值）。"""
+    raw = _invoke_ipc(
+        page,
+        "jiangchang:resolve-skill-data-dir",
+        {"skillSlug": skill_slug},
+    )
+    missing = [k for k in ("root", "path", "userId", "skillSlug") if not raw.get(k)]
+    if missing:
+        raise RuntimeError(
+            "jiangchang:resolve-skill-data-dir 返回结构不完整，"
+            f"缺少字段 {missing}，实际: {raw!r}"
+        )
+
+    data_path = Path(str(raw["path"]))
+    if create:
+        data_path.mkdir(parents=True, exist_ok=True)
+
+    return SkillDataDir(
+        root=Path(str(raw["root"])),
+        path=data_path,
+        user_id=str(raw["userId"]),
+        skill_slug=str(raw["skillSlug"]),
+    )
+
+
+def require_logged_in_skill_data_dir(page: Page, skill_slug: str) -> SkillDataDir:
+    """解析 skill 数据目录，并断言当前宿主已同步真实登录用户（非匿名）。"""
+    ctx = resolve_skill_data_dir(page, skill_slug)
+    uid = (ctx.user_id or "").strip().lower()
+    if uid in _ANONYMOUS_USER_IDS:
+        raise AssertionError(
+            "当前宿主没有同步登录用户 ID（收到 "
+            f"{ctx.user_id!r}），不能运行桌面 E2E。"
+            "请先在匠厂客户端登录，或检查 jiangchangUserId 是否已同步到设置。"
+        )
+    return ctx
+
+
+def get_runtime_env(page: Page) -> dict:
+    """诊断用：读取宿主主进程解析的运行时环境（勿用于拼路径）。"""
+    return _invoke_ipc(page, "jiangchang:get-runtime-env", None)
+
+
+def _strip_env_value(raw: str) -> str:
+    value = raw.strip()
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
+        return value[1:-1]
+    return value
+
+
+def _parse_env_file(env_path: Path) -> dict[str, str]:
+    """解析简单 KEY=value .env（无 shell 展开）。"""
+    parsed: dict[str, str] = {}
+    text = env_path.read_text(encoding="utf-8")
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#"):
+            continue
+        if "=" not in stripped:
+            continue
+        key, _, raw_value = stripped.partition("=")
+        key = key.strip()
+        if not key:
+            continue
+        parsed[key] = _strip_env_value(raw_value)
+    return parsed
+
+
+def _is_placeholder_value(value: str, extra: set[str] | None) -> bool:
+    normalized = value.strip().lower()
+    placeholders = _DEFAULT_PLACEHOLDER_VALUES
+    if extra:
+        placeholders = placeholders | {p.strip().lower() for p in extra}
+    return normalized in placeholders
+
+
+def assert_skill_env_file(
+    skill_data_dir: SkillDataDir,
+    required_keys: list[str] | tuple[str, ...],
+    *,
+    env_rel_path: str = "config/.env",
+    placeholder_values: set[str] | None = None,
+) -> Path:
+    """Preflight：检查 skill 用户数据目录下 config/.env 是否已配置所需密钥。"""
+    env_path = skill_data_dir.path / env_rel_path
+    if not env_path.is_file():
+        raise AssertionError(
+            f"缺少配置文件: {env_path}\n"
+            "E2E 测试不会自动生成真实密钥，请先一次性配置用户数据目录 config/.env"
+        )
+
+    env_map = _parse_env_file(env_path)
+    invalid: list[str] = []
+    for key in required_keys:
+        if key not in env_map:
+            invalid.append(key)
+            continue
+        value = env_map[key]
+        if not value.strip() or _is_placeholder_value(value, placeholder_values):
+            invalid.append(key)
+
+    if invalid:
+        raise AssertionError(
+            f"config/.env 配置不完整或仍为占位符: {env_path}\n"
+            f"缺失/无效的 key: {', '.join(invalid)}\n"
+            "E2E 测试不会自动生成真实密钥，请先一次性配置用户数据目录 config/.env"
+        )
+    return env_path
+
+
+def _iter_text_and_newlines(prompt: str) -> Iterator[str | object]:
+    """按真实输入顺序产出文本片段与换行标记（\\r\\n / \\n / \\r 均视为单次换行）。"""
+    i = 0
+    n = len(prompt)
+    text_start = 0
+    while i < n:
+        if i + 1 < n and prompt[i] == "\r" and prompt[i + 1] == "\n":
+            if i > text_start:
+                yield prompt[text_start:i]
+            yield _NEWLINE
+            i += 2
+            text_start = i
+        elif prompt[i] in "\n\r":
+            if i > text_start:
+                yield prompt[text_start:i]
+            yield _NEWLINE
+            i += 1
+            text_start = i
+        else:
+            i += 1
+    if text_start < n:
+        yield prompt[text_start:]
+
+
+# ─── 附件上传（绕开 SDK send_file 未实现） ──────────────────────────
+
+def drop_file_into_composer(page: Page, file_path: Path) -> None:
+    """通过 DragEvent + DataTransfer 把文件拖入聊天输入框。
+
+    匠厂 ChatInput 的 onDrop 挂在外层 div 上，但 React 17+ 事件代理到 root，
+    所以在 textarea 上派发 bubbles:true 的 drop 事件能正常触发 stageBufferFiles
+    → /api/files/stage-buffer → 真实附件链路。
+    """
+    payload_b64 = base64.b64encode(file_path.read_bytes()).decode("ascii")
+    page.evaluate(
+        """
+        ({ base64, fileName, mimeType }) => {
+          const binary = atob(base64);
+          const bytes = new Uint8Array(binary.length);
+          for (let i = 0; i < binary.length; i++) {
+            bytes[i] = binary.charCodeAt(i);
+          }
+          const file = new File([bytes], fileName, { type: mimeType });
+          const dt = new DataTransfer();
+          dt.items.add(file);
+          const textarea = document.querySelector('[data-testid="chat-composer-input"]');
+          if (!textarea) {
+            throw new Error('chat-composer-input not found');
+          }
+          const opts = { bubbles: true, cancelable: true, dataTransfer: dt };
+          textarea.dispatchEvent(new DragEvent('dragenter', opts));
+          textarea.dispatchEvent(new DragEvent('dragover', opts));
+          textarea.dispatchEvent(new DragEvent('drop', opts));
+        }
+        """,
+        {
+            "base64": payload_b64,
+            "fileName": file_path.name,
+            "mimeType": _guess_mime(file_path),
+        },
+    )
+
+
+def _guess_mime(file_path: Path) -> str:
+    suffix = file_path.suffix.lower()
+    return {
+        ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ".xls": "application/vnd.ms-excel",
+        ".csv": "text/csv",
+        ".pdf": "application/pdf",
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".txt": "text/plain",
+        ".json": "application/json",
+    }.get(suffix, "application/octet-stream")
+
+
+def wait_for_attachment_ready(page: Page, file_name: str, timeout_s: float = 30.0) -> None:
+    """stage-buffer 成功后附件 chip 会展示 fileName，轮询直到可见。"""
+    deadline = time.time() + timeout_s
+    name_loc = page.get_by_text(file_name, exact=False)
+    while time.time() < deadline:
+        try:
+            n = name_loc.count()
+            for i in range(n):
+                if name_loc.nth(i).is_visible():
+                    return
+        except Exception:
+            pass
+        time.sleep(0.5)
+    raise TimeoutError(
+        f"附件 {file_name!r} 在 {timeout_s}s 内未变为 ready"
+    )
+
+
+# ─── 发送 prompt（绕开 SDK ask 的 15s 锁死） ────────────────────────
+
+def wait_for_composer_enabled(page: Page, timeout_s: float = 15.0) -> None:
+    """等输入框可见且非 disabled（避免 canSend=false 静默 no-op）。"""
+    page.wait_for_selector(
+        '[data-testid="chat-composer-input"]',
+        state="visible",
+        timeout=int(timeout_s * 1000),
+    )
+    textarea = page.locator('[data-testid="chat-composer-input"]')
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        if not textarea.evaluate("(el) => el.disabled"):
+            return
+        time.sleep(0.3)
+    raise TimeoutError(
+        f"chat-composer-input 在 {timeout_s}s 内仍为 disabled（gateway 可能未就绪）"
+    )
+
+
+def send_prompt_via_composer(page: Page, prompt: str, *, delay_ms: int = 25) -> None:
+    """拟人逐字输入 + Shift+Enter 换行 + Enter 发送（禁止 DOM/剪贴板/伪造事件）。"""
+    textarea = page.locator('[data-testid="chat-composer-input"]')
+    textarea.click()
+    textarea.press("Control+A")
+    textarea.press("Backspace")
+    for segment in _iter_text_and_newlines(prompt):
+        if segment is _NEWLINE:
+            textarea.press("Shift+Enter")
+        else:
+            textarea.press_sequentially(str(segment), delay=delay_ms)
+    textarea.press("Enter")
+
+
+def wait_for_user_message(page: Page, prompt: str, timeout_s: float = 60.0) -> int:
+    """等我们这条 user 消息进入 DOM，返回它在 [data-role] 列表里的 index。
+
+    绕开 SDK ask() 只给 15s 窗口的脆弱判据：用 prompt 文本反向匹配，
+    宽容到 60s。脏 session（new_task 后 gateway 载回历史）也能正确定位。
+    """
+    prompt_norm = _normalize_ws(prompt)
+    deadline = time.time() + timeout_s
+    items = page.locator('[data-role]')
+    while time.time() < deadline:
+        total = items.count()
+        for i in range(total - 1, -1, -1):
+            node = items.nth(i)
+            role = (node.get_attribute("data-role") or "").lower()
+            if role != "user":
+                continue
+            text = node.inner_text(timeout=2000) or ""
+            text_norm = _normalize_ws(text)
+            if prompt_norm in text_norm or text_norm in prompt_norm:
+                return i
+        time.sleep(0.5)
+    raise TimeoutError(
+        f"在 {timeout_s}s 内未找到匹配 user 消息（prompt 归一化前 80 字="
+        f"{prompt_norm[:80]!r})"
+    )
+
+
+# ─── 等 agent 完成（多信号合流） ────────────────────────────────
+
+def _chat_sending_false(page: Page) -> bool:
+    val = page.locator('[data-testid="chat-page"]').first.get_attribute("data-sending")
+    return val is not None and val.lower() == "false"
+
+
+def _window_sending_not_true(page: Page) -> bool:
+    val = page.evaluate("() => window.__jc_sending__")
+    return val is not True
+
+
+def _no_expanded_execution_graph(page: Page) -> bool:
+    return (
+        page.locator('[data-testid="chat-execution-graph"][data-collapsed="false"]').count()
+        == 0
+    )
+
+
+def _latest_assistant_after(page: Page, user_idx: int) -> Optional[Tuple[int, str, bool]]:
+    items = page.locator('[data-role]')
+    total = items.count()
+    for i in range(total - 1, user_idx, -1):
+        node = items.nth(i)
+        role = (node.get_attribute("data-role") or "").lower()
+        if role != "assistant":
+            continue
+        body = node.inner_text(timeout=2000) or ""
+        streaming = (node.get_attribute("data-streaming") or "false").lower() == "true"
+        return i, body, streaming
+    return None
+
+
+def wait_for_agent_complete(
+    page: Page,
+    user_idx: int,
+    *,
+    overall_timeout_s: float = 1200.0,
+    stable_seconds: float = 5.0,
+    poll_interval: float = 1.0,
+) -> str:
+    """多信号合流等本轮 agent 完成，返回最末 assistant 气泡文本。
+
+    完成条件（全部满足并稳定 stable_seconds 秒）：
+    1) chat-page data-sending="false"
+    2) window.__jc_sending__ 不为 True
+    3) 不存在未折叠的 execution graph
+    4) user_idx 之后存在 assistant 节点，有非空 body，data-streaming != "true"
+    5) 上面 assistant 节点的文本长度连续 stable_seconds 秒无变化
+    """
+    deadline = time.time() + overall_timeout_s
+    last_body_len = -1
+    last_change_at = time.time()
+
+    while time.time() < deadline:
+        chat_ok = _chat_sending_false(page)
+        win_ok = _window_sending_not_true(page)
+        graph_ok = _no_expanded_execution_graph(page)
+        picked = _latest_assistant_after(page, user_idx)
+
+        body_text = ""
+        streaming = True
+        has_body = False
+        if picked is not None:
+            _, body_text, streaming = picked
+            has_body = len(body_text.strip()) > 0
+
+        all_ready = chat_ok and win_ok and graph_ok and has_body and not streaming
+
+        if not all_ready:
+            last_body_len = -1
+            last_change_at = time.time()
+        else:
+            cur_len = len(body_text)
+            if cur_len != last_body_len:
+                last_body_len = cur_len
+                last_change_at = time.time()
+            elif time.time() - last_change_at >= stable_seconds:
+                return body_text
+
+        time.sleep(poll_interval)
+
+    raise TimeoutError(
+        f"agent 在 {overall_timeout_s}s 内未完成（user_idx={user_idx} "
+        f"last_body_len={last_body_len}）"
+    )

src/jiangchang_desktop_sdk/window_win32.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""
+Win32 窗口工具：跨 SDK / screencast 共享的"激活并最大化匠厂主窗口"实现。
+
+为什么单独抽出来：
+- SDK 的 bring_to_front 需要 OS 级窗口前置（page.bring_to_front 只切 tab）
+- screencast 的 run_screencast 录屏前也要把匠厂浮到前面并铺满屏幕
+- 两者用同一份 EnumWindows + ShowWindow(SW_MAXIMIZE) + SetForegroundWindow 实现，
+  避免代码重复维护
+
+跨平台行为：
+- Windows: 真正执行
+- 其他系统: 直接返回 False（调用方按需 fallback 到协议激活）
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Optional, Tuple
+
+_logger = logging.getLogger("jiangchang_desktop_sdk.window_win32")
+
+_DEFAULT_KEYWORDS: Tuple[str, ...] = ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
+SW_MAXIMIZE = 3
+
+
+def activate_and_maximize_jiangchang_window(
+    keywords: Optional[Tuple[str, ...]] = None,
+) -> bool:
+    """Windows: EnumWindows 找匠厂主窗口 → ShowWindow(SW_MAXIMIZE) + SetForegroundWindow。
+
+    Args:
+        keywords: 标题模糊匹配的关键词；不传则用默认
+                  ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
+
+    Returns:
+        True: 找到窗口并执行了 ShowWindow + SetForegroundWindow
+        False: 非 Windows / ctypes 不可用 / 找不到匹配窗口 / 任何异常
+    """
+    if os.name != "nt":
+        return False
+
+    try:
+        import ctypes
+        from ctypes import wintypes
+    except Exception as exc:
+        _logger.debug("ctypes 不可用：%s", exc)
+        return False
+
+    user32 = ctypes.windll.user32
+    kw = keywords or _DEFAULT_KEYWORDS
+    matching: list = []
+
+    @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
+    def _enum_proc(hwnd, _lparam):
+        if not user32.IsWindowVisible(hwnd):
+            return True
+        length = user32.GetWindowTextLengthW(hwnd)
+        if length == 0:
+            return True
+        buf = ctypes.create_unicode_buffer(length + 1)
+        user32.GetWindowTextW(hwnd, buf, length + 1)
+        if any(k in buf.value for k in kw):
+            matching.append((hwnd, buf.value))
+        return True
+
+    try:
+        user32.EnumWindows(_enum_proc, 0)
+    except Exception as exc:
+        _logger.debug("EnumWindows 失败：%s", exc)
+        return False
+
+    if not matching:
+        _logger.debug("未找到匠厂主窗口（keywords=%s）", kw)
+        return False
+
+    hwnd, title = matching[0]
+    try:
+        user32.ShowWindow(hwnd, SW_MAXIMIZE)
+        user32.SetForegroundWindow(hwnd)
+        _logger.debug("已激活+最大化 hwnd=%s title=%s", hwnd, title)
+        return True
+    except Exception as exc:
+        _logger.debug("ShowWindow/SetForegroundWindow 失败：%s", exc)
+        return False

src/jiangchang_skill_core/__init__.py

@@ -1,9 +1,11 @@
 from .client import EntitlementClient
+from .config import ensure_env_file, get, get_bool, get_float, get_int
 from .errors import EntitlementDeniedError, EntitlementError, EntitlementServiceError
 from .guard import enforce_entitlement
 from .models import EntitlementResult
 from .runtime_env import (
     apply_cli_local_defaults,
+    find_chrome_executable,
     get_data_root,
     get_sibling_skills_root,
     get_skills_root,
@@ -20,6 +22,11 @@ from .unified_logging import (
     subprocess_env_with_trace,
 )
 
+try:
+    from . import rpa
+except ImportError:
+    rpa = None  # type: ignore[assignment,misc]
+
 __all__ = [
     "EntitlementClient",
     "EntitlementDeniedError",
@@ -28,9 +35,15 @@ __all__ = [
     "EntitlementServiceError",
     "apply_cli_local_defaults",
     "attach_unified_file_handler",
+    "ensure_env_file",
     "enforce_entitlement",
     "ensure_trace_for_process",
+    "find_chrome_executable",
+    "get",
+    "get_bool",
     "get_data_root",
+    "get_float",
+    "get_int",
     "get_sibling_skills_root",
     "get_skills_root",
     "get_skill_log_file_path",
@@ -38,6 +51,7 @@ __all__ = [
     "get_unified_logs_dir",
     "get_user_id",
     "platform_default_data_root",
+    "rpa",
     "setup_skill_logging",
     "subprocess_env_with_trace",
 ]

src/jiangchang_skill_core/config.py

@@ -0,0 +1,125 @@
+"""三级优先级配置读取 + 首次 .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

src/jiangchang_skill_core/rpa/__init__.py

@@ -0,0 +1,59 @@
+"""浏览器 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]
+
+__all__ = [
+    "anti_detect",
+    "artifacts",
+    "errors",
+    "human_login",
+    "stealth",
+    "launch_persistent_browser",
+    # 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,
+)

src/jiangchang_skill_core/rpa/anti_detect.py

@@ -0,0 +1,215 @@
+"""防反爬：随机延迟、人工鼠标轨迹模拟、拟人滚动。
+
+所有延迟使用 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))

src/jiangchang_skill_core/rpa/artifacts.py

@@ -0,0 +1,28 @@
+"""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

src/jiangchang_skill_core/rpa/browser.py

@@ -0,0 +1,61 @@
+"""统一 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

src/jiangchang_skill_core/rpa/errors.py

@@ -0,0 +1,19 @@
+"""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",
+]

src/jiangchang_skill_core/rpa/human_login.py

@@ -0,0 +1,96 @@
+"""验证码/登录 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

src/jiangchang_skill_core/rpa/stealth.py

@@ -0,0 +1,66 @@
+"""淡化 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"]

src/jiangchang_skill_core/runtime_env.py

@@ -134,3 +134,74 @@ def apply_cli_local_defaults() -> None:
         os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
     if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
         os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()
+def find_chrome_executable() -> "str | None":
+    """查找系统已安装的 Chrome 或 Edge 可执行文件路径。
+
+    查找顺序：
+    1. 常见文件路径（PROGRAMFILES、LOCALAPPDATA、macOS、Linux）
+    2. Windows 注册表 App Paths（HKLM 优先，HKCU 其次）
+    3. PATH shutil.which 兜底
+
+    返回找到的路径字符串，找不到返回 None。
+    宿主/技能调用 Playwright launch 时应传 executable_path=find_chrome_executable()
+    而非 channel="chrome"，以避免 Playwright 内部检测在系统级安装时失效。
+    """
+    import shutil
+
+    if sys.platform == "win32":
+        local_app_data = os.environ.get("LOCALAPPDATA", "")
+        prog_files = os.environ.get("PROGRAMFILES", r"C:\Program Files")
+        prog_files_x86 = os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)")
+        candidates = [
+            os.path.join(prog_files, r"Google\Chrome\Application\chrome.exe"),
+            os.path.join(prog_files_x86, r"Google\Chrome\Application\chrome.exe"),
+            os.path.join(local_app_data, r"Google\Chrome\Application\chrome.exe"),
+            os.path.join(prog_files, r"Microsoft\Edge\Application\msedge.exe"),
+            os.path.join(prog_files_x86, r"Microsoft\Edge\Application\msedge.exe"),
+            os.path.join(local_app_data, r"Microsoft\Edge\Application\msedge.exe"),
+        ]
+    elif sys.platform == "darwin":
+        candidates = [
+            "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
+            "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
+            "/Applications/Chromium.app/Contents/MacOS/Chromium",
+        ]
+    else:
+        candidates = [
+            "/usr/bin/google-chrome-stable",
+            "/usr/bin/google-chrome",
+            "/usr/bin/chromium-browser",
+            "/usr/bin/chromium",
+            "/usr/bin/microsoft-edge",
+        ]
+
+    for path in candidates:
+        if path and os.path.isfile(path):
+            return path
+
+    # Windows 注册表兜底（覆盖系统级 HKLM 安装）
+    if sys.platform == "win32":
+        try:
+            import winreg
+            for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
+                for reg_path in (
+                    r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\chrome.exe",
+                    r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\msedge.exe",
+                ):
+                    try:
+                        with winreg.OpenKey(hive, reg_path) as key:
+                            val, _ = winreg.QueryValueEx(key, "")
+                            if val and os.path.isfile(val):
+                                return val
+                    except OSError:
+                        continue
+        except ImportError:
+            pass
+
+    # PATH 兜底
+    for name in ("google-chrome-stable", "google-chrome", "chrome", "msedge", "chromium"):
+        found = shutil.which(name)
+        if found:
+            return found
+
+    return None

src/screencast/__init__.py

@@ -0,0 +1,3 @@
+from .runner import run_screencast
+
+__all__ = ["run_screencast"]

src/screencast/composer.py

@@ -0,0 +1,80 @@
+"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
+from __future__ import annotations
+
+import random
+import subprocess
+import sys
+from pathlib import Path
+
+
+# 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性
+# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感
+# Outline 2：14 号字配 Outline 3 会显胖，2 已足够在浅色背景上保持可读
+# 修改请评估对所有 skill 录屏的影响。
+_JIANGCHANG_SUBTITLE_STYLE = (
+    "FontName=Arial,FontSize=14,"
+    "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
+    "Outline=2,Shadow=1,Alignment=2"
+)
+
+
+def compose_video(
+    frames_dir: str,
+    fps: int,
+    subtitle_path: str,
+    music_dir: str,
+    output_path: str,
+    music_volume: float = 0.15,
+) -> str:
+    frames_dir = Path(frames_dir)
+    subtitle_path = Path(subtitle_path)
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # 随机选一首背景音乐
+    music_files = list(Path(music_dir).rglob("*.mp3"))
+    if not music_files:
+        raise FileNotFoundError(f"music_dir 下没有找到 MP3 文件: {music_dir}")
+    music_file = random.choice(music_files)
+    print(f"[screencast] 背景音乐: {music_file.name}")
+
+    # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法）
+    srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:")
+
+    subtitle_filter = (
+        f"subtitles='{srt_path_str}'"
+        f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
+    )
+
+    # 视频和音频都放进 filter_complex，避免 -vf 与 -filter_complex 混用报错
+    filter_complex = (
+        f"[0:v]{subtitle_filter}[vout];"
+        f"[1:a]volume={music_volume},apad[aout]"
+    )
+
+    cmd = [
+        "ffmpeg", "-y",
+        "-framerate", str(fps),
+        "-i", str(frames_dir / "frame_%06d.png"),
+        "-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),
+    ]
+
+    print(f"[screencast] 运行 FFmpeg 合成...")
+    result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
+    if result.returncode != 0:
+        print(result.stderr, file=sys.stderr)
+        raise RuntimeError(f"FFmpeg 合成失败，退出码 {result.returncode}")
+
+    print(f"[screencast] 输出: {output_path}")
+    return str(output_path)

src/screencast/recorder.py

@@ -0,0 +1,63 @@
+"""全屏截帧引擎，后台线程持续截图存为 PNG 帧序列。"""
+from __future__ import annotations
+
+import threading
+import time
+from pathlib import Path
+from typing import Optional
+
+import mss
+import mss.tools
+
+
+class ScreenRecorder:
+    def __init__(
+        self,
+        frames_dir: str,
+        fps: int = 10,
+        monitor_index: Optional[int] = None,
+    ):
+        """
+        Args:
+            monitor_index: mss 显示器索引；None=monitors[0]（所有屏合并区域，
+                          旧行为），1=主屏，2+=第 N 屏。多显示器场景推荐显式传 1。
+        """
+        self._frames_dir = Path(frames_dir)
+        self._fps = fps
+        self._monitor_index = monitor_index
+        self._stop_event = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._frame_count = 0
+
+    def start(self) -> None:
+        self._frames_dir.mkdir(parents=True, exist_ok=True)
+        self._stop_event.clear()
+        self._frame_count = 0
+        self._thread = threading.Thread(target=self._capture_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop_event.set()
+        if self._thread:
+            self._thread.join(timeout=10)
+
+    @property
+    def frame_count(self) -> int:
+        return self._frame_count
+
+    def _capture_loop(self) -> None:
+        interval = 1.0 / self._fps
+        with mss.mss() as sct:
+            idx = self._monitor_index if self._monitor_index is not None else 0
+            if idx < 0 or idx >= len(sct.monitors):
+                idx = 0  # 越界兜底
+            monitor = sct.monitors[idx]
+            while not self._stop_event.is_set():
+                t0 = time.perf_counter()
+                img = sct.grab(monitor)
+                path = self._frames_dir / f"frame_{self._frame_count:06d}.png"
+                mss.tools.to_png(img.rgb, img.size, output=str(path))
+                self._frame_count += 1
+                elapsed = time.perf_counter() - t0
+                sleep = max(0.0, interval - elapsed)
+                time.sleep(sleep)

src/screencast/runner.py

@@ -0,0 +1,142 @@
+"""主编排器：录屏 + 运行 pytest + 生成字幕 + FFmpeg 合成。"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+
+def _ensure_sdk_on_path() -> None:
+    """pip 安装后应能直接 import；仅源码开发态才把 src 加入 sys.path。"""
+    import importlib.util
+
+    if importlib.util.find_spec("jiangchang_desktop_sdk") is not None:
+        return
+    # src/screencast/runner.py → src/screencast → src
+    here = Path(__file__).resolve()
+    src = here.parent.parent
+    sdk_dir = src / "jiangchang_desktop_sdk"
+    if sdk_dir.is_dir():
+        src_str = str(src)
+        if src_str not in sys.path:
+            sys.path.insert(0, src_str)
+
+
+_ensure_sdk_on_path()
+from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window  # noqa: E402
+
+from .composer import compose_video
+from .recorder import ScreenRecorder
+from .subtitle import SubtitleEngine
+
+
+def run_screencast(
+    skill_slug: str,
+    subtitle_script: List[Tuple[str, str]],
+    pytest_args: List[str],
+    output_dir: str,
+    media_assets_root: Optional[str] = None,
+    music_subdir: str = "music",
+    fps: int = 10,
+    activate_window: bool = True,
+    monitor_index: Optional[int] = None,
+) -> str:
+    """
+    录制一次技能桌面 E2E 演示视频。
+
+    Args:
+        skill_slug:       技能唯一标识，如 "query-balance-icbc"
+        subtitle_script:  [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout
+        pytest_args:      传递给 pytest 的参数列表
+        output_dir:       最终 MP4 输出目录
+        media_assets_root: media-assets 仓库本地路径；
+                           不传时读 MEDIA_ASSETS_ROOT 环境变量，
+                           再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
+        music_subdir:     music_assets_root 下音乐子目录，默认 "music"
+        fps:               截帧帧率，默认 10
+        activate_window:   录屏开始前是否自动激活+最大化匠厂窗口（默认 True；
+                          Windows 走 ctypes，非 Windows 走 jiangchang:// 协议）
+        monitor_index:     指定录哪个显示器；None=所有显示器合并（旧行为），
+                          1=主屏，2+=第 N 屏（推荐多显示器场景显式传 1）
+
+    Returns:
+        输出 MP4 的绝对路径
+    """
+    if media_assets_root is None:
+        media_assets_root = os.environ.get(
+            "MEDIA_ASSETS_ROOT",
+            r"D:\OpenClaw\client-commons\media-assets",
+        )
+
+    music_dir = str(Path(media_assets_root) / music_subdir)
+    output_dir_path = Path(output_dir)
+    output_dir_path.mkdir(parents=True, exist_ok=True)
+
+    date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
+    base_name = f"{skill_slug}_{date_str}"
+    output_mp4 = output_dir_path / f"{base_name}.mp4"
+    subtitle_file = output_dir_path / f"{base_name}.srt"
+
+    engine = SubtitleEngine(subtitle_script)
+
+    with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir:
+        recorder = ScreenRecorder(frames_dir, fps=fps, monitor_index=monitor_index)
+
+        if activate_window:
+            ok = activate_and_maximize_jiangchang_window()
+            if ok:
+                print("[screencast] 已激活+最大化匠厂窗口")
+                time.sleep(2)  # 等窗口浮起稳定
+            else:
+                print("[screencast] 未能激活匠厂窗口，继续录制（可能录到桌面边角）")
+
+        print(f"[screencast] 启动录制 → {output_mp4.name}")
+        recorder.start()
+        engine.set_start_time(time.time())
+
+        try:
+            # Windows 子进程 stdout 默认走系统 ANSI code page（中文 Windows 是 cp936/GBK），
+            # 这边用 encoding="utf-8" 解码会乱码。注入 PYTHONIOENCODING + PYTHONUTF8 让
+            # 子进程 Python 强制以 UTF-8 输出，两端编码对齐。非 Windows 默认就 UTF-8，无副作用。
+            env = os.environ.copy()
+            env["PYTHONIOENCODING"] = "utf-8"
+            env["PYTHONUTF8"] = "1"
+
+            proc = subprocess.Popen(
+                [sys.executable, "-m", "pytest", *pytest_args],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                env=env,
+            )
+            assert proc.stdout is not None
+            for line in proc.stdout:
+                sys.stdout.write(line)
+                sys.stdout.flush()
+                engine.process_line(line)
+            exit_code = proc.wait()
+            if exit_code != 0:
+                print(f"[screencast] pytest 退出码 {exit_code}，录制仍继续合成")
+        finally:
+            recorder.stop()
+            print(f"[screencast] 录制停止，共 {recorder.frame_count} 帧")
+
+        engine.generate_srt(str(subtitle_file))
+        print(f"[screencast] 字幕 → {subtitle_file.name}")
+
+        compose_video(
+            frames_dir=frames_dir,
+            fps=fps,
+            subtitle_path=str(subtitle_file),
+            music_dir=music_dir,
+            output_path=str(output_mp4),
+        )
+
+    return str(output_mp4)

src/screencast/subtitle.py

@@ -0,0 +1,94 @@
+"""字幕引擎：根据 stdout 关键词打时间戳，生成 SRT 字幕文件。"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Tuple
+
+
+@dataclass
+class _Entry:
+    start: float   # 距录制开始的秒数
+    text: str
+    duration: float = 5.0  # 每条字幕默认显示 5 秒
+
+
+class SubtitleEngine:
+    def __init__(
+        self,
+        script: List[Tuple],
+        default_duration: float = 5.0,
+    ):
+        """
+        script: [(关键词, 字幕文案) | (关键词, 字幕文案, 显示时长秒), ...]
+                - 二元组：用 default_duration（向后兼容）
+                - 三元组：本条字幕独立指定显示秒数（用于长等待场景延长字幕停留）
+        关键词大小写不敏感，每个关键词只触发一次。
+        """
+        self._script = script
+        self._default_duration = default_duration
+        self._entries: List[_Entry] = []
+        self._matched: set[str] = set()
+        self._t0: float = 0.0
+
+    def set_start_time(self, t: float) -> None:
+        self._t0 = t
+
+    def process_line(self, line: str) -> None:
+        lower = line.lower()
+        for entry_tuple in self._script:
+            if len(entry_tuple) == 2:
+                keyword, text = entry_tuple
+                duration = self._default_duration
+            elif len(entry_tuple) == 3:
+                keyword, text, duration = entry_tuple
+            else:
+                continue
+            if keyword in self._matched:
+                continue
+            if keyword.lower() in lower:
+                elapsed = time.time() - self._t0
+                self._entries.append(_Entry(start=elapsed, text=text, duration=float(duration)))
+                self._matched.add(keyword)
+
+    def generate_srt(self, output_path: str) -> None:
+        """生成 SRT 文件，自动处理字幕重叠。
+
+        字幕重叠修复策略（避免屏幕上同时显示 2+ 条叠成两行）：
+        - 字幕按 start 时间升序排序后；
+        - 相邻字幕保留 GAP=0.1s 间隔；
+        - 每条字幕保证至少 MIN_DURATION=1.0s 显示时间（短到这个值仍读不完是字幕脚本设计问题）；
+        - 若当前条让位给下一条后剩余时间 < MIN_DURATION，则把下一条 start 往后推。
+        这样任意时刻屏幕上最多 1 条字幕。
+        """
+        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}"
+
+        # 防重叠 post-process
+        MIN_DURATION = 1.0
+        GAP = 0.1
+        entries = sorted(self._entries, key=lambda e: e.start)
+        for i in range(len(entries) - 1):
+            cur, nxt = entries[i], entries[i + 1]
+            max_end = nxt.start - GAP
+            new_dur = max_end - cur.start
+            if new_dur < MIN_DURATION:
+                # 太挤：cur 至少撑 MIN_DURATION 秒，把 nxt 推迟
+                cur.duration = MIN_DURATION
+                nxt.start = cur.start + MIN_DURATION + GAP
+            else:
+                # 正常让位：cur 显示到 (nxt.start - GAP)
+                cur.duration = min(cur.duration, new_dur)
+        # 最后一条不需要让位，保持原 duration
+
+        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(entries, 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")

tests/__init__.py

@@ -0,0 +1 @@
+# Package marker for unittest discovery.

tests/test_anti_detect_import.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""anti_detect / rpa 子包导入冒烟。"""
+from __future__ import annotations
+
+import importlib
+import unittest
+
+
+class TestAntiDetectImport(unittest.TestCase):
+    def test_import_anti_detect(self) -> None:
+        mod = importlib.import_module("jiangchang_skill_core.rpa.anti_detect")
+        self.assertTrue(callable(mod.human_click))
+        self.assertTrue(callable(mod.random_delay))
+
+    def test_import_rpa_package(self) -> None:
+        mod = importlib.import_module("jiangchang_skill_core.rpa")
+        self.assertTrue(callable(mod.wait_for_captcha_pass))
+        self.assertIsNotNone(mod.stealth_enabled)
+
+    def test_top_level_import_with_rpa(self) -> None:
+        pkg = importlib.import_module("jiangchang_skill_core")
+        self.assertIsNotNone(getattr(pkg, "config", None))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_config.py

@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-
+"""config 三级优先级与 ensure_env_file 测试。"""
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+from jiangchang_skill_core import config
+
+
+class TestConfig(unittest.TestCase):
+    def setUp(self) -> None:
+        config.reset_cache()
+        self._saved_env = dict(os.environ)
+
+    def tearDown(self) -> None:
+        os.environ.clear()
+        os.environ.update(self._saved_env)
+        config.reset_cache()
+
+    def test_ensure_env_file_copies_once(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            example = os.path.join(tmp, ".env.example")
+            with open(example, "w", encoding="utf-8") as f:
+                f.write("FOO=from_example\nBAR=2\n")
+
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "testuser"
+
+            dest = config.ensure_env_file("demo-skill", example)
+            self.assertTrue(os.path.isfile(dest))
+            with open(dest, encoding="utf-8") as f:
+                content = f.read()
+            self.assertIn("FOO=from_example", content)
+
+            with open(dest, "w", encoding="utf-8") as f:
+                f.write("FOO=user_edited\n")
+
+            dest2 = config.ensure_env_file("demo-skill", example)
+            self.assertEqual(dest, dest2)
+            with open(dest2, encoding="utf-8") as f:
+                self.assertIn("user_edited", f.read())
+
+    def test_three_level_priority(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            example = os.path.join(tmp, ".env.example")
+            with open(example, "w", encoding="utf-8") as f:
+                f.write("KEY_A=example\nKEY_B=example\nKEY_C=example\n")
+
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "u1"
+            dest = config.ensure_env_file("sk", example)
+
+            with open(dest, "w", encoding="utf-8") as f:
+                f.write("KEY_B=user_env\n")
+
+            config.reset_cache()
+            self.assertEqual(config.get("KEY_A"), "example")
+            self.assertEqual(config.get("KEY_B"), "user_env")
+            self.assertEqual(config.get("KEY_C"), "example")
+            self.assertIsNone(config.get("MISSING"))
+
+            os.environ["KEY_C"] = "process_env"
+            self.assertEqual(config.get("KEY_C"), "process_env")
+
+    def test_get_bool_float_int(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            example = os.path.join(tmp, ".env.example")
+            with open(example, "w", encoding="utf-8") as f:
+                f.write("FLAG=1\nRATE=2.5\nCOUNT=42\nBAD=x\n")
+
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "u1"
+            config.ensure_env_file("sk", example)
+
+            self.assertTrue(config.get_bool("FLAG"))
+            self.assertFalse(config.get_bool("MISSING", default=False))
+            self.assertAlmostEqual(config.get_float("RATE", 0.0), 2.5)
+            self.assertEqual(config.get_int("COUNT", 0), 42)
+            self.assertEqual(config.get_float("BAD", 9.9), 9.9)
+            self.assertEqual(config.get_int("BAD", 7), 7)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_e2e_helpers.py

@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+"""e2e_helpers 轻量单元测试（不依赖匠厂桌面 / IPC）。"""
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from jiangchang_desktop_sdk.e2e_helpers import (
+    SkillDataDir,
+    _iter_text_and_newlines,
+    _is_placeholder_value,
+    _parse_env_file,
+    assert_skill_env_file,
+)
+
+
+def _segments(prompt: str) -> list[str | None]:
+    out: list[str | None] = []
+    for seg in _iter_text_and_newlines(prompt):
+        if seg is not None and not isinstance(seg, str):
+            out.append(None)
+        else:
+            out.append(seg)  # type: ignore[arg-type]
+    return out
+
+
+class TestIterTextAndNewlines(unittest.TestCase):
+    def test_lf(self) -> None:
+        self.assertEqual(_segments("a\nb"), ["a", None, "b"])
+
+    def test_crlf(self) -> None:
+        self.assertEqual(_segments("a\r\nb"), ["a", None, "b"])
+
+    def test_cr(self) -> None:
+        self.assertEqual(_segments("a\rb"), ["a", None, "b"])
+
+    def test_multiple_newlines(self) -> None:
+        self.assertEqual(_segments("x\n\ny"), ["x", None, None, "y"])
+
+
+class TestParseEnvFile(unittest.TestCase):
+    def _write(self, path: Path, content: str) -> None:
+        path.write_text(content, encoding="utf-8")
+
+    def test_basic_and_comments(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            env_path = Path(tmp) / ".env"
+            self._write(
+                env_path,
+                "# comment\n\nAPI_KEY=real-secret\n# KEY=ignored\n",
+            )
+            parsed = _parse_env_file(env_path)
+            self.assertEqual(parsed["API_KEY"], "real-secret")
+            self.assertNotIn("KEY", parsed)
+
+    def test_quoted_values(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            env_path = Path(tmp) / ".env"
+            self._write(env_path, 'A="double"\nB=\'single\'\n')
+            parsed = _parse_env_file(env_path)
+            self.assertEqual(parsed["A"], "double")
+            self.assertEqual(parsed["B"], "single")
+
+    def test_placeholder_detection(self) -> None:
+        self.assertTrue(_is_placeholder_value("TODO", None))
+        self.assertTrue(_is_placeholder_value("sk-xxx", None))
+        self.assertTrue(_is_placeholder_value("  Changeme  ", None))
+        self.assertFalse(_is_placeholder_value("sk-live-abc", None))
+
+
+class TestAssertSkillEnvFile(unittest.TestCase):
+    def test_missing_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            ctx = SkillDataDir(
+                root=Path(tmp),
+                path=Path(tmp) / "user" / "demo-skill",
+                user_id="12428",
+                skill_slug="demo-skill",
+            )
+            with self.assertRaises(AssertionError) as cm:
+                assert_skill_env_file(ctx, ["API_KEY"])
+            self.assertIn("config/.env", str(cm.exception))
+
+    def test_missing_and_placeholder_keys(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            skill_dir = Path(tmp) / "12428" / "demo-skill"
+            env_dir = skill_dir / "config"
+            env_dir.mkdir(parents=True)
+            (env_dir / ".env").write_text(
+                "API_KEY=your-key\nOTHER=real\n",
+                encoding="utf-8",
+            )
+            ctx = SkillDataDir(
+                root=Path(tmp),
+                path=skill_dir,
+                user_id="12428",
+                skill_slug="demo-skill",
+            )
+            with self.assertRaises(AssertionError) as cm:
+                assert_skill_env_file(ctx, ["API_KEY", "MISSING", "OTHER"])
+            msg = str(cm.exception)
+            self.assertIn("API_KEY", msg)
+            self.assertIn("MISSING", msg)
+            self.assertNotIn("OTHER", msg)
+
+    def test_pass_returns_path(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            skill_dir = Path(tmp) / "1" / "demo-skill"
+            env_dir = skill_dir / "config"
+            env_dir.mkdir(parents=True)
+            env_file = env_dir / ".env"
+            env_file.write_text("API_KEY=sk-live-ok\n", encoding="utf-8")
+            ctx = SkillDataDir(
+                root=Path(tmp),
+                path=skill_dir,
+                user_id="1",
+                skill_slug="demo-skill",
+            )
+            got = assert_skill_env_file(ctx, ["API_KEY"])
+            self.assertEqual(got, env_file)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_release_workflow.py

@@ -0,0 +1,42 @@
+"""Text-level checks that reusable-release-skill packages .env.example at package root."""
+
+from __future__ import annotations
+
+import os
+import re
+
+import pytest
+
+_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+_WORKFLOW_PATH = os.path.join(
+    _REPO_ROOT, ".github", "workflows", "reusable-release-skill.yaml"
+)
+
+
+@pytest.fixture(scope="module")
+def workflow_text() -> str:
+    with open(_WORKFLOW_PATH, encoding="utf-8") as f:
+        return f.read()
+
+
+def test_workflow_packages_env_example_at_root(workflow_text: str) -> None:
+    assert ".env.example" in workflow_text
+    assert "Copied .env.example" in workflow_text
+    assert "env_example_src = os.path.abspath('.env.example')" in workflow_text
+    assert "env_example_dst = os.path.join(PACKAGE, '.env.example')" in workflow_text
+    assert (
+        "raise RuntimeError('.env.example exists in skill root but was not packaged')"
+        in workflow_text
+    )
+
+
+def test_workflow_does_not_package_dot_env(workflow_text: str) -> None:
+    # Must not copy live .env to package root (only .env.example).
+    assert not re.search(
+        r"shutil\.copy2\([^)]*['\"]\.env['\"]",
+        workflow_text,
+    )
+    assert not re.search(
+        r"os\.path\.join\(PACKAGE,\s*['\"]\.env['\"]\)",
+        workflow_text,
+    )

tests/test_stealth.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+"""stealth 启动参数与开关测试。"""
+from __future__ import annotations
+
+import os
+import unittest
+
+from jiangchang_skill_core.rpa.stealth import (
+    persistent_context_launch_parts,
+    stealth_enabled,
+)
+
+
+class TestStealth(unittest.TestCase):
+    def setUp(self) -> None:
+        self._saved = os.environ.get("OPENCLAW_PLAYWRIGHT_STEALTH")
+
+    def tearDown(self) -> None:
+        if self._saved is None:
+            os.environ.pop("OPENCLAW_PLAYWRIGHT_STEALTH", None)
+        else:
+            os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = self._saved
+
+    def test_stealth_enabled_default_on(self) -> None:
+        os.environ.pop("OPENCLAW_PLAYWRIGHT_STEALTH", None)
+        self.assertTrue(stealth_enabled())
+
+    def test_stealth_enabled_off_values(self) -> None:
+        for off in ("0", "false", "no", "off", "FALSE"):
+            os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = off
+            self.assertFalse(stealth_enabled(), msg=off)
+
+    def test_launch_parts_stealth_on(self) -> None:
+        os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "1"
+        args, ignore = persistent_context_launch_parts()
+        self.assertIn("--start-maximized", args)
+        self.assertIn("--disable-blink-features=AutomationControlled", args)
+        self.assertEqual(ignore, ["--enable-automation"])
+
+    def test_launch_parts_stealth_off(self) -> None:
+        os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "0"
+        args, ignore = persistent_context_launch_parts()
+        self.assertIn("--start-maximized", args)
+        self.assertNotIn("--disable-blink-features=AutomationControlled", args)
+        self.assertIsNone(ignore)
+
+    def test_launch_parts_extra_args(self) -> None:
+        os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "1"
+        args, _ = persistent_context_launch_parts(extra_args=["--foo"])
+        self.assertIn("--foo", args)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/ci_set_package_version.py

@@ -0,0 +1,79 @@
+"""Patch pyproject.toml and __init__.py version for CI builds (not committed)."""
+from __future__ import annotations
+
+import os
+import re
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _read_base_version() -> str:
+    text = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
+    match = re.search(r'^version\s*=\s*"([^"]+)"', text, re.MULTILINE)
+    if not match:
+        raise SystemExit("ERROR: version not found in pyproject.toml")
+    base = match.group(1)
+    return re.sub(r"\.post\d+$", "", base)
+
+
+def resolve_version() -> str:
+    ref = os.environ.get("GITHUB_REF", "")
+    if ref.startswith("refs/tags/v"):
+        return ref.removeprefix("refs/tags/v")
+    if ref.startswith("refs/tags/"):
+        return ref.removeprefix("refs/tags/")
+
+    run_number = (
+        os.environ.get("GITHUB_RUN_NUMBER")
+        or os.environ.get("GITHUB_RUN_ID")
+        or "0"
+    )
+    base = _read_base_version()
+    return f"{base}.post{run_number}"
+
+
+def patch_version(version: str) -> None:
+    pyproject = ROOT / "pyproject.toml"
+    py_text = pyproject.read_text(encoding="utf-8")
+    py_text, n = re.subn(
+        r'^version\s*=\s*"[^"]*"',
+        f'version = "{version}"',
+        py_text,
+        count=1,
+        flags=re.MULTILINE,
+    )
+    if n != 1:
+        raise SystemExit("ERROR: failed to patch version in pyproject.toml")
+    pyproject.write_text(py_text, encoding="utf-8")
+
+    init_py = ROOT / "src" / "jiangchang_desktop_sdk" / "__init__.py"
+    init_text = init_py.read_text(encoding="utf-8")
+    init_text, n = re.subn(
+        r'^__version__\s*=\s*"[^"]*"',
+        f'__version__ = "{version}"',
+        init_text,
+        count=1,
+        flags=re.MULTILINE,
+    )
+    if n != 1:
+        raise SystemExit("ERROR: failed to patch __version__ in __init__.py")
+    init_py.write_text(init_text, encoding="utf-8")
+
+
+def main() -> None:
+    version = resolve_version()
+    print(f"Resolved package version: {version}")
+    patch_version(version)
+    if os.environ.get("GITHUB_OUTPUT"):
+        with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as fh:
+            fh.write(f"package_version={version}\n")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except SystemExit as exc:
+        print(exc, file=sys.stderr)
+        raise

tools/release.ps1

@@ -32,6 +32,8 @@
   - 复制 assets/
   - 复制 tests/
   - 复制 evals/（除 scripts 外均为明文；若目录不存在则跳过；复制时排除常见缓存与运行产物）
+  - 若根目录存在 requirements.txt，则复制到包根（明文，不 PyArmor）
+  - 若根目录存在 .env.example，则复制到包根（明文，不 PyArmor，用于首次运行时落盘用户 .env）
   本脚本在打 tag 前会做一次 scripts/ 结构自检，避免子目录未提交却仍发布。
 #>
 

tools/screencast/__init__.py

@@ -1,3 +1,6 @@
-from .runner import run_screencast
+from ._bootstrap import bootstrap_src
+
+bootstrap_src()
+from screencast import run_screencast
 
 __all__ = ["run_screencast"]

tools/screencast/_bootstrap.py

@@ -0,0 +1,15 @@
+"""源码仓库内运行 tools/screencast 时，确保 src 在 sys.path（pip 安装后无需）。"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+
+def bootstrap_src() -> None:
+    try:
+        import screencast  # noqa: F401
+    except ImportError:
+        src = Path(__file__).resolve().parent.parent.parent / "src"
+        src_str = str(src)
+        if src.is_dir() and src_str not in sys.path:
+            sys.path.insert(0, src_str)

tools/screencast/composer.py

@@ -1,80 +1,6 @@
-"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
-from __future__ import annotations
+from ._bootstrap import bootstrap_src
 
-import random
-import subprocess
-import sys
-from pathlib import Path
+bootstrap_src()
+from screencast.composer import compose_video
 
-
-# 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性
-# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感
-# Outline 2：14 号字配 Outline 3 会显胖，2 已足够在浅色背景上保持可读
-# 修改请评估对所有 skill 录屏的影响。
-_JIANGCHANG_SUBTITLE_STYLE = (
-    "FontName=Arial,FontSize=14,"
-    "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
-    "Outline=2,Shadow=1,Alignment=2"
-)
-
-
-def compose_video(
-    frames_dir: str,
-    fps: int,
-    subtitle_path: str,
-    music_dir: str,
-    output_path: str,
-    music_volume: float = 0.15,
-) -> str:
-    frames_dir = Path(frames_dir)
-    subtitle_path = Path(subtitle_path)
-    output_path = Path(output_path)
-    output_path.parent.mkdir(parents=True, exist_ok=True)
-
-    # 随机选一首背景音乐
-    music_files = list(Path(music_dir).rglob("*.mp3"))
-    if not music_files:
-        raise FileNotFoundError(f"music_dir 下没有找到 MP3 文件: {music_dir}")
-    music_file = random.choice(music_files)
-    print(f"[screencast] 背景音乐: {music_file.name}")
-
-    # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法）
-    srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:")
-
-    subtitle_filter = (
-        f"subtitles='{srt_path_str}'"
-        f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
-    )
-
-    # 视频和音频都放进 filter_complex，避免 -vf 与 -filter_complex 混用报错
-    filter_complex = (
-        f"[0:v]{subtitle_filter}[vout];"
-        f"[1:a]volume={music_volume},apad[aout]"
-    )
-
-    cmd = [
-        "ffmpeg", "-y",
-        "-framerate", str(fps),
-        "-i", str(frames_dir / "frame_%06d.png"),
-        "-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),
-    ]
-
-    print(f"[screencast] 运行 FFmpeg 合成...")
-    result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
-    if result.returncode != 0:
-        print(result.stderr, file=sys.stderr)
-        raise RuntimeError(f"FFmpeg 合成失败，退出码 {result.returncode}")
-
-    print(f"[screencast] 输出: {output_path}")
-    return str(output_path)
+__all__ = ["compose_video"]

tools/screencast/recorder.py

@@ -1,48 +1,6 @@
-"""全屏截帧引擎，后台线程持续截图存为 PNG 帧序列。"""
-from __future__ import annotations
+from ._bootstrap import bootstrap_src
 
-import threading
-import time
-from pathlib import Path
+bootstrap_src()
+from screencast.recorder import ScreenRecorder
 
-import mss
-import mss.tools
-
-
-class ScreenRecorder:
-    def __init__(self, frames_dir: str, fps: int = 10):
-        self._frames_dir = Path(frames_dir)
-        self._fps = fps
-        self._stop_event = threading.Event()
-        self._thread: threading.Thread | None = None
-        self._frame_count = 0
-
-    def start(self) -> None:
-        self._frames_dir.mkdir(parents=True, exist_ok=True)
-        self._stop_event.clear()
-        self._frame_count = 0
-        self._thread = threading.Thread(target=self._capture_loop, daemon=True)
-        self._thread.start()
-
-    def stop(self) -> None:
-        self._stop_event.set()
-        if self._thread:
-            self._thread.join(timeout=10)
-
-    @property
-    def frame_count(self) -> int:
-        return self._frame_count
-
-    def _capture_loop(self) -> None:
-        interval = 1.0 / self._fps
-        with mss.mss() as sct:
-            monitor = sct.monitors[0]  # 全屏（monitor[0] 是所有显示器合并区域）
-            while not self._stop_event.is_set():
-                t0 = time.perf_counter()
-                img = sct.grab(monitor)
-                path = self._frames_dir / f"frame_{self._frame_count:06d}.png"
-                mss.tools.to_png(img.rgb, img.size, output=str(path))
-                self._frame_count += 1
-                elapsed = time.perf_counter() - t0
-                sleep = max(0.0, interval - elapsed)
-                time.sleep(sleep)
+__all__ = ["ScreenRecorder"]

tools/screencast/runner.py

@@ -1,108 +1,6 @@
-"""主编排器：录屏 + 运行 pytest + 生成字幕 + FFmpeg 合成。"""
-from __future__ import annotations
+from ._bootstrap import bootstrap_src
 
-import os
-import subprocess
-import sys
-import tempfile
-import time
-from datetime import datetime
-from pathlib import Path
-from typing import List, Optional, Tuple
+bootstrap_src()
+from screencast.runner import run_screencast
 
-from .composer import compose_video
-from .recorder import ScreenRecorder
-from .subtitle import SubtitleEngine
-
-
-def run_screencast(
-    skill_slug: str,
-    subtitle_script: List[Tuple[str, str]],
-    pytest_args: List[str],
-    output_dir: str,
-    media_assets_root: Optional[str] = None,
-    music_subdir: str = "music",
-    fps: int = 10,
-) -> str:
-    """
-    录制一次技能桌面 E2E 演示视频。
-
-    Args:
-        skill_slug:       技能唯一标识，如 "query-balance-icbc"
-        subtitle_script:  [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout
-        pytest_args:      传递给 pytest 的参数列表
-        output_dir:       最终 MP4 输出目录
-        media_assets_root: media-assets 仓库本地路径；
-                           不传时读 MEDIA_ASSETS_ROOT 环境变量，
-                           再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
-        music_subdir:     music_assets_root 下音乐子目录，默认 "music"
-        fps:              截帧帧率，默认 10
-
-    Returns:
-        输出 MP4 的绝对路径
-    """
-    if media_assets_root is None:
-        media_assets_root = os.environ.get(
-            "MEDIA_ASSETS_ROOT",
-            r"D:\OpenClaw\client-commons\media-assets",
-        )
-
-    music_dir = str(Path(media_assets_root) / music_subdir)
-    output_dir_path = Path(output_dir)
-    output_dir_path.mkdir(parents=True, exist_ok=True)
-
-    date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
-    base_name = f"{skill_slug}_{date_str}"
-    output_mp4 = output_dir_path / f"{base_name}.mp4"
-    subtitle_file = output_dir_path / f"{base_name}.srt"
-
-    engine = SubtitleEngine(subtitle_script)
-
-    with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir:
-        recorder = ScreenRecorder(frames_dir, fps=fps)
-
-        print(f"[screencast] 启动录制 → {output_mp4.name}")
-        recorder.start()
-        engine.set_start_time(time.time())
-
-        try:
-            # Windows 子进程 stdout 默认走系统 ANSI code page（中文 Windows 是 cp936/GBK），
-            # 这边用 encoding="utf-8" 解码会乱码。注入 PYTHONIOENCODING + PYTHONUTF8 让
-            # 子进程 Python 强制以 UTF-8 输出，两端编码对齐。非 Windows 默认就 UTF-8，无副作用。
-            env = os.environ.copy()
-            env["PYTHONIOENCODING"] = "utf-8"
-            env["PYTHONUTF8"] = "1"
-
-            proc = subprocess.Popen(
-                [sys.executable, "-m", "pytest", *pytest_args],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                text=True,
-                encoding="utf-8",
-                errors="replace",
-                env=env,
-            )
-            assert proc.stdout is not None
-            for line in proc.stdout:
-                sys.stdout.write(line)
-                sys.stdout.flush()
-                engine.process_line(line)
-            exit_code = proc.wait()
-            if exit_code != 0:
-                print(f"[screencast] pytest 退出码 {exit_code}，录制仍继续合成")
-        finally:
-            recorder.stop()
-            print(f"[screencast] 录制停止，共 {recorder.frame_count} 帧")
-
-        engine.generate_srt(str(subtitle_file))
-        print(f"[screencast] 字幕 → {subtitle_file.name}")
-
-        compose_video(
-            frames_dir=frames_dir,
-            fps=fps,
-            subtitle_path=str(subtitle_file),
-            music_dir=music_dir,
-            output_path=str(output_mp4),
-        )
-
-    return str(output_mp4)
+__all__ = ["run_screencast"]

tools/screencast/subtitle.py

@@ -1,81 +1,6 @@
-"""字幕引擎：根据 stdout 关键词打时间戳，生成 SRT 字幕文件。"""
-from __future__ import annotations
+from ._bootstrap import bootstrap_src
 
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import List, Tuple
+bootstrap_src()
+from screencast.subtitle import SubtitleEngine
 
-
-@dataclass
-class _Entry:
-    start: float   # 距录制开始的秒数
-    text: str
-    duration: float = 5.0  # 每条字幕默认显示 5 秒
-
-
-class SubtitleEngine:
-    def __init__(self, script: List[Tuple[str, str]], default_duration: float = 5.0):
-        """
-        script: [(关键词, 字幕文案), ...]
-        关键词大小写不敏感，每个关键词只触发一次。
-        """
-        self._script = script
-        self._default_duration = default_duration
-        self._entries: List[_Entry] = []
-        self._matched: set[str] = set()
-        self._t0: float = 0.0
-
-    def set_start_time(self, t: float) -> None:
-        self._t0 = t
-
-    def process_line(self, line: str) -> None:
-        lower = line.lower()
-        for keyword, text in self._script:
-            if keyword in self._matched:
-                continue
-            if keyword.lower() in lower:
-                elapsed = time.time() - self._t0
-                self._entries.append(_Entry(start=elapsed, text=text, duration=self._default_duration))
-                self._matched.add(keyword)
-
-    def generate_srt(self, output_path: str) -> None:
-        """生成 SRT 文件，自动处理字幕重叠。
-
-        字幕重叠修复策略（避免屏幕上同时显示 2+ 条叠成两行）：
-        - 字幕按 start 时间升序排序后；
-        - 相邻字幕保留 GAP=0.1s 间隔；
-        - 每条字幕保证至少 MIN_DURATION=1.0s 显示时间（短到这个值仍读不完是字幕脚本设计问题）；
-        - 若当前条让位给下一条后剩余时间 < MIN_DURATION，则把下一条 start 往后推。
-        这样任意时刻屏幕上最多 1 条字幕。
-        """
-        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}"
-
-        # 防重叠 post-process
-        MIN_DURATION = 1.0
-        GAP = 0.1
-        entries = sorted(self._entries, key=lambda e: e.start)
-        for i in range(len(entries) - 1):
-            cur, nxt = entries[i], entries[i + 1]
-            max_end = nxt.start - GAP
-            new_dur = max_end - cur.start
-            if new_dur < MIN_DURATION:
-                # 太挤：cur 至少撑 MIN_DURATION 秒，把 nxt 推迟
-                cur.duration = MIN_DURATION
-                nxt.start = cur.start + MIN_DURATION + GAP
-            else:
-                # 正常让位：cur 显示到 (nxt.start - GAP)
-                cur.duration = min(cur.duration, new_dur)
-        # 最后一条不需要让位，保持原 duration
-
-        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(entries, 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")
+__all__ = ["SubtitleEngine"]