00f2dc98dd
等同网页「指派」表单：--to 接受账号或姓名，状态保持激活，提交后回读校验指派人； 多行备注用 --comment-file，换行转 <br /> 以便网页分行显示。 SKILL.md 补充用法，约定转给后端时备注照搬《后端接口对接问题清单》对应行。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
679 lines
30 KiB
Python
679 lines
30 KiB
Python
#!/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 post_form(self, path, fields):
|
||
"""提交页面表单（与网页点击「保存」相同）；返回响应 HTML（禅道用 <script>alert/parent 跳转回执）。"""
|
||
body = urllib.parse.urlencode(fields).encode()
|
||
for _ in range(2):
|
||
req = urllib.request.Request(self.url + path, data=body, headers={
|
||
"Cookie": f"zentaosid={self.cfg['token']}", "Token": self.cfg["token"],
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Referer": self.url + path,
|
||
})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||
text = resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as e:
|
||
if e.code != 401:
|
||
raise
|
||
text = "user-login"
|
||
if "user-login" not in text:
|
||
return text
|
||
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""
|
||
if dest not in self.files:
|
||
self.files.append(dest)
|
||
return f""
|
||
|
||
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 form_defaults(client, bug_id):
|
||
"""读取网页「解决」表单渲染出的默认值：解决日期（服务器时间，北京时间）与指派给（禅道默认选中创建人）。
|
||
接口提交不会自动补默认值，必须先取表单再原样带上；不能用本机时间，本机可能在其他时区。"""
|
||
with client._open(f"/bug-resolve-{bug_id}.html?onlybody=yes") as resp:
|
||
page = resp.read().decode("utf-8", "replace")
|
||
date = re.search(r"name='resolvedDate'[^>]*value='([^']+)'", page)
|
||
select = re.search(r"<select name='assignedTo'.*?</select>", page, re.S)
|
||
assigned = select and re.search(r"<option value='([^']*)' selected", select.group(0))
|
||
if not date or not assigned or not assigned.group(1):
|
||
die(f"#{bug_id} 未能从解决表单读取默认的解决日期 / 指派给，可用 --date / --assign-to 指定")
|
||
return date.group(1), assigned.group(1)
|
||
|
||
|
||
def cmd_resolve(args):
|
||
"""解决 Bug：等同网页「解决」表单。默认解决方案=已解决、解决版本=主干(trunk)、
|
||
解决日期与指派给取网页表单默认值（服务器时间 / 创建人）；只处理激活状态的单子，提交后回读校验。"""
|
||
client = Client(load_config())
|
||
failed = 0
|
||
for bug_id in args.ids:
|
||
form = client.page(f"/bug-resolve-{bug_id}.json")
|
||
bug = (form or {}).get("bug")
|
||
if not bug:
|
||
print(f"#{bug_id} 跳过：不存在或无权查看")
|
||
failed += 1
|
||
continue
|
||
title = html.unescape(bug["title"])
|
||
if bug["status"] != "active":
|
||
print(f"#{bug_id} 跳过：当前状态为「{STATUS.get(bug['status'], bug['status'])}」，只解决激活的 Bug（{title}）")
|
||
continue
|
||
builds = form.get("builds") or {}
|
||
if args.build not in builds:
|
||
die(f"解决版本 {args.build} 不在可选列表中：{', '.join(k for k in builds if k)}")
|
||
default_date, default_assigned = form_defaults(client, bug_id)
|
||
assigned = args.assign_to or default_assigned
|
||
fields = {
|
||
"resolution": args.resolution, "duplicateBug": "", "buildExecution": "",
|
||
"resolvedBuild": args.build, "buildName": "",
|
||
"resolvedDate": args.date or default_date,
|
||
"assignedTo": assigned, "status": "resolved", "comment": args.comment or "",
|
||
}
|
||
desc = (f"#{bug_id} {title} → {RESOLUTIONS.get(args.resolution, args.resolution)}，"
|
||
f"版本 {builds[args.build]}，指派给 {(form.get('users') or {}).get(assigned, assigned)}")
|
||
if args.dry_run:
|
||
print(f"[dry-run] {desc}")
|
||
continue
|
||
resp = client.post_form(f"/bug-resolve-{bug_id}.html?onlybody=yes", fields)
|
||
after = client.page(f"/bug-view-{bug_id}.json").get("bug") or {}
|
||
if after.get("status") == "resolved":
|
||
print(f"已解决 {desc}")
|
||
else:
|
||
alerts = re.findall(r"alert\((['\"])(.*?)\1\)", resp)
|
||
print(f"#{bug_id} 失败：提交后状态仍为 {after.get('status')}"
|
||
+ (f"，禅道提示：{'; '.join(html.unescape(a[1]) for a in alerts)}" if alerts else ""))
|
||
failed += 1
|
||
if failed:
|
||
sys.exit(1)
|
||
|
||
|
||
def resolve_user(users, who):
|
||
"""按账号或姓名定位用户：先精确匹配账号，再按「姓名」匹配（禅道显示名形如 T:陶雨佳）。"""
|
||
if who in users:
|
||
return who
|
||
hits = [k for k, v in users.items() if k and v.split(":", 1)[-1] == who]
|
||
if len(hits) == 1:
|
||
return hits[0]
|
||
die(f"找不到唯一的用户「{who}」" + (f"，候选：{', '.join(hits)}" if hits else ""))
|
||
|
||
|
||
def comment_html(text):
|
||
"""备注是富文本，纯文本换行在网页上会被合并，逐行转义后用 <br /> 连接。"""
|
||
return "<br />".join(html.escape(line) for line in text.splitlines())
|
||
|
||
|
||
def cmd_assign(args):
|
||
"""指派 Bug：等同网页「指派」表单（指派给 + 备注），状态保持激活；提交后回读校验指派人。"""
|
||
client = Client(load_config())
|
||
comment = args.comment
|
||
if args.comment_file:
|
||
comment = Path(args.comment_file).read_text(encoding="utf-8").strip()
|
||
failed = 0
|
||
for bug_id in args.ids:
|
||
form = client.page(f"/bug-assignTo-{bug_id}.json")
|
||
bug = (form or {}).get("bug")
|
||
if not bug:
|
||
print(f"#{bug_id} 跳过：不存在或无权查看")
|
||
failed += 1
|
||
continue
|
||
users = form.get("users") or {}
|
||
account = resolve_user(users, args.to)
|
||
title = html.unescape(bug["title"])
|
||
desc = f"#{bug_id} {title} → 指派给 {users.get(account, account)}"
|
||
if args.dry_run:
|
||
print(f"[dry-run] {desc}")
|
||
if comment:
|
||
print(" 备注：\n " + comment.replace("\n", "\n "))
|
||
continue
|
||
fields = {"assignedTo": account, "status": bug["status"], "mailto[]": "",
|
||
"comment": comment_html(comment) if comment else ""}
|
||
resp = client.post_form(f"/bug-assignTo-{bug_id}.html?onlybody=yes", fields)
|
||
after = client.page(f"/bug-view-{bug_id}.json").get("bug") or {}
|
||
if after.get("assignedTo") == account:
|
||
print(f"已指派 {desc}")
|
||
else:
|
||
alerts = re.findall(r"alert\((['\"])(.*?)\1\)", resp)
|
||
print(f"#{bug_id} 失败：提交后指派人仍为 {after.get('assignedTo')}"
|
||
+ (f"，禅道提示：{'; '.join(html.unescape(a[1]) for a in alerts)}" if alerts else ""))
|
||
failed += 1
|
||
if failed:
|
||
sys.exit(1)
|
||
|
||
|
||
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)
|
||
|
||
p = sub.add_parser("resolve", help="解决 Bug（写操作，会改动禅道线上状态）")
|
||
p.add_argument("ids", type=int, nargs="+")
|
||
p.add_argument("--resolution", choices=tuple(RESOLUTIONS), default="fixed", help="默认 fixed（已解决）")
|
||
p.add_argument("--build", default="trunk", help="解决版本，默认 trunk（主干）")
|
||
p.add_argument("--assign-to", help="指派给（账号），默认取表单默认值（即创建人）")
|
||
p.add_argument("--date", help="解决日期 YYYY-MM-DD HH:MM:SS，默认取服务器当前时间")
|
||
p.add_argument("--comment", default="", help="备注，默认留空")
|
||
p.add_argument("--dry-run", action="store_true", help="只打印将要做的修改，不提交")
|
||
p.set_defaults(func=cmd_resolve)
|
||
|
||
p = sub.add_parser("assign", help="指派 Bug 并附备注（写操作，会改动禅道线上状态）")
|
||
p.add_argument("ids", type=int, nargs="+")
|
||
p.add_argument("--to", required=True, help="指派给：账号（如 TYJ）或姓名（如 陶雨佳）")
|
||
p.add_argument("--comment", default="", help="备注，多行用换行分隔")
|
||
p.add_argument("--comment-file", help="从文件读取备注（多行内容推荐用这个）")
|
||
p.add_argument("--dry-run", action="store_true", help="只打印将要做的修改，不提交")
|
||
p.set_defaults(func=cmd_assign)
|
||
|
||
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()
|