feat(skill): 新增 zentao-bug skill,读取禅道 18.x Bug 单并按项目规范修复

禅道 18.8 无 RESTful API 2.0,官方 zentao-cli/MCP 不可用。zentao.py 以 v1 tokens 换会话,
走页面 JSON 模式读取「我的 Bug」列表与 Bug 详情(描述/截图/评论/修改历史/附件),
截图下载到本地供 Agent 查看;Token 存 ~/.config(0600),密码仅存 macOS 钥匙串。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
YANG JIANKUAN
2026-10-14 20:20:40 -07:00
parent f41d4186ea
commit 7f1a413172
2 changed files with 585 additions and 0 deletions

View File

@@ -0,0 +1,528 @@
#!/usr/bin/env python3
"""禅道 18.x Bug 读取工具(仅依赖 Python 标准库)。
禅道 18.8 没有 RESTful API 2.0,官方 zentao-cli 无法使用;本脚本组合两条 18.x 可用的通道:
- 认证:POST /api.php/v1/tokens 换取 Token(即会话 ID)
- 数据:页面 JSON 模式(把页面地址 .html 换成 .json),Cookie 带 zentaosid=<Token>
- 我的 Bug:/my-work-bug-<type>-0-id_desc-0-<perPage>-<page>.json
- Bug 详情:/bug-view-<id>.json(含描述、修改记录、评论)
- 图片:/file-read-<fileID>.<ext>
凭据:Token 存 ~/.config/zentao-lite/config.json(0600);密码只存 macOS 钥匙串,
用于会话过期(禅道默认 24 分钟空闲失效)后自动重新登录,任何文件里都不落明文密码。
"""
import argparse
import base64
import getpass
import html
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from html.parser import HTMLParser
from pathlib import Path
CONFIG_DIR = Path(os.environ.get("ZENTAO_LITE_HOME", Path.home() / ".config" / "zentao-lite"))
CONFIG_FILE = CONFIG_DIR / "config.json"
DEFAULT_OUT = Path.home() / ".cache" / "zentao-lite" / "bugs"
KEYCHAIN_SERVICE = "zentao-lite"
DEFAULT_URL = "https://njlkairport.chandao.net"
STATUS = {"active": "激活", "resolved": "已解决", "closed": "已关闭"}
TYPES = {
"codeerror": "代码错误", "config": "配置相关", "install": "安装部署", "security": "安全相关",
"performance": "性能问题", "standard": "标准规范", "automation": "测试脚本",
"designdefect": "设计缺陷", "others": "其他",
}
RESOLUTIONS = {
"fixed": "已解决", "bydesign": "设计如此", "duplicate": "重复Bug", "external": "外部原因",
"notrepro": "无法重现", "postponed": "延期处理", "willnotfix": "不予解决", "tostory": "转为需求",
}
ACTIONS = {
"opened": "创建", "edited": "编辑", "commented": "备注", "resolved": "解决", "activated": "激活",
"closed": "关闭", "assigned": "指派", "confirmed": "确认", "bugconfirmed": "确认",
"tostory": "转需求", "totask": "转任务", "linked2build": "关联版本", "linked2release": "关联发布",
}
FIELDS = {
"steps": "重现步骤", "title": "标题", "assignedTo": "指派给", "status": "状态", "severity": "严重程度",
"pri": "优先级", "type": "类型", "resolution": "解决方案", "resolvedBuild": "解决版本",
"module": "模块", "deadline": "截止日期", "keywords": "关键词", "openedBuild": "影响版本",
"resolvedBy": "解决者", "resolvedDate": "解决日期", "closedBy": "关闭者", "closedDate": "关闭日期",
"confirmed": "是否确认", "mailto": "抄送给", "linkBug": "相关Bug", "activatedCount": "激活次数",
}
USER_FIELDS = {"assignedTo", "resolvedBy", "closedBy", "openedBy"}
BUG_TYPES = ("assignedTo", "openedBy", "resolvedBy", "closedBy")
class AuthError(Exception):
pass
def die(msg, code=1):
print(f"错误:{msg}", file=sys.stderr)
sys.exit(code)
# ---------------------------------------------------------------- 配置与凭据
def load_config():
if not CONFIG_FILE.exists():
die("尚未登录,请先执行:python3 .claude/skills/zentao-bug/zentao.py login", 2)
return json.loads(CONFIG_FILE.read_text())
def save_config(cfg):
CONFIG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
tmp = CONFIG_FILE.with_suffix(".tmp")
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as f:
json.dump(cfg, f, ensure_ascii=False, indent=2)
os.replace(tmp, CONFIG_FILE)
def keychain_get(account):
if sys.platform != "darwin":
return None
r = subprocess.run(
["security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-a", account, "-w"],
capture_output=True, text=True,
)
return r.stdout.rstrip("\n") if r.returncode == 0 else None
def keychain_set(account, password):
if sys.platform != "darwin":
return False
r = subprocess.run(
["security", "add-generic-password", "-U", "-s", KEYCHAIN_SERVICE, "-a", account, "-w", password],
capture_output=True, text=True,
)
return r.returncode == 0
def keychain_delete(account):
if sys.platform == "darwin":
subprocess.run(["security", "delete-generic-password", "-s", KEYCHAIN_SERVICE, "-a", account],
capture_output=True)
def request_token(url, account, password):
body = json.dumps({"account": account, "password": password}).encode()
req = urllib.request.Request(url + "/api.php/v1/tokens", data=body,
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.load(resp)
except urllib.error.HTTPError as e:
try:
data = json.load(e)
except Exception:
data = {"error": f"HTTP {e.code}"}
if not data.get("token"):
raise AuthError(data.get("error") or "登录失败")
return data["token"]
# ---------------------------------------------------------------- 请求
class Client:
def __init__(self, cfg):
self.cfg = cfg
self.url = cfg["url"].rstrip("/")
self.relogged = False
def _relogin(self):
if self.relogged:
raise AuthError("重新登录后仍无权限")
password = keychain_get(self.cfg["account"])
if not password:
raise AuthError("会话已过期,且钥匙串中没有保存密码,请重新执行 login")
self.cfg["token"] = request_token(self.url, self.cfg["account"], password)
save_config(self.cfg)
self.relogged = True
def _open(self, path):
req = urllib.request.Request(self.url + path, headers={
"Cookie": f"zentaosid={self.cfg['token']}", "Token": self.cfg["token"],
})
try:
return urllib.request.urlopen(req, timeout=60)
except urllib.error.HTTPError as e:
if e.code == 401:
raise AuthError("Token 无效")
raise
def page(self, path):
"""读取页面 JSON;会话失效时自动用钥匙串密码重登一次。"""
for _ in range(2):
try:
with self._open(path) as resp:
outer = json.load(resp)
data = outer.get("data")
if isinstance(data, str):
data = json.loads(data)
if outer.get("status") != "success" or (isinstance(data, dict) and "user-login" in str(data.get("locate", ""))):
raise AuthError("会话已过期")
return data
except AuthError:
self._relogin()
raise AuthError("无法获取数据")
def download(self, path, dest):
for _ in range(2):
try:
with self._open(path) as resp:
ctype = resp.headers.get("Content-Type", "")
content = resp.read()
if "text/html" in ctype or "application/json" in ctype:
if b"user-login" in content:
raise AuthError("会话已过期")
raise RuntimeError(f"返回的不是文件内容({ctype})")
dest.write_bytes(content)
return
except AuthError:
self._relogin()
# ---------------------------------------------------------------- HTML → Markdown
class Html2Md(HTMLParser):
BLOCK = {"p", "div", "h1", "h2", "h3", "h4", "h5", "h6", "tr", "table", "blockquote", "pre"}
def __init__(self, on_image):
super().__init__(convert_charrefs=True)
self.out = []
self.lists = []
self.on_image = on_image
self.href = None
self.skip = 0
def _nl(self, n=1):
text = "".join(self.out)
have = len(text) - len(text.rstrip("\n"))
if text.strip():
self.out.append("\n" * max(0, n - have))
def handle_starttag(self, tag, attrs):
a = dict(attrs)
if tag in ("style", "script"):
self.skip += 1
elif tag in self.BLOCK:
self._nl(2 if tag != "tr" else 1)
if tag[0] == "h" and tag[1:].isdigit():
self.out.append("**") # 标题降级为加粗,避免与外层 Markdown 标题层级混淆
elif tag == "br":
self.out.append("\n")
elif tag in ("ul", "ol"):
self._nl(1)
self.lists.append([tag, 0])
elif tag == "li":
self._nl(1)
indent = " " * max(0, len(self.lists) - 1)
if self.lists and self.lists[-1][0] == "ol":
self.lists[-1][1] += 1
self.out.append(f"{indent}{self.lists[-1][1]}. ")
else:
self.out.append(f"{indent}- ")
elif tag in ("strong", "b"):
self.out.append("**")
elif tag == "td" or tag == "th":
self.out.append("| ")
elif tag == "a":
self.href = a.get("href")
self.out.append("[")
elif tag == "img":
src = a.get("src") or a.get("data-src") or ""
if src:
self.out.append(self.on_image(src))
def handle_endtag(self, tag):
if tag in ("style", "script"):
self.skip = max(0, self.skip - 1)
elif tag in self.BLOCK:
if tag[0] == "h" and tag[1:].isdigit():
self.out.append("**")
self._nl(2 if tag != "tr" else 1)
elif tag in ("ul", "ol"):
if self.lists:
self.lists.pop()
self._nl(2)
elif tag in ("strong", "b"):
self.out.append("**")
elif tag in ("td", "th"):
self.out.append(" ")
elif tag == "a":
self.out.append(f"]({self.href})" if self.href else "]")
self.href = None
def handle_data(self, data):
if not self.skip:
self.out.append(re.sub(r"[ \t\r\n]+", " ", data) if not data.isspace() else (" " if self.out and not self.out[-1].endswith(("\n", " ")) else ""))
def markdown(self):
text = "".join(self.out).replace("****", "")
text = "\n".join(line.rstrip() for line in text.splitlines())
return re.sub(r"\n{3,}", "\n\n", text).strip()
class ImageFetcher:
"""把富文本中的图片下载到本地,返回 Markdown 图片引用(绝对路径,可直接用 Read 查看)。"""
def __init__(self, client, img_dir):
self.client = client
self.dir = img_dir
self.files = []
self.errors = []
def __call__(self, src):
src = html.unescape(src.strip())
m = re.fullmatch(r"\{(\d+)\.(\w+)\}", src) # 数据库原始占位符 {1154.jpeg}
if m:
src = f"/file-read-{m.group(1)}.{m.group(2)}"
try:
dest = self._fetch(src)
except Exception as e: # 单张图片失败不影响整体
self.errors.append(f"{src}:{e}")
return f"![图片下载失败]({src})"
if dest not in self.files:
self.files.append(dest)
return f"![图{self.files.index(dest) + 1}]({dest})"
def _fetch(self, src):
self.dir.mkdir(parents=True, exist_ok=True)
if src.startswith("data:"):
head, b64 = src.split(",", 1)
ext = re.search(r"image/(\w+)", head)
dest = self.dir / f"inline-{len(self.files) + 1}.{ext.group(1) if ext else 'png'}"
dest.write_bytes(base64.b64decode(b64))
return dest
parsed = urllib.parse.urlparse(src)
if parsed.netloc and parsed.netloc not in self.client.url:
dest = self.dir / Path(parsed.path).name
if not dest.exists():
urllib.request.urlretrieve(src, dest)
return dest
path = parsed.path + (f"?{parsed.query}" if parsed.query else "")
m = re.search(r"file-read-(\d+)(?:\.(\w+))?", path)
name = f"{m.group(1)}.{m.group(2) or 'bin'}" if m else Path(parsed.path).name
dest = self.dir / name
if not dest.exists() or dest.stat().st_size == 0:
self.client.download(path, dest)
return dest
# ---------------------------------------------------------------- 命令
def cmd_login(args):
url = (args.url or os.environ.get("ZENTAO_URL") or DEFAULT_URL).rstrip("/")
account = args.account or os.environ.get("ZENTAO_ACCOUNT") or input("禅道账号:").strip()
password = os.environ.get("ZENTAO_PASSWORD") or getpass.getpass("禅道密码(输入不回显):")
try:
token = request_token(url, account, password)
except AuthError as e:
die(f"登录失败:{e}")
cfg = {"url": url, "account": account, "token": token}
save_config(cfg)
saved = False if args.no_keychain else keychain_set(account, password)
print(f"登录成功:{account} @ {url}")
print(f"Token 已保存到 {CONFIG_FILE}(权限 0600)")
print("密码已存入 macOS 钥匙串,会话过期时自动重新登录" if saved
else "未保存密码:会话过期(默认 24 分钟空闲)后需重新 login")
def cmd_logout(args):
if CONFIG_FILE.exists():
keychain_delete(json.loads(CONFIG_FILE.read_text()).get("account", ""))
CONFIG_FILE.unlink()
print("已退出并清除本地 Token 与钥匙串密码")
def cmd_whoami(args):
client = Client(load_config())
data = client.page("/my-work-bug-assignedTo-0-id_desc-0-1-1.json") # 顺带验证会话可用
users = data.get("users", {})
account = client.cfg["account"]
print(f"{account}({users.get(account, '')}) @ {client.url},会话有效")
def fetch_bug_list(client, bug_type, per_page=100):
bugs, page, users = [], 1, {}
while True:
data = client.page(f"/my-work-bug-{bug_type}-0-id_desc-0-{per_page}-{page}.json")
users.update(data.get("users") or {})
bugs.extend(data.get("bugs") or [])
pager = data.get("pager") or {}
if page >= int(pager.get("pageTotal") or 1):
return bugs, users
page += 1
def cmd_list(args):
client = Client(load_config())
bugs, users = fetch_bug_list(client, args.type)
if args.status != "all":
bugs = [b for b in bugs if b.get("status") == args.status]
if args.json:
print(json.dumps([{
"id": int(b["id"]), "title": html.unescape(b["title"]), "status": b["status"],
"severity": int(b["severity"]), "pri": int(b["pri"]), "type": b["type"],
"product": b.get("productName"), "openedBy": b["openedBy"], "openedDate": b["openedDate"],
"assignedTo": b["assignedTo"], "deadline": b.get("deadline"),
} for b in bugs], ensure_ascii=False, indent=2))
return
title = {"assignedTo": "指派给我", "openedBy": "由我创建", "resolvedBy": "由我解决", "closedBy": "由我关闭"}[args.type]
print(f"# 我的 Bug({title},状态:{args.status})共 {len(bugs)} 条\n")
print("| ID | 严重 | 优先级 | 状态 | 类型 | 标题 | 创建人 | 创建时间 |")
print("|---|---|---|---|---|---|---|---|")
for b in bugs:
t = html.unescape(b["title"]).replace("|", "\\|")
print(f"| {b['id']} | {b['severity']} | {b['pri']} | {STATUS.get(b['status'], b['status'])} | "
f"{TYPES.get(b['type'], b['type'])} | {t} | {users.get(b['openedBy'], b['openedBy'])} | {b['openedDate'][:16]} |")
def cmd_view(args):
client = Client(load_config())
data = client.page(f"/bug-view-{args.id}.json")
bug = data.get("bug")
if not bug:
die(f"Bug #{args.id} 不存在或无权查看")
users = data.get("users") or {}
who = lambda a: f"{users.get(a, a)}({a})" if a and users.get(a) and users.get(a) != a else (a or "-")
def show(field, value):
if not value:
return "空"
if field in USER_FIELDS:
return who(value)
if field == "mailto":
return "、".join(who(v) for v in value.split(",") if v) or "空"
if field == "confirmed":
return "是" if value == "1" else "否"
return {"status": STATUS, "resolution": RESOLUTIONS, "type": TYPES}.get(field, {}).get(value, value)
out_dir = Path(args.out or DEFAULT_OUT) / str(args.id)
fetcher = ImageFetcher(client, out_dir / "images")
def md(html_text):
p = Html2Md(fetcher)
p.feed(html_text or "")
p.close()
return p.markdown()
module = " / ".join(m.get("name", "") for m in (data.get("modulePath") or [])) or "/"
product = (data.get("product") or {}).get("name") or (data.get("products") or {}).get(bug.get("product"), "")
lines = [
f"# Bug #{bug['id']} {html.unescape(bug['title'])}", "",
f"- 链接:{client.url}/bug-view-{bug['id']}.html",
f"- 产品 / 模块:{product} / {module}",
f"- 状态:{STATUS.get(bug['status'], bug['status'])}"
+ (f"({RESOLUTIONS.get(bug['resolution'], bug['resolution'])})" if bug.get("resolution") else "")
+ f" 确认:{'是' if bug.get('confirmed') == '1' else '否'}",
f"- 严重程度:{bug['severity']} 优先级:{bug['pri']} 类型:{TYPES.get(bug['type'], bug['type'])}",
f"- 创建:{who(bug['openedBy'])} {bug['openedDate']} 影响版本:{bug.get('openedBuild', '')}",
f"- 指派给:{who(bug['assignedTo'])} {bug.get('assignedDate', '')}",
]
if bug.get("resolvedBy"):
lines.append(f"- 解决:{who(bug['resolvedBy'])} {bug.get('resolvedDate', '')} 解决版本:{bug.get('resolvedBuild', '')}")
if bug.get("deadline") and not bug["deadline"].startswith("0000"):
lines.append(f"- 截止日期:{bug['deadline']}")
if bug.get("keywords"):
lines.append(f"- 关键词:{bug['keywords']}")
lines += ["", "## 重现步骤", "", md(bug.get("steps")) or "(空)"]
files = list((bug.get("files") or {}).values()) if isinstance(bug.get("files"), dict) else (bug.get("files") or [])
if files:
lines += ["", "## 附件", ""]
att_dir = out_dir / "attachments"
att_dir.mkdir(parents=True, exist_ok=True)
for f in files:
name = f"{f['id']}-{f.get('title', 'file')}"
if not name.lower().endswith("." + f.get("extension", "").lower()):
name += "." + f.get("extension", "bin")
dest = att_dir / re.sub(r"[/\\]", "_", name)
try:
if not dest.exists():
client.download(f"/file-read-{f['id']}.{f.get('extension', '')}", dest)
lines.append(f"- [{f.get('title')}]({dest})({f.get('size', '?')} 字节)")
except Exception as e:
lines.append(f"- {f.get('title')}:下载失败({e})")
actions = sorted((data.get("actions") or {}).values(), key=lambda a: (a.get("date", ""), int(a.get("id", 0))))
comment_count = sum(1 for a in actions if (a.get("comment") or "").strip())
lines += ["", f"## 历史记录与评论({len(actions)} 条,含评论 {comment_count} 条)", ""]
for a in actions:
name = ACTIONS.get(a.get("action", ""), a.get("action", ""))
extra = a.get("extra") or ""
if a.get("action") == "resolved" and extra:
extra = RESOLUTIONS.get(extra.split(",")[0], extra)
elif a.get("action") == "assigned" and extra:
extra = f"→ {who(extra)}"
lines.append(f"### {a.get('date', '')} {who(a.get('actor'))} {name}{(' ' + extra) if extra else ''}")
for h in a.get("history") or []:
field = FIELDS.get(h.get("field"), h.get("field"))
if h.get("field") == "steps":
lines.append(f"- 修改了 {field}(以上方当前内容为准)")
else:
lines.append(f"- 修改了 {field}:{show(h.get('field'), h.get('old'))} → {show(h.get('field'), h.get('new'))}")
if (a.get("comment") or "").strip():
lines += ["", "**评论:**", ""] + [f"> {l}" if l else ">" for l in md(a["comment"]).splitlines()]
lines.append("")
if fetcher.files:
lines += ["## 本地图片", ""] + [f"- 图{i + 1}:{p}" for i, p in enumerate(fetcher.files)] + [""]
if fetcher.errors:
lines += ["## 图片下载失败", ""] + [f"- {e}" for e in fetcher.errors] + [""]
text = "\n".join(lines).rstrip() + "\n"
out_dir.mkdir(parents=True, exist_ok=True)
(out_dir / "bug.md").write_text(text)
if args.json:
bug["_images"] = [str(p) for p in fetcher.files]
bug["_actions"] = actions
print(json.dumps(bug, ensure_ascii=False, indent=2))
else:
print(text)
print(f"(已保存到 {out_dir / 'bug.md'},图片 {len(fetcher.files)} 张)")
def main():
ap = argparse.ArgumentParser(description="禅道 18.x Bug 读取工具")
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("login", help="登录并保存 Token(密码存 macOS 钥匙串)")
p.add_argument("--url", help=f"禅道地址,默认 {DEFAULT_URL}")
p.add_argument("--account", help="账号")
p.add_argument("--no-keychain", action="store_true", help="不在钥匙串保存密码")
p.set_defaults(func=cmd_login)
sub.add_parser("logout", help="清除 Token 与钥匙串密码").set_defaults(func=cmd_logout)
sub.add_parser("whoami", help="查看当前账号并验证会话").set_defaults(func=cmd_whoami)
p = sub.add_parser("list", help="我的 Bug 列表")
p.add_argument("--type", choices=BUG_TYPES, default="assignedTo", help="默认 assignedTo(指派给我)")
p.add_argument("--status", choices=("active", "resolved", "closed", "all"), default="all")
p.add_argument("--json", action="store_true")
p.set_defaults(func=cmd_list)
p = sub.add_parser("view", help="Bug 详情(描述、图片、历史、评论、附件)")
p.add_argument("id", type=int)
p.add_argument("--out", help=f"输出目录,默认 {DEFAULT_OUT}")
p.add_argument("--json", action="store_true")
p.set_defaults(func=cmd_view)
args = ap.parse_args()
try:
args.func(args)
except AuthError as e:
die(f"{e}。请执行:python3 .claude/skills/zentao-bug/zentao.py login", 2)
except urllib.error.URLError as e:
die(f"无法访问禅道:{e}")
if __name__ == "__main__":
main()