#!/usr/bin/env python3 """ 安全地更新 user-story.csv 中某个用户故事的状态和链接字段。 设计目标： 1. 不破坏 CSV 的编码、引号、列顺序。 2. 按状态机校验合法跳转，避免 Agent 随意跳状态。 3. 自动更新 last_updated。 4. 支持 --dry-run 预览。 用法示例： python3 update_story_status.py \ --file docs/00-discovery/user-story.csv \ --story US-004 \ --status integ_done \ --owner backend-agent-3 \ --branch feat/US-004-create-task \ --report docs/03-dev/reports/US-004-创建任务.md \ --unit-test-status pass \ --integ-test-status pass # 强制跳转（仅在人工确认后使用） python3 update_story_status.py ... --force """ from __future__ import annotations import argparse import csv import os import sys import tempfile from datetime import datetime from pathlib import Path REQUIRED_COLUMNS = [ "story_id", "module", "title", "as_a", "i_want", "so_that", "priority", "status", "owner", "branch", "spec_ref", "details_ref", "api_ref", "db_tables", "feature_report", "e2e_manifest", "unit_test_status", "integ_test_status", "dev_url", "staging_url", "prod_url", "dev_deploy_status", "staging_deploy_status", "prod_deploy_status", "estimate_sp", "actual_sp", "dependencies", "last_updated", "notes", ] STATUSES = [ "draft", "ready", "designing", "dev", "unit_done", "integ_done", "qa_pass", "deployed_dev", "deployed_prod", "verified_prod", "blocked", "failed", "cancelled", ] # 允许的“向前”跳转。blocked/failed/cancelled 作为异常状态单独处理。 TRANSITIONS = { "draft": {"ready", "cancelled"}, "ready": {"designing", "dev", "blocked", "cancelled"}, "designing": {"dev", "blocked", "cancelled"}, "dev": {"unit_done", "failed", "blocked", "cancelled"}, "unit_done": {"integ_done", "failed", "blocked", "cancelled"}, "integ_done": {"qa_pass", "failed", "blocked", "cancelled"}, "qa_pass": {"deployed_dev", "failed", "blocked", "cancelled"}, "deployed_dev": {"deployed_prod", "failed", "blocked", "cancelled"}, "deployed_prod": {"verified_prod", "failed", "blocked"}, "verified_prod": {"failed"}, "blocked": {"ready", "designing", "dev", "failed", "cancelled"}, "failed": {"dev", "unit_done", "integ_done", "qa_pass", "blocked", "cancelled"}, "cancelled": {"ready"}, } PATH_FIELDS = ("feature_report", "e2e_manifest") def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def load_csv(path: Path): with path.open("r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) if reader.fieldnames is None: raise SystemExit(f"错误：{path} 没有表头") fieldnames = list(reader.fieldnames) rows = [dict(row) for row in reader] return fieldnames, rows def save_csv(path: Path, fieldnames: list[str], rows: list[dict]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp( dir=str(path.parent), prefix=path.name + ".", suffix=".tmp" ) os.close(fd) try: with open(tmp_name, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) os.replace(tmp_name, path) except Exception: try: os.unlink(tmp_name) except OSError: pass raise def parse_args(argv: list[str] | None = None) -> argparse.Namespace: p = argparse.ArgumentParser(description="更新 user-story.csv 中的故事状态") p.add_argument("--file", required=True, type=Path, help="CSV 文件路径") p.add_argument("--story", required=True, help="故事 ID，例如 US-004") p.add_argument("--status", required=True, choices=STATUSES, help="新状态") p.add_argument("--owner", help="负责人/Agent") p.add_argument("--branch", help="Git 分支") p.add_argument("--report", help="功能实现报告路径") p.add_argument("--e2e-manifest", help="E2E manifest 路径") p.add_argument("--unit-test-status", choices=["pending", "pass", "fail"], help="单元测试状态") p.add_argument("--integ-test-status", choices=["pending", "pass", "fail"], help="集成测试状态") p.add_argument("--dev-deploy-status", choices=["not_deployed", "deployed", "failed"], help="开发环境部署状态") p.add_argument("--staging-deploy-status", choices=["not_deployed", "deployed", "failed"], help="预发环境部署状态") p.add_argument("--prod-deploy-status", choices=["not_deployed", "deployed", "failed"], help="生产环境部署状态") p.add_argument("--notes", help="替换备注") p.add_argument("--append-note", help="追加备注（带时间戳）") p.add_argument("--force", action="store_true", help="跳过状态机校验") p.add_argument("--dry-run", action="store_true", help="只打印，不写文件") p.add_argument("--check-paths", action="store_true", help="检查 report/manifest 路径是否存在") return p.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if not args.file.exists(): print(f"错误：文件不存在：{args.file}", file=sys.stderr) return 2 fieldnames, rows = load_csv(args.file) missing_cols = [c for c in REQUIRED_COLUMNS if c not in fieldnames] if missing_cols: print("错误：CSV 缺少列：" + ", ".join(missing_cols), file=sys.stderr) return 2 targets = [r for r in rows if r.get("story_id", "").strip() == args.story] if not targets: print(f"错误：找不到故事 {args.story}", file=sys.stderr) return 3 if len(targets) > 1: print(f"错误：故事 {args.story} 有重复行，请先修复 CSV", file=sys.stderr) return 3 row = targets[0] old_status = (row.get("status") or "").strip() if old_status == args.status: pass # 幂等，允许重复设置 elif not args.force and args.status not in TRANSITIONS.get(old_status, set()): print( f"错误：不允许从 {old_status!r} 跳到 {args.status!r}。\n" f"允许的目标：{sorted(TRANSITIONS.get(old_status, set()))}\n" f"如确需跳转，请人工确认后加 --force。", file=sys.stderr, ) return 4 updates: dict[str, str] = {"status": args.status, "last_updated": now_iso()} mapping = { "owner": args.owner, "branch": args.branch, "feature_report": args.report, "e2e_manifest": args.e2e_manifest, "unit_test_status": args.unit_test_status, "integ_test_status": args.integ_test_status, "dev_deploy_status": args.dev_deploy_status, "staging_deploy_status": args.staging_deploy_status, "prod_deploy_status": args.prod_deploy_status, "notes": args.notes, } for col, value in mapping.items(): if value is not None: updates[col] = value if args.append_note: old_note = (row.get("notes") or "").strip() stamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M") new_note = f"{old_note} | [{stamp}] {args.append_note}" if old_note else f"[{stamp}] {args.append_note}" updates["notes"] = new_note if args.check_paths: for col in PATH_FIELDS: value = updates.get(col, row.get(col, "")).strip() if value and not Path(value).exists(): print(f"警告：{col} 路径不存在：{value}", file=sys.stderr) print(f"故事：{args.story}") print(f"状态：{old_status} -> {args.status}") for col, value in updates.items(): if col == "last_updated": continue old = row.get(col, "") if old != value: print(f" {col}: {old!r} -> {value!r}") if args.dry_run: print("（dry-run，未写文件）") return 0 row.update(updates) save_csv(args.file, fieldnames, rows) print(f"已写入：{args.file}") return 0 if __name__ == "__main__": raise SystemExit(main())