84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
"""发布编排、日志查询模板。"""
|
||
|
||
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
|