lookatbar/add-todo-all
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始化测试通过

Browse Source
This commit is contained in:
lookatbar
2026-09-10 13:04:13 +08:00
parent ccb12dad7c
commit 9c7b27c675
40 changed files with 141 additions and 1790 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
examples/todo_skill/README.md
View File

@@ -1,126 +0,0 @@
# 待办事项 demo（todo-skill）
> 这是匠厂 `skill-template` 的**教学型 demo**，复刻模板的分层与 Action 契约，
> 自带一份最小 mock 平台库，可以脱离宿主在本机直接跑通。
> 真实业务技能**不要** vendor `jiangchang_skill_core`（见模板 `SKILL.md`）。
## 这个 demo 演示什么
- **四层架构** `cli → service → db → util`（与 [../../scripts/](../../scripts) 一致）
- **CLI 入口**：`scripts/main.py` 子命令分发（argparse + 中文错误）
- **业务表 + 数据管理元数据**：`todos` 表 + `_jiangchang_tables` / `_jiangchang_columns`
- **Action 契约**：[assets/actions.json](assets/actions.json) 覆盖 `toolbar`/`row`/`batch`/`cron`/`agent`/`skill-detail` 六个 placement
- **sync / async 双形态**：CRUD 是 sync（当场返回）；`remind` 是 async（演示 `job_context` / `emit` / `finish` 任务流）
- **mock 平台库**：`scripts/_mock_platform/jiangchang_skill_core/`，在没有宿主时自动注入 sys.path
## 快速开始
```bash
# 用独立的数据目录跑（避免污染默认位置）
# Windows PowerShell
$env:JIANGCHANG_USER_ID="demo_test"
$env:JIANGCHANG_DATA_ROOT="$env:TEMP\jc_demo_run"
# 1. 初始化数据库（幂等创建 todos 表 + 展示元数据）
python examples/todo_skill/scripts/main.py init-db
# 2. 新增三条待办
python examples/todo_skill/scripts/main.py todo add "写文档" -p high
python examples/todo_skill/scripts/main.py todo add "回邮件" -p normal
python examples/todo_skill/scripts/main.py todo add "买咖啡" -p low
# 3. 列出 / 按状态过滤
python examples/todo_skill/scripts/main.py todo list
python examples/todo_skill/scripts/main.py todo list -s pending
# 4. 批量标记完成（演示逗号分隔 ids，对应 actions.json 的 batch 入口）
python examples/todo_skill/scripts/main.py todo done 1,2
# 5. 统计
python examples/todo_skill/scripts/main.py todo stats
# 6. async 提醒任务（演示 job_context/emit/finish 任务流）
python examples/todo_skill/scripts/main.py remind -s 2
# 7. 删除 / 健康检查 / 版本
python examples/todo_skill/scripts/main.py todo delete 3
python examples/todo_skill/scripts/main.py health
python examples/todo_skill/scripts/main.py version
python examples/todo_skill/scripts/main.py config-path
```
数据落在 `{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/todo-skill/todo-skill.db`，
不设环境变量时回退到 `{系统临时目录}/jiangchang_demo_cache/demo_user/todo-skill/`。
## 目录结构
```
examples/todo_skill/
├── README.md # 本文（用户说明）
├── SKILL.md # 技能入口（Agent / LLM 读，对齐模板 frontmatter）
├── assets/
│ └── actions.json # Action 清单（6 placements × sync/async）
└── scripts/
├── main.py # CLI 入口；含 mock 平台库回退注入
├── _mock_platform/ # mock jiangchang_skill_core（仅 demo 独立运行用）
│ └── jiangchang_skill_core/
│ ├── __init__.py
│ ├── runtime_env.py # 数据根 / 用户 ID / apply_cli_local_defaults
│ ├── config.py # .env 加载与合并
│ ├── activity.py # job_context / emit / finish
│ └── diagnostics.py # health 体检
├── cli/
│ └── app.py # argparse 子命令分发
├── service/
│ ├── task_service.py # 标准命令 + remind async 编排
│ └── todo_service.py # 待办 CRUD 业务内核（多入口复用）
├── db/
│ ├── connection.py # SQLite 连接 + init_db（todos 表 + 元数据）
│ └── display_metadata.py # _jiangchang_tables/_columns 元数据 DDL
└── util/
├── constants.py # SKILL_SLUG / SKILL_VERSION
├── runtime_paths.py # 数据目录与 DB 路径
├── argparse_zh.py # 中文 argparse 错误
├── logging_config.py # stdlib logging 薄封装
└── config_bootstrap.py # .env.example → 用户 .env 合并
```
## 与真实模板的差异
| 维度 | 真实模板 | 本 demo |
|------|----------|----------|
| `jiangchang_skill_core` 来源 | 宿主共享 Python runtime（`jiangchang-platform-kit`） | 自带 mock（`scripts/_mock_platform/`） |
| RPA / 视频录制 | `RpaVideoSession` 完整链路 | 未实现（remind 仅模拟进度，不开浏览器） |
| 任务中心 UI | 宿主「Run Journal」推送 + 进度面板 | stdout 打印 JSON Lines 模拟 |
| 鉴权 | `entitlement_service` 调宿主接口 | 完全省略 |
| HTML 报告 | `render-html` action + `html/*.html` | 未实现（教学范围外） |
| 日志 | `jiangchang_skill_core.unified_logging` 统一文件日志 | stdlib logging，仅 stdout |
## 三大契约在 demo 中的对应
### 1. CLI 契约（[scripts/cli/app.py](scripts/cli/app.py)）
固定子命令：`init-db` / `todo add|list|done|delete|stats` / `remind` / `health` / `version` / `config-path`。
### 2. Actions 契约（[assets/actions.json](assets/actions.json)）
| action id | executionProfile | placements | 说明 |
|-----------|------------------|------------|------|
| `health` / `version` | sync | skill-detail | 标准 |
| `todo-add` | sync | toolbar, agent, skill-detail | 表单新增 |
| `todo-list` | sync | skill-detail, agent | 只读查询 |
| `todo-done` | sync | row, batch, agent | 行级 / 批量，含 `confirmation` |
| `todo-delete` | sync | row, batch | 行级 / 批量删除 |
| `todo-stats` | sync | skill-detail, agent | 只读统计 |
| `remind` | **async** | toolbar, cron, agent, skill-detail | 演示任务中心入口 |
### 3. 数据管理元数据（[scripts/db/connection.py](scripts/db/connection.py)）
- 业务表：`todos`（id/title/status/priority/due_at/created_at/updated_at，字段顺序即列展示顺序）
- 元数据表：`_jiangchang_tables` + `_jiangchang_columns`
- `todos` 表 `readonly=1`：增删改通过技能 Action 走，宿主不直写
## 冒烟测试
```bash
cd examples/todo_skill
python -m unittest discover -s tests
```

49
examples/todo_skill/SKILL.md
View File

@@ -1,49 +0,0 @@
---
name: 待办事项 demo
description: "匠厂 skill-template 的教学型 demo：用待办事项场景演示分层架构、CLI 契约、Action manifest 与 sync/async 双形态。"
version: 0.1.0
author: demo
metadata:
openclaw:
slug: todo-skill
platform_kit_min_version: "1.2.11"
emoji: "✅"
category: "通用"
developer_ids:
- 1
allowed-tools:
- bash
---
# 待办事项 demo（todo-skill）
这是匠厂 [`skill-template`](../../SKILL.md) 的**教学型 demo**，用最简单的待办事项业务
复刻模板的所有关键模式。它**不是**真实业务技能，自带一份 mock 平台库以便脱离宿主运行。
## 关键模式映射
| 模板铁律 | 本 demo 落地 |
|----------|-------------|
| `cli → service → db → util` 四层 | [scripts/](scripts/) 完整复刻同结构 |
| 单业务内核、多入口 | `service/todo_service.py` 是唯一 CRUD 内核，CLI/Action/数据管理按钮都调它 |
| CLI 固定入口 | [scripts/main.py](scripts/main.py) + `cli/app.py` argparse 分发 |
| Action manifest 正交字段 | [assets/actions.json](assets/actions.json) 覆盖 6 placements × sync/async |
| 数据管理元数据 | [scripts/db/connection.py](scripts/db/connection.py) 建 `todos` + `_jiangchang_*` |
| 长任务 async 范式 | [service/task_service.py](scripts/service/task_service.py) `cmd_remind` 用 `job_context`/`emit`/`finish` |
## Agent 触发路由表（demo）
| 用户意图 | Agent 怎么做 | 进任务中心？ |
|----------|--------------|--------------|
| 看版本 / 检查环境 | bash + 共享 Python `version` / `health` | 否 |
| 列出 / 统计 / 新增 / 完成 / 删除 | bash + `todo list/stats/add/done/delete` | 否（sync） |
| 扫描并提醒 | `run_skill_action` → `remind`（async） | **是** |
## 独立运行
详见 [README.md](README.md)。设 `JIANGCHANG_DATA_ROOT` + `JIANGCHANG_USER_ID` 后可直接 `python scripts/main.py <cmd>` 跑通。
## 与真实模板的差异
本 demo 自带 `_mock_platform/jiangchang_skill_core`，**真实技能不得 vendor 平台库**。
其它省略项：RPA/视频、鉴权、HTML 报告、统一文件日志（详见 README）。

189
examples/todo_skill/assets/actions.json
View File

@@ -1,189 +0,0 @@
{
"schemaVersion": 1,
"skill": "todo-skill",
"actions": [
{
"id": "health",
"label": "运行检查",
"description": "检查 demo 运行环境（Python / mock 平台库 / 数据库路径），不启动浏览器。",
"placements": ["skill-detail", "agent"],
"executionProfile": "sync",
"entrypoint": {
"type": "cli",
"command": "health",
"args": []
}
},
{
"id": "version",
"label": "查看版本",
"description": "返回 demo 版本与 slug。",
"placements": ["skill-detail"],
"executionProfile": "sync",
"entrypoint": {
"type": "cli",
"command": "version",
"args": []
}
},
{
"id": "todo-add",
"label": "新增待办",
"description": "在 todos 表新增一条待办事项（走技能 CLI，不直接写库）。",
"placements": ["toolbar", "agent", "skill-detail"],
"executionProfile": "sync",
"interaction": "form",
"bind": { "tables": ["todos"] },
"hostEffects": { "onSuccess": ["reloadTable"] },
"inputSchema": {
"type": "object",
"required": ["title"],
"properties": {
"title": {
"type": "string",
"title": "标题",
"description": "待办事项内容"
},
"priority": {
"type": "string",
"title": "优先级",
"enum": ["high", "normal", "low"],
"default": "normal"
},
"dueAt": {
"type": "number",
"title": "截止时间",
"description": "Unix 秒；可空",
"display_type": "datetime_unix_seconds"
}
}
},
"entrypoint": {
"type": "cli",
"command": "todo",
"args": ["add", "{{title}}", "--priority", "{{priority}}", "--due-at", "{{dueAt}}"]
}
},
{
"id": "todo-list",
"label": "查看列表",
"description": "列出待办事项（可选按状态过滤），返回 JSON。",
"placements": ["skill-detail", "agent"],
"executionProfile": "sync",
"inputSchema": {
"type": "object",
"properties": {
"status": {
"type": "string",
"title": "状态过滤",
"enum": ["pending", "done"],
"description": "留空则返回全部"
},
"limit": {
"type": "number",
"title": "返回数量",
"default": 100
}
}
},
"entrypoint": {
"type": "cli",
"command": "todo",
"args": ["list", "--status", "{{status}}", "--limit", "{{limit}}"]
}
},
{
"id": "todo-done",
"label": "标记完成",
"description": "将当前行待办标记为完成。",
"placements": ["row", "batch", "agent"],
"executionProfile": "sync",
"interaction": "confirm",
"confirmation": { "message": "将所选待办标记为完成？" },
"bind": { "tables": ["todos"] },
"hostEffects": { "onSuccess": ["reloadTable", "clearSelection"] },
"inputSchema": {
"type": "object",
"required": ["ids"],
"properties": {
"ids": {
"type": "string",
"title": "待办编号",
"description": "逗号分隔的待办 id（batch 入口由勾选映射）",
"readOnly": true
}
}
},
"entrypoint": {
"type": "cli",
"command": "todo",
"args": ["done", "{{ids}}"]
}
},
{
"id": "todo-delete",
"label": "删除待办",
"description": "删除当前行待办（技能级删除，会更新宿主表格）。",
"placements": ["row", "batch"],
"executionProfile": "sync",
"interaction": "confirm",
"confirmation": { "message": "确认删除所选待办？此操作不可撤销。" },
"bind": { "tables": ["todos"] },
"hostEffects": { "onSuccess": ["reloadTable", "clearSelection"] },
"inputSchema": {
"type": "object",
"required": ["ids"],
"properties": {
"ids": {
"type": "string",
"title": "待办编号",
"description": "逗号分隔的待办 id（batch 入口由勾选映射）",
"readOnly": true
}
}
},
"entrypoint": {
"type": "cli",
"command": "todo",
"args": ["delete", "{{ids}}"]
}
},
{
"id": "todo-stats",
"label": "查看统计",
"description": "返回待办事项简单统计（总数 / 待办 / 完成）。",
"placements": ["skill-detail", "agent"],
"executionProfile": "sync",
"entrypoint": {
"type": "cli",
"command": "todo",
"args": ["stats"]
}
},
{
"id": "remind",
"label": "提醒扫描",
"description": "演示 async 任务：扫描待办事项并模拟发出提醒。进入任务中心，支持进度与取消。",
"placements": ["toolbar", "cron", "agent", "skill-detail"],
"executionProfile": "async",
"bind": { "tables": ["todos"] },
"inputSchema": {
"type": "object",
"properties": {
"seconds": {
"type": "number",
"title": "运行秒数",
"description": "模拟任务持续秒数，用于演示进度推送",
"default": 3,
"minimum": 1
}
}
},
"entrypoint": {
"type": "cli",
"command": "remind",
"args": ["--seconds", "{{seconds}}"]
}
}
]
}

14
examples/todo_skill/scripts/_mock_platform/jiangchang_skill_core/__init__.py
View File

@@ -1,14 +0,0 @@
"""jiangchang_skill_core mock 包（仅供 demo 独立运行使用）。
真实环境中本包来自宿主共享 Python runtime 的 ``jiangchang-platform-kit``，
不在技能仓库内 vendored。本 mock 只为让 demo 脱离宿主也能跑通，行为不
等价于真实平台库。
"""
from __future__ import annotations
from . import config as config # noqa: F401
from . import runtime_env as runtime_env # noqa: F401
from .activity import JobStopped, emit, finish, job_context # noqa: F401
from .diagnostics import collect_runtime_diagnostics, format_runtime_health_lines # noqa: F401
__version__ = "1.2.11-mock"

61
examples/todo_skill/scripts/_mock_platform/jiangchang_skill_core/activity.py
View File

@@ -1,61 +0,0 @@
"""activity mock：job_context / emit / finish 任务中心契约的最小实现。
真实平台 ``emit``/``finish`` 会写入 Run Journal（``{DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl``）
并推送到任务中心 UI；本 mock 只把进度打印到 stdout，便于在终端看到任务流。
"""
from __future__ import annotations
import json
import sys
import time
import uuid
from contextlib import contextmanager
from typing import Any, Iterator
class JobStopped(Exception):
"""用户/宿主主动停止任务时抛出。真实平台由心跳检查触发；demo 不会自动抛。"""
def _emit_record(payload: dict[str, Any]) -> None:
"""统一以 JSON Lines 打印到 stdout，模拟 Run Journal 一行。"""
line = json.dumps(payload, ensure_ascii=False, default=str)
print(f"[activity] {line}", flush=True)
def emit(message: str, *, type: str = "info", skill: str = "", **fields: Any) -> None:
"""报告进度事件。"""
_emit_record({
"ts": int(time.time()),
"event": "emit",
"type": type,
"skill": skill,
"message": message,
**fields,
})
def finish(*, status: str, message: str = "", skill: str = "", **fields: Any) -> None:
"""报告任务终态。``status`` ∈ ``ok``/``failed``/``stopped``。"""
_emit_record({
"ts": int(time.time()),
"event": "finish",
"status": status,
"skill": skill,
"message": message,
**fields,
})
# 真实平台 finish 后任务中心记录终态；mock 额外在 stderr 提示一句。
print(f"[activity] task_finished status={status} message={message or '(empty)'}", file=sys.stderr)
@contextmanager
def job_context(*, skill: str) -> Iterator[dict[str, Any]]:
"""任务上下文：开始时生成 jobId，结束时打印边界。"""
job_id = uuid.uuid4().hex
ctx = {"skill": skill, "job_id": job_id}
_emit_record({"ts": int(time.time()), "event": "job_start", "skill": skill, "job_id": job_id})
try:
yield ctx
finally:
_emit_record({"ts": int(time.time()), "event": "job_end", "skill": skill, "job_id": job_id})

89
examples/todo_skill/scripts/_mock_platform/jiangchang_skill_core/config.py
View File

@@ -1,89 +0,0 @@
"""config mock：只读环境变量 + 简单 .env 合并。
真实平台有更完整的多层配置与缓存；本 mock 仅演示优先级：
进程环境变量 > 用户 .env > .env.example。
"""
from __future__ import annotations
import os
from typing import Optional
_cache: dict[str, str] = {}
_env_path: Optional[str] = None
def _load_env_file(path: str) -> dict[str, str]:
if not path or not os.path.isfile(path):
return {}
data: dict[str, str] = {}
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
data[key.strip()] = _strip_comment(value.strip())
return data
def _strip_comment(value: str) -> str:
"""去掉行尾 `` # 注释``（保持简单，不处理引号内 #）。"""
if " #" in value:
value = value.split(" #", 1)[0].rstrip()
return value
def get(key: str, default: Optional[str] = None) -> Optional[str]:
"""读配置：进程环境变量优先，其次已加载的 .env。"""
if key in os.environ:
return os.environ[key]
if key in _cache:
return _cache[key]
return default
def get_env_file_path() -> Optional[str]:
return _env_path
def ensure_env_file(skill_slug: str, example_path: str) -> str:
"""确保用户 .env 存在，缺失时从 example 复制。"""
global _env_path
from .runtime_env import get_data_root, get_user_id
user_env_dir = os.path.join(get_data_root(), get_user_id(), skill_slug)
os.makedirs(user_env_dir, exist_ok=True)
path = os.path.join(user_env_dir, ".env")
if not os.path.isfile(path) and os.path.isfile(example_path):
with open(example_path, encoding="utf-8") as src:
with open(path, "w", encoding="utf-8") as dst:
dst.write(src.read())
_env_path = path
_refresh_cache(path)
return path
def merge_missing_env_keys(example_path: str, env_path: str, comment_skill: Optional[str] = None) -> None:
"""把 example 中存在但 .env 缺失的键追加到 .env 末尾（带版本注释）。"""
if not os.path.isfile(example_path) or not os.path.isfile(env_path):
return
existing = _load_env_file(env_path)
example = _load_env_file(example_path)
missing = {k: v for k, v in example.items() if k not in existing}
if not missing:
return
with open(env_path, "a", encoding="utf-8") as f:
if comment_skill:
f.write(f"\n# === merged by {comment_skill} ===\n")
for k, v in missing.items():
f.write(f"{k}={v}\n")
_refresh_cache(env_path)
def reset_cache() -> None:
_cache.clear()
def _refresh_cache(path: str) -> None:
_cache.clear()
_cache.update(_load_env_file(path))

89
examples/todo_skill/scripts/_mock_platform/jiangchang_skill_core/diagnostics.py
View File

@@ -1,89 +0,0 @@
"""runtime diagnostics mock：``health`` 命令的只读环境体检。"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass, field
from typing import Any
from . import config
from .runtime_env import get_data_root, get_user_id
# 避免循环导入：与 __init__.py 的 __version__ 保持同步
_CORE_VERSION = "1.2.11-mock"
@dataclass
class RuntimeDiagnostics:
skill_slug: str
python_executable: str
platform_kit_version: str
platform_kit_min_version: str
platform_kit_version_ok: bool
jiangchang_skill_core_file: str
data_root: str
user_id: str
env_path: str
env_exists: bool
has_fatal_issues: bool = False
issues: list[dict[str, str]] = field(default_factory=list)
def collect_runtime_diagnostics(
*,
skill_slug: str,
platform_kit_min_version: str,
skill_root: str,
) -> RuntimeDiagnostics:
"""收集只读运行时诊断。致命问题（缺 python、版本不符等）置 has_fatal_issues。"""
issues: list[dict[str, str]] = []
# 版本检查（mock 总是自报合格，便于 demo 跑通）
version_ok = True
if platform_kit_min_version and _CORE_VERSION < platform_kit_min_version:
version_ok = False
issues.append({
"level": "error",
"code": "PLATFORM_KIT_VERSION_TOO_LOW",
"message": f"jiangchang_skill_core {_CORE_VERSION} < 要求 {platform_kit_min_version}",
})
env_path = config.get_env_file_path() or ""
env_exists = bool(env_path and os.path.isfile(env_path))
if not env_exists:
issues.append({
"level": "warning",
"code": "ENV_FILE_MISSING",
"message": f"用户 .env 不存在：{env_path or '(未设置)'}",
})
return RuntimeDiagnostics(
skill_slug=skill_slug,
python_executable=sys.executable,
platform_kit_version=_CORE_VERSION,
platform_kit_min_version=platform_kit_min_version,
platform_kit_version_ok=version_ok,
jiangchang_skill_core_file=__file__,
data_root=get_data_root(),
user_id=get_user_id(),
env_path=env_path,
env_exists=env_exists,
has_fatal_issues=any(i["level"] == "error" for i in issues),
issues=issues,
)
def format_runtime_health_lines(runtime: RuntimeDiagnostics) -> list[str]:
"""格式化为可打印行，对齐真实 ``health`` 输出风格。"""
lines = [
f"python_executable: {runtime.python_executable}",
f"platform_kit_version: {runtime.platform_kit_version}",
f"platform_kit_min_version: {runtime.platform_kit_min_version}",
f"platform_kit_version_ok: {runtime.platform_kit_version_ok}",
f"jiangchang_skill_core_file: {runtime.jiangchang_skill_core_file}",
f"data_root: {runtime.data_root}",
f"user_id: {runtime.user_id}",
]
for issue in runtime.issues:
lines.append(f"runtime_issue[{issue['level']}]: {issue['code']} - {issue['message']}")
return lines

34
examples/todo_skill/scripts/_mock_platform/jiangchang_skill_core/runtime_env.py
View File

@@ -1,34 +0,0 @@
"""runtime_env mock：解析数据根目录与用户 ID。
真实平台用宿主注入的环境变量；本 mock 在没有宿主时用本地临时目录兜底。
"""
from __future__ import annotations
import os
import tempfile
_DEFAULT_USER_ID = "demo_user"
def get_data_root() -> str:
"""数据根目录：真实平台来自 ``JIANGCHANG_DATA_ROOT``，demo 用本地缓存目录。"""
root = os.environ.get("JIANGCHANG_DATA_ROOT")
if root:
return root
cache = os.path.join(tempfile.gettempdir(), "jiangchang_demo_cache")
os.makedirs(cache, exist_ok=True)
return cache
def get_user_id() -> str:
return os.environ.get("JIANGCHANG_USER_ID") or _DEFAULT_USER_ID
def get_sibling_skills_root(scripts_dir: str) -> str:
"""同级技能根：父目录的父目录（即 ``examples/`` 之类）。"""
return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(scripts_dir))))
def apply_cli_local_defaults() -> None:
"""CLI 本地默认配置：真实平台会调整 PATH/VIRTUAL_ENV 等；mock 仅占位。"""
return None

0
examples/todo_skill/scripts/cli/__init__.py
View File

128
examples/todo_skill/scripts/cli/app.py
View File

@@ -1,128 +0,0 @@
"""CLI：argparse 解析与分发（业务内核在 service，本层只做参数→service 映射）。"""
from __future__ import annotations
import argparse
import sys
from typing import List, Optional
from service.task_service import (
cmd_add,
cmd_config_path,
cmd_delete,
cmd_done,
cmd_health,
cmd_init_db,
cmd_list,
cmd_remind,
cmd_stats,
cmd_version,
)
from util.config_bootstrap import bootstrap_skill_config
from util.argparse_zh import ZhArgumentParser
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.logging_config import get_skill_logger, setup_skill_logging
def _cli_str_or_none(raw: Optional[str]) -> Optional[str]:
if raw is None:
return None
v = str(raw).strip()
return v or None
def _parse_ids(raw: str) -> list[int]:
"""解析 ``1`` 或 ``1,2,3`` 为 int 列表。"""
parts = [p.strip() for p in (raw or "").split(",") if p.strip()]
try:
return [int(p) for p in parts]
except ValueError as exc:
raise SystemExit(f"参数错误：ids 必须是数字或逗号分隔的数字列表，收到 {raw!r}\n{exc}")
def _print_full_usage() -> None:
print("待办事项 demo（main.py）可用命令：")
print(" python main.py init-db # 初始化数据库与元数据")
print(" python main.py todo add <title> [-p high|normal|low] [-d <unix秒>")
print(" python main.py todo list [-s pending|done] [-n <limit>]")
print(" python main.py todo done <id>")
print(" python main.py todo delete <id>")
print(" python main.py todo stats")
print(" python main.py remind [-s <秒>]")
print(" python main.py health / version / config-path")
def _add_todo_subparser(sub) -> None:
sp = sub.add_parser("todo", help="待办事项 CRUD")
todo_sub = sp.add_subparsers(dest="todo_cmd", required=True, parser_class=ZhArgumentParser)
p = todo_sub.add_parser("add", help="新增一条待办")
p.add_argument("title", help="待办标题（必填）")
p.add_argument("-p", "--priority", default="normal",
choices=["high", "normal", "low"], help="优先级，默认 normal")
p.add_argument("-d", "--due-at", type=int, default=None,
help="截止时间（Unix 秒，可选）")
p.set_defaults(handler=lambda a: cmd_add(
title=a.title, priority=a.priority, due_at=a.due_at))
p = todo_sub.add_parser("list", help="列出待办")
p.add_argument("-s", "--status", default=None,
choices=["pending", "done"], help="按状态过滤")
p.add_argument("-n", "--limit", type=int, default=100, help="返回数量上限")
p.set_defaults(handler=lambda a: cmd_list(status=a.status, limit=a.limit))
p = todo_sub.add_parser("done", help="标记完成（可传逗号分隔的多个 id）")
p.add_argument("ids", help="待办 id 或逗号分隔的 id 列表（如 1 或 1,2,3）")
p.set_defaults(handler=lambda a: cmd_done(todo_ids=_parse_ids(a.ids)))
p = todo_sub.add_parser("delete", help="删除（可传逗号分隔的多个 id）")
p.add_argument("ids", help="待办 id 或逗号分隔的 id 列表")
p.set_defaults(handler=lambda a: cmd_delete(todo_ids=_parse_ids(a.ids)))
p = todo_sub.add_parser("stats", help="简单统计")
p.set_defaults(handler=lambda _a: cmd_stats())
def _add_remind_subparser(sub) -> None:
sp = sub.add_parser("remind", help="演示 async 提醒任务（job_context/emit/finish）")
sp.add_argument("-s", "--seconds", type=int, default=3,
help="模拟运行秒数，默认 3")
sp.set_defaults(handler=lambda a: cmd_remind(seconds=a.seconds))
def build_parser() -> ZhArgumentParser:
p = ZhArgumentParser(
prog="main.py",
description="待办事项 demo：演示匠厂 skill-template 的分层与 Action 契约。",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
sp = sub.add_parser("init-db", help="幂等初始化 SQLite 与数据管理元数据")
sp.set_defaults(handler=lambda _a: cmd_init_db())
_add_todo_subparser(sub)
_add_remind_subparser(sub)
sp = sub.add_parser("health", help="健康检查")
sp.set_defaults(handler=lambda _a: cmd_health())
sp = sub.add_parser("version", help="版本信息(JSON)")
sp.set_defaults(handler=lambda _a: cmd_version())
sp = sub.add_parser("config-path", help="输出 .env 与模板路径")
sp.set_defaults(handler=lambda _a: cmd_config_path())
return p
def main(argv: Optional[List[str]] = None) -> int:
argv = argv if argv is not None else sys.argv[1:]
bootstrap_skill_config()
setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
get_skill_logger().info("cli_start argv=%s", sys.argv)
if not argv:
_print_full_usage()
return 1
parser = build_parser()
args = parser.parse_args(argv)
return int(args.handler(args))

0
examples/todo_skill/scripts/db/__init__.py
View File

96
examples/todo_skill/scripts/db/connection.py
View File

@@ -1,96 +0,0 @@
"""SQLite 连接与幂等初始化。
``init_db`` 建两张表：
1. ``todos`` — 业务表（demo：待办事项）
2. ``_jiangchang_tables`` / ``_jiangchang_columns`` — 宿主数据管理展示元数据
业务表字段顺序就是宿主列展示顺序（cid 顺序）。
"""
from __future__ import annotations
import sqlite3
from db.display_metadata import (
init_display_metadata_tables,
seed_column_metadata_batch,
upsert_table_metadata,
)
from util.runtime_paths import get_db_path
# ---- 业务表 DDL（待办事项）----
# 字段顺序 = 数据管理列展示顺序；状态/优先级用 options_json 给宿主做下拉筛选
_TODOS_DDL = """
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
priority TEXT NOT NULL DEFAULT 'normal',
due_at INTEGER,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
)
"""
_TODOS_INDEX = """
CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status)
"""
def get_conn() -> sqlite3.Connection:
return sqlite3.connect(get_db_path())
def _seed_todos_display_metadata(cursor) -> None:
"""写 ``todos`` 表与字段的中文展示元数据。"""
upsert_table_metadata(
cursor,
table_name="todos",
display_name="待办事项",
description="本技能的核心业务表：每条待办事项一行。",
grid_help="状态变更走技能 CLI（todo done / todo delete），不要在表里直接改。",
visible=1,
readonly=1, # 整表只读：增删改通过技能 Action 走，宿主不直写
default_sort_json='[{"column":"created_at","direction":"desc"}]',
default_visible_columns_json='["title","status","priority","due_at","created_at"]',
)
seed_column_metadata_batch(
cursor,
table_name="todos",
columns=[
{"column_name": "id", "display_name": "编号", "description": "主键",
"visible": 1, "searchable": 1, "editable": 0, "display_type": "number"},
{"column_name": "title", "display_name": "标题", "description": "待办事项内容",
"grid_help": "简明描述要做的事", "visible": 1, "searchable": 1, "editable": 0,
"display_type": "text"},
{"column_name": "status", "display_name": "状态", "description": "pending/done",
"grid_help": "pending=待办，done=完成", "visible": 1, "searchable": 1, "editable": 0,
"display_type": "enum",
"options_json": '[{"value":"pending","label":"待办"},{"value":"done","label":"完成"}]'},
{"column_name": "priority", "display_name": "优先级", "description": "high/normal/low",
"visible": 1, "searchable": 1, "editable": 0, "display_type": "enum",
"options_json": '[{"value":"high","label":"高"},{"value":"normal","label":"中"},{"value":"low","label":"低"}]'},
{"column_name": "due_at", "display_name": "截止时间", "description": "Unix 秒；可空",
"visible": 1, "searchable": 1, "editable": 0, "display_type": "datetime_unix_seconds"},
{"column_name": "created_at", "display_name": "创建时间", "description": "Unix 秒",
"visible": 1, "searchable": 0, "editable": 0, "display_type": "datetime_unix_seconds"},
{"column_name": "updated_at", "display_name": "更新时间", "description": "Unix 秒",
"visible": 0, "searchable": 0, "editable": 0, "display_type": "datetime_unix_seconds"},
],
)
def init_db() -> None:
"""幂等创建/迁移本地 SQLite 与数据管理元数据。"""
conn = get_conn()
try:
cur = conn.cursor()
# 1. 元数据表（先建，方便其它步骤 upsert）
init_display_metadata_tables(cur)
# 2. 业务表
cur.execute(_TODOS_DDL)
cur.execute(_TODOS_INDEX)
# 3. 业务表展示元数据
_seed_todos_display_metadata(cur)
conn.commit()
finally:
conn.close()

160
examples/todo_skill/scripts/db/display_metadata.py
View File

@@ -1,160 +0,0 @@
"""匠厂数据管理展示元数据（``_jiangchang_tables`` / ``_jiangchang_columns``）。
SQLite 没有原生表/字段 COMMENT，宿主「数据管理」页面靠这两张元数据表
拿到中文表名/列名/可编辑性等。字段顺序由 ``CREATE TABLE`` + ``PRAGMA table_info``
的 ``cid`` 决定，不读 ``display_order``。
本文件是模板 ``scripts/db/display_metadata.py`` 的最小可用子集：
- ``init_display_metadata_tables``：建元数据表
- ``upsert_table_metadata`` / ``upsert_column_metadata``：写入展示元数据
- ``list_physical_column_names``：按 cid 列出物理字段名
"""
from __future__ import annotations
from typing import Iterable
# 宿主隐藏这些表，不作为业务表展示
INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns"})
_JIANGCHANG_TABLES_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_tables (
table_name TEXT PRIMARY KEY,
display_name TEXT NOT NULL,
description TEXT,
grid_help TEXT,
visible INTEGER NOT NULL DEFAULT 1,
readonly INTEGER NOT NULL DEFAULT 0,
default_sort_json TEXT,
default_visible_columns_json TEXT
)
"""
_JIANGCHANG_COLUMNS_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_columns (
table_name TEXT NOT NULL,
column_name TEXT NOT NULL,
display_name TEXT NOT NULL,
description TEXT,
grid_help TEXT,
display_order INTEGER,
visible INTEGER NOT NULL DEFAULT 1,
searchable INTEGER NOT NULL DEFAULT 1,
editable INTEGER NOT NULL DEFAULT 1,
display_type TEXT,
options_json TEXT,
PRIMARY KEY (table_name, column_name)
)
"""
def init_display_metadata_tables(cursor) -> None:
cursor.execute(_JIANGCHANG_TABLES_DDL)
cursor.execute(_JIANGCHANG_COLUMNS_DDL)
def upsert_table_metadata(
cursor,
*,
table_name: str,
display_name: str,
description: str | None = None,
grid_help: str | None = None,
visible: int = 1,
readonly: int = 0,
default_sort_json: str | None = None,
default_visible_columns_json: str | None = None,
) -> None:
cursor.execute(
"""
INSERT INTO _jiangchang_tables (
table_name, display_name, description, grid_help, visible, readonly,
default_sort_json, default_visible_columns_json
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(table_name) DO UPDATE SET
display_name = excluded.display_name,
description = excluded.description,
grid_help = excluded.grid_help,
visible = excluded.visible,
readonly = excluded.readonly,
default_sort_json = excluded.default_sort_json,
default_visible_columns_json = excluded.default_visible_columns_json
""",
(
table_name, display_name, description, grid_help, visible, readonly,
default_sort_json, default_visible_columns_json,
),
)
def upsert_column_metadata(
cursor,
*,
table_name: str,
column_name: str,
display_name: str,
description: str | None = None,
grid_help: str | None = None,
visible: int = 1,
searchable: int = 1,
editable: int = 1,
display_type: str | None = None,
options_json: str | None = None,
) -> None:
cursor.execute(
"""
INSERT INTO _jiangchang_columns (
table_name, column_name, display_name, description, grid_help,
display_order, visible, searchable, editable, display_type, options_json
)
VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?)
ON CONFLICT(table_name, column_name) DO UPDATE SET
display_name = excluded.display_name,
description = excluded.description,
grid_help = excluded.grid_help,
display_order = NULL,
visible = excluded.visible,
searchable = excluded.searchable,
editable = excluded.editable,
display_type = excluded.display_type,
options_json = excluded.options_json
""",
(
table_name, column_name, display_name, description, grid_help,
visible, searchable, editable, display_type, options_json,
),
)
def seed_column_metadata_batch(
cursor,
*,
table_name: str,
columns: Iterable[dict[str, object]],
) -> None:
"""批量 upsert 字段元数据。"""
for column in columns:
upsert_column_metadata(
cursor,
table_name=table_name,
column_name=str(column["column_name"]),
display_name=str(column["display_name"]),
description=str(column["description"]) if column.get("description") else None,
grid_help=str(column["grid_help"]) if column.get("grid_help") else None,
visible=int(column.get("visible", 1)),
searchable=int(column.get("searchable", 1)),
editable=int(column.get("editable", 1)),
display_type=str(column["display_type"]) if column.get("display_type") else None,
options_json=str(column["options_json"]) if column.get("options_json") else None,
)
def _quote_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def list_physical_column_names(cursor, table_name: str) -> list[str]:
cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
rows = cursor.fetchall()
rows.sort(key=lambda row: int(row[0]))
return [str(row[1]) for row in rows]

49
examples/todo_skill/scripts/main.py
View File

@@ -1,49 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""待办事项 demo CLI 入口（结构对齐 skill-template 的 main.py）。
分层：cli（argv）→ service（业务编排）→ db（持久化）→ util（通用工具）。
独立运行说明：
真实技能依赖宿主共享 Python runtime 的 ``jiangchang_skill_core``。
本 demo 自带一份最小 mock（``scripts/_mock_platform/``），在没有宿主时
由本文件自动注入 sys.path，使 demo 在任意 Python 3.10+ 环境可直接跑通。
设置环境变量 ``JIANGCHANG_DATA_ROOT`` 即可切到真实平台 runtime 模式。
"""
from __future__ import annotations
import os
import sys
# Windows 终端 UTF-8（对齐模板）
if sys.platform == "win32":
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
_scripts_dir = os.path.dirname(os.path.abspath(__file__))
if _scripts_dir not in sys.path:
sys.path.insert(0, _scripts_dir)
# 没有真实平台库时启用 mock（demo 自带，便于脱离宿主独立运行）；
# 设环境变量 ``TODO_SKILL_USE_REAL_PLATFORM=1`` 可强制走真实 runtime。
if not os.environ.get("TODO_SKILL_USE_REAL_PLATFORM"):
try:
import jiangchang_skill_core # noqa: F401
except ImportError:
_mock_platform = os.path.join(_scripts_dir, "_mock_platform")
if _mock_platform not in sys.path:
sys.path.insert(0, _mock_platform)
from jiangchang_skill_core.runtime_env import apply_cli_local_defaults # noqa: E402
apply_cli_local_defaults()
from util.config_bootstrap import bootstrap_skill_config # noqa: E402
bootstrap_skill_config()
from cli.app import main # noqa: E402
if __name__ == "__main__":
raise SystemExit(main())

0
examples/todo_skill/scripts/service/__init__.py
View File

189
examples/todo_skill/scripts/service/task_service.py
View File

@@ -1,189 +0,0 @@
"""任务编排层（对齐模板 task_service.py 的职责分工）。
- ``cmd_health`` / ``cmd_version`` / ``cmd_config_path`` / ``cmd_init_db``：标准 CLI；
- ``cmd_remind``：演示 async 任务范式（``job_context`` + ``emit`` + ``finish``），
真实平台会推送到任务中心 UI；本 demo 在 stdout 打印 JSON Lines 模拟。
业务内核（CRUD）在 ``todo_service``，本模块只做编排，不复制业务逻辑。
"""
from __future__ import annotations
import asyncio
import json
import os
import time
from typing import Any, Optional
from jiangchang_skill_core import collect_runtime_diagnostics, config, format_runtime_health_lines
from jiangchang_skill_core.activity import JobStopped, emit, finish, job_context
from db.connection import init_db
from service import todo_service
from util.constants import LOG_LOGGER_NAME, PLATFORM_KIT_MIN_VERSION, SKILL_SLUG, SKILL_VERSION
from util.logging_config import get_skill_logger, setup_skill_logging
from util.runtime_paths import get_db_path, get_skill_root
def _get_task_logger():
try:
return get_skill_logger()
except RuntimeError:
setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
return get_skill_logger()
# ---------------- 标准命令（sync，当场返回） ----------------
def cmd_config_path() -> int:
example_path = os.path.join(get_skill_root(), ".env.example")
env_path = config.get_env_file_path() or ""
print(json.dumps(
{"skill": SKILL_SLUG, "env_path": env_path, "example_path": os.path.abspath(example_path)},
ensure_ascii=False,
))
return 0
def cmd_health() -> int:
runtime = collect_runtime_diagnostics(
skill_slug=SKILL_SLUG,
platform_kit_min_version=PLATFORM_KIT_MIN_VERSION,
skill_root=get_skill_root(),
)
example_path = os.path.join(get_skill_root(), ".env.example")
env_path = config.get_env_file_path() or ""
env_exists = bool(env_path and os.path.isfile(env_path))
health_status = "failed" if runtime.has_fatal_issues else "ok"
lines = [
f"{SKILL_SLUG} health: {health_status}",
*format_runtime_health_lines(runtime),
f"env_path: {env_path}",
f"env_exists: {env_exists}",
f"example_path: {os.path.abspath(example_path)}",
f"db_path: {get_db_path()}",
]
for line in lines:
print(line)
return 1 if runtime.has_fatal_issues else 0
def cmd_version() -> int:
print(json.dumps({"version": SKILL_VERSION, "skill": SKILL_SLUG}, ensure_ascii=False))
return 0
def cmd_init_db() -> int:
"""幂等创建本地 SQLite 与数据管理元数据。"""
try:
init_db()
except Exception as exc: # CLI 边界统一成可机读失败
print(json.dumps(
{"ok": False, "skill": SKILL_SLUG, "error": str(exc)},
ensure_ascii=False,
))
return 1
print(json.dumps(
{"ok": True, "skill": SKILL_SLUG, "db_path": get_db_path()},
ensure_ascii=False,
))
return 0
# ---------------- 业务命令（sync） ----------------
def cmd_list(*, status: Optional[str] = None, limit: int = 100) -> int:
todos = todo_service.list_todos(status=status, limit=limit)
print(json.dumps(
{"count": len(todos), "items": [t.to_dict() for t in todos]},
ensure_ascii=False,
))
return 0
def cmd_add(*, title: str, priority: str = "normal", due_at: Optional[int] = None) -> int:
try:
todo = todo_service.add_todo(title=title, priority=priority, due_at=due_at)
except ValueError as exc:
print(f"ERROR:TODO_INVALID: {exc}")
return 1
print(json.dumps(todo.to_dict(), ensure_ascii=False))
return 0
def cmd_done(*, todo_ids: list[int]) -> int:
"""批量标记完成。返回 JSON 列表，每项含 ok/id/error。"""
results: list[dict] = []
for tid in todo_ids:
try:
todo = todo_service.mark_done(tid)
results.append({"ok": True, "id": tid, "todo": todo.to_dict()})
except ValueError as exc:
results.append({"ok": False, "id": tid, "error": str(exc)})
print(json.dumps({"count": len(results), "items": results}, ensure_ascii=False))
return 0 if all(r["ok"] for r in results) else 1
def cmd_delete(*, todo_ids: list[int]) -> int:
"""批量删除。"""
results: list[dict] = []
for tid in todo_ids:
deleted = todo_service.delete_todo(tid)
results.append({"ok": deleted, "id": tid})
print(json.dumps({"count": len(results), "items": results}, ensure_ascii=False))
return 0 if all(r["ok"] for r in results) else 1
def cmd_stats() -> int:
print(json.dumps(todo_service.stats(), ensure_ascii=False))
return 0
# ---------------- async 任务范式（demo） ----------------
async def _run_remind_async(*, seconds: int) -> dict[str, Any]:
"""模拟长耗时任务：每秒 emit 一次进度，最后 finish 终态。
真实平台会推送到任务中心 UI 并支持取消；本 demo 在 stdout 打印 JSON Lines。
"""
steps = max(1, int(seconds))
for i in range(1, steps + 1):
await asyncio.sleep(1)
emit(f"提醒任务进度 {i}/{steps}", skill=SKILL_SLUG, stage="remind", progress=i, total=steps)
# 真实业务在这里会查 todos 表挑出 due/pending 项发出提醒；demo 不做真实推送
summary = todo_service.stats()
return {"remind": True, "duration_seconds": steps, "stats": summary}
def cmd_remind(*, seconds: int = 3) -> int:
"""演示 async 任务编排。CLI 入口仍走同一内核；Agent 真实调度时
通过 ``run_skill_action``（executionProfile: async）触发。"""
log = _get_task_logger()
task_type = "remind"
with job_context(skill=SKILL_SLUG):
log.info("task_start task_type=%s stage=remind seconds=%s", task_type, seconds)
emit("开始执行提醒任务", skill=SKILL_SLUG, stage="remind")
try:
result = asyncio.run(_run_remind_async(seconds=seconds))
log.info("task_done task_type=%s status=ok", task_type)
finish(
status="ok",
message=f"提醒任务完成，处理 {seconds} 秒",
skill=SKILL_SLUG,
stage="remind",
**result,
)
print(json.dumps({"ok": True, "task_type": task_type, **result}, ensure_ascii=False))
return 0
except JobStopped:
raise
except Exception:
log.exception("task_failed task_type=%s", task_type)
emit("任务失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="remind")
finish(
status="failed",
message="提醒任务执行异常，详见统一日志",
skill=SKILL_SLUG,
stage="remind",
)
return 1

167
examples/todo_skill/scripts/service/todo_service.py
View File

@@ -1,167 +0,0 @@
"""待办事项领域服务（业务内核）。
按模板铁律：「单业务内核、多入口」——CLI / Action / 数据管理按钮 / 定时任务
都调用这里的同一个函数，禁止按入口分叉业务行为。
直接用 sqlite3 做持久化（demo 简化，未抽 repository）。
"""
from __future__ import annotations
import time
from dataclasses import dataclass
from typing import Optional
from db.connection import get_conn
# 业务常量（不暴露给宿主，技能自定）
STATUS_PENDING = "pending"
STATUS_DONE = "done"
VALID_STATUS = {STATUS_PENDING, STATUS_DONE}
PRIORITY_HIGH = "high"
PRIORITY_NORMAL = "normal"
PRIORITY_LOW = "low"
VALID_PRIORITY = {PRIORITY_HIGH, PRIORITY_NORMAL, PRIORITY_LOW}
@dataclass
class Todo:
id: int
title: str
status: str
priority: str
due_at: Optional[int]
created_at: int
updated_at: int
def to_dict(self) -> dict:
return {
"id": self.id,
"title": self.title,
"status": self.status,
"priority": self.priority,
"due_at": self.due_at,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
def _now() -> int:
return int(time.time())
def add_todo(
*,
title: str,
priority: str = PRIORITY_NORMAL,
due_at: Optional[int] = None,
) -> Todo:
"""新增一条待办。领域校验：标题非空、优先级合法。"""
title = (title or "").strip()
if not title:
raise ValueError("标题不能为空")
if priority not in VALID_PRIORITY:
raise ValueError(f"优先级不合法，可选：{sorted(VALID_PRIORITY)}")
now = _now()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"INSERT INTO todos (title, status, priority, due_at, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?)",
(title, STATUS_PENDING, priority, due_at, now, now),
)
new_id = cur.lastrowid
conn.commit()
return Todo(new_id, title, STATUS_PENDING, priority, due_at, now, now)
finally:
conn.close()
def list_todos(*, status: Optional[str] = None, limit: int = 100) -> list[Todo]:
"""列出待办；可按状态过滤。默认按创建时间倒序。"""
sql = "SELECT id, title, status, priority, due_at, created_at, updated_at FROM todos"
params: list = []
if status:
if status not in VALID_STATUS:
raise ValueError(f"状态不合法，可选：{sorted(VALID_STATUS)}")
sql += " WHERE status = ?"
params.append(status)
sql += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(sql, params)
rows = cur.fetchall()
return [Todo(*row) for row in rows]
finally:
conn.close()
def get_todo(todo_id: int) -> Optional[Todo]:
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"SELECT id, title, status, priority, due_at, created_at, updated_at "
"FROM todos WHERE id = ?",
(todo_id,),
)
row = cur.fetchone()
return Todo(*row) if row else None
finally:
conn.close()
def mark_done(todo_id: int) -> Todo:
"""标记一条待办为完成。"""
todo = get_todo(todo_id)
if not todo:
raise ValueError(f"待办 {todo_id} 不存在")
if todo.status == STATUS_DONE:
raise ValueError(f"待办 {todo_id} 已是完成状态")
now = _now()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"UPDATE todos SET status = ?, updated_at = ? WHERE id = ?",
(STATUS_DONE, now, todo_id),
)
conn.commit()
todo.status = STATUS_DONE
todo.updated_at = now
return todo
finally:
conn.close()
def delete_todo(todo_id: int) -> bool:
"""删除一条待办；返回是否真的删了一行。"""
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
conn.commit()
return cur.rowcount > 0
finally:
conn.close()
def stats() -> dict:
"""简单统计：用于 ``todo stats`` 命令与 demo 报告。"""
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("SELECT status, COUNT(*) FROM todos GROUP BY status")
by_status = {row[0]: int(row[1]) for row in cur.fetchall()}
cur.execute("SELECT COUNT(*) FROM todos")
total = int(cur.fetchone()[0])
return {
"total": total,
"pending": by_status.get(STATUS_PENDING, 0),
"done": by_status.get(STATUS_DONE, 0),
}
finally:
conn.close()

0
examples/todo_skill/scripts/util/__init__.py
View File

11
examples/todo_skill/scripts/util/argparse_zh.py
View File

@@ -1,11 +0,0 @@
"""argparse 中文错误说明（与模板一致）。"""
from __future__ import annotations
import argparse
import sys
class ZhArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
print(f"参数错误：{message}\n请执行：python main.py -h 查看帮助", file=sys.stderr)
self.exit(2)

24
examples/todo_skill/scripts/util/config_bootstrap.py
View File

@@ -1,24 +0,0 @@
"""配置初始化：从 .env.example 复制到用户 .env，并合并缺失键。
行为对齐模板：``main.py`` 与 ``cli.app.main()`` 启动时均会调用一次。
"""
from __future__ import annotations
import os
from jiangchang_skill_core import config
from util.constants import SKILL_SLUG, SKILL_VERSION
from util.runtime_paths import get_skill_root
def bootstrap_skill_config() -> str:
"""确保用户 .env 存在，并追加 .env.example 中的新配置项。"""
example_path = os.path.join(get_skill_root(), ".env.example")
if not os.path.isfile(example_path):
# demo 没有强制 .env.example，直接返回空路径
return ""
env_path = config.ensure_env_file(SKILL_SLUG, example_path)
comment = f"{SKILL_SLUG} v{SKILL_VERSION}" if SKILL_VERSION else None
config.merge_missing_env_keys(example_path, env_path, comment_skill=comment)
return env_path

11
examples/todo_skill/scripts/util/constants.py
View File

@@ -1,11 +0,0 @@
"""技能标识与版本（demo：待办事项）。
真实技能需保证 slug = 目录名 = DB 文件名 = ``metadata.openclaw.slug``。
"""
from __future__ import annotations
# 英文 kebab-case 机器标识；与目录名、actions.json 的 ``skill`` 字段一致
SKILL_SLUG = "todo-skill"
SKILL_VERSION = "0.1.0"
LOG_LOGGER_NAME = "openclaw.skill.todo_skill"
PLATFORM_KIT_MIN_VERSION = "1.2.11"

29
examples/todo_skill/scripts/util/logging_config.py
View File

@@ -1,29 +0,0 @@
"""日志：直接用 stdlib logging（真实模板用 jiangchang_skill_core.unified_logging）。
为保持与模板 ``from util.logging_config import get_skill_logger, setup_skill_logging``
一致的调用面，本 demo 提供同名薄封装。
"""
from __future__ import annotations
import logging
from util.constants import LOG_LOGGER_NAME
_configured = False
def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
global _configured
if _configured:
return
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
logger = logging.getLogger(logger_name)
logger.setLevel(logging.INFO)
if not logger.handlers:
logger.addHandler(handler)
_configured = True
def get_skill_logger() -> logging.Logger:
return logging.getLogger(LOG_LOGGER_NAME)

32
examples/todo_skill/scripts/util/runtime_paths.py
View File

@@ -1,32 +0,0 @@
"""技能数据目录与 DB 路径解析（简化版）。
真实模板用 ``jiangchang_skill_core.runtime_env``；本 demo 复用其 mock。
数据目录 = ``{data_root}/{user_id}/{slug}/``，DB 即 ``{slug}.db``。
"""
from __future__ import annotations
import os
from jiangchang_skill_core.runtime_env import get_data_root, get_user_id
from util.constants import SKILL_SLUG
_SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def get_skill_root() -> str:
"""技能根目录（含 SKILL.md / scripts/ 的目录）。"""
return os.path.dirname(_SCRIPTS_DIR)
def get_skill_data_dir() -> str:
"""本技能数据目录：``{data_root}/{user_id}/{slug}/``。"""
path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG)
os.makedirs(path, exist_ok=True)
return path
def get_db_path(filename: str | None = None) -> str:
"""SQLite 文件路径，默认 ``{slug}.db``。"""
name = filename or f"{SKILL_SLUG}.db"
return os.path.join(get_skill_data_dir(), name)

0
examples/todo_skill/tests/__init__.py
View File

25
examples/todo_skill/tests/conftest.py
View File

@@ -1,25 +0,0 @@
"""pytest 全局 fixture：把 ``scripts/`` 注入 sys.path 并隔离数据目录。"""
from __future__ import annotations
import os
import sys
import tempfile
import pytest
_SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")
if _SCRIPTS_DIR not in sys.path:
sys.path.insert(0, _SCRIPTS_DIR)
@pytest.fixture
def isolated_env(tmp_path, monkeypatch):
"""隔离的数据根目录，避免污染本机真实数据。"""
data_root = tmp_path / "data_root"
data_root.mkdir()
monkeypatch.setenv("JIANGCHANG_DATA_ROOT", str(data_root))
monkeypatch.setenv("JIANGCHANG_USER_ID", "test_user")
# 重置 config mock 的缓存，让下一次 get 重新读 env
from jiangchang_skill_core import config
config.reset_cache()
return data_root

168
examples/todo_skill/tests/test_smoke.py
View File

@@ -1,168 +0,0 @@
"""冒烟测试：覆盖 demo 关键路径。
不依赖 pytest，可用 ``python -m unittest discover -s tests`` 跑。
但 fixture 用法对齐 pytest；这里把 fixture 用 ``setUp`` 模拟。
"""
from __future__ import annotations
import io
import json
import os
import sys
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
# 注入 scripts/ 到 sys.path
_HERE = os.path.dirname(os.path.abspath(__file__))
_SCRIPTS_DIR = os.path.join(os.path.dirname(_HERE), "scripts")
if _SCRIPTS_DIR not in sys.path:
sys.path.insert(0, _SCRIPTS_DIR)
# 与 main.py 同样的 mock 注入：没有真实平台库时启用 mock
if not os.environ.get("TODO_SKILL_USE_REAL_PLATFORM"):
try:
import jiangchang_skill_core # noqa: F401
except ImportError:
_mock_platform = os.path.join(_SCRIPTS_DIR, "_mock_platform")
if _mock_platform not in sys.path:
sys.path.insert(0, _mock_platform)
class IsolatedEnv:
"""临时数据根目录，测试结束自动清理。"""
def __init__(self):
self._tmp = tempfile.mkdtemp(prefix="todo_test_")
self._prev_root = os.environ.get("JIANGCHANG_DATA_ROOT")
self._prev_user = os.environ.get("JIANGCHANG_USER_ID")
os.environ["JIANGCHANG_DATA_ROOT"] = os.path.join(self._tmp, "data")
os.environ["JIANGCHANG_USER_ID"] = "test_user"
os.makedirs(os.environ["JIANGCHANG_DATA_ROOT"], exist_ok=True)
def reset_config(self):
from jiangchang_skill_core import config
config.reset_cache()
def __enter__(self):
self.reset_config()
return self
def __exit__(self, *exc):
if self._prev_root is None:
os.environ.pop("JIANGCHANG_DATA_ROOT", None)
else:
os.environ["JIANGCHANG_DATA_ROOT"] = self._prev_root
if self._prev_user is None:
os.environ.pop("JIANGCHANG_USER_ID", None)
else:
os.environ["JIANGCHANG_USER_ID"] = self._prev_user
return False
def _run_cli(argv: list[str]) -> tuple[int, str, str]:
"""调用 cli.app.main，返回 (rc, stdout, stderr)。"""
from cli.app import main
out = io.StringIO()
err = io.StringIO()
with redirect_stdout(out), redirect_stderr(err):
rc = main(argv)
return rc, out.getvalue(), err.getvalue()
class TestCliSmoke(unittest.TestCase):
def test_empty_argv_shows_usage_and_nonzero(self):
rc, out, _ = _run_cli([])
self.assertEqual(rc, 1)
self.assertIn("待办事项 demo", out)
def test_version_json(self):
with IsolatedEnv():
rc, out, _ = _run_cli(["version"])
self.assertEqual(rc, 0)
payload = json.loads(out.strip())
self.assertEqual(payload["skill"], "todo-skill")
self.assertIn("version", payload)
def test_init_db_creates_tables_and_is_idempotent(self):
with IsolatedEnv():
rc, out, _ = _run_cli(["init-db"])
self.assertEqual(rc, 0)
payload = json.loads(out.strip())
self.assertTrue(payload["ok"])
self.assertTrue(os.path.isfile(payload["db_path"]))
# 再来一次应仍成功
rc2, out2, _ = _run_cli(["init-db"])
self.assertEqual(rc2, 0)
self.assertEqual(json.loads(out2.strip())["db_path"], payload["db_path"])
def test_add_list_done_stats_full_flow(self):
with IsolatedEnv():
_run_cli(["init-db"])
rc, out, _ = _run_cli(["todo", "add", "任务A", "-p", "high"])
self.assertEqual(rc, 0)
item = json.loads(out.strip())
self.assertEqual(item["title"], "任务A")
self.assertEqual(item["status"], "pending")
todo_id = item["id"]
rc, out, _ = _run_cli(["todo", "list"])
self.assertEqual(rc, 0)
listing = json.loads(out.strip())
self.assertEqual(listing["count"], 1)
rc, out, _ = _run_cli(["todo", "done", str(todo_id)])
self.assertEqual(rc, 0)
result = json.loads(out.strip())
self.assertTrue(result["items"][0]["ok"])
self.assertEqual(result["items"][0]["todo"]["status"], "done")
rc, out, _ = _run_cli(["todo", "stats"])
stats = json.loads(out.strip())
self.assertEqual(stats, {"total": 1, "pending": 0, "done": 1})
def test_done_batch_with_comma_ids(self):
with IsolatedEnv():
_run_cli(["init-db"])
_run_cli(["todo", "add", "T1"])
_run_cli(["todo", "add", "T2"])
rc, out, _ = _run_cli(["todo", "done", "1,2"])
self.assertEqual(rc, 0)
result = json.loads(out.strip())
self.assertEqual(result["count"], 2)
self.assertTrue(all(r["ok"] for r in result["items"]))
def test_done_invalid_id_returns_error(self):
with IsolatedEnv():
_run_cli(["init-db"])
rc, out, _ = _run_cli(["todo", "done", "999"])
self.assertEqual(rc, 1)
result = json.loads(out.strip())
self.assertFalse(result["items"][0]["ok"])
def test_remind_async_pattern_emits_progress(self):
with IsolatedEnv():
_run_cli(["init-db"])
_run_cli(["todo", "add", "T1"])
rc, out, _ = _run_cli(["remind", "-s", "1"])
self.assertEqual(rc, 0)
# stdout 应含 emit 进度事件（mock 把 activity 打到 stdout）
self.assertIn('"event": "emit"', out)
self.assertIn('"event": "finish"', out)
self.assertIn('"status": "ok"', out)
# 最后一行是 CLI 结果 JSON
last_line = [l for l in out.strip().splitlines() if l.startswith("{")][-1]
payload = json.loads(last_line)
self.assertTrue(payload["ok"])
self.assertEqual(payload["task_type"], "remind")
def test_health_returns_zero(self):
with IsolatedEnv():
rc, out, _ = _run_cli(["health"])
self.assertEqual(rc, 0)
self.assertIn("todo-skill health: ok", out)
self.assertIn("platform_kit_version:", out)
if __name__ == "__main__":
unittest.main()