#!/usr/bin/env python3 """Shared ship-id and YAML helpers. Zero third-party deps.""" from __future__ import annotations import re import sys from datetime import date, datetime, timezone from pathlib import Path from typing import Any _REPO_ROOT = Path(__file__).resolve().parents[3] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from pouch.yaml_io import safe_dump, safe_load # noqa: E402 SHIP_ID_RE = re.compile( r"^(?:[a-z][a-z0-9]*(?:-[a-z0-9]+)*|\d{8}-[a-z][a-z0-9]*(?:-[a-z0-9]+)*)$" ) DATED_SHIP_ID_RE = re.compile(r"^\d{8}-[a-z][a-z0-9]*(?:-[a-z0-9]+)*$") DATE_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$|^\d{8}$") KIND = frozenset({"short", "long"}) SHIP_STATUS = frozenset( { "draft", "spec_review", "approved", "in_progress", "blocked", "done", "parked", "superseded", } ) TASK_STATUS = frozenset( {"open", "in_progress", "blocked", "verified", "merged", "done"} ) DELIVERY_STATE = frozenset( { "change_ready", "verified", "pr_opened", "merged", "regression_passed", "artifact_built", "published", "deployed", } ) # User-facing stop points. The two legacy names remain accepted so existing # ship records continue to resolve to the new phase meanings. STOP_AT_STATE = { "develop_verified": "verified", "review_ready": "pr_opened", "release_ready": "regression_passed", "released": "published", "deployed": "deployed", "verified": "verified", # legacy alias for develop_verified "validation_ready": "regression_passed", # legacy alias for release_ready } STOP_AT = frozenset(STOP_AT_STATE) SOURCE_KIND = frozenset({"user", "hunt", "ack", "pr", "feishu"}) COMMIT_ROLE = frozenset({"baseline", "merged", "head"}) # These directories are the human-readable audit trail for the orchestration # phases. The step list is intentionally kept here so create_ship and # next_step cannot drift apart. STAGE_DEFINITIONS = ( ("1-spec", "规格与审批", ("write_spec", "specifier_new", "specifier_review", "specifier_handoff", "materialize_tasks", "hunt")), ("2-develop", "开发", ("implement", "implement_task",)), ("3-test", "验证与回归", ("verify", "verify_task", "regress")), ("4-review", "PR 与合并", ("open_pr", "check_pr", "fix_pr", "merge_pr")), ("5-build", "构建", ("obtain",)), ("6-publish", "发布", ("publish",)), ("7-deploy", "部署", ("deploy",)), ) STAGE_BY_STEP = { step: (directory, title) for directory, title, steps in STAGE_DEFINITIONS for step in steps } def skill_dir() -> Path: return Path(__file__).resolve().parents[1] def load_yaml(path: Path) -> Any: return safe_load(path.read_text(encoding="utf-8")) def dump_yaml(path: Path, data: Any) -> None: path.write_text(safe_dump(data), encoding="utf-8") def check_ship_id(ship_id: str) -> str | None: if not ship_id: return "shipId 为空" if len(ship_id) > 64: return f"shipId 超过 64 字符: {ship_id}" if DATE_ID_RE.fullmatch(ship_id): return f"shipId 不能是日期: {ship_id}" if not SHIP_ID_RE.fullmatch(ship_id): return ( f"shipId 无效: {ship_id!r}（小写字母开头，仅小写字母、数字、连字符）" ) return None def check_dated_ship_id(ship_id: str) -> str | None: """Validate the convention for newly-created records. check_ship_id remains backward compatible so existing ship records can be inspected. New entry points should use this stricter check. """ err = check_ship_id(ship_id) if err: return err if not DATED_SHIP_ID_RE.fullmatch(ship_id): return ( f"新 ship-id 必须是 YYYYMMDD-短语格式: {ship_id!r}" ) return None def dated_ship_id(phrase: str, *, today: date | None = None) -> str: """Build a sortable ship id from a short lowercase phrase.""" phrase = re.sub(r"[^a-z0-9]+", "-", phrase.strip().lower()).strip("-") if not phrase: raise ValueError("ship-id 短语不能为空") value = f"{(today or date.today()):%Y%m%d}-{phrase}" err = check_dated_ship_id(value) if err: raise ValueError(err) return value def stage_for_step(step: str | None) -> tuple[str, str] | None: return STAGE_BY_STEP.get(str(step)) if step else None def ensure_stage_layout(ship_dir: Path) -> None: """Create the visible phase folders and their small status manifests.""" for order, (directory, title, steps) in enumerate(STAGE_DEFINITIONS, start=1): stage_dir = ship_dir / directory stage_dir.mkdir(parents=True, exist_ok=True) manifest = stage_dir / "stage.yaml" if not manifest.exists(): manifest.write_text( safe_dump({ "order": order, "stage": directory, "title": title, "status": "pending", "steps": list(steps), "startedAt": None, "finishedAt": None, }), encoding="utf-8", ) activity = stage_dir / "activity.md" if not activity.exists(): activity.write_text( f"# {directory} {title}\n\n状态：pending\n\n" "## 活动记录\n\n", encoding="utf-8", ) def _stage_manifest(path: Path) -> dict[str, Any]: try: value = safe_load(path.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 value = None return value if isinstance(value, dict) else {} def record_stage_event( ship_dir: Path, *, step: str | None, status: str, summary: str = "", task_id: str | None = None, ) -> None: """Append one dispatch/result event to the visible stage audit trail.""" resolved = stage_for_step(step) if resolved is None: return ensure_stage_layout(ship_dir) directory, title = resolved stage_dir = ship_dir / directory manifest_path = stage_dir / "stage.yaml" manifest = _stage_manifest(manifest_path) now = datetime.now(timezone.utc).isoformat(timespec="seconds") event_key = f"{step}|{status}|{task_id or ''}|{summary.strip()}" if manifest.get("lastEventKey") == event_key: return manifest["lastEventKey"] = event_key if status in {"ready", "in_progress"}: if manifest.get("status") == "pending": manifest["startedAt"] = now manifest["status"] = "in_progress" elif status in {"pass", "fail", "blocked", "ask"}: manifest["lastResult"] = status manifest["lastStep"] = step if status == "pass": manifest["status"] = "completed" manifest["finishedAt"] = now elif status in {"fail", "blocked"}: manifest["status"] = "blocked" manifest_path.write_text(safe_dump(manifest), encoding="utf-8") line = f"- `{now}` `{step}` status=`{status}`" if task_id: line += f" task=`{task_id}`" if summary: line += f"：{summary.strip()}" with (stage_dir / "activity.md").open("a", encoding="utf-8") as handle: handle.write(line + "\n")