324 lines
12 KiB
Python
324 lines
12 KiB
Python
#!/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())
|