168 lines
5.4 KiB
Python
Executable File
168 lines
5.4 KiB
Python
Executable File
#!/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())
|