307 lines
11 KiB
Python
307 lines
11 KiB
Python
#!/usr/bin/env python
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
DEFAULT_PATH = Path("docs/ai/BACKLOG.yaml")
|
|
VALID_STATUSES = {"done", "ready", "blocked-external", "deferred"}
|
|
VALID_PRIORITIES = {"P0", "P1", "P2", "P3"}
|
|
ID_PATTERN = re.compile(r"^VR-\d{3}$")
|
|
REQUIREMENT_PATTERN = re.compile(r"^(PR|NFR|AC)-\d{3}$")
|
|
REQUIRED_TASK_FIELDS = {
|
|
"id",
|
|
"title",
|
|
"status",
|
|
"priority",
|
|
"requirement_ids",
|
|
"depends_on",
|
|
"summary",
|
|
"acceptance_criteria",
|
|
"verification",
|
|
"primary_paths",
|
|
}
|
|
|
|
|
|
class LedgerError(ValueError):
|
|
pass
|
|
|
|
|
|
def load_ledger(path: Path = DEFAULT_PATH) -> dict[str, Any]:
|
|
try:
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise LedgerError(f"Taakledger ontbreekt: {path}") from exc
|
|
except yaml.YAMLError as exc:
|
|
raise LedgerError(f"Ongeldige YAML in {path}: {exc}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise LedgerError("Taakledger moet een mapping op topniveau zijn")
|
|
return payload
|
|
|
|
|
|
def _require_nonempty_list(task: dict[str, Any], field: str) -> None:
|
|
value = task.get(field)
|
|
if not isinstance(value, list) or not value:
|
|
raise LedgerError(f"{task.get('id', '<zonder-id>')}: {field} moet een niet-lege lijst zijn")
|
|
if any(not isinstance(item, str) or not item.strip() for item in value):
|
|
task_id = task.get("id", "<zonder-id>")
|
|
raise LedgerError(f"{task_id}: {field} bevat een lege/niet-tekst waarde")
|
|
|
|
|
|
def validate_ledger(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
if payload.get("schema_version") != 1:
|
|
raise LedgerError("Alleen schema_version 1 wordt ondersteund")
|
|
tasks = payload.get("tasks")
|
|
if not isinstance(tasks, list) or not tasks:
|
|
raise LedgerError("BACKLOG.yaml moet minstens één taak bevatten")
|
|
|
|
ids: list[str] = []
|
|
task_by_id: dict[str, dict[str, Any]] = {}
|
|
for raw in tasks:
|
|
if not isinstance(raw, dict):
|
|
raise LedgerError("Iedere taak moet een mapping zijn")
|
|
missing = REQUIRED_TASK_FIELDS - raw.keys()
|
|
if missing:
|
|
task_id = raw.get("id", "<zonder-id>")
|
|
raise LedgerError(f"{task_id}: ontbrekende velden {sorted(missing)}")
|
|
task_id = raw["id"]
|
|
if not isinstance(task_id, str) or not ID_PATTERN.fullmatch(task_id):
|
|
raise LedgerError(f"Ongeldige taak-ID: {task_id!r}")
|
|
if task_id in task_by_id:
|
|
raise LedgerError(f"Dubbele taak-ID: {task_id}")
|
|
if raw["status"] not in VALID_STATUSES:
|
|
raise LedgerError(f"{task_id}: ongeldige status {raw['status']!r}")
|
|
if raw["priority"] not in VALID_PRIORITIES:
|
|
raise LedgerError(f"{task_id}: ongeldige priority {raw['priority']!r}")
|
|
if not isinstance(raw["title"], str) or not raw["title"].strip():
|
|
raise LedgerError(f"{task_id}: title ontbreekt")
|
|
if not isinstance(raw["summary"], str) or len(raw["summary"].strip()) < 20:
|
|
raise LedgerError(f"{task_id}: summary is te kort")
|
|
for field in ("requirement_ids", "acceptance_criteria", "verification", "primary_paths"):
|
|
_require_nonempty_list(raw, field)
|
|
for requirement in raw["requirement_ids"]:
|
|
if not REQUIREMENT_PATTERN.fullmatch(requirement):
|
|
raise LedgerError(f"{task_id}: ongeldige requirement-ID {requirement!r}")
|
|
if not isinstance(raw["depends_on"], list) or any(
|
|
not isinstance(item, str) for item in raw["depends_on"]
|
|
):
|
|
raise LedgerError(f"{task_id}: depends_on moet een lijst van IDs zijn")
|
|
if task_id in raw["depends_on"]:
|
|
raise LedgerError(f"{task_id}: taak mag niet van zichzelf afhangen")
|
|
if raw["status"] == "done":
|
|
result = raw.get("result")
|
|
has_result = (
|
|
isinstance(result, dict)
|
|
and bool(result.get("completed_at"))
|
|
and bool(result.get("note"))
|
|
)
|
|
if not has_result:
|
|
raise LedgerError(
|
|
f"{task_id}: done taak vereist result.completed_at en result.note"
|
|
)
|
|
if raw["status"] == "blocked-external" and not raw.get("external_blocker"):
|
|
raise LedgerError(f"{task_id}: blocked-external vereist external_blocker")
|
|
ids.append(task_id)
|
|
task_by_id[task_id] = raw
|
|
|
|
for task in tasks:
|
|
for dependency in task["depends_on"]:
|
|
if dependency not in task_by_id:
|
|
raise LedgerError(f"{task['id']}: onbekende dependency {dependency}")
|
|
|
|
visiting: set[str] = set()
|
|
visited: set[str] = set()
|
|
|
|
def visit(task_id: str, trail: list[str]) -> None:
|
|
if task_id in visiting:
|
|
cycle = " -> ".join([*trail, task_id])
|
|
raise LedgerError(f"Dependencycyclus: {cycle}")
|
|
if task_id in visited:
|
|
return
|
|
visiting.add(task_id)
|
|
for dependency in task_by_id[task_id]["depends_on"]:
|
|
visit(dependency, [*trail, task_id])
|
|
visiting.remove(task_id)
|
|
visited.add(task_id)
|
|
|
|
for task_id in ids:
|
|
visit(task_id, [])
|
|
|
|
for task in tasks:
|
|
if task["status"] == "done":
|
|
incomplete_dependencies = [
|
|
dep for dep in task["depends_on"] if task_by_id[dep]["status"] != "done"
|
|
]
|
|
if incomplete_dependencies:
|
|
raise LedgerError(
|
|
f"{task['id']}: done maar dependencies niet done: {incomplete_dependencies}"
|
|
)
|
|
return tasks
|
|
|
|
|
|
def save_ledger(payload: dict[str, Any], path: Path = DEFAULT_PATH) -> None:
|
|
payload["updated_at"] = datetime.now(UTC).date().isoformat()
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
yaml.safe_dump(payload, sort_keys=False, allow_unicode=True, width=110),
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(path)
|
|
|
|
|
|
def executable_tasks(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
status_by_id = {task["id"]: task["status"] for task in tasks}
|
|
return [
|
|
task
|
|
for task in tasks
|
|
if task["status"] == "ready"
|
|
and all(status_by_id[dependency] == "done" for dependency in task["depends_on"])
|
|
]
|
|
|
|
|
|
def print_task(task: dict[str, Any], *, verbose: bool = True) -> None:
|
|
print(f"{task['id']} [{task['priority']}/{task['status']}] {task['title']}")
|
|
if not verbose:
|
|
return
|
|
print(f"\n{task['summary']}")
|
|
if task["depends_on"]:
|
|
print(f"\nDependencies: {', '.join(task['depends_on'])}")
|
|
print("\nAcceptatiecriteria:")
|
|
for item in task["acceptance_criteria"]:
|
|
print(f"- {item}")
|
|
print("\nVerificatie:")
|
|
for item in task["verification"]:
|
|
print(f"- {item}")
|
|
print("\nPrimaire paden:")
|
|
for item in task["primary_paths"]:
|
|
print(f"- {item}")
|
|
if task.get("external_blocker"):
|
|
print(f"\nExterne blokkade: {task['external_blocker']}")
|
|
if task.get("result"):
|
|
print(f"\nResultaat: {task['result'].get('note', '')}")
|
|
|
|
|
|
def command_validate(args: argparse.Namespace) -> int:
|
|
tasks = validate_ledger(load_ledger(args.path))
|
|
counts = {status: sum(task["status"] == status for task in tasks) for status in VALID_STATUSES}
|
|
executable = executable_tasks(tasks)
|
|
print(
|
|
"Taakledger geldig: "
|
|
f"{len(tasks)} taken; "
|
|
+ ", ".join(f"{status}={counts[status]}" for status in sorted(counts))
|
|
+ f"; uitvoerbaar={len(executable)}"
|
|
)
|
|
return 0
|
|
|
|
|
|
def command_next(args: argparse.Namespace) -> int:
|
|
tasks = validate_ledger(load_ledger(args.path))
|
|
executable = executable_tasks(tasks)
|
|
if not executable:
|
|
print("Geen uitvoerbare ready taak. Controleer blocked-external en deferred taken.")
|
|
return 1
|
|
print_task(executable[0], verbose=not args.id_only)
|
|
return 0
|
|
|
|
|
|
def command_list(args: argparse.Namespace) -> int:
|
|
tasks = validate_ledger(load_ledger(args.path))
|
|
selected = [task for task in tasks if not args.status or task["status"] == args.status]
|
|
for task in selected:
|
|
print_task(task, verbose=False)
|
|
return 0
|
|
|
|
|
|
def command_show(args: argparse.Namespace) -> int:
|
|
tasks = validate_ledger(load_ledger(args.path))
|
|
for task in tasks:
|
|
if task["id"] == args.task_id:
|
|
print_task(task)
|
|
return 0
|
|
raise LedgerError(f"Onbekende taak: {args.task_id}")
|
|
|
|
|
|
def command_set(args: argparse.Namespace) -> int:
|
|
payload = load_ledger(args.path)
|
|
tasks = validate_ledger(payload)
|
|
task_by_id = {task["id"]: task for task in tasks}
|
|
try:
|
|
target = task_by_id[args.task_id]
|
|
except KeyError as exc:
|
|
raise LedgerError(f"Onbekende taak: {args.task_id}") from exc
|
|
if args.status == "done":
|
|
incomplete = [dep for dep in target["depends_on"] if task_by_id[dep]["status"] != "done"]
|
|
if incomplete:
|
|
raise LedgerError(
|
|
f"Kan {args.task_id} niet afronden; dependencies niet done: {incomplete}"
|
|
)
|
|
if not args.note:
|
|
raise LedgerError("Status done vereist --note met het geverifieerde resultaat")
|
|
target["result"] = {
|
|
"completed_at": datetime.now(UTC).date().isoformat(),
|
|
"note": args.note.strip(),
|
|
}
|
|
target.pop("external_blocker", None)
|
|
elif args.status == "blocked-external":
|
|
if not args.note:
|
|
raise LedgerError("blocked-external vereist --note met de exacte externe blokkade")
|
|
target["external_blocker"] = args.note.strip()
|
|
target.pop("result", None)
|
|
else:
|
|
target.pop("result", None)
|
|
if args.status != "blocked-external":
|
|
target.pop("external_blocker", None)
|
|
target["status"] = args.status
|
|
save_ledger(payload, args.path)
|
|
validate_ledger(load_ledger(args.path))
|
|
print_task(target, verbose=False)
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="VacatureRadar machineleesbare backlog beheren")
|
|
parser.add_argument("--path", type=Path, default=DEFAULT_PATH)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
validate_parser = subparsers.add_parser("validate", help="Valideer schema en dependencygraaf")
|
|
validate_parser.set_defaults(func=command_validate)
|
|
|
|
next_parser = subparsers.add_parser("next", help="Toon de eerste uitvoerbare ready taak")
|
|
next_parser.add_argument("--id-only", action="store_true", help="Toon alleen de kopregel")
|
|
next_parser.set_defaults(func=command_next)
|
|
|
|
list_parser = subparsers.add_parser("list", help="Toon taken")
|
|
list_parser.add_argument("--status", choices=sorted(VALID_STATUSES))
|
|
list_parser.set_defaults(func=command_list)
|
|
|
|
show_parser = subparsers.add_parser("show", help="Toon één taak volledig")
|
|
show_parser.add_argument("task_id")
|
|
show_parser.set_defaults(func=command_show)
|
|
|
|
set_parser = subparsers.add_parser("set", help="Wijzig een taakstatus")
|
|
set_parser.add_argument("task_id")
|
|
set_parser.add_argument("status", choices=sorted(VALID_STATUSES))
|
|
set_parser.add_argument("--note", default="")
|
|
set_parser.set_defaults(func=command_set)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
return int(args.func(args))
|
|
except LedgerError as exc:
|
|
print(f"Taakledgerfout: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|