cmshoppe/scripts/gen_task_board.py

"""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())