jiangchang-platform-kit/README.md

jiangchang-platform-kit

1. 项目简介

匠厂平台共享 Python SDK：在同一发行包中提供 Skill 侧通用能力（权益校验、统一日志与运行环境解析）与 桌面应用自动化能力（通过 CDP 连接匠厂客户端，驱动聊天与断言）。

2. 安装方式

pip install jiangchang-platform-kit --index-url https://git.jc2009.com/api/packages/client-jiangchang/pypi/simple/

3. 包含的模块

• jiangchang_skill_core：权益 HTTP 客户端、enforce_entitlement、统一文件日志、JIANGCHANG_* 数据根与技能根路径解析、Skill Activity Stream / Job 运行时（emit / finish / Run Journal）、兄弟技能 CLI 桥（sibling_bridge）、共享媒体资源（media_assets）等，供 Skill CLI / 服务进程使用。
• jiangchang_desktop_sdk：匠厂桌面应用自动化客户端（基于 CDP + Playwright 同步 API）。本包不声明 Playwright 依赖；请在运行环境中自行安装 Playwright，以便 import playwright 可用。

4. 快速开始

jiangchang_skill_core

import os
from jiangchang_skill_core import (
    EntitlementClient,
    enforce_entitlement,
    setup_skill_logging,
    get_skill_logger,
)

os.environ.setdefault("JIANGCHANG_AUTH_BASE_URL", "https://auth.example.com")
os.environ.setdefault("JIANGCHANG_AUTH_API_KEY", "secret")

setup_skill_logging(skill_slug="demo-skill", logger_name="demo")
log = get_skill_logger()
log.info("skill started")

# client = EntitlementClient()
# enforce_entitlement(user_id="u1", skill_slug="demo-skill")

jiangchang_desktop_sdk

from jiangchang_desktop_sdk import JiangchangDesktopClient, AskOptions

# 需已安装 playwright，且匠厂客户端可按 CDP 端口暴露调试接口
client = JiangchangDesktopClient()
client.ensure_app_running(wait_timeout_s=30.0)
try:
    answer = client.ask("你好", options=AskOptions(new_task=True))
    print(answer)
finally:
    client.disconnect()

桌面 E2E（jiangchang_desktop_sdk.e2e_helpers）

Skill 的桌面端到端测试应通过宿主 IPC 获取真实当前用户数据目录，不要在 pytest 进程里手工设置 JIANGCHANG_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) 查看主进程解析的环境，不要用它拼用户数据路径。

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第二行")

Skill Activity Stream / Job 运行时

长任务（批量生成、RPA）需要把逐步进度推到匠厂宿主 UI，而不能依赖 Agent 轮询或 OpenClaw stdout patch。jiangchang_skill_core.activity 提供 Skill Activity Stream (SAS) v1 契约：

API	作用
emit(text, ...)	写 Run Journal（activity / progress / warn / debug）
step(text)	emit(type='activity') 别名
finish(status=..., **fields)	写终态事件 + 唯一向 stdout 输出单行 result JSON
rpa_step(name)	装饰器：进入/成功/失败自动 emit
job_context(...)	可选包裹 main；异常时自动 finish(failed)

Run Journal 路径（宿主 tail 此文件）

{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl

• job_id：优先 JIANGCHANG_JOB_ID，否则进程内 uuid4().hex（首次 emit 时生成并缓存）
• JIANGCHANG_RUN_MODE：job | cli（默认 cli），首行 meta 事件记录
• NDJSON，UTF-8，append-only，每行 flush
• debug 类型仅当 JIANGCHANG_ACTIVITY_DEBUG=1 时写入

与 Agent / OpenClaw 的分工

• emit() / journal：永不 print 到 stdout，避免 bash tool 结果淹没 Agent context
• finish()：唯一允许向 stdout 写业务结果的 API；单行 compact JSON，status + 摘要字段，默认 ≤4096 字符
• 技能里手写 print("OK")、pretty JSON 应逐步改为 finish()；RPA 步骤用 emit("打开浏览器") 或 @rpa_step

GEO batch 示例

from jiangchang_skill_core.activity import emit, finish

for i, item in enumerate(items, 1):
    emit(f"生成 {item['title']}", type="progress", current=i, total=len(items), skill="geo-batch")
    # ... business logic ...

finish(status="success", skill="geo-batch", generated=len(items))

RPA 步骤示例

from jiangchang_skill_core.activity import emit, rpa_step

@rpa_step("打开浏览器")
def open_browser(page):
    page.goto("https://example.com")

emit("手动步骤说明", skill="publish-toutiao", stage="rpa")

RpaVideoSession.add_step() 在录屏字幕之外默认同步 emit(..., stage="rpa.video")；可通过 emit_activity=False 关闭。

兄弟技能 CLI 桥

from jiangchang_skill_core import call_sibling_json, get_sibling_main_path

accounts = call_sibling_json("account-manager", ["list", "all"])

同步 subprocess.run，capture_output=True，自动带上 subprocess_env_with_trace()。

共享媒体资源

jiangchang_skill_core.media_assets 提供共享媒体资源解析能力。默认会从：

{JIANGCHANG_DATA_ROOT}/shared/media-assets

读取背景音乐、字体、水印和 ffmpeg 工具。

如果目录不存在，会优先下载稳定 release bundle（不依赖用户电脑安装 Git）：

https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip

bundle 内含背景音乐、字体与水印；不包含 ffmpeg 二进制。ffmpeg / ffprobe 仍由 bundle 内 manifest.json 的 tools.ffmpeg 下载源单独拉取。

可通过环境变量覆盖：

• MEDIA_ASSETS_ROOT=/path/to/custom/media-assets — 本地资源目录
• MEDIA_ASSETS_BUNDLE_URL=https://example.com/media-assets.zip — bundle 下载地址

Runtime diagnostics（统一 health 诊断）

jiangchang_skill_core.runtime_diagnostics 提供各 Skill 共用的 Runtime / media-assets / ffmpeg 探测，避免每个技能各自复制检查逻辑。Skill 的 health 命令可调用：

from jiangchang_skill_core import (
    collect_runtime_diagnostics,
    format_runtime_health_lines,
    runtime_diagnostics_dict,
)

diag = collect_runtime_diagnostics(
    skill_slug="my-skill",
    platform_kit_min_version="1.0.10",  # 可选；省略则只报告当前版本
    skill_root="/path/to/my-skill",     # 可选；用于检测 vendored jiangchang_skill_core
)
for line in format_runtime_health_lines(diag):
    print(line)
print(runtime_diagnostics_dict(diag))  # 可 JSON 序列化

诊断只读探测本地状态，不会触发 media-assets 下载。

录屏合成（screencast）

run_screencast() / compose_video() 通过 jiangchang_skill_core.media_assets 解析 ffmpeg 与背景音乐，不使用系统 PATH 中的 ffmpeg。

• 默认：不传 media_assets_root 时，使用 {JIANGCHANG_DATA_ROOT}/shared/media-assets（或进程环境中已有的 MEDIA_ASSETS_ROOT）。
• 覆盖：传入 media_assets_root="/path/to/media-assets"，或在 compose_video(..., media_assets_env={"MEDIA_ASSETS_ROOT": "..."}) 中设置；参数 media_assets_root 优先于 media_assets_env 里已有的 MEDIA_ASSETS_ROOT。
• run_screencast(media_assets_root=...) 会把该路径原样传给 compose_video()；music_subdir 已废弃（会触发 DeprecationWarning），音乐选择统一由 pick_background_music() 完成。

from screencast import run_screencast

run_screencast(
    skill_slug="demo-skill",
    subtitle_script=[("PASSED", "测试通过")],
    pytest_args=["-q", "tests/test_demo.py"],
    output_dir="./out",
    media_assets_root="/path/to/media-assets",  # 可选
)

5. 发版流程（维护者）

发布规则

• jiangchang-platform-kit 只发布正式版本。
• 每次需要发布公共能力时，先 bump pyproject.toml 的 version（例如 1.0.10 → 1.0.11）。
• 推送 main 后，.github/workflows/publish.yml 会构建并上传该正式版本到 Gitea PyPI。
• 发布包版本必须等于 pyproject.toml 中的 version；若该版本已存在，上传会失败，需继续 bump 版本号后再推送。

git push origin main

升级安装：

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

安装后验证

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)

TL;DR

本次内部重构:SDK 不再依赖匠厂主仓库的 data-jcid 属性。 所有公开 API(ask / read / new_task / wait_gateway_ready 等) 签名与行为保持不变,只有内部选择器实现切换到新的语义锚点 + ClawX testid。

改变了什么

• client.py 内所有 [data-jcid="..."] 选择器已移除
• 改用匠厂主仓库为 SDK 承诺的语义锚点:data-role / data-message-id / data-streaming(在 ChatMessage 上)、data-sending / data-message-count (在 chat-page 根上)
• window.__jc_sending__ 仍是核心 sending 信号,保持不变
• 这些锚点由匠厂主仓库的以下机制守护:
  • docs/JIANGCHANG_CUSTOM_ANCHORS.md 中央清单
  • 源码内 JIANGCHANG CUSTOM ANCHOR 边界注释
  • harness/specs/jiangchang-custom-anchors.spec.ts E2E 守护测试
  • Gitea CI grep 检查 step

已知限制

• send_file() 暂未实现,抛 NotImplementedError。 原因:附件上传走 Electron IPC,SDK 需要主仓库提供新 hook。
• _gateway_state() 永远返回 'unknown'。 原因:gateway 状态不再暴露 DOM data-state;本方法已预留未来通过 window.__jc_gateway_state__ 读取的实现路径。

这两个限制对常规 ask/read 工作流无影响,sending 完成判定依靠 window.__jc_sending__ + data-sending + 文本稳定性合流即可。

升级影响

• 在 jiangchang v2.0.17(含)以上版本上跑:应正常工作
• 在 jiangchang v2.0.16(及更早含 data-jcid 的版本)上跑:不再兼容 如果需要在旧版本上跑,固定 SDK 版本到 v0.1.x

验证清单(集成方建议跑一次)

打开匠厂客户端 → 用 SDK 跑下面 3 步,确认基础流程通:

from jiangchang_desktop_sdk import JiangchangDesktopClient

with JiangchangDesktopClient() as c:
    c.ensure_app_running()
    c.wait_gateway_ready()
    c.new_task()
    answer = c.ask("你好,请回复确认 SDK 可用")
    assert answer, f"Empty answer: {answer!r}"
    print("OK:", answer[:200])