✨ feat(plan_progress): track plan status in progress.md
This commit is contained in:
+156
-119
@@ -1,29 +1,31 @@
|
||||
#!/usr/bin/env python3
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
PLAN_PREFIX = "[PLAN]"
|
||||
PLAN_SECTION_HEADER = "## Plan 状态记录"
|
||||
PLAN_STATUS_START = "<!-- plan-status:start -->"
|
||||
PLAN_STATUS_END = "<!-- plan-status:end -->"
|
||||
PLAN_FILE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})-.+\.md$")
|
||||
VALID_STATUSES = {"in-progress", "done", "blocked"}
|
||||
PLAN_LINE_RE = re.compile(
|
||||
r"^- \[(?P<check>[ xX])\] `(?P<plan>[^`]+)` (?P<status>done|blocked|pending)(?:: (?P<note>.*))?$"
|
||||
)
|
||||
VALID_STATUSES = {"done", "blocked", "pending"}
|
||||
|
||||
|
||||
def usage() -> str:
|
||||
return (
|
||||
"Usage:\\n"
|
||||
" python scripts/plan_progress.py select -plans <dir> -progress <file>\\n"
|
||||
" python scripts/plan_progress.py record -plan <path> -status <status> -progress <file> [-note <text>]\\n"
|
||||
" python scripts/plan_progress.py -h\\n"
|
||||
"Options:\\n"
|
||||
" -plans DIR\\n"
|
||||
" -plan PATH\\n"
|
||||
" -status in-progress|done|blocked\\n"
|
||||
" -progress FILE\\n"
|
||||
" -note TEXT\\n"
|
||||
" -h, -help Show this help.\\n"
|
||||
"Usage:\n"
|
||||
" python scripts/plan_progress.py select -plans <dir> -progress <file>\n"
|
||||
" python scripts/plan_progress.py record -plan <path> -status <status> -progress <file> [-note <text>]\n"
|
||||
" python scripts/plan_progress.py -h\n"
|
||||
"Options:\n"
|
||||
" -plans DIR\n"
|
||||
" -plan PATH\n"
|
||||
" -status done|blocked|pending\n"
|
||||
" -progress FILE\n"
|
||||
" -note TEXT\n"
|
||||
" -h, -help Show this help.\n"
|
||||
)
|
||||
|
||||
|
||||
@@ -43,102 +45,123 @@ def parse_flags(args: list[str]) -> dict[str, str]:
|
||||
return flags
|
||||
|
||||
|
||||
def normalize_plan_key(plan_value: str, cwd: Path) -> str:
|
||||
try:
|
||||
return Path(plan_value).resolve().relative_to(cwd.resolve()).as_posix()
|
||||
except ValueError:
|
||||
return Path(plan_value).as_posix()
|
||||
def normalize_plan_key(plan_value: str) -> str:
|
||||
raw = plan_value.strip().replace("\\", "/")
|
||||
raw = raw.lstrip("./")
|
||||
if raw.startswith("docs/plans/"):
|
||||
return raw[len("docs/plans/") :]
|
||||
marker = "/docs/plans/"
|
||||
if marker in raw:
|
||||
return raw.split(marker, 1)[1]
|
||||
return raw
|
||||
|
||||
|
||||
def load_plan_records(progress_path: Path, cwd: Path) -> dict[str, str]:
|
||||
if not progress_path.exists():
|
||||
return {}
|
||||
text = progress_path.read_text(encoding="utf-8")
|
||||
records: dict[str, str] = {}
|
||||
for line in text.splitlines():
|
||||
if not line.startswith(PLAN_PREFIX):
|
||||
continue
|
||||
payload = line[len(PLAN_PREFIX) :].strip()
|
||||
if not payload:
|
||||
continue
|
||||
segments = [seg.strip() for seg in payload.split("|")]
|
||||
if not segments:
|
||||
continue
|
||||
plan_path = segments[0]
|
||||
status = None
|
||||
for seg in segments[1:]:
|
||||
if "=" not in seg:
|
||||
continue
|
||||
key, value = seg.split("=", 1)
|
||||
if key.strip() == "status":
|
||||
status = value.strip()
|
||||
if not plan_path or status is None:
|
||||
continue
|
||||
records[normalize_plan_key(plan_path, cwd)] = status
|
||||
return records
|
||||
|
||||
|
||||
def list_plan_files(plans_dir: Path, cwd: Path) -> list[tuple[str, Path, str]]:
|
||||
entries: list[tuple[str, Path, str]] = []
|
||||
for path in plans_dir.iterdir():
|
||||
if not path.is_file():
|
||||
continue
|
||||
match = PLAN_FILE_RE.match(path.name)
|
||||
if not match:
|
||||
continue
|
||||
date_value = match.group(1)
|
||||
try:
|
||||
rel = path.resolve().relative_to(cwd.resolve()).as_posix()
|
||||
except ValueError:
|
||||
rel = path.as_posix()
|
||||
entries.append((date_value, path, rel))
|
||||
return entries
|
||||
|
||||
|
||||
def select_plan(plans_dir: Path, progress_path: Path) -> tuple[int, str]:
|
||||
cwd = Path.cwd()
|
||||
if not plans_dir.is_dir():
|
||||
return 2, f"ERROR: plans dir not found: {plans_dir}"
|
||||
plans = list_plan_files(plans_dir, cwd)
|
||||
if not plans:
|
||||
return 2, "ERROR: no plan files found"
|
||||
|
||||
records = load_plan_records(progress_path, cwd)
|
||||
|
||||
in_progress = [item for item in plans if records.get(item[2]) == "in-progress"]
|
||||
if in_progress:
|
||||
in_progress.sort(key=lambda item: (item[0], item[2]))
|
||||
return 0, in_progress[-1][2]
|
||||
|
||||
pending = [
|
||||
item
|
||||
for item in plans
|
||||
if records.get(item[2]) not in ("done", "blocked")
|
||||
]
|
||||
if not pending:
|
||||
return 2, "ERROR: no pending plans"
|
||||
|
||||
pending.sort(key=lambda item: (item[0], item[2]))
|
||||
return 0, pending[-1][2]
|
||||
|
||||
|
||||
def ensure_plan_section(text: str) -> str:
|
||||
if PLAN_SECTION_HEADER in text:
|
||||
return text
|
||||
suffix = text
|
||||
if suffix and not suffix.endswith("\n"):
|
||||
suffix += "\n"
|
||||
if suffix:
|
||||
suffix += "\n"
|
||||
suffix += PLAN_SECTION_HEADER + "\n"
|
||||
return suffix
|
||||
def render_plan_line(plan_key: str, status: str, note: Optional[str]) -> str:
|
||||
checked = "x" if status == "done" else " "
|
||||
if status == "blocked":
|
||||
suffix = "blocked"
|
||||
if note:
|
||||
suffix += f": {note}"
|
||||
elif status == "pending":
|
||||
suffix = "pending"
|
||||
else:
|
||||
suffix = "done"
|
||||
return f"- [{checked}] `{plan_key}` {suffix}"
|
||||
|
||||
|
||||
def normalize_note(note: str) -> str:
|
||||
cleaned = note.replace("\n", " ").replace("|", " ").strip()
|
||||
cleaned = note.replace("\n", " ").replace("\r", " ").replace("`", "'").strip()
|
||||
return cleaned
|
||||
|
||||
|
||||
def list_plan_files(plans_dir: Path) -> list[str]:
|
||||
entries: list[str] = []
|
||||
for path in plans_dir.iterdir():
|
||||
if not path.is_file():
|
||||
continue
|
||||
if not PLAN_FILE_RE.match(path.name):
|
||||
continue
|
||||
try:
|
||||
rel = path.resolve().relative_to(plans_dir.resolve()).as_posix()
|
||||
except ValueError:
|
||||
rel = path.name
|
||||
entries.append(rel)
|
||||
return sorted(entries)
|
||||
|
||||
|
||||
def find_block(lines: list[str]) -> Optional[tuple[int, int]]:
|
||||
start_idx = None
|
||||
for idx, line in enumerate(lines):
|
||||
if line.strip() == PLAN_STATUS_START:
|
||||
start_idx = idx
|
||||
break
|
||||
if start_idx is None:
|
||||
return None
|
||||
for idx in range(start_idx + 1, len(lines)):
|
||||
if lines[idx].strip() == PLAN_STATUS_END:
|
||||
return start_idx, idx
|
||||
return None
|
||||
|
||||
|
||||
def parse_entries(lines: list[str], start_idx: int, end_idx: int) -> list[tuple[str, str, Optional[str], int]]:
|
||||
entries: list[tuple[str, str, Optional[str], int]] = []
|
||||
for idx in range(start_idx + 1, end_idx):
|
||||
line = lines[idx].strip()
|
||||
match = PLAN_LINE_RE.match(line)
|
||||
if not match:
|
||||
continue
|
||||
plan_key = normalize_plan_key(match.group("plan"))
|
||||
status = match.group("status")
|
||||
note = match.group("note")
|
||||
entries.append((plan_key, status, note, idx))
|
||||
return entries
|
||||
|
||||
|
||||
def render_progress_lines(plans: list[str]) -> list[str]:
|
||||
lines = ["# Plan 状态", "", PLAN_STATUS_START]
|
||||
for plan_key in plans:
|
||||
lines.append(render_plan_line(plan_key, "pending", None))
|
||||
lines.append(PLAN_STATUS_END)
|
||||
return lines
|
||||
|
||||
|
||||
def select_plan(plans_dir: Path, progress_path: Path) -> tuple[int, str]:
|
||||
if not plans_dir.is_dir():
|
||||
return 2, f"ERROR: plans dir not found: {plans_dir}"
|
||||
plan_keys = list_plan_files(plans_dir)
|
||||
if not plan_keys:
|
||||
return 2, "ERROR: no plan files found"
|
||||
|
||||
progress_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if progress_path.exists():
|
||||
lines = progress_path.read_text(encoding="utf-8").splitlines()
|
||||
else:
|
||||
lines = []
|
||||
|
||||
block = find_block(lines)
|
||||
if not block:
|
||||
lines = render_progress_lines(plan_keys)
|
||||
progress_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
return 0, (plans_dir / plan_keys[0]).as_posix()
|
||||
|
||||
start_idx, end_idx = block
|
||||
entries = parse_entries(lines, start_idx, end_idx)
|
||||
existing = {plan for plan, _, _, _ in entries}
|
||||
missing = [plan for plan in plan_keys if plan not in existing]
|
||||
if missing:
|
||||
insert_lines = [render_plan_line(plan, "pending", None) for plan in missing]
|
||||
lines[end_idx:end_idx] = insert_lines
|
||||
end_idx += len(insert_lines)
|
||||
progress_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
entries = parse_entries(lines, start_idx, end_idx)
|
||||
|
||||
for plan_key, status, _, _ in entries:
|
||||
if status == "pending":
|
||||
return 0, (plans_dir / plan_key).as_posix()
|
||||
|
||||
return 2, "ERROR: no pending plans"
|
||||
|
||||
|
||||
def record_status(plan: str, status: str, progress_path: Path, note: Optional[str]) -> tuple[int, str]:
|
||||
if status not in VALID_STATUSES:
|
||||
return 2, f"ERROR: invalid status: {status}"
|
||||
@@ -147,24 +170,38 @@ def record_status(plan: str, status: str, progress_path: Path, note: Optional[st
|
||||
|
||||
progress_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if progress_path.exists():
|
||||
text = progress_path.read_text(encoding="utf-8")
|
||||
lines = progress_path.read_text(encoding="utf-8").splitlines()
|
||||
else:
|
||||
text = "# 开发进度追踪\n"
|
||||
lines = []
|
||||
|
||||
text = ensure_plan_section(text)
|
||||
if not text.endswith("\n"):
|
||||
text += "\n"
|
||||
plan_key = normalize_plan_key(plan)
|
||||
block = find_block(lines)
|
||||
if not block:
|
||||
lines = render_progress_lines([plan_key])
|
||||
block = find_block(lines)
|
||||
if not block:
|
||||
return 2, "ERROR: failed to create plan status block"
|
||||
|
||||
date_value = datetime.now().strftime("%Y-%m-%d")
|
||||
plan_path = Path(plan).as_posix()
|
||||
line = f"{PLAN_PREFIX} {plan_path} | status={status} | date={date_value}"
|
||||
if note:
|
||||
cleaned = normalize_note(note)
|
||||
if cleaned:
|
||||
line += f" | note={cleaned}"
|
||||
text += line + "\n"
|
||||
progress_path.write_text(text, encoding="utf-8")
|
||||
return 0, line
|
||||
start_idx, end_idx = block
|
||||
entries = parse_entries(lines, start_idx, end_idx)
|
||||
|
||||
rendered_note = None
|
||||
if status == "blocked" and note:
|
||||
rendered_note = normalize_note(note)
|
||||
|
||||
updated_line = render_plan_line(plan_key, status, rendered_note)
|
||||
updated = False
|
||||
for entry_plan, _, _, idx in entries:
|
||||
if entry_plan == plan_key:
|
||||
lines[idx] = updated_line
|
||||
updated = True
|
||||
break
|
||||
|
||||
if not updated:
|
||||
lines[end_idx:end_idx] = [updated_line]
|
||||
|
||||
progress_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
return 0, updated_line
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
@@ -222,4 +259,4 @@ def main(argv: list[str]) -> int:
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
|
||||
Reference in New Issue
Block a user