feat: PLANNING体系首版（00路线图/02任务总表/03执行协议/I-03/I-04/registry42口径）+M1止血交付
This commit is contained in:
@@ -0,0 +1,323 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PLANNING 任务调度器（参考实现，Python 标准库，零依赖）。
|
||||
|
||||
用法：
|
||||
python tasks.py list # 全部任务 + 状态
|
||||
python tasks.py show C-01 # 任务卡摘要
|
||||
python tasks.py dispatch C-01 # 产出可直接粘贴的派单提示词
|
||||
python tasks.py dispatch C-01 --model deepseek-v4.1-flash
|
||||
python tasks.py bundle C-01 # 交接包（卡 + 协议 + 项目上下文）
|
||||
python tasks.py collect # 扫描各项目根 PROGRESS_*.md
|
||||
python tasks.py status # 看板（里程碑分组）
|
||||
python tasks.py start C-01 # 标记执行中
|
||||
python tasks.py accept C-01 # 标记完成
|
||||
python tasks.py block C-01 "原因" # 标记阻塞
|
||||
python tasks.py reset C-01 # 状态还原为 todo
|
||||
|
||||
状态存储：manifest 为源 + state.json 覆盖层（本脚本只写 state.json）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
MANIFEST = ROOT / "task-manifest.json"
|
||||
STATE = ROOT / "state.json"
|
||||
PROTOCOL = ROOT / "03-执行协议.md"
|
||||
|
||||
STATUS_ICON = {
|
||||
"todo": "○",
|
||||
"doing": "◐",
|
||||
"review": "◑",
|
||||
"done": "●",
|
||||
"blocked": "✗",
|
||||
}
|
||||
|
||||
PROGRESS_RE = re.compile(r"PROGRESS[_-]([A-Z]-\d{2})\.md$", re.IGNORECASE)
|
||||
STATUS_LINE_RE = re.compile(r"状态[：:]\s*(done|partial|blocked|doing|review)", re.IGNORECASE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 基础读写
|
||||
|
||||
def load_manifest() -> dict:
|
||||
return json.loads(MANIFEST.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def load_state() -> dict:
|
||||
if STATE.exists():
|
||||
return json.loads(STATE.read_text(encoding="utf-8"))
|
||||
return {}
|
||||
|
||||
|
||||
def save_state(state: dict) -> None:
|
||||
STATE.write_text(
|
||||
json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def effective_status(task: dict, state: dict) -> str:
|
||||
return state.get(task["id"], {}).get("status", task.get("status", "todo"))
|
||||
|
||||
|
||||
def find_task(manifest: dict, tid: str) -> dict | None:
|
||||
for t in manifest["tasks"]:
|
||||
if t["id"].upper() == tid.upper():
|
||||
return t
|
||||
return None
|
||||
|
||||
|
||||
def card_path(task: dict) -> Path:
|
||||
return ROOT.parent / task["card"]
|
||||
|
||||
|
||||
def project_root(manifest: dict, task: dict) -> Path:
|
||||
return Path(manifest["project_roots"].get(task["project"], "?"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 命令实现
|
||||
|
||||
def cmd_list(manifest: dict, state: dict, args) -> int:
|
||||
tasks = manifest["tasks"]
|
||||
if args.project:
|
||||
tasks = [t for t in tasks if t["project"] == args.project]
|
||||
print(f"{'ID':<5} {'状态':<4} {'优先级':<5} {'里程碑':<5} {'项目':<14} 标题")
|
||||
print("-" * 78)
|
||||
for t in tasks:
|
||||
st = effective_status(t, state)
|
||||
icon = STATUS_ICON.get(st, "?")
|
||||
print(f"{t['id']:<5} {icon}{st:<3} {t['priority']:<5} {t['milestone']:<5} {t['project']:<14} {t['title']}")
|
||||
n_todo = sum(1 for t in tasks if effective_status(t, state) == "todo")
|
||||
print("-" * 78)
|
||||
print(f"共 {len(tasks)} 项，待派 {n_todo} 项。图例：○待派 ◐执行中 ◑待验收 ●完成 ✗阻塞")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_show(manifest: dict, state: dict, args) -> int:
|
||||
t = find_task(manifest, args.id)
|
||||
if not t:
|
||||
print(f"未找到任务 {args.id}")
|
||||
return 1
|
||||
p = card_path(t)
|
||||
print(f"# {t['id']} {t['title']}")
|
||||
print(f"项目={t['project']} 优先级={t['priority']} 里程碑={t['milestone']} "
|
||||
f"状态={effective_status(t, state)} 预估={t['estimate_days']}天")
|
||||
print(f"建议模型={t['model_primary']} 复核={t.get('model_review', '—')}")
|
||||
print(f"依赖={t.get('depends_on') or '无'}")
|
||||
print(f"卡文件={p}")
|
||||
if p.exists():
|
||||
print("-" * 78)
|
||||
print(p.read_text(encoding="utf-8"))
|
||||
else:
|
||||
print("!! 卡文件缺失")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
def _protocol_excerpt() -> str:
|
||||
if not PROTOCOL.exists():
|
||||
return "（协议文件缺失）"
|
||||
text = PROTOCOL.read_text(encoding="utf-8")
|
||||
# 取「红线」章节 + 接单流程；无章节则给全文前 60 行
|
||||
out: list[str] = []
|
||||
for block in re.split(r"\n(?=## )", text):
|
||||
if block.startswith("## 一、接单流程") or block.startswith("## 三、质量红线"):
|
||||
out.append(block.strip())
|
||||
return "\n\n".join(out) if out else "\n".join(text.splitlines()[:60])
|
||||
|
||||
|
||||
def cmd_dispatch(manifest: dict, state: dict, args) -> int:
|
||||
t = find_task(manifest, args.id)
|
||||
if not t:
|
||||
print(f"未找到任务 {args.id}")
|
||||
return 1
|
||||
p = card_path(t)
|
||||
if not p.exists():
|
||||
print(f"卡文件缺失：{p}")
|
||||
return 1
|
||||
model = args.model or t["model_primary"]
|
||||
root = project_root(manifest, t)
|
||||
card = p.read_text(encoding="utf-8")
|
||||
|
||||
print(f"""你是执行模型 {model}。下面是你的任务卡与作业规范，按规范执行到验收全绿。
|
||||
|
||||
==================== 任务卡 ====================
|
||||
{card}
|
||||
==================== 作业规范（要点） ====================
|
||||
{_protocol_excerpt()}
|
||||
|
||||
==================== 项目上下文 ====================
|
||||
项目根：{root}
|
||||
开工前：先读项目根 README / AGENTS 与最近 2 份 PROGRESS（或迭代报告），然后跑基线测试确认开工状态。
|
||||
交付：代码 + 项目根 {t['progress_file']}（格式见协议）+ 关键命令的原始输出。
|
||||
完成后请逐条对照任务卡「验收标准」自检，并在 PROGRESS 中逐条标注验证状态（未验证项必须写明）。""")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_bundle(manifest: dict, state: dict, args) -> int:
|
||||
"""交接包：卡 + 协议 + 项目上下文（README 摘要 + 最近 PROGRESS 清单 + 源码目录树）。"""
|
||||
t = find_task(manifest, args.id)
|
||||
if not t:
|
||||
print(f"未找到任务 {args.id}")
|
||||
return 1
|
||||
p = card_path(t)
|
||||
root = project_root(manifest, t)
|
||||
parts: list[str] = []
|
||||
parts.append(f"# 交接包（Handoff Bundle）— {t['id']} {t['title']}")
|
||||
parts.append(f"生成时间：{date.today().isoformat()} · 项目根：{root}")
|
||||
|
||||
parts.append("\n## 1. 任务卡全文\n")
|
||||
parts.append(p.read_text(encoding="utf-8") if p.exists() else "（卡缺失）")
|
||||
|
||||
parts.append("\n## 2. 作业规范要点\n")
|
||||
parts.append(_protocol_excerpt())
|
||||
|
||||
parts.append("\n## 3. 项目上下文\n")
|
||||
if root.exists():
|
||||
readme = next(
|
||||
(root / n for n in ("README.md", "readme.md", "AGENTS.md") if (root / n).exists()),
|
||||
None,
|
||||
)
|
||||
parts.append(f"### 3.1 README：{readme.name if readme else '（未找到）'}\n")
|
||||
if readme:
|
||||
text = readme.read_text(encoding="utf-8", errors="replace")
|
||||
parts.append("\n".join(text.splitlines()[:80]))
|
||||
|
||||
prog = sorted(root.glob("PROGRESS*.md"), key=lambda f: f.stat().st_mtime, reverse=True)
|
||||
parts.append(f"\n### 3.2 最近 PROGRESS（{len(prog)} 份，取最近 2 份）\n")
|
||||
for f in prog[:2]:
|
||||
text = f.read_text(encoding="utf-8", errors="replace")
|
||||
parts.append(f"#### {f.name}\n")
|
||||
parts.append("\n".join(text.splitlines()[:60]))
|
||||
parts.append("")
|
||||
else:
|
||||
parts.append(f"（项目根不存在：{root}——交接前请核对路径）")
|
||||
|
||||
print("\n".join(parts))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_collect(manifest: dict, state: dict, args) -> int:
|
||||
print("扫描各项目根 PROGRESS 文件……\n")
|
||||
found: dict[str, Path] = {}
|
||||
for proj, root_s in manifest["project_roots"].items():
|
||||
root = Path(root_s)
|
||||
if not root.exists():
|
||||
continue
|
||||
for f in root.glob("PROGRESS*.md"):
|
||||
m = PROGRESS_RE.search(f.name)
|
||||
if m:
|
||||
found[m.group(1).upper()] = f
|
||||
# 也扫一层子目录（如 backend/、frontend/）
|
||||
for f in root.glob("*/PROGRESS*.md"):
|
||||
m = PROGRESS_RE.search(f.name)
|
||||
if m and m.group(1).upper() not in found:
|
||||
found[m.group(1).upper()] = f
|
||||
|
||||
if not found:
|
||||
print("未发现任何 PROGRESS 文件（任务尚未开始执行）。")
|
||||
return 0
|
||||
|
||||
print(f"{'ID':<5} {'PROGRESS 声明':<14} {'解析':<10} 文件")
|
||||
print("-" * 78)
|
||||
for tid in sorted(found):
|
||||
f = found[tid]
|
||||
text = f.read_text(encoding="utf-8", errors="replace")
|
||||
m = STATUS_LINE_RE.search(text)
|
||||
declared = m.group(1).lower() if m else "未识别"
|
||||
cur = state.get(tid, {}).get("status", "todo")
|
||||
print(f"{tid:<5} {declared:<14} {cur:<10} {f}")
|
||||
print("-" * 78)
|
||||
print("提示：PROGRESS 存在即表示模型已交付；请由规划者按任务卡「验收标准」核对后再 accept。")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_status(manifest: dict, state: dict, args) -> int:
|
||||
by_ms: dict[str, list[dict]] = {}
|
||||
for t in manifest["tasks"]:
|
||||
by_ms.setdefault(t["milestone"], []).append(t)
|
||||
total = len(manifest["tasks"])
|
||||
done = sum(1 for t in manifest["tasks"] if effective_status(t, state) == "done")
|
||||
doing = sum(1 for t in manifest["tasks"] if effective_status(t, state) == "doing")
|
||||
print(f"== PLANNING 看板 == 总 {total} · 完成 {done} · 执行中 {doing}\n")
|
||||
for ms in sorted(by_ms):
|
||||
tasks = by_ms[ms]
|
||||
d = sum(1 for t in tasks if effective_status(t, state) == "done")
|
||||
print(f"[{ms}] {d}/{len(tasks)}")
|
||||
for t in tasks:
|
||||
st = effective_status(t, state)
|
||||
print(f" {STATUS_ICON.get(st, '?')} {t['id']:<5} {t['title']}")
|
||||
print()
|
||||
return 0
|
||||
|
||||
|
||||
def _set_status(tid: str, manifest: dict, state: dict, status: str, note: str = "") -> int:
|
||||
t = find_task(manifest, tid)
|
||||
if not t:
|
||||
print(f"未找到任务 {tid}")
|
||||
return 1
|
||||
entry = state.get(t["id"], {})
|
||||
entry["status"] = status
|
||||
entry["updated"] = date.today().isoformat()
|
||||
if note:
|
||||
entry["note"] = note
|
||||
state[t["id"]] = entry
|
||||
save_state(state)
|
||||
print(f"{t['id']} → {status}" + (f"（{note}）" if note else ""))
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 入口
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="PLANNING 任务调度器")
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p = sub.add_parser("list", help="列出任务")
|
||||
p.add_argument("--project", help="按项目过滤（chunyu/englishdrill/dealerhub/dsp/dsh）")
|
||||
|
||||
p = sub.add_parser("show", help="显示任务卡")
|
||||
p.add_argument("id")
|
||||
|
||||
p = sub.add_parser("dispatch", help="产出派单提示词")
|
||||
p.add_argument("id")
|
||||
p.add_argument("--model", help="覆盖建议模型")
|
||||
|
||||
p = sub.add_parser("bundle", help="生成交接包")
|
||||
p.add_argument("id")
|
||||
|
||||
sub.add_parser("collect", help="扫描 PROGRESS")
|
||||
sub.add_parser("status", help="看板")
|
||||
|
||||
p = sub.add_parser("start", help="标记执行中")
|
||||
p.add_argument("id")
|
||||
p = sub.add_parser("accept", help="标记完成")
|
||||
p.add_argument("id")
|
||||
p = sub.add_parser("block", help="标记阻塞")
|
||||
p.add_argument("id")
|
||||
p.add_argument("reason", nargs="?", default="")
|
||||
p = sub.add_parser("reset", help="还原为 todo")
|
||||
p.add_argument("id")
|
||||
|
||||
args = ap.parse_args()
|
||||
manifest = load_manifest()
|
||||
state = load_state()
|
||||
|
||||
handlers = {
|
||||
"list": cmd_list, "show": cmd_show, "dispatch": cmd_dispatch,
|
||||
"bundle": cmd_bundle, "collect": cmd_collect, "status": cmd_status,
|
||||
"start": lambda m, s, a: _set_status(a.id, m, s, "doing"),
|
||||
"accept": lambda m, s, a: _set_status(a.id, m, s, "done"),
|
||||
"block": lambda m, s, a: _set_status(a.id, m, s, "blocked", a.reason),
|
||||
"reset": lambda m, s, a: _set_status(a.id, m, s, "todo"),
|
||||
}
|
||||
return handlers[args.cmd](manifest, state, args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user