729 lines
26 KiB
Python
Executable File
729 lines
26 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""校验 tasks.yaml 是否符合 ack 任务板结构。
|
||
|
||
权威结构是同目录上层的 templates/tasks.schema.json（跨语言可用）。
|
||
本脚本是参考实现：
|
||
- 始终执行内置语义校验；
|
||
- 安装了 jsonschema 时，再叠加 schema 结构校验；
|
||
- knowledge 字段会检查引用格式、候选结构和 Test 检查结果。
|
||
YAML 优先使用 PyYAML；未安装时使用 fail-closed 的 ACK YAML 子集。
|
||
JSON 任务板只使用标准库，两种格式都拒绝重复键。
|
||
|
||
用法:
|
||
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 re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from yaml_subset import (
|
||
DuplicateKeyError,
|
||
YamlSubsetError,
|
||
load_json_unique,
|
||
load_yaml_subset,
|
||
make_unique_pyyaml_loader,
|
||
)
|
||
from worker_profiles import validate_routing_document
|
||
|
||
STATUS_ENUM = {
|
||
"open",
|
||
"dispatched",
|
||
"fixed_by_dev",
|
||
"retesting",
|
||
"failed_retest",
|
||
"verified",
|
||
"blocked",
|
||
"leftover",
|
||
}
|
||
MAX_ROUNDS = 3
|
||
KNOWLEDGE_KINDS = {"guardrail", "pitfall", "verification"}
|
||
KNOWLEDGE_CHECK_RESULTS = {"passed", "failed", "not_applicable"}
|
||
KNOWLEDGE_REF_RE = re.compile(r"^K-[A-Z0-9][A-Z0-9-]*@[1-9][0-9]*$")
|
||
TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
||
ATTEMPT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*-A[1-9][0-9]*$")
|
||
SEMVER_RE = re.compile(
|
||
r"^(0|[1-9][0-9]*)\."
|
||
r"(0|[1-9][0-9]*)\."
|
||
r"(0|[1-9][0-9]*)"
|
||
r"(?:-(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)"
|
||
r"(?:\.(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*)?"
|
||
r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$"
|
||
)
|
||
KNOWLEDGE_SCOPE_FIELDS = {
|
||
"components",
|
||
"paths",
|
||
"dependencies",
|
||
"versions",
|
||
"tags",
|
||
"symbols",
|
||
"errorSignatures",
|
||
}
|
||
KNOWLEDGE_APPLICATION_FIELDS = {"ref", "result", "evidence"}
|
||
KNOWLEDGE_CANDIDATE_FIELDS = {
|
||
"kind",
|
||
"title",
|
||
"claim",
|
||
"scope",
|
||
"appliesWhen",
|
||
"directive",
|
||
"rationale",
|
||
"evidenceRefs",
|
||
"proposedBy",
|
||
"proposedAt",
|
||
}
|
||
KNOWLEDGE_CHECK_FIELDS = {
|
||
"ref",
|
||
"result",
|
||
"evidence",
|
||
"checkedBy",
|
||
"checkedAt",
|
||
}
|
||
DISPATCH_FIELDS = {
|
||
"taskId",
|
||
"dispatchId",
|
||
"worker",
|
||
"developer",
|
||
"test",
|
||
"rounds",
|
||
}
|
||
KNOWLEDGE_CANDIDATE_REQUIRED_FIELDS = {
|
||
"kind",
|
||
"title",
|
||
"claim",
|
||
"scope",
|
||
"appliesWhen",
|
||
"directive",
|
||
"rationale",
|
||
"evidenceRefs",
|
||
}
|
||
KNOWLEDGE_CANDIDATE_TEXT_FIELDS = {
|
||
"title",
|
||
"claim",
|
||
"appliesWhen",
|
||
"directive",
|
||
"rationale",
|
||
}
|
||
|
||
|
||
def _nonempty_string(value: object) -> bool:
|
||
return isinstance(value, str) and bool(value.strip())
|
||
|
||
|
||
def load_document(path: Path) -> dict:
|
||
try:
|
||
content = path.read_text(encoding="utf-8")
|
||
except OSError as exc:
|
||
sys.stderr.write(f"任务板读取失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
|
||
if path.suffix.lower() == ".json":
|
||
try:
|
||
data = load_json_unique(content)
|
||
except (json.JSONDecodeError, DuplicateKeyError) as exc:
|
||
sys.stderr.write(f"JSON 解析失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
else:
|
||
try:
|
||
import yaml # type: ignore
|
||
except ImportError:
|
||
try:
|
||
data = load_yaml_subset(content)
|
||
except YamlSubsetError as exc:
|
||
sys.stderr.write(f"YAML 子集解析失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
else:
|
||
try:
|
||
data = yaml.load(
|
||
content,
|
||
Loader=make_unique_pyyaml_loader(yaml),
|
||
)
|
||
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_knowledge_ref_list(
|
||
value: object,
|
||
where: str,
|
||
errors: list[str],
|
||
) -> list[str]:
|
||
if not isinstance(value, list):
|
||
errors.append(f"{where}: 必须是列表")
|
||
return []
|
||
|
||
refs: list[str] = []
|
||
seen: set[str] = set()
|
||
for index, ref in enumerate(value):
|
||
item_where = f"{where}[{index}]"
|
||
if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
|
||
errors.append(f"{item_where}: 必须使用 K-<id>@<revision> 格式")
|
||
continue
|
||
if ref in seen:
|
||
errors.append(f"{item_where}: 引用重复: {ref}")
|
||
seen.add(ref)
|
||
refs.append(ref)
|
||
return refs
|
||
|
||
|
||
def reject_unknown_fields(
|
||
value: dict,
|
||
allowed: set[str],
|
||
where: str,
|
||
errors: list[str],
|
||
) -> None:
|
||
"""Mirror additionalProperties=false for builtin knowledge validation."""
|
||
for field in sorted(set(value) - allowed):
|
||
errors.append(f"{where}: 未知字段 {field!r}")
|
||
|
||
|
||
def validate_optional_string_fields(
|
||
value: dict,
|
||
fields: set[str],
|
||
where: str,
|
||
errors: list[str],
|
||
) -> None:
|
||
for field in sorted(fields):
|
||
if field in value and not isinstance(value[field], str):
|
||
errors.append(f"{where}.{field}: 必须是字符串")
|
||
|
||
|
||
def validate_knowledge_fields(
|
||
task: dict,
|
||
where: str,
|
||
status: object,
|
||
errors: list[str],
|
||
) -> None:
|
||
refs = (
|
||
validate_knowledge_ref_list(
|
||
task["knowledgeRefs"],
|
||
f"{where}.knowledgeRefs",
|
||
errors,
|
||
)
|
||
if "knowledgeRefs" in task
|
||
else []
|
||
)
|
||
if "knowledgeApplied" in task:
|
||
applications = task["knowledgeApplied"]
|
||
if not isinstance(applications, list):
|
||
errors.append(f"{where}.knowledgeApplied: 必须是列表")
|
||
else:
|
||
seen_applications: set[str] = set()
|
||
for index, application in enumerate(applications):
|
||
application_where = f"{where}.knowledgeApplied[{index}]"
|
||
if not isinstance(application, dict):
|
||
errors.append(f"{application_where}: 必须是对象")
|
||
continue
|
||
reject_unknown_fields(
|
||
application,
|
||
KNOWLEDGE_APPLICATION_FIELDS,
|
||
application_where,
|
||
errors,
|
||
)
|
||
ref = application.get("ref")
|
||
if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
|
||
errors.append(
|
||
f"{application_where}.ref: 必须使用 K-<id>@<revision> 格式"
|
||
)
|
||
else:
|
||
if ref in seen_applications:
|
||
errors.append(f"{application_where}.ref: 应用结果重复: {ref}")
|
||
seen_applications.add(ref)
|
||
if ref not in refs:
|
||
errors.append(f"{application_where}.ref: {ref} 不在 knowledgeRefs 中")
|
||
if application.get("result") not in {"applied", "not_applicable"}:
|
||
errors.append(
|
||
f"{application_where}.result: 必须是 applied/not_applicable"
|
||
)
|
||
evidence = application.get("evidence")
|
||
if not isinstance(evidence, str) or not evidence.strip():
|
||
errors.append(f"{application_where}.evidence: 必须提供非空证据")
|
||
|
||
if "knowledgeCandidates" in task:
|
||
candidates = task["knowledgeCandidates"]
|
||
if not isinstance(candidates, list):
|
||
errors.append(f"{where}.knowledgeCandidates: 必须是列表")
|
||
else:
|
||
for index, candidate in enumerate(candidates):
|
||
candidate_where = f"{where}.knowledgeCandidates[{index}]"
|
||
if not isinstance(candidate, dict):
|
||
errors.append(f"{candidate_where}: 必须是对象")
|
||
continue
|
||
reject_unknown_fields(
|
||
candidate,
|
||
KNOWLEDGE_CANDIDATE_FIELDS,
|
||
candidate_where,
|
||
errors,
|
||
)
|
||
missing = sorted(
|
||
key
|
||
for key in KNOWLEDGE_CANDIDATE_REQUIRED_FIELDS
|
||
if key not in candidate
|
||
)
|
||
if missing:
|
||
errors.append(
|
||
f"{candidate_where}: 缺少必填字段 {', '.join(missing)}"
|
||
)
|
||
if candidate.get("kind") not in KNOWLEDGE_KINDS:
|
||
errors.append(
|
||
f"{candidate_where}.kind: 必须是 {sorted(KNOWLEDGE_KINDS)}"
|
||
)
|
||
for field in sorted(KNOWLEDGE_CANDIDATE_TEXT_FIELDS):
|
||
value = candidate.get(field)
|
||
if not isinstance(value, str) or not value.strip():
|
||
errors.append(
|
||
f"{candidate_where}.{field}: 必须是非空字符串"
|
||
)
|
||
validate_optional_string_fields(
|
||
candidate,
|
||
{"proposedBy", "proposedAt"},
|
||
candidate_where,
|
||
errors,
|
||
)
|
||
scope = candidate.get("scope")
|
||
if not isinstance(scope, dict):
|
||
errors.append(f"{candidate_where}.scope: 至少包含一个非空作用域")
|
||
else:
|
||
unknown_scope_fields = sorted(
|
||
set(scope) - KNOWLEDGE_SCOPE_FIELDS - {"all"}
|
||
)
|
||
for field in unknown_scope_fields:
|
||
errors.append(
|
||
f"{candidate_where}.scope: 未知字段 {field!r}"
|
||
)
|
||
if "all" in scope and not isinstance(scope.get("all"), bool):
|
||
errors.append(f"{candidate_where}.scope.all: 必须是布尔值")
|
||
populated_scope_fields: list[str] = []
|
||
for field in sorted(KNOWLEDGE_SCOPE_FIELDS & set(scope)):
|
||
values = scope.get(field)
|
||
if not isinstance(values, list) or any(
|
||
not isinstance(item, str) or not item.strip()
|
||
for item in values
|
||
):
|
||
errors.append(
|
||
f"{candidate_where}.scope.{field}: "
|
||
"必须是非空字符串列表"
|
||
)
|
||
elif len(values) != len(set(values)):
|
||
errors.append(
|
||
f"{candidate_where}.scope.{field}: 不能包含重复值"
|
||
)
|
||
elif values:
|
||
populated_scope_fields.append(field)
|
||
if scope.get("all") is True and populated_scope_fields:
|
||
errors.append(
|
||
f"{candidate_where}.scope: all=true 时不能同时填写作用域维度"
|
||
)
|
||
if scope.get("all") is not True and not populated_scope_fields:
|
||
errors.append(
|
||
f"{candidate_where}.scope: 至少包含一个非空作用域"
|
||
)
|
||
evidence_refs = candidate.get("evidenceRefs")
|
||
if (
|
||
not isinstance(evidence_refs, list)
|
||
or not evidence_refs
|
||
or not all(
|
||
isinstance(ref, str) and ref.strip()
|
||
for ref in evidence_refs
|
||
)
|
||
):
|
||
errors.append(
|
||
f"{candidate_where}.evidenceRefs: 必须是非空字符串列表"
|
||
)
|
||
elif len(evidence_refs) != len(set(evidence_refs)):
|
||
errors.append(
|
||
f"{candidate_where}.evidenceRefs: 不能包含重复值"
|
||
)
|
||
|
||
if "knowledgeChecks" not in task:
|
||
return
|
||
checks = task["knowledgeChecks"]
|
||
if not isinstance(checks, list):
|
||
errors.append(f"{where}.knowledgeChecks: 必须是列表")
|
||
return
|
||
|
||
seen_checks: set[str] = set()
|
||
for index, check in enumerate(checks):
|
||
check_where = f"{where}.knowledgeChecks[{index}]"
|
||
if not isinstance(check, dict):
|
||
errors.append(f"{check_where}: 必须是对象")
|
||
continue
|
||
reject_unknown_fields(
|
||
check,
|
||
KNOWLEDGE_CHECK_FIELDS,
|
||
check_where,
|
||
errors,
|
||
)
|
||
validate_optional_string_fields(
|
||
check,
|
||
{"checkedBy", "checkedAt"},
|
||
check_where,
|
||
errors,
|
||
)
|
||
ref = check.get("ref")
|
||
result = check.get("result")
|
||
evidence = check.get("evidence")
|
||
if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
|
||
errors.append(f"{check_where}.ref: 必须使用 K-<id>@<revision> 格式")
|
||
else:
|
||
if ref in seen_checks:
|
||
errors.append(f"{check_where}.ref: 检查结果重复: {ref}")
|
||
seen_checks.add(ref)
|
||
if ref not in refs:
|
||
errors.append(f"{check_where}.ref: {ref} 不在 knowledgeRefs 中")
|
||
if result not in KNOWLEDGE_CHECK_RESULTS:
|
||
errors.append(
|
||
f"{check_where}.result: 必须是 {sorted(KNOWLEDGE_CHECK_RESULTS)}"
|
||
)
|
||
if not isinstance(evidence, str) or not evidence.strip():
|
||
errors.append(f"{check_where}.evidence: 必须提供非空证据")
|
||
if status == "verified" and result == "failed":
|
||
errors.append(f"{check_where}: verified 任务不能保留失败的知识检查")
|
||
|
||
|
||
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] = []
|
||
|
||
def validate_string_fields(
|
||
value: dict,
|
||
fields: set[str],
|
||
where: str,
|
||
*,
|
||
nullable: bool = False,
|
||
) -> None:
|
||
expected = "必须是字符串或 null" if nullable else "必须是字符串"
|
||
for field in sorted(fields):
|
||
if field not in value:
|
||
continue
|
||
field_value = value[field]
|
||
if not isinstance(field_value, str) and not (
|
||
nullable and field_value is None
|
||
):
|
||
errors.append(f"{where}.{field}: {expected}")
|
||
|
||
def validate_string_lists(
|
||
value: dict,
|
||
fields: set[str],
|
||
where: str,
|
||
) -> None:
|
||
for field in sorted(fields):
|
||
if field not in value:
|
||
continue
|
||
items = value[field]
|
||
if not isinstance(items, list):
|
||
errors.append(f"{where}.{field}: 必须是列表")
|
||
elif any(not isinstance(item, str) for item in items):
|
||
errors.append(f"{where}.{field}: 列表项必须是字符串")
|
||
|
||
def validate_object_fields(
|
||
value: dict,
|
||
fields: set[str],
|
||
where: str,
|
||
) -> None:
|
||
for field in sorted(fields):
|
||
if field in value and not isinstance(value[field], dict):
|
||
errors.append(f"{where}.{field}: 必须是对象")
|
||
|
||
version = data.get("version")
|
||
if (
|
||
not isinstance(version, int)
|
||
or isinstance(version, bool)
|
||
or version < 1
|
||
):
|
||
errors.append("version 必须是 >=1 的整数")
|
||
validate_string_fields(
|
||
data,
|
||
{"updatedAt", "source", "ackVersion", "kitVersion"},
|
||
"<root>",
|
||
)
|
||
|
||
project = data.get("project")
|
||
if not isinstance(project, dict):
|
||
errors.append("project 必须是对象")
|
||
else:
|
||
if not _nonempty_string(project.get("name")):
|
||
errors.append("project.name 必须是非空字符串")
|
||
validate_string_fields(
|
||
project,
|
||
{"repoPath", "baseUrl", "devWorktree", "overlayFile"},
|
||
"project",
|
||
)
|
||
if (
|
||
"knowledgeFile" in project
|
||
and project.get("knowledgeFile") != "docs/ack/knowledge.yaml"
|
||
):
|
||
errors.append(
|
||
"project.knowledgeFile 必须固定为 docs/ack/knowledge.yaml"
|
||
)
|
||
|
||
ack_version = data.get("ackVersion")
|
||
version_match = SEMVER_RE.fullmatch(ack_version) if isinstance(ack_version, str) else None
|
||
if "ackVersion" in data and version_match is None:
|
||
errors.append("ackVersion 必须是合法 SemVer（例如 0.10.0）")
|
||
routing_required = (
|
||
(isinstance(project, dict) and "orchestration" in project)
|
||
or "workerReceipts" in data
|
||
or (
|
||
version_match is not None
|
||
and (int(version_match.group(1)), int(version_match.group(2)))
|
||
>= (0, 10)
|
||
)
|
||
)
|
||
if routing_required:
|
||
errors.extend(validate_routing_document(data))
|
||
|
||
if "summary" in data:
|
||
summary = data["summary"]
|
||
if not isinstance(summary, dict):
|
||
errors.append("summary 必须是对象")
|
||
else:
|
||
validate_string_lists(
|
||
summary,
|
||
{"verified", "open", "failedRetest", "leftovers"},
|
||
"summary",
|
||
)
|
||
if "statusReference" in data and not isinstance(
|
||
data["statusReference"], dict
|
||
):
|
||
errors.append("statusReference 必须是对象")
|
||
|
||
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 _nonempty_string(tid):
|
||
errors.append(f"{where}: id 必须是非空字符串")
|
||
else:
|
||
where = f"tasks[{i}] {tid}"
|
||
if routing_required and TASK_ID_RE.fullmatch(tid) is None:
|
||
errors.append(
|
||
f"{where}: v0.10 自动路由 id 只允许字母、数字、点、下划线和连字符"
|
||
)
|
||
if tid in seen_ids:
|
||
errors.append(f"{where}: id 重复")
|
||
seen_ids.add(tid)
|
||
if not _nonempty_string(title):
|
||
errors.append(f"{where}: title 必须是非空字符串")
|
||
if status not in STATUS_ENUM:
|
||
errors.append(
|
||
f"{where}: status={status!r} 非法，应为 {sorted(STATUS_ENUM)}"
|
||
)
|
||
|
||
validate_string_fields(
|
||
task,
|
||
{
|
||
"type",
|
||
"priority",
|
||
"assignee",
|
||
"component",
|
||
"description",
|
||
"expected",
|
||
"actual",
|
||
},
|
||
where,
|
||
)
|
||
validate_string_lists(
|
||
task,
|
||
{"specRefs", "testRefs", "stepsToReproduce"},
|
||
where,
|
||
)
|
||
validate_object_fields(task, {"evidence", "verification"}, where)
|
||
|
||
validate_knowledge_fields(task, where, status, errors)
|
||
|
||
if "dispatch" not in task:
|
||
dispatch = {}
|
||
elif not isinstance(task["dispatch"], dict):
|
||
errors.append(f"{where}.dispatch: 必须是对象")
|
||
dispatch = {}
|
||
else:
|
||
dispatch = task["dispatch"]
|
||
reject_unknown_fields(
|
||
dispatch,
|
||
DISPATCH_FIELDS,
|
||
f"{where}.dispatch",
|
||
errors,
|
||
)
|
||
validate_string_fields(
|
||
dispatch,
|
||
{"taskId", "dispatchId", "worker"},
|
||
f"{where}.dispatch",
|
||
nullable=True,
|
||
)
|
||
|
||
rounds = dispatch.get("rounds", [])
|
||
if not isinstance(rounds, list):
|
||
errors.append(f"{where}.dispatch.rounds: 必须是列表")
|
||
else:
|
||
if len(rounds) > MAX_ROUNDS:
|
||
errors.append(
|
||
f"{where}: 派发轮次 {len(rounds)} 超过上限 {MAX_ROUNDS}"
|
||
)
|
||
seen_attempt_ids: set[str] = set()
|
||
round_numbers: list[int] = []
|
||
for round_index, round_item in enumerate(rounds):
|
||
round_where = f"{where}.dispatch.rounds[{round_index}]"
|
||
if not isinstance(round_item, dict):
|
||
errors.append(f"{round_where}: 必须是对象")
|
||
continue
|
||
if round_item.get("result") not in {"passed", "failed"}:
|
||
errors.append(f"{round_where}.result: 必须是 passed/failed")
|
||
if "evidence" in round_item and not isinstance(
|
||
round_item["evidence"], str
|
||
):
|
||
errors.append(f"{round_where}.evidence: 必须是字符串")
|
||
round_number = round_item.get("round")
|
||
round_number_is_valid = (
|
||
isinstance(round_number, int)
|
||
and not isinstance(round_number, bool)
|
||
and 1 <= round_number <= MAX_ROUNDS
|
||
)
|
||
if not round_number_is_valid:
|
||
errors.append(
|
||
f"{round_where}.round: 必须是 1..{MAX_ROUNDS} 的整数"
|
||
)
|
||
else:
|
||
round_numbers.append(round_number)
|
||
if "attemptId" in round_item:
|
||
attempt_id = round_item["attemptId"]
|
||
if (
|
||
not isinstance(attempt_id, str)
|
||
or not ATTEMPT_ID_RE.fullmatch(attempt_id)
|
||
):
|
||
errors.append(
|
||
f"{round_where}.attemptId: "
|
||
"必须使用 <task-id>-A<round> 格式"
|
||
)
|
||
else:
|
||
if attempt_id in seen_attempt_ids:
|
||
errors.append(
|
||
f"{round_where}.attemptId: "
|
||
f"轮次内不能重复: {attempt_id}"
|
||
)
|
||
seen_attempt_ids.add(attempt_id)
|
||
if (
|
||
isinstance(tid, str)
|
||
and round_number_is_valid
|
||
and attempt_id != f"{tid}-A{round_number}"
|
||
):
|
||
errors.append(
|
||
f"{round_where}.attemptId: 应为 "
|
||
f"{tid}-A{round_number}"
|
||
)
|
||
expected_rounds = list(range(1, len(rounds) + 1))
|
||
if round_numbers != expected_rounds:
|
||
errors.append(
|
||
f"{where}.dispatch.rounds: round 必须从 1 连续递增且不重复"
|
||
)
|
||
|
||
resolution = task.get("resolution")
|
||
if "resolution" in task:
|
||
if not isinstance(resolution, dict):
|
||
errors.append(f"{where}.resolution: 必须是对象")
|
||
else:
|
||
validate_string_fields(
|
||
resolution,
|
||
{
|
||
"fixedBy",
|
||
"verifiedBy",
|
||
"verifiedAt",
|
||
"leftoverReason",
|
||
},
|
||
f"{where}.resolution",
|
||
nullable=True,
|
||
)
|
||
validate_object_fields(
|
||
resolution,
|
||
{"evidence"},
|
||
f"{where}.resolution",
|
||
)
|
||
|
||
if status == "leftover":
|
||
if (
|
||
not isinstance(resolution, dict)
|
||
or not _nonempty_string(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_document(tasks_path)
|
||
|
||
schema_path = Path(args.schema) if args.schema else (
|
||
Path(__file__).resolve().parent.parent / "templates" / "tasks.schema.json"
|
||
)
|
||
if args.schema and not schema_path.is_file():
|
||
sys.stderr.write(f"找不到指定的 schema 文件: {schema_path}\n")
|
||
return 2
|
||
|
||
errors = validate_builtin(data)
|
||
mode = "内置语义规则"
|
||
try:
|
||
import jsonschema # type: ignore # noqa: F401
|
||
|
||
if schema_path.is_file():
|
||
errors = validate_with_schema(data, schema_path) + errors
|
||
mode = f"schema ({schema_path.name}) + 内置语义规则"
|
||
else:
|
||
mode = "内置语义规则（未找到 schema 文件）"
|
||
except ImportError:
|
||
pass
|
||
|
||
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())
|