feat: generate task board from task frontmatter
This commit is contained in:
@@ -0,0 +1,194 @@
|
||||
"""Generate a read-only task board from docs/tasks frontmatter."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_TASKS_DIR = REPO_ROOT / "docs" / "tasks"
|
||||
DEFAULT_OUTPUT_PATH = REPO_ROOT / "docs" / "tasks-board.md"
|
||||
REQUIRED_FIELDS = {"id", "title", "phase", "deps", "status", "created"}
|
||||
VALID_STATUSES = {"TODO", "DOING", "DONE", "BLOCKED"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Task:
|
||||
id: str
|
||||
title: str
|
||||
phase: int
|
||||
deps: tuple[str, ...]
|
||||
status: str
|
||||
created: str
|
||||
path: Path
|
||||
|
||||
|
||||
def parse_deps(raw_value):
|
||||
value = str(raw_value or "").strip()
|
||||
if value == "[]":
|
||||
return ()
|
||||
if not (value.startswith("[") and value.endswith("]")):
|
||||
raise ValueError("deps 必须使用 [T-xxx, ...] 格式")
|
||||
inner = value[1:-1].strip()
|
||||
if not inner:
|
||||
return ()
|
||||
deps = []
|
||||
for item in inner.split(","):
|
||||
dep = item.strip().strip("'\"")
|
||||
if not dep:
|
||||
raise ValueError("deps 中存在空依赖")
|
||||
deps.append(dep)
|
||||
return tuple(deps)
|
||||
|
||||
|
||||
def parse_frontmatter(path):
|
||||
lines = path.read_text(encoding="utf-8").splitlines()
|
||||
if not lines or lines[0].strip() != "---":
|
||||
raise ValueError("缺少 frontmatter 起始 ---")
|
||||
end_index = None
|
||||
for index, line in enumerate(lines[1:], start=1):
|
||||
if line.strip() == "---":
|
||||
end_index = index
|
||||
break
|
||||
if end_index is None:
|
||||
raise ValueError("缺少 frontmatter 结束 ---")
|
||||
|
||||
data = {}
|
||||
for line in lines[1:end_index]:
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped.startswith("#"):
|
||||
continue
|
||||
if ":" not in stripped:
|
||||
raise ValueError(f"frontmatter 行格式错误:{stripped}")
|
||||
key, value = stripped.split(":", 1)
|
||||
data[key.strip()] = value.strip()
|
||||
missing = sorted(REQUIRED_FIELDS - set(data))
|
||||
if missing:
|
||||
raise ValueError("缺少字段:" + "、".join(missing))
|
||||
try:
|
||||
phase = int(str(data["phase"]).strip())
|
||||
except ValueError as exc:
|
||||
raise ValueError("phase 必须是整数") from exc
|
||||
deps = parse_deps(data["deps"])
|
||||
status = str(data["status"]).strip()
|
||||
if status not in VALID_STATUSES:
|
||||
raise ValueError("status 必须是 TODO/DOING/DONE/BLOCKED")
|
||||
task_id = str(data["id"]).strip()
|
||||
if not re.fullmatch(r"T-\d+[A-Za-z]*", task_id):
|
||||
raise ValueError("id 必须是 T-数字 或 T-数字后缀")
|
||||
return Task(
|
||||
id=task_id,
|
||||
title=str(data["title"]).strip(),
|
||||
phase=phase,
|
||||
deps=deps,
|
||||
status=status,
|
||||
created=str(data["created"]).strip(),
|
||||
path=path,
|
||||
)
|
||||
|
||||
|
||||
def task_sort_key(task_or_id):
|
||||
task_id = task_or_id.id if isinstance(task_or_id, Task) else str(task_or_id)
|
||||
match = re.fullmatch(r"T-(\d+)([A-Za-z]*)", task_id)
|
||||
if not match:
|
||||
return (10**9, task_id)
|
||||
return (int(match.group(1)), match.group(2))
|
||||
|
||||
|
||||
def load_tasks(tasks_dir):
|
||||
tasks = []
|
||||
warnings = []
|
||||
for path in sorted(Path(tasks_dir).glob("T-*.md")):
|
||||
try:
|
||||
tasks.append(parse_frontmatter(path))
|
||||
except ValueError as exc:
|
||||
warnings.append(f"{path.name}: {exc}")
|
||||
tasks.sort(key=task_sort_key)
|
||||
return tasks, warnings
|
||||
|
||||
|
||||
def next_claimable_task(tasks):
|
||||
done_ids = {task.id for task in tasks if task.status == "DONE"}
|
||||
for task in sorted(tasks, key=task_sort_key):
|
||||
if task.status == "TODO" and all(dep in done_ids for dep in task.deps):
|
||||
return task
|
||||
return None
|
||||
|
||||
|
||||
def markdown_escape(value):
|
||||
return str(value).replace("|", "\\|").replace("\n", " ")
|
||||
|
||||
|
||||
def render_board(tasks):
|
||||
status_counts = Counter(task.status for task in tasks)
|
||||
next_task = next_claimable_task(tasks)
|
||||
lines = [
|
||||
"# 任务看板",
|
||||
"",
|
||||
"> 本文件由 `python scripts/gen_task_board.py` 自动生成,请勿手改。",
|
||||
"> 数据源只包含 `docs/tasks/T-*.md`;`docs/06-tasks.md` 是 T-000~T-549 历史归档,不纳入本看板。",
|
||||
"",
|
||||
"## 汇总",
|
||||
"",
|
||||
f"- 总任务:{len(tasks)}",
|
||||
"- 状态统计:"
|
||||
+ " · ".join(f"{status} {status_counts.get(status, 0)}" for status in sorted(VALID_STATUSES)),
|
||||
"- 下一个可领取:"
|
||||
+ (f"{next_task.id}({next_task.title})" if next_task is not None else "暂无"),
|
||||
"",
|
||||
]
|
||||
grouped = defaultdict(list)
|
||||
for task in tasks:
|
||||
grouped[task.phase].append(task)
|
||||
for phase in sorted(grouped):
|
||||
lines.extend(
|
||||
[
|
||||
f"## Phase {phase}",
|
||||
"",
|
||||
"| ID | 任务 | 依赖 | 状态 |",
|
||||
"| --- | --- | --- | --- |",
|
||||
]
|
||||
)
|
||||
for task in sorted(grouped[phase], key=task_sort_key):
|
||||
deps = ", ".join(task.deps) if task.deps else "-"
|
||||
lines.append(
|
||||
"| {id} | {title} | {deps} | {status} |".format(
|
||||
id=markdown_escape(task.id),
|
||||
title=markdown_escape(task.title),
|
||||
deps=markdown_escape(deps),
|
||||
status=markdown_escape(task.status),
|
||||
)
|
||||
)
|
||||
lines.append("")
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
def generate_board(tasks_dir=DEFAULT_TASKS_DIR, output_path=DEFAULT_OUTPUT_PATH):
|
||||
tasks, warnings = load_tasks(tasks_dir)
|
||||
output = render_board(tasks)
|
||||
output_path = Path(output_path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(output, encoding="utf-8")
|
||||
return tasks, warnings, output
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(description="从 docs/tasks frontmatter 生成只读任务看板。")
|
||||
parser.add_argument("--tasks-dir", default=str(DEFAULT_TASKS_DIR), help="任务文件目录")
|
||||
parser.add_argument("--output", default=str(DEFAULT_OUTPUT_PATH), help="输出 Markdown 文件")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
tasks, warnings, _output = generate_board(args.tasks_dir, args.output)
|
||||
for warning in warnings:
|
||||
print(f"跳过任务文件:{warning}", file=sys.stderr)
|
||||
print(f"已生成 {args.output},任务 {len(tasks)} 个")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user