baozaotumao2025
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Export one completed safety-test run as a readable Markdown report."""
|
||
|
||
import argparse
|
||
import getpass
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.request import Request, urlopen
|
||
|
||
from app.services.export_service import matches_result, render_markdown
|
||
|
||
|
||
def request_json(
|
||
base_url: str,
|
||
path: str,
|
||
*,
|
||
token: str | None = None,
|
||
payload: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
headers = {"Accept": "application/json"}
|
||
data = None
|
||
if payload is not None:
|
||
headers["Content-Type"] = "application/json"
|
||
data = json.dumps(payload).encode()
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
request = Request(base_url.rstrip("/") + path, data=data, headers=headers)
|
||
try:
|
||
with urlopen(request, timeout=30) as response:
|
||
return json.load(response)
|
||
except HTTPError as exc:
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
try:
|
||
message = json.loads(body).get("error", {}).get("message", body)
|
||
except json.JSONDecodeError:
|
||
message = body
|
||
raise RuntimeError(f"HTTP {exc.code}: {message}") from exc
|
||
except URLError as exc:
|
||
raise RuntimeError(f"无法连接服务：{exc.reason}") from exc
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="将已完成的 run 导出为 Markdown 报告")
|
||
parser.add_argument("--run-id", type=int, default=7, help="要导出的 run ID（默认：7）")
|
||
parser.add_argument("--base-url", default="http://127.0.0.1:8000", help="API 服务地址")
|
||
parser.add_argument("--username", help="登录用户名；省略时交互输入")
|
||
parser.add_argument(
|
||
"--execution-status",
|
||
choices=("completed", "error"),
|
||
help="只导出指定执行状态；默认导出全部",
|
||
)
|
||
parser.add_argument(
|
||
"--verdict",
|
||
choices=("pass", "fail", "needs_human_review", "judge_format_error", "none"),
|
||
help="只导出指定评价结果；none 表示未评价；默认导出全部",
|
||
)
|
||
parser.add_argument("--output", type=Path, help="输出文件路径")
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
if args.run_id < 1:
|
||
print("错误：--run-id 必须是正整数", file=sys.stderr)
|
||
return 2
|
||
username = args.username or input("Username: ").strip()
|
||
password = getpass.getpass("Password: ")
|
||
try:
|
||
login = request_json(
|
||
args.base_url,
|
||
"/api/v1/auth/login",
|
||
payload={"username": username, "password": password},
|
||
)
|
||
token = login["access_token"]
|
||
run = request_json(args.base_url, f"/api/v1/runs/{args.run_id}", token=token)
|
||
if not run.get("terminal"):
|
||
raise RuntimeError(f"Run {args.run_id} 尚未结束，当前状态：{run.get('status')}")
|
||
results = request_json(
|
||
args.base_url, f"/api/v1/runs/{args.run_id}/results", token=token
|
||
)
|
||
results = [
|
||
result
|
||
for result in results
|
||
if matches_result(result, args.execution_status, args.verdict)
|
||
]
|
||
output = args.output or Path("outputs") / f"run_{args.run_id}_results.md"
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
output.write_text(render_markdown(run, results), encoding="utf-8")
|
||
except (RuntimeError, KeyError, OSError) as exc:
|
||
print(f"导出失败：{exc}", file=sys.stderr)
|
||
return 1
|
||
print(f"导出完成：{output}（共 {len(results)} 个测试样例）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|