client-commons/skill-template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: task_logs, task_service, run CLI; remove platform_playwright

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
chendelian
2026-05-10 10:00:37 +08:00
parent 4730cd527d
commit f3f59278b4
14 changed files with 240 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
assets/README.md
View File

@@ -3,5 +3,4 @@
- `examples/`：CLI 成功输出形状示例（虚构路径与数据）。
- `schemas/`：轻量 JSON Schema 示例（`additionalProperties: true` 允许扩展字段）。
- 面向用户的介绍见 `references/README.md`。
- 面向编排/CLI 的细节见 `references/CLI.md`、`RUNTIME.md`、`SCHEMA.md`。
面向编排与用户文档见 `references/README.md`、`references/CLI.md`、`RUNTIME.md`、`SCHEMA.md`。

10
assets/examples/log-get-response.json
View File

@@ -1,10 +1,12 @@
{
"id": 1,
"account_id": "demo_account_1",
"article_id": 12,
"article_title": "示例标题",
"status": "published",
"task_type": "demo_task",
"target_id": "demo_target_1",
"input_id": "demo_input_42",
"input_title": "示例任务输入标题",
"status": "success",
"error_msg": null,
"result_summary": "{\"processed\": 10, \"skipped\": 0}",
"created_at": "2026-04-01T10:00:00",
"updated_at": "2026-04-01T10:00:00"
}

19
assets/schemas/publish-log-record.schema.json
View File

@@ -1,19 +0,0 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://openclaw.local/skill-template/publish-log-record.schema.json",
"title": "PublishLogRecord",
"description": "发布型技能 log-get 返回的单条日志对象模板",
"type": "object",
"required": ["id", "account_id", "article_id", "article_title", "status", "created_at", "updated_at"],
"properties": {
"id": { "type": "integer" },
"account_id": { "type": ["string", "integer"] },
"article_id": { "type": "integer" },
"article_title": { "type": "string" },
"status": { "type": "string" },
"error_msg": { "type": ["string", "null"] },
"created_at": { "type": ["string", "null"] },
"updated_at": { "type": ["string", "null"] }
},
"additionalProperties": true
}

21
assets/schemas/task-log-record.schema.json Normal file
View File

@@ -0,0 +1,21 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://openclaw.local/skill-template/task-log-record.schema.json",
"title": "TaskLogRecord",
"description": "通用业务技能 log-get 返回的单条任务日志对象模板",
"type": "object",
"required": ["id", "task_type", "status", "created_at", "updated_at"],
"properties": {
"id": { "type": "integer" },
"task_type": { "type": "string" },
"target_id": { "type": ["string", "null"] },
"input_id": { "type": ["string", "null"] },
"input_title": { "type": ["string", "null"] },
"status": { "type": "string" },
"error_msg": { "type": ["string", "null"] },
"result_summary": { "type": ["string", "null"] },
"created_at": { "type": ["string", "null"] },
"updated_at": { "type": ["string", "null"] }
},
"additionalProperties": true
}

63
scripts/cli/app.py
View File

@@ -6,11 +6,11 @@ import argparse
import sys
from typing import List, Optional
from service.publish_service import (
from service.task_service import (
cmd_health,
cmd_log_get,
cmd_logs,
cmd_publish,
cmd_run,
cmd_version,
)
from util.argparse_zh import ZhArgumentParser
@@ -25,34 +25,34 @@ def _cli_str_or_none(raw: Optional[str]) -> Optional[str]:
return v or None
def _handle_publish(args: argparse.Namespace) -> int:
tail = [str(x).strip() for x in (args.publish_tail or []) if str(x).strip()]
def _handle_run(args: argparse.Namespace) -> int:
tail = [str(x).strip() for x in (args.run_tail or []) if str(x).strip()]
if len(tail) > 2:
print("❌ 参数过多。")
print("用法：python main.py publish [账号id [文章id]] | publish [-a 账号id] [-i 文章id]")
print("用法：python main.py run [target [input_id]] | run [-t target] [-i input_id]")
return 1
t_acc: Optional[str] = None
t_art: Optional[str] = None
t_target: Optional[str] = None
t_input: Optional[str] = None
if len(tail) == 2:
t_acc, t_art = tail[0], tail[1]
t_target, t_input = tail[0], tail[1]
elif len(tail) == 1:
if tail[0].isdigit():
t_art = tail[0]
t_input = tail[0]
else:
t_acc = tail[0]
t_target = tail[0]
pick_a = _cli_str_or_none(getattr(args, "account_id", None))
pick_i = _cli_str_or_none(getattr(args, "article_id", None))
acc = pick_a or t_acc
art = pick_i or t_art
return cmd_publish(account_id=acc, article_id=art)
pick_t = _cli_str_or_none(getattr(args, "target", None))
pick_i = _cli_str_or_none(getattr(args, "input_id", None))
target = pick_t or t_target
input_id = pick_i or t_input
return cmd_run(target=target, input_id=input_id)
def _print_full_usage() -> None:
print("模板技能（main.py）可用命令：")
print(" python main.py publish [账号id [文章id]] [-a 账号] [-i 文章id]")
print(" python main.py logs [--limit N] [--status s] [--account-id a]")
print("通用业务技能模板（main.py）可用命令：")
print(" python main.py run [target [input_id]] [-t target] [-i input_id]")
print(" python main.py logs [--limit N] [--status s] [--task-type t] [--target-id tid]")
print(" python main.py log-get <log_id>")
print(" python main.py health")
print(" python main.py version")
@@ -61,24 +61,27 @@ def _print_full_usage() -> None:
def build_parser() -> ZhArgumentParser:
p = ZhArgumentParser(
prog="main.py",
description="模板技能：发布命令骨架、日志查询、健康检查、版本输出。",
description="通用业务技能模板：任务执行命令骨架、日志查询、健康检查、版本输出。",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
sp = sub.add_parser("publish", help="发布型技能命令骨架")
sp.add_argument("--account-id", "-a", default=None, metavar="账号id")
sp.add_argument("--article-id", "-i", default=None, metavar="文章id")
sp.add_argument("publish_tail", nargs="*", metavar="位置参数")
sp.set_defaults(handler=_handle_publish)
sp = sub.add_parser("run", help="任务执行命令骨架")
sp.add_argument("--target", "-t", default=None, metavar="目标")
sp.add_argument("--input-id", "-i", default=None, metavar="输入id", dest="input_id")
sp.add_argument("run_tail", nargs="*", metavar="位置参数")
sp.set_defaults(handler=_handle_run)
sp = sub.add_parser("logs", help="查看发布记录")
sp = sub.add_parser("logs", help="查看任务日志")
sp.add_argument("--limit", type=int, default=10)
sp.add_argument("--status", default=None)
sp.add_argument("--account-id", default=None)
sp.set_defaults(handler=lambda a: cmd_logs(limit=a.limit, status=a.status, account_id=a.account_id))
sp.add_argument("--task-type", default=None, dest="task_type")
sp.add_argument("--target-id", default=None, dest="target_id")
sp.set_defaults(handler=lambda a: cmd_logs(
limit=a.limit, status=a.status, task_type=a.task_type, target_id=a.target_id
))
sp = sub.add_parser("log-get", help="按 log_id 查看单条发布记录(JSON)")
sp = sub.add_parser("log-get", help="按 log_id 查看单条任务日志(JSON)")
sp.add_argument("log_id")
sp.set_defaults(handler=lambda a: cmd_log_get(a.log_id))
@@ -97,8 +100,8 @@ def main(argv: Optional[List[str]] = None) -> int:
if not argv:
_print_full_usage()
return 1
if len(argv) == 2 and argv[0] not in {"publish", "logs", "log-get", "health", "version", "-h", "--help"}:
return cmd_publish(account_id=argv[0], article_id=argv[1])
if len(argv) == 2 and argv[0] not in {"run", "logs", "log-get", "health", "version", "-h", "--help"}:
return cmd_run(target=argv[0], input_id=argv[1])
parser = build_parser()
args = parser.parse_args(argv)
return int(args.handler(args))

12
scripts/db/connection.py
View File

@@ -1,4 +1,4 @@
"""SQLite 连接与日志表迁移模板。"""
"""SQLite 连接与任务日志表迁移模板。"""
from __future__ import annotations
@@ -17,13 +17,15 @@ def init_db() -> None:
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE IF NOT EXISTS publish_logs (
CREATE TABLE IF NOT EXISTS task_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL,
article_id INTEGER NOT NULL,
article_title TEXT,
task_type TEXT NOT NULL,
target_id TEXT,
input_id TEXT,
input_title TEXT,
status TEXT NOT NULL,
error_msg TEXT,
result_summary TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
)

76
scripts/db/publish_logs_repository.py
View File

@@ -1,76 +0,0 @@
"""publish_logs 表读写模板。"""
from __future__ import annotations
from typing import Any, List, Optional, Tuple
from db.connection import get_conn, init_db
from util.timeutil import now_unix
def save_publish_log(
account_id: str,
article_id: int,
article_title: str,
status: str,
error_msg: Optional[str] = None,
) -> int:
init_db()
now = now_unix()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
INSERT INTO publish_logs (account_id, article_id, article_title, status, error_msg, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(account_id, int(article_id), article_title or "", status, error_msg, now, now),
)
new_id = int(cur.lastrowid)
conn.commit()
finally:
conn.close()
return new_id
def list_publish_logs(
limit: int,
status: Optional[str] = None,
account_id: Optional[str] = None,
) -> List[Tuple[Any, ...]]:
init_db()
conn = get_conn()
try:
cur = conn.cursor()
sql = (
"SELECT id, account_id, article_id, article_title, status, error_msg, created_at, updated_at "
"FROM publish_logs WHERE 1=1 "
)
params: List[Any] = []
if status:
sql += "AND status = ? "
params.append(status)
if account_id:
sql += "AND account_id = ? "
params.append(account_id)
sql += "ORDER BY created_at DESC, id DESC LIMIT ?"
params.append(int(limit))
cur.execute(sql, tuple(params))
return list(cur.fetchall())
finally:
conn.close()
def get_publish_log_by_id(log_id: int) -> Optional[Tuple[Any, ...]]:
init_db()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"SELECT id, account_id, article_id, article_title, status, error_msg, created_at, updated_at FROM publish_logs WHERE id = ?",
(int(log_id),),
)
return cur.fetchone()
finally:
conn.close()

88
scripts/db/task_logs_repository.py Normal file
View File

@@ -0,0 +1,88 @@
"""task_logs 表读写模板。
通用业务日志仓储：记录每次任务执行的 task_type / target / input / 状态 / 结果摘要。
不假设任何特定业务（发布、对账、审核等都可复用）。
"""
from __future__ import annotations
from typing import Any, List, Optional, Tuple
from db.connection import get_conn, init_db
from util.timeutil import now_unix
def save_task_log(
task_type: str,
target_id: Optional[str] = None,
input_id: Optional[str] = None,
input_title: Optional[str] = None,
status: str = "success",
error_msg: Optional[str] = None,
result_summary: Optional[str] = None,
) -> int:
init_db()
now = now_unix()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
INSERT INTO task_logs
(task_type, target_id, input_id, input_title, status, error_msg, result_summary, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(task_type, target_id, input_id, input_title or "", status, error_msg, result_summary, now, now),
)
new_id = int(cur.lastrowid)
conn.commit()
finally:
conn.close()
return new_id
def list_task_logs(
limit: int,
status: Optional[str] = None,
task_type: Optional[str] = None,
target_id: Optional[str] = None,
) -> List[Tuple[Any, ...]]:
init_db()
conn = get_conn()
try:
cur = conn.cursor()
sql = (
"SELECT id, task_type, target_id, input_id, input_title, status, error_msg, result_summary, "
"created_at, updated_at FROM task_logs WHERE 1=1 "
)
params: List[Any] = []
if status:
sql += "AND status = ? "
params.append(status)
if task_type:
sql += "AND task_type = ? "
params.append(task_type)
if target_id:
sql += "AND target_id = ? "
params.append(target_id)
sql += "ORDER BY created_at DESC, id DESC LIMIT ?"
params.append(int(limit))
cur.execute(sql, tuple(params))
return list(cur.fetchall())
finally:
conn.close()
def get_task_log_by_id(log_id: int) -> Optional[Tuple[Any, ...]]:
init_db()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"SELECT id, task_type, target_id, input_id, input_title, status, error_msg, result_summary, "
"created_at, updated_at FROM task_logs WHERE id = ?",
(int(log_id),),
)
return cur.fetchone()
finally:
conn.close()

8
scripts/jiangchang_skill_core/runtime_env.py
View File

@@ -43,14 +43,10 @@ def get_user_id() -> str:
def _looks_like_skills_root(path: str) -> bool:
if not path or not os.path.isdir(path):
return False
# 通过常见基础技能存在与否，判断这是不是 skills 根目录
# 这里列举的是平台基础设施类技能，不包含具体业务技能
for marker in (
"llm-manager",
"content-manager",
"account-manager",
"sohu-publisher",
"toutiao-publisher",
"gongzhonghao-publisher",
"weibo-publisher",
"skill-template",
):
if os.path.isdir(os.path.join(path, marker)):

10
scripts/service/platform_playwright.py
View File

@@ -1,10 +0,0 @@
"""后台自动化占位模块模板。"""
from __future__ import annotations
from typing import Any, Dict
async def publish(account: Dict[str, Any], article: Dict[str, Any], account_id: str) -> str:
_ = (account, article, account_id)
return "ERROR:NOT_IMPLEMENTED 请复制模板后将本文件改名为具体平台模块并实现后台自动化逻辑"

38
scripts/service/sibling_bridge.py
View File

@@ -1,4 +1,8 @@
"""兄弟技能 CLI 调用模板。"""
"""兄弟技能 CLI 调用模板。
通用工具：通过子进程调用同级其他 skill 的 main.py，并解析 JSON 输出。
具体调用哪些兄弟技能，由具体业务 skill 决定（在 task_service.py 中按需 import 调用）。
"""
from __future__ import annotations
@@ -12,7 +16,19 @@ from util.logging_config import subprocess_env_with_trace
from util.runtime_paths import get_skills_root
def _call_json_script(script_path: str, args: List[str]) -> Optional[Dict[str, Any]]:
def call_sibling_json(skill_slug: str, args: List[str]) -> Optional[Dict[str, Any]]:
"""通用兄弟技能调用器。
Args:
skill_slug: 兄弟技能的 slug（即同级目录名）。
args: 传给 main.py 的命令行参数列表。
Returns:
若兄弟技能输出可解析的 JSON 对象则返回 dict；否则返回 None。
"""
script_path = os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
if not os.path.isfile(script_path):
return None
proc = subprocess.run(
[sys.executable, script_path, *args],
capture_output=True,
@@ -31,9 +47,19 @@ def _call_json_script(script_path: str, args: List[str]) -> Optional[Dict[str, A
return data if isinstance(data, dict) else None
def get_account_manager_main_path() -> str:
return os.path.join(get_skills_root(), "account-manager", "scripts", "main.py")
def get_sibling_main_path(skill_slug: str) -> str:
"""获取兄弟技能的 main.py 绝对路径。
使用示例（在你的 task_service.py 中）：
from service.sibling_bridge import get_sibling_main_path, call_sibling_json
# 调用名为 account-manager 的兄弟技能：
result = call_sibling_json("account-manager", ["list", "--limit", "10"])
"""
return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
def get_content_manager_main_path() -> str:
return os.path.join(get_skills_root(), "content-manager", "scripts", "main.py")
# ============================================================
# 复制本模板后，按需在你自己的 task_service.py 中调用上面的工具。
# 不要在本文件中硬编码具体兄弟技能函数（如 get_account_manager_main_path），
# 那是发布类技能的特定需求，不属于通用模板。
# ============================================================

50
scripts/service/publish_service.py → scripts/service/task_service.py
View File

@@ -1,4 +1,8 @@
"""发布编排、日志查询模板。"""
"""任务编排、日志查询模板。
通用业务编排层：负责参数校验、鉴权、调用具体业务实现、写入任务日志。
复制后在 cmd_run 中实现真正的业务逻辑。
"""
from __future__ import annotations
@@ -6,39 +10,47 @@ import json
import sys
from typing import Optional
from db import publish_logs_repository as plr
from db import task_logs_repository as tlr
from service.entitlement_service import check_entitlement
from util.constants import SKILL_SLUG, SKILL_VERSION
from util.timeutil import unix_to_iso
def cmd_publish(account_id: Optional[str] = None, article_id: Optional[str] = None) -> int:
_ = (account_id, article_id)
def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int:
"""通用任务执行入口模板。复制后请实现真实业务逻辑。"""
_ = (target, input_id)
ok, reason = check_entitlement(SKILL_SLUG)
if not ok:
print(f"❌ {reason}")
return 1
print("❌ 这是模板仓库，请复制后在 scripts/service/ 中实现真正的发布逻辑。")
print("❌ 这是模板仓库，请复制后在 scripts/service/task_service.py 中实现 cmd_run 的真实业务逻辑。")
return 1
def cmd_logs(limit: int = 10, status: Optional[str] = None, account_id: Optional[str] = None) -> int:
def cmd_logs(
limit: int = 10,
status: Optional[str] = None,
task_type: Optional[str] = None,
target_id: Optional[str] = None,
) -> int:
if limit <= 0:
limit = 10
rows = plr.list_publish_logs(limit, status, account_id)
rows = tlr.list_task_logs(limit, status, task_type, target_id)
if not rows:
print("暂无发布记录")
print("暂无任务日志")
return 0
sep_line = "_" * 39
for idx, r in enumerate(rows):
rid, aid, arid, title, st, err, cat, uat = r
rid, ttype, tid, iid, ititle, st, err, rsum, cat, uat = r
print(f"id：{rid}")
print(f"account_id：{aid or ''}")
print(f"article_id：{arid}")
print(f"article_title：{title or ''}")
print(f"task_type：{ttype or ''}")
print(f"target_id：{tid or ''}")
print(f"input_id：{iid or ''}")
print(f"input_title：{ititle or ''}")
print(f"status：{st or ''}")
print(f"error_msg：{err or ''}")
print(f"result_summary：{rsum or ''}")
print(f"created_at：{unix_to_iso(cat) or str(cat or '')}")
print(f"updated_at：{unix_to_iso(uat) or str(uat or '')}")
if idx != len(rows) - 1:
@@ -51,20 +63,22 @@ def cmd_log_get(log_id: str) -> int:
if not str(log_id).isdigit():
print("❌ log_id 必须是数字")
return 1
row = plr.get_publish_log_by_id(int(log_id))
row = tlr.get_task_log_by_id(int(log_id))
if not row:
print("❌ 没有这条发布记录")
print("❌ 没有这条任务日志")
return 1
rid, aid, arid, title, st, err, cat, uat = row
rid, ttype, tid, iid, ititle, st, err, rsum, cat, uat = row
print(
json.dumps(
{
"id": int(rid),
"account_id": aid,
"article_id": int(arid),
"article_title": title,
"task_type": ttype,
"target_id": tid,
"input_id": iid,
"input_title": ititle,
"status": st,
"error_msg": err,
"result_summary": rsum,
"created_at": unix_to_iso(cat),
"updated_at": unix_to_iso(uat),
},

10
tests/test_cli_smoke.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""通用 CLI 冒烟：用法、health、version、logs、log-get（不触网、不深测 publish 占位）。"""
"""通用 CLI 冒烟：用法、health、version、logs、log-get（不触网、不深测 run 占位）。"""
from __future__ import annotations
import io
@@ -23,7 +23,7 @@ class TestCliSmoke(unittest.TestCase):
rc = main([])
self.assertEqual(rc, 1)
out = buf.getvalue()
self.assertIn("模板技能", out)
self.assertIn("通用业务技能模板", out)
self.assertIn("health", out)
def test_health_zero(self) -> None:
@@ -57,13 +57,13 @@ class TestCliSmoke(unittest.TestCase):
self.assertNotEqual(rc, 0)
self.assertIn("数字", buf.getvalue())
def test_publish_placeholder_returns_nonzero_without_network(self) -> None:
"""占位 publish：不验证业务成功，仅确认模板提示且退出码非 0（无 AUTH_BASE 时不发 HTTP）。"""
def test_run_placeholder_returns_nonzero_without_network(self) -> None:
"""占位 run：不验证业务成功，仅确认模板提示且退出码非 0（无 AUTH_BASE 时不发 HTTP）。"""
old_auth = os.environ.pop("JIANGCHANG_AUTH_BASE_URL", None)
try:
buf = io.StringIO()
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
rc = main(["publish"])
rc = main(["run"])
self.assertNotEqual(rc, 0)
self.assertIn("模板", buf.getvalue())
finally:

24
tests/test_db_smoke.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""publish_logs 表：init 幂等、写入与查询（全程隔离数据目录）。"""
"""task_logs 表：init 幂等、写入与查询（全程隔离数据目录）。"""
from __future__ import annotations
import unittest
@@ -11,29 +11,31 @@ class TestDbSmoke(unittest.TestCase):
def test_init_db_idempotent_and_crud_smoke(self) -> None:
with IsolatedDataRoot():
from db.connection import init_db
from db import publish_logs_repository as plr
from db import task_logs_repository as tlr
init_db()
init_db()
init_db() # 幂等
new_id = plr.save_publish_log(
account_id="acc-1",
article_id=42,
article_title="t",
status="ok",
new_id = tlr.save_task_log(
task_type="demo_task",
target_id="tgt-1",
input_id="42",
input_title="t",
status="success",
error_msg=None,
result_summary=None,
)
self.assertIsInstance(new_id, int)
self.assertGreater(new_id, 0)
rows = plr.list_publish_logs(limit=10)
rows = tlr.list_task_logs(limit=10)
self.assertTrue(any(r[0] == new_id for r in rows))
row = plr.get_publish_log_by_id(new_id)
row = tlr.get_task_log_by_id(new_id)
self.assertIsNotNone(row)
self.assertEqual(int(row[0]), new_id)
missing = plr.get_publish_log_by_id(999_999)
missing = tlr.get_task_log_by_id(999_999)
self.assertIsNone(missing)