docs: standardize skill-template and add development guide
All checks were successful
技能自动化发布 / release (push) Successful in 22s

This commit is contained in:
2026-04-13 13:46:23 +08:00
parent f11c596bde
commit 298448840d
40 changed files with 1455 additions and 533 deletions

View File

@@ -0,0 +1,46 @@
"""在线鉴权(可选模板)。"""
from __future__ import annotations
import os
from typing import Tuple
import requests
def check_entitlement(skill_slug: str) -> Tuple[bool, str]:
auth_base = (os.getenv("JIANGCHANG_AUTH_BASE_URL") or "").strip().rstrip("/")
if not auth_base:
return True, ""
user_id = (os.getenv("CLAW_USER_ID") or os.getenv("JIANGCHANG_USER_ID") or "").strip()
if not user_id:
return False, "鉴权失败缺少用户身份CLAW_USER_ID / JIANGCHANG_USER_ID"
auth_api_key = (os.getenv("JIANGCHANG_AUTH_API_KEY") or "").strip()
timeout = int((os.getenv("JIANGCHANG_AUTH_TIMEOUT_SECONDS") or "5").strip())
headers = {"Content-Type": "application/json"}
if auth_api_key:
headers["Authorization"] = f"Bearer {auth_api_key}"
payload = {
"user_id": user_id,
"skill_slug": skill_slug,
"trace_id": (os.getenv("JIANGCHANG_TRACE_ID") or "").strip(),
"context": {"entry": "main.py"},
}
try:
res = requests.post(f"{auth_base}/api/entitlements/check", json=payload, headers=headers, timeout=timeout)
except requests.RequestException as exc:
return False, f"鉴权请求失败:{exc}"
if res.status_code != 200:
return False, f"鉴权服务异常HTTP {res.status_code}"
try:
body = res.json()
except ValueError:
return False, "鉴权服务异常:返回非 JSON"
code = body.get("code")
data = body.get("data") or {}
if code != 200:
return False, str(body.get("msg") or "鉴权失败")
if not data.get("allow", False):
return False, str(data.get("reason") or "未购买或已过期")
return True, ""

View File

@@ -0,0 +1,10 @@
"""后台自动化占位模块模板。"""
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 请复制模板后将本文件改名为具体平台模块并实现后台自动化逻辑"

View File

@@ -0,0 +1,83 @@
"""发布编排、日志查询模板。"""
from __future__ import annotations
import json
import sys
from typing import Optional
from db import publish_logs_repository as plr
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)
ok, reason = check_entitlement(SKILL_SLUG)
if not ok:
print(f"{reason}")
return 1
print("❌ 这是模板仓库,请复制后在 scripts/service/ 中实现真正的发布逻辑。")
return 1
def cmd_logs(limit: int = 10, status: Optional[str] = None, account_id: Optional[str] = None) -> int:
if limit <= 0:
limit = 10
rows = plr.list_publish_logs(limit, status, account_id)
if not rows:
print("暂无发布记录")
return 0
sep_line = "_" * 39
for idx, r in enumerate(rows):
rid, aid, arid, title, st, err, 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"status{st or ''}")
print(f"error_msg{err 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:
print(sep_line)
print()
return 0
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))
if not row:
print("❌ 没有这条发布记录")
return 1
rid, aid, arid, title, st, err, cat, uat = row
print(
json.dumps(
{
"id": int(rid),
"account_id": aid,
"article_id": int(arid),
"article_title": title,
"status": st,
"error_msg": err,
"created_at": unix_to_iso(cat),
"updated_at": unix_to_iso(uat),
},
ensure_ascii=False,
)
)
return 0
def cmd_health() -> int:
return 0 if sys.version_info >= (3, 10) else 1
def cmd_version() -> int:
print(json.dumps({"version": SKILL_VERSION, "skill": SKILL_SLUG}, ensure_ascii=False))
return 0

View File

@@ -0,0 +1,39 @@
"""兄弟技能 CLI 调用模板。"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from typing import Any, Dict, List, Optional
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]]:
proc = subprocess.run(
[sys.executable, script_path, *args],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
env=subprocess_env_with_trace(),
)
raw = (proc.stdout or "").strip()
if not raw or raw.startswith("ERROR"):
return None
try:
data = json.loads(raw)
except json.JSONDecodeError:
return None
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_content_manager_main_path() -> str:
return os.path.join(get_skills_root(), "content-manager", "scripts", "main.py")