refactor: fold ack kit into skill
This commit is contained in:
Executable
+167
@@ -0,0 +1,167 @@
|
||||
#!/usr/bin/env python3
|
||||
"""校验 tasks.yaml 是否符合 ack 任务板结构。
|
||||
|
||||
权威结构是同目录上层的 templates/tasks.schema.json(跨语言可用)。
|
||||
本脚本是参考实现:
|
||||
- 若安装了 jsonschema,则用 schema 做完整校验;
|
||||
- 否则回退到内置的关键规则校验(必填字段、状态枚举、三轮上限、leftover 留档)。
|
||||
YAML 解析优先用 pyyaml;未安装时给出提示而非崩溃。
|
||||
|
||||
用法:
|
||||
python3 validate_tasks.py [tasks.yaml]
|
||||
python3 validate_tasks.py --schema path/to/tasks.schema.json tasks.yaml
|
||||
|
||||
退出码: 0 通过 / 1 校验失败 / 2 环境或用法错误。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
STATUS_ENUM = {
|
||||
"open",
|
||||
"dispatched",
|
||||
"fixed_by_dev",
|
||||
"retesting",
|
||||
"failed_retest",
|
||||
"verified",
|
||||
"blocked",
|
||||
"leftover",
|
||||
}
|
||||
MAX_ROUNDS = 3
|
||||
|
||||
|
||||
def load_yaml(path: Path) -> dict:
|
||||
try:
|
||||
import yaml # type: ignore
|
||||
except ImportError:
|
||||
sys.stderr.write(
|
||||
"需要 PyYAML 才能解析 YAML:pip install pyyaml\n"
|
||||
"(或把任务板导出为 JSON 后再校验)\n"
|
||||
)
|
||||
raise SystemExit(2)
|
||||
try:
|
||||
with path.open(encoding="utf-8") as fh:
|
||||
data = yaml.safe_load(fh)
|
||||
except yaml.YAMLError as exc: # type: ignore
|
||||
sys.stderr.write(f"YAML 解析失败: {exc}\n")
|
||||
raise SystemExit(1)
|
||||
if not isinstance(data, dict):
|
||||
sys.stderr.write("任务板顶层必须是对象(mapping)\n")
|
||||
raise SystemExit(1)
|
||||
return data
|
||||
|
||||
|
||||
def validate_with_schema(data: dict, schema_path: Path) -> list[str]:
|
||||
import jsonschema # type: ignore
|
||||
|
||||
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
||||
validator = jsonschema.Draft7Validator(schema)
|
||||
errors = []
|
||||
for err in sorted(validator.iter_errors(data), key=lambda e: list(e.path)):
|
||||
loc = "/".join(str(p) for p in err.path) or "<root>"
|
||||
errors.append(f"[schema] {loc}: {err.message}")
|
||||
return errors
|
||||
|
||||
|
||||
def validate_builtin(data: dict) -> list[str]:
|
||||
errors: list[str] = []
|
||||
|
||||
if not isinstance(data.get("version"), int) or data.get("version", 0) < 1:
|
||||
errors.append("version 必须是 >=1 的整数")
|
||||
project = data.get("project")
|
||||
if not isinstance(project, dict) or not project.get("name"):
|
||||
errors.append("project.name 必填")
|
||||
|
||||
tasks = data.get("tasks")
|
||||
if not isinstance(tasks, list):
|
||||
errors.append("tasks 必须是列表")
|
||||
return errors
|
||||
|
||||
seen_ids: set[str] = set()
|
||||
for i, task in enumerate(tasks):
|
||||
where = f"tasks[{i}]"
|
||||
if not isinstance(task, dict):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
tid = task.get("id")
|
||||
title = task.get("title")
|
||||
status = task.get("status")
|
||||
if not tid:
|
||||
errors.append(f"{where}: id 必填")
|
||||
else:
|
||||
where = f"tasks[{i}] {tid}"
|
||||
if tid in seen_ids:
|
||||
errors.append(f"{where}: id 重复")
|
||||
seen_ids.add(tid)
|
||||
if not title:
|
||||
errors.append(f"{where}: title 必填")
|
||||
if status not in STATUS_ENUM:
|
||||
errors.append(
|
||||
f"{where}: status={status!r} 非法,应为 {sorted(STATUS_ENUM)}"
|
||||
)
|
||||
|
||||
dispatch = task.get("dispatch") or {}
|
||||
rounds = dispatch.get("rounds") or []
|
||||
if isinstance(rounds, list):
|
||||
if len(rounds) > MAX_ROUNDS:
|
||||
errors.append(
|
||||
f"{where}: 派发轮次 {len(rounds)} 超过上限 {MAX_ROUNDS}"
|
||||
)
|
||||
for r in rounds:
|
||||
if isinstance(r, dict) and r.get("result") not in {"passed", "failed"}:
|
||||
errors.append(f"{where}: round.result 必须是 passed/failed")
|
||||
|
||||
if status == "leftover":
|
||||
resolution = task.get("resolution") or {}
|
||||
if not resolution.get("leftoverReason"):
|
||||
errors.append(f"{where}: leftover 必须填 resolution.leftoverReason")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="校验 tasks.yaml 结构")
|
||||
parser.add_argument("tasks", nargs="?", default="tasks.yaml", help="任务板路径")
|
||||
parser.add_argument("--schema", help="tasks.schema.json 路径(默认自动探测)")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
tasks_path = Path(args.tasks)
|
||||
if not tasks_path.is_file():
|
||||
sys.stderr.write(f"找不到任务板文件: {tasks_path}\n")
|
||||
return 2
|
||||
|
||||
data = load_yaml(tasks_path)
|
||||
|
||||
schema_path = Path(args.schema) if args.schema else (
|
||||
Path(__file__).resolve().parent.parent / "templates" / "tasks.schema.json"
|
||||
)
|
||||
|
||||
mode = "内置规则"
|
||||
try:
|
||||
import jsonschema # type: ignore # noqa: F401
|
||||
|
||||
if schema_path.is_file():
|
||||
errors = validate_with_schema(data, schema_path)
|
||||
mode = f"schema ({schema_path.name})"
|
||||
else:
|
||||
errors = validate_builtin(data)
|
||||
mode = "内置规则(未找到 schema 文件)"
|
||||
except ImportError:
|
||||
errors = validate_builtin(data)
|
||||
|
||||
if errors:
|
||||
sys.stderr.write(f"任务板校验失败({mode}),共 {len(errors)} 项:\n")
|
||||
for e in errors:
|
||||
sys.stderr.write(f" - {e}\n")
|
||||
return 1
|
||||
|
||||
sys.stdout.write(f"任务板校验通过({mode}):{tasks_path}\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user