1453 lines
52 KiB
Python
1453 lines
52 KiB
Python
#!/usr/bin/env python3
|
||
"""按项目类型物化 Cursor Rules、Skills 与运行配置。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from pathlib import Path, PurePosixPath
|
||
import sys
|
||
import tempfile
|
||
from typing import Any, Iterable
|
||
|
||
|
||
SCHEMA_VERSION = 1
|
||
GITIGNORE_START = "# >>> cursor-init managed >>>"
|
||
GITIGNORE_END = "# <<< cursor-init managed <<<"
|
||
GITATTRIBUTES_START = "# >>> cursor-init managed >>>"
|
||
GITATTRIBUTES_END = "# <<< cursor-init managed <<<"
|
||
SENTINEL_PATH = ".cursor/.init-done"
|
||
STATE_PATH = ".cursor/.init-state.json"
|
||
SCAN_CONFIG_PATH = ".cursor/hooks/changelog-scan.json"
|
||
REGISTRY_PATH = ".cursor/skills/epee-orchestrator/registry.md"
|
||
LOCAL_ENV_EXAMPLE_PATH = ".cursor/local-env.example.json"
|
||
LOCAL_ENV_PATH = ".cursor/local-env.json"
|
||
|
||
|
||
class InitError(RuntimeError):
|
||
"""可向用户直接展示的初始化错误。"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TypeEntry:
|
||
id: str
|
||
label: str
|
||
status: str
|
||
manifest_path: Path
|
||
manifest: dict[str, Any]
|
||
|
||
|
||
@dataclass
|
||
class Context:
|
||
root: Path
|
||
bootstrap: Path
|
||
catalog_path: Path
|
||
catalog: dict[str, Any]
|
||
common_path: Path
|
||
common: dict[str, Any]
|
||
skills_path: Path
|
||
skills: dict[str, dict[str, Any]]
|
||
types: dict[str, TypeEntry]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FileSpec:
|
||
source: Path
|
||
target: str
|
||
owner: str
|
||
|
||
|
||
@dataclass
|
||
class Action:
|
||
operation: str
|
||
path: str
|
||
reason: str
|
||
content: bytes | None = None
|
||
mode: int | None = None
|
||
redact_digest: bool = False
|
||
|
||
def digest_record(self) -> dict[str, str]:
|
||
record = {
|
||
"operation": self.operation,
|
||
"path": self.path,
|
||
"reason": self.reason,
|
||
}
|
||
if self.content is not None:
|
||
record["sha256"] = (
|
||
"<local-sensitive-content>"
|
||
if self.redact_digest
|
||
else hashlib.sha256(self.content).hexdigest()
|
||
)
|
||
if self.mode is not None:
|
||
record["mode"] = oct(self.mode)
|
||
return record
|
||
|
||
|
||
@dataclass
|
||
class Plan:
|
||
project_type: str
|
||
label: str
|
||
manifest_digest: str
|
||
bundle_version: int
|
||
gitignore_generated: bool
|
||
actions: list[Action] = field(default_factory=list)
|
||
blockers: list[str] = field(default_factory=list)
|
||
foreign: list[str] = field(default_factory=list)
|
||
required_inputs: list[dict[str, Any]] = field(default_factory=list)
|
||
|
||
@property
|
||
def digest(self) -> str:
|
||
payload = {
|
||
"project_type": self.project_type,
|
||
"manifest_digest": self.manifest_digest,
|
||
"bundle_version": self.bundle_version,
|
||
"gitignore_generated": self.gitignore_generated,
|
||
"actions": [action.digest_record() for action in self.actions],
|
||
"blockers": sorted(self.blockers),
|
||
"foreign": sorted(self.foreign),
|
||
}
|
||
encoded = json.dumps(
|
||
payload,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
).encode("utf-8")
|
||
return hashlib.sha256(encoded).hexdigest()
|
||
|
||
|
||
def _read_json(path: Path) -> dict[str, Any]:
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except FileNotFoundError as exc:
|
||
raise InitError(f"缺少 JSON 文件：{path}") from exc
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||
raise InitError(f"JSON 无法解析：{path}：{exc}") from exc
|
||
if not isinstance(data, dict):
|
||
raise InitError(f"JSON 顶层必须是对象：{path}")
|
||
return data
|
||
|
||
|
||
def _safe_relative(value: str, field_name: str) -> str:
|
||
if not isinstance(value, str) or not value.strip():
|
||
raise InitError(f"{field_name} 必须是非空相对路径")
|
||
normalized = value.replace("\\", "/")
|
||
path = PurePosixPath(normalized)
|
||
if path.is_absolute() or ".." in path.parts:
|
||
raise InitError(f"{field_name} 不能越出目标根目录：{value}")
|
||
return path.as_posix()
|
||
|
||
|
||
def _target_path(root: Path, relative: str) -> Path:
|
||
safe = _safe_relative(relative, "target")
|
||
candidate = (root / PurePosixPath(safe)).resolve()
|
||
root_resolved = root.resolve()
|
||
try:
|
||
candidate.relative_to(root_resolved)
|
||
except ValueError as exc:
|
||
raise InitError(f"目标路径越界：{relative}") from exc
|
||
return candidate
|
||
|
||
|
||
def _source_path(manifest_path: Path, relative: str) -> Path:
|
||
safe = _safe_relative(relative, "source")
|
||
candidate = (manifest_path.parent / PurePosixPath(safe)).resolve()
|
||
manifest_root = manifest_path.parent.resolve()
|
||
try:
|
||
candidate.relative_to(manifest_root)
|
||
except ValueError as exc:
|
||
raise InitError(f"payload 路径越界：{relative}") from exc
|
||
return candidate
|
||
|
||
|
||
def _require_schema(data: dict[str, Any], path: Path) -> None:
|
||
if data.get("schema_version") != SCHEMA_VERSION:
|
||
raise InitError(
|
||
f"不支持的 schema_version：{path}（期望 {SCHEMA_VERSION}）"
|
||
)
|
||
|
||
|
||
def _list_of_strings(data: dict[str, Any], key: str, path: Path) -> list[str]:
|
||
value = data.get(key, [])
|
||
if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
|
||
raise InitError(f"{path} 的 {key} 必须是字符串数组")
|
||
return value
|
||
|
||
|
||
def _validate_scan(scan: Any, path: Path) -> None:
|
||
if not isinstance(scan, dict):
|
||
raise InitError(f"{path} 的 scan 必须是对象")
|
||
for key in ("included_names", "included_extensions", "excluded_directories"):
|
||
value = scan.get(key, [])
|
||
if not isinstance(value, list) or not all(
|
||
isinstance(item, str) and item for item in value
|
||
):
|
||
raise InitError(f"{path} 的 scan.{key} 必须是非空字符串数组")
|
||
|
||
|
||
def _validate_env_fields(fields: Any, path: Path) -> None:
|
||
if not isinstance(fields, list):
|
||
raise InitError(f"{path} 的 local_env_fields 必须是数组")
|
||
seen: set[str] = set()
|
||
for item in fields:
|
||
if not isinstance(item, dict):
|
||
raise InitError(f"{path} 的 local_env_fields 条目必须是对象")
|
||
key = item.get("key")
|
||
if not isinstance(key, str) or not key or key in seen:
|
||
raise InitError(f"{path} 包含无效或重复的环境字段：{key!r}")
|
||
seen.add(key)
|
||
if not isinstance(item.get("required", False), bool):
|
||
raise InitError(f"{path} 的环境字段 required 必须是布尔值：{key}")
|
||
for flag in ("sensitive", "collect_at_init"):
|
||
if not isinstance(item.get(flag, False if flag == "sensitive" else True), bool):
|
||
raise InitError(f"{path} 的环境字段 {flag} 必须是布尔值：{key}")
|
||
if item.get("sensitive", False):
|
||
default = str(item.get("default", ""))
|
||
if default and not default.startswith("<"):
|
||
raise InitError(f"{path} 的敏感环境字段不得包含真实 default：{key}")
|
||
example = str(item.get("example", ""))
|
||
placeholder_wrappers = (("<", ">"), ("${", "}"), ("{{", "}}"))
|
||
if example and not any(
|
||
example.startswith(start) and example.endswith(end)
|
||
for start, end in placeholder_wrappers
|
||
):
|
||
raise InitError(
|
||
f"{path} 的敏感环境字段 example 必须是占位值：{key}"
|
||
)
|
||
|
||
|
||
def _validate_manifest(
|
||
manifest_path: Path,
|
||
manifest: dict[str, Any],
|
||
*,
|
||
require_payload: bool,
|
||
) -> None:
|
||
_require_schema(manifest, manifest_path)
|
||
manifest_id = manifest.get("id")
|
||
if not isinstance(manifest_id, str) or not manifest_id:
|
||
raise InitError(f"{manifest_path} 缺少有效 id")
|
||
|
||
files = manifest.get("files", [])
|
||
trees = manifest.get("trees", [])
|
||
if not isinstance(files, list) or not isinstance(trees, list):
|
||
raise InitError(f"{manifest_path} 的 files/trees 必须是数组")
|
||
readiness = manifest.get("readiness", {})
|
||
if not isinstance(readiness, dict):
|
||
raise InitError(f"{manifest_path} 的 readiness 必须是对象")
|
||
|
||
seen_targets: set[str] = set()
|
||
for item in files:
|
||
if not isinstance(item, dict):
|
||
raise InitError(f"{manifest_path} 的 files 条目必须是对象")
|
||
source = _source_path(manifest_path, item.get("source", ""))
|
||
target = _safe_relative(item.get("target", ""), "target")
|
||
if target in seen_targets:
|
||
raise InitError(f"{manifest_path} 重复声明 target：{target}")
|
||
seen_targets.add(target)
|
||
if require_payload and not source.is_file():
|
||
raise InitError(f"ready bundle 缺少 payload 文件：{source}")
|
||
|
||
for item in trees:
|
||
if not isinstance(item, dict):
|
||
raise InitError(f"{manifest_path} 的 trees 条目必须是对象")
|
||
source = _source_path(manifest_path, item.get("source", ""))
|
||
_safe_relative(item.get("target", ""), "target")
|
||
if require_payload and not source.is_dir():
|
||
raise InitError(f"ready bundle 缺少 payload 目录：{source}")
|
||
|
||
_validate_env_fields(manifest.get("local_env_fields", []), manifest_path)
|
||
_validate_scan(manifest.get("scan", {}), manifest_path)
|
||
_list_of_strings(manifest, "skills", manifest_path)
|
||
|
||
for key in ("gitignore_fragments", "gitattributes_fragments"):
|
||
for relative in _list_of_strings(manifest, key, manifest_path):
|
||
fragment = _source_path(manifest_path, relative)
|
||
if require_payload and not fragment.is_file():
|
||
raise InitError(f"ready bundle 缺少片段：{fragment}")
|
||
|
||
if require_payload:
|
||
if not readiness.get("allow_empty_payload", False) and not files and not trees:
|
||
raise InitError(f"ready bundle 不允许空 payload：{manifest_path}")
|
||
if readiness.get("require_scan", False):
|
||
scan = manifest.get("scan", {})
|
||
if not scan.get("included_names") and not scan.get("included_extensions"):
|
||
raise InitError(f"ready bundle 缺少扫描策略：{manifest_path}")
|
||
if readiness.get("require_gitignore", False) and not manifest.get(
|
||
"gitignore_fragments"
|
||
):
|
||
raise InitError(f"ready bundle 缺少 gitignore 片段：{manifest_path}")
|
||
|
||
|
||
def _validate_common_operations(manifest_path: Path, manifest: dict[str, Any]) -> None:
|
||
reset_files = manifest.get("reset_files", [])
|
||
if not isinstance(reset_files, list):
|
||
raise InitError(f"{manifest_path} 的 reset_files 必须是数组")
|
||
seen_targets: set[str] = set()
|
||
for item in reset_files:
|
||
if not isinstance(item, dict):
|
||
raise InitError(f"{manifest_path} 的 reset_files 条目必须是对象")
|
||
source = _source_path(manifest_path, item.get("source", ""))
|
||
target = _safe_relative(item.get("target", ""), "reset target")
|
||
if target in seen_targets:
|
||
raise InitError(f"{manifest_path} 重复声明 reset target：{target}")
|
||
if not source.is_file():
|
||
raise InitError(f"reset 模板不存在：{source}")
|
||
seen_targets.add(target)
|
||
for pattern in _list_of_strings(manifest, "cleanup_globs", manifest_path):
|
||
_safe_relative(pattern, "cleanup_globs")
|
||
|
||
|
||
def load_context(root: Path) -> Context:
|
||
root = root.resolve()
|
||
bootstrap = root / ".cursor" / "bootstrap"
|
||
catalog_path = bootstrap / "catalog.json"
|
||
catalog = _read_json(catalog_path)
|
||
_require_schema(catalog, catalog_path)
|
||
|
||
common_path = bootstrap / _safe_relative(
|
||
catalog.get("common_manifest", ""),
|
||
"common_manifest",
|
||
)
|
||
skills_path = bootstrap / _safe_relative(
|
||
catalog.get("skills_catalog", ""),
|
||
"skills_catalog",
|
||
)
|
||
common = _read_json(common_path)
|
||
skills_data = _read_json(skills_path)
|
||
_require_schema(common, common_path)
|
||
_require_schema(skills_data, skills_path)
|
||
if common.get("id") != "common":
|
||
raise InitError(f"{common_path} 的 id 必须是 common")
|
||
_validate_common_operations(common_path, common)
|
||
_validate_env_fields(common.get("local_env_fields", []), common_path)
|
||
_validate_scan(common.get("scan", {}), common_path)
|
||
|
||
canonical_files = _list_of_strings(common, "canonical_files", common_path)
|
||
canonical_normalized = [
|
||
_safe_relative(path, "canonical_files") for path in canonical_files
|
||
]
|
||
if len(set(canonical_normalized)) != len(canonical_normalized):
|
||
raise InitError(f"{common_path} 包含重复 canonical_files")
|
||
for relative in _list_of_strings(common, "bootstrap_files", common_path):
|
||
path = _source_path(common_path, relative)
|
||
if not path.is_file():
|
||
raise InitError(f"bootstrap 文件不存在：{path}")
|
||
_list_of_strings(common, "skills", common_path)
|
||
for key in ("gitignore_fragments", "gitattributes_fragments"):
|
||
for relative in _list_of_strings(common, key, common_path):
|
||
fragment = _source_path(common_path, relative)
|
||
if not fragment.is_file():
|
||
raise InitError(f"common 片段不存在：{fragment}")
|
||
|
||
raw_skills = skills_data.get("skills")
|
||
if not isinstance(raw_skills, list):
|
||
raise InitError(f"{skills_path} 的 skills 必须是数组")
|
||
skills: dict[str, dict[str, Any]] = {}
|
||
targets: set[str] = set()
|
||
for item in raw_skills:
|
||
if not isinstance(item, dict):
|
||
raise InitError(f"{skills_path} 的 Skill 条目必须是对象")
|
||
skill_id = item.get("id")
|
||
target = _safe_relative(item.get("target", ""), "skill.target")
|
||
if not isinstance(skill_id, str) or not skill_id or skill_id in skills:
|
||
raise InitError(f"{skills_path} 包含无效或重复 Skill ID：{skill_id!r}")
|
||
if target in targets:
|
||
raise InitError(f"{skills_path} 包含重复 Skill target：{target}")
|
||
for key in ("scope", "type", "capability", "triggers", "output"):
|
||
if not isinstance(item.get(key), str) or not item[key].strip():
|
||
raise InitError(f"{skills_path} 的 {skill_id}.{key} 不能为空")
|
||
targets.add(target)
|
||
skills[skill_id] = item
|
||
|
||
raw_types = catalog.get("types")
|
||
if not isinstance(raw_types, list) or not raw_types:
|
||
raise InitError(f"{catalog_path} 的 types 必须是非空数组")
|
||
types: dict[str, TypeEntry] = {}
|
||
for raw in raw_types:
|
||
if not isinstance(raw, dict):
|
||
raise InitError(f"{catalog_path} 的 type 条目必须是对象")
|
||
type_id = raw.get("id")
|
||
label = raw.get("label")
|
||
status = raw.get("status")
|
||
if (
|
||
not isinstance(type_id, str)
|
||
or not type_id
|
||
or type_id in types
|
||
or not isinstance(label, str)
|
||
or not label
|
||
or status not in {"ready", "draft"}
|
||
):
|
||
raise InitError(f"{catalog_path} 包含无效 type 条目：{raw!r}")
|
||
manifest_path = bootstrap / _safe_relative(
|
||
raw.get("manifest", ""),
|
||
f"types.{type_id}.manifest",
|
||
)
|
||
manifest = _read_json(manifest_path)
|
||
if manifest.get("id") != type_id:
|
||
raise InitError(f"{manifest_path} 的 id 与 catalog 不一致")
|
||
_validate_manifest(
|
||
manifest_path,
|
||
manifest,
|
||
require_payload=status == "ready",
|
||
)
|
||
types[type_id] = TypeEntry(
|
||
id=type_id,
|
||
label=label,
|
||
status=status,
|
||
manifest_path=manifest_path,
|
||
manifest=manifest,
|
||
)
|
||
|
||
common_skill_ids = _list_of_strings(common, "skills", common_path)
|
||
for skill_id in common_skill_ids:
|
||
skill = skills.get(skill_id)
|
||
if skill is None or skill["scope"] != "common":
|
||
raise InitError(f"common manifest 引用了无效 common Skill：{skill_id}")
|
||
for entry in types.values():
|
||
type_skill_ids = _list_of_strings(
|
||
entry.manifest,
|
||
"skills",
|
||
entry.manifest_path,
|
||
)
|
||
for skill_id in type_skill_ids:
|
||
skill = skills.get(skill_id)
|
||
if skill is None or skill["scope"] != entry.id:
|
||
raise InitError(f"{entry.id} manifest 引用了无效 Skill：{skill_id}")
|
||
if entry.status == "ready":
|
||
expanded_targets = {
|
||
spec.target for spec in _expand_manifest_files(entry)
|
||
}
|
||
expected_skill_files = {
|
||
(
|
||
PurePosixPath(skills[skill_id]["target"]) / "SKILL.md"
|
||
).as_posix()
|
||
for skill_id in type_skill_ids
|
||
}
|
||
missing_skill_files = expected_skill_files - expanded_targets
|
||
if missing_skill_files:
|
||
raise InitError(
|
||
f"ready bundle 缺少 Skill payload：{sorted(missing_skill_files)}"
|
||
)
|
||
declared_skill_roots = {
|
||
PurePosixPath(skills[skill_id]["target"]).as_posix()
|
||
for skill_id in type_skill_ids
|
||
}
|
||
for target in expanded_targets:
|
||
if not target.startswith(".cursor/skills/"):
|
||
continue
|
||
if not any(
|
||
target == root_path or target.startswith(root_path + "/")
|
||
for root_path in declared_skill_roots
|
||
):
|
||
raise InitError(
|
||
f"ready bundle 含未登记的 Skill payload：{target}"
|
||
)
|
||
|
||
return Context(
|
||
root=root,
|
||
bootstrap=bootstrap,
|
||
catalog_path=catalog_path,
|
||
catalog=catalog,
|
||
common_path=common_path,
|
||
common=common,
|
||
skills_path=skills_path,
|
||
skills=skills,
|
||
types=types,
|
||
)
|
||
|
||
|
||
def list_ready_types(context: Context) -> list[dict[str, str]]:
|
||
return [
|
||
{"id": entry.id, "label": entry.label}
|
||
for entry in context.types.values()
|
||
if entry.status == "ready"
|
||
]
|
||
|
||
|
||
def describe_type(context: Context, project_type: str) -> dict[str, Any]:
|
||
selected = context.types.get(project_type)
|
||
if selected is None or selected.status != "ready":
|
||
raise InitError(f"项目类型不可用：{project_type}")
|
||
return {
|
||
"id": selected.id,
|
||
"label": selected.label,
|
||
"local_env_fields": _all_env_fields(context, selected),
|
||
}
|
||
|
||
|
||
def _expand_manifest_files(entry: TypeEntry) -> list[FileSpec]:
|
||
specs: list[FileSpec] = []
|
||
for item in entry.manifest.get("files", []):
|
||
specs.append(
|
||
FileSpec(
|
||
source=_source_path(entry.manifest_path, item["source"]),
|
||
target=_safe_relative(item["target"], "target"),
|
||
owner=entry.id,
|
||
)
|
||
)
|
||
for item in entry.manifest.get("trees", []):
|
||
source_root = _source_path(entry.manifest_path, item["source"])
|
||
target_root = PurePosixPath(_safe_relative(item["target"], "target"))
|
||
if not source_root.is_dir():
|
||
continue
|
||
for source in sorted(path for path in source_root.rglob("*") if path.is_file()):
|
||
relative = source.relative_to(source_root)
|
||
target = (target_root / PurePosixPath(relative.as_posix())).as_posix()
|
||
specs.append(FileSpec(source=source, target=target, owner=entry.id))
|
||
targets = [spec.target for spec in specs]
|
||
if len(set(targets)) != len(targets):
|
||
raise InitError(f"{entry.manifest_path} 展开后包含重复 target")
|
||
return specs
|
||
|
||
|
||
def _expand_reset_files(context: Context) -> list[FileSpec]:
|
||
return [
|
||
FileSpec(
|
||
source=_source_path(context.common_path, item["source"]),
|
||
target=_safe_relative(item["target"], "reset target"),
|
||
owner="common-reset",
|
||
)
|
||
for item in context.common.get("reset_files", [])
|
||
]
|
||
|
||
|
||
def _merge_unique(*groups: Iterable[str]) -> list[str]:
|
||
result: list[str] = []
|
||
seen: set[str] = set()
|
||
for group in groups:
|
||
for item in group:
|
||
if item not in seen:
|
||
seen.add(item)
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
def _all_env_fields(context: Context, selected: TypeEntry) -> list[dict[str, Any]]:
|
||
fields = [
|
||
*context.common.get("local_env_fields", []),
|
||
*selected.manifest.get("local_env_fields", []),
|
||
]
|
||
keys = [field["key"] for field in fields]
|
||
if len(keys) != len(set(keys)):
|
||
raise InitError(f"{selected.id} 与 common 声明了重复环境字段")
|
||
return fields
|
||
|
||
|
||
def _render_example(fields: list[dict[str, Any]]) -> bytes:
|
||
data: dict[str, Any] = {
|
||
"_comment": (
|
||
"复制为 .cursor/local-env.json 后填写本机值；"
|
||
"local-env.json 仅供本机使用，不应提交。"
|
||
)
|
||
}
|
||
for field in fields:
|
||
if field.get("sensitive", False):
|
||
data[field["key"]] = field.get("example", "")
|
||
else:
|
||
data[field["key"]] = field.get("example", field.get("default", ""))
|
||
comment_key = field.get("comment_key")
|
||
if comment_key:
|
||
data[comment_key] = field.get("comment", "")
|
||
return _json_bytes(data)
|
||
|
||
|
||
def _render_local_env(
|
||
root: Path,
|
||
fields: list[dict[str, Any]],
|
||
values: dict[str, str],
|
||
) -> tuple[bytes, list[dict[str, Any]]]:
|
||
existing_path = _target_path(root, LOCAL_ENV_PATH)
|
||
existing: dict[str, Any] = {}
|
||
if existing_path.is_file():
|
||
try:
|
||
loaded = json.loads(existing_path.read_text(encoding="utf-8"))
|
||
if isinstance(loaded, dict):
|
||
existing = loaded
|
||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||
pass
|
||
|
||
rendered = json.loads(_render_example(fields).decode("utf-8"))
|
||
rendered.update(existing)
|
||
required_inputs: list[dict[str, Any]] = []
|
||
for field in fields:
|
||
key = field["key"]
|
||
if key in values:
|
||
rendered[key] = values[key]
|
||
elif key not in rendered or str(rendered[key]).startswith("<"):
|
||
rendered[key] = field.get("default", "")
|
||
value = rendered.get(key, "")
|
||
collect_at_init = field.get("collect_at_init", True)
|
||
if field.get("required") and collect_at_init and not str(value).strip():
|
||
required_inputs.append(
|
||
{
|
||
"key": key,
|
||
"prompt": field.get("prompt", key),
|
||
"default": field.get("default", ""),
|
||
}
|
||
)
|
||
return _json_bytes(rendered), required_inputs
|
||
|
||
|
||
def _json_bytes(data: Any) -> bytes:
|
||
text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
|
||
return text.encode("utf-8")
|
||
|
||
|
||
def _read_fragments(manifest_path: Path, manifest: dict[str, Any], key: str) -> list[str]:
|
||
result: list[str] = []
|
||
for relative in manifest.get(key, []):
|
||
path = _source_path(manifest_path, relative)
|
||
result.append(path.read_text(encoding="utf-8").strip())
|
||
return [item for item in result if item]
|
||
|
||
|
||
def _render_managed_block(
|
||
path: Path,
|
||
body: str,
|
||
start_marker: str,
|
||
end_marker: str,
|
||
) -> bytes:
|
||
existing = ""
|
||
if path.is_file():
|
||
existing = path.read_text(encoding="utf-8").replace("\r\n", "\n")
|
||
start_count = existing.count(start_marker)
|
||
end_count = existing.count(end_marker)
|
||
if start_count != end_count or start_count > 1:
|
||
raise InitError(f"managed block 标记不完整或重复：{path}")
|
||
|
||
block = f"{start_marker}\n{body.rstrip()}\n{end_marker}\n"
|
||
if start_count == 1:
|
||
start = existing.index(start_marker)
|
||
end = existing.index(end_marker, start) + len(end_marker)
|
||
prefix = existing[:start].rstrip("\n")
|
||
suffix = existing[end:].lstrip("\n")
|
||
pieces = [piece for piece in (prefix, block.rstrip("\n"), suffix) if piece]
|
||
rendered = "\n\n".join(pieces) + "\n"
|
||
elif existing.strip():
|
||
rendered = existing.rstrip("\n") + "\n\n" + block
|
||
else:
|
||
rendered = block
|
||
return rendered.encode("utf-8")
|
||
|
||
|
||
def _render_scan(context: Context, selected: TypeEntry) -> bytes:
|
||
common_scan = context.common.get("scan", {})
|
||
type_scan = selected.manifest.get("scan", {})
|
||
data = {
|
||
"schema_version": SCHEMA_VERSION,
|
||
"project_type": selected.id,
|
||
"display_name": selected.label,
|
||
"included_names": _merge_unique(
|
||
common_scan.get("included_names", []),
|
||
type_scan.get("included_names", []),
|
||
),
|
||
"included_extensions": _merge_unique(
|
||
common_scan.get("included_extensions", []),
|
||
type_scan.get("included_extensions", []),
|
||
),
|
||
"excluded_directories": _merge_unique(
|
||
common_scan.get("excluded_directories", []),
|
||
type_scan.get("excluded_directories", []),
|
||
),
|
||
}
|
||
return _json_bytes(data)
|
||
|
||
|
||
def _active_skill_ids(context: Context, selected: TypeEntry) -> list[str]:
|
||
return _merge_unique(
|
||
context.common.get("skills", []),
|
||
selected.manifest.get("skills", []),
|
||
)
|
||
|
||
|
||
def _render_registry(context: Context, selected: TypeEntry) -> bytes:
|
||
lines = [
|
||
"# EPEE Skill Registry",
|
||
"",
|
||
"> 本文件由 `.cursor/bootstrap/tools/cursor_init.py` 根据 `skills.json` 生成。",
|
||
"> 创建、删除或实质修改 Skill 后必须更新机读源并重新生成。",
|
||
"",
|
||
"## 已注册 Skill",
|
||
"",
|
||
]
|
||
for skill_id in _active_skill_ids(context, selected):
|
||
skill = context.skills[skill_id]
|
||
skill_md = PurePosixPath(skill["target"]) / "SKILL.md"
|
||
lines.extend(
|
||
[
|
||
f"### {skill_id}",
|
||
f"- **类型**: {skill['type']}",
|
||
f"- **能力**: {skill['capability']}",
|
||
f"- **触发场景**: {skill['triggers']}",
|
||
f"- **输出**: {skill['output']}",
|
||
f"- **路径**: {skill_md.as_posix()}",
|
||
"",
|
||
]
|
||
)
|
||
return ("\n".join(lines).rstrip() + "\n").encode("utf-8")
|
||
|
||
|
||
def _manifest_digest(context: Context, selected: TypeEntry) -> str:
|
||
paths: set[Path] = {
|
||
context.catalog_path,
|
||
context.common_path,
|
||
context.skills_path,
|
||
selected.manifest_path,
|
||
}
|
||
for manifest_path, manifest in (
|
||
(context.common_path, context.common),
|
||
(selected.manifest_path, selected.manifest),
|
||
):
|
||
for key in ("gitignore_fragments", "gitattributes_fragments"):
|
||
for relative in manifest.get(key, []):
|
||
paths.add(_source_path(manifest_path, relative))
|
||
for relative in context.common.get("bootstrap_files", []):
|
||
paths.add(_source_path(context.common_path, relative))
|
||
for spec in _expand_manifest_files(selected):
|
||
paths.add(spec.source)
|
||
for spec in _expand_reset_files(context):
|
||
paths.add(spec.source)
|
||
|
||
digest = hashlib.sha256()
|
||
for path in sorted(paths):
|
||
digest.update(path.relative_to(context.root).as_posix().encode("utf-8"))
|
||
digest.update(b"\0")
|
||
digest.update(path.read_bytes())
|
||
digest.update(b"\0")
|
||
return digest.hexdigest()
|
||
|
||
|
||
def _add_content_action(
|
||
plan: Plan,
|
||
root: Path,
|
||
relative: str,
|
||
content: bytes,
|
||
reason: str,
|
||
*,
|
||
redact_digest: bool = False,
|
||
) -> None:
|
||
path = _target_path(root, relative)
|
||
if not path.exists():
|
||
plan.actions.append(
|
||
Action(
|
||
"create",
|
||
relative,
|
||
reason,
|
||
content,
|
||
redact_digest=redact_digest,
|
||
)
|
||
)
|
||
elif not path.is_file():
|
||
plan.blockers.append(f"目标不是文件：{relative}")
|
||
elif path.read_bytes() != content:
|
||
plan.actions.append(
|
||
Action(
|
||
"update",
|
||
relative,
|
||
reason,
|
||
content,
|
||
redact_digest=redact_digest,
|
||
)
|
||
)
|
||
|
||
|
||
def _load_state(root: Path) -> dict[str, Any]:
|
||
path = _target_path(root, STATE_PATH)
|
||
if not path.is_file():
|
||
return {}
|
||
try:
|
||
state = json.loads(path.read_text(encoding="utf-8"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||
return {}
|
||
return state if isinstance(state, dict) else {}
|
||
|
||
|
||
def _sha256_file(path: Path) -> str:
|
||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
||
|
||
def _managed_file_is_unchanged(
|
||
path: Path,
|
||
target: str,
|
||
spec: FileSpec | None,
|
||
state_hashes: dict[str, str],
|
||
) -> bool:
|
||
current_hash = _sha256_file(path)
|
||
if state_hashes.get(target) == current_hash:
|
||
return True
|
||
return spec is not None and _sha256_file(spec.source) == current_hash
|
||
|
||
|
||
def _detect_foreign(
|
||
context: Context,
|
||
all_managed_targets: set[str],
|
||
) -> list[str]:
|
||
foreign: list[str] = []
|
||
project_rules = context.root / ".cursor" / "rules" / "project"
|
||
if project_rules.is_dir():
|
||
for path in sorted(item for item in project_rules.rglob("*") if item.is_file()):
|
||
relative = path.relative_to(context.root).as_posix()
|
||
if relative.endswith("/.gitkeep"):
|
||
continue
|
||
if relative not in all_managed_targets:
|
||
foreign.append(relative)
|
||
|
||
skills_root = context.root / ".cursor" / "skills"
|
||
known_skill_dirs = {
|
||
PurePosixPath(skill["target"]).parts[-1] for skill in context.skills.values()
|
||
}
|
||
if skills_root.is_dir():
|
||
for path in sorted(item for item in skills_root.iterdir() if item.is_dir()):
|
||
if path.name not in known_skill_dirs:
|
||
foreign.append(path.relative_to(context.root).as_posix() + "/")
|
||
return foreign
|
||
|
||
|
||
def build_plan(
|
||
context: Context,
|
||
project_type: str,
|
||
*,
|
||
values: dict[str, str] | None = None,
|
||
skip_git_files: bool = False,
|
||
) -> Plan:
|
||
selected = context.types.get(project_type)
|
||
if selected is None or selected.status != "ready":
|
||
raise InitError(f"项目类型不可用：{project_type}")
|
||
values = values or {}
|
||
plan = Plan(
|
||
project_type=selected.id,
|
||
label=selected.label,
|
||
manifest_digest=_manifest_digest(context, selected),
|
||
bundle_version=int(selected.manifest.get("version", 0)),
|
||
gitignore_generated=not skip_git_files,
|
||
)
|
||
|
||
selected_specs = _expand_manifest_files(selected)
|
||
selected_targets = {spec.target for spec in selected_specs}
|
||
all_specs = [
|
||
spec
|
||
for entry in context.types.values()
|
||
for spec in _expand_manifest_files(entry)
|
||
]
|
||
specs_by_target = {spec.target: spec for spec in all_specs}
|
||
state = _load_state(context.root)
|
||
raw_state_hashes = state.get("managed_files", {})
|
||
state_hashes = (
|
||
{
|
||
str(path): str(digest)
|
||
for path, digest in raw_state_hashes.items()
|
||
if isinstance(path, str) and isinstance(digest, str)
|
||
}
|
||
if isinstance(raw_state_hashes, dict)
|
||
else {}
|
||
)
|
||
all_managed_targets = {spec.target for spec in all_specs} | set(state_hashes)
|
||
plan.foreign.extend(_detect_foreign(context, all_managed_targets))
|
||
if plan.foreign:
|
||
plan.blockers.append("存在未由 manifest 管理的活动 Rule/Skill，请先逐项处理")
|
||
|
||
reset_specs = _expand_reset_files(context)
|
||
generated_targets = {
|
||
SCAN_CONFIG_PATH,
|
||
REGISTRY_PATH,
|
||
LOCAL_ENV_EXAMPLE_PATH,
|
||
LOCAL_ENV_PATH,
|
||
*(spec.target for spec in reset_specs),
|
||
}
|
||
for relative in context.common.get("canonical_files", []):
|
||
normalized = _safe_relative(relative, "canonical_files")
|
||
if normalized in generated_targets:
|
||
continue
|
||
if not _target_path(context.root, normalized).is_file():
|
||
plan.blockers.append(f"缺少 common canonical：{normalized}")
|
||
|
||
for spec in reset_specs:
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
spec.target,
|
||
spec.source.read_bytes(),
|
||
"重置模板运行数据",
|
||
)
|
||
|
||
cleanup_paths: set[str] = set()
|
||
for raw_pattern in context.common.get("cleanup_globs", []):
|
||
pattern = _safe_relative(raw_pattern, "cleanup_globs")
|
||
for path in context.root.glob(pattern):
|
||
if not path.is_file():
|
||
continue
|
||
relative = path.relative_to(context.root).as_posix()
|
||
if relative.startswith(".cursor/bootstrap/"):
|
||
continue
|
||
cleanup_paths.add(relative)
|
||
for relative in sorted(cleanup_paths):
|
||
plan.actions.append(Action("delete", relative, "清理历史运行数据"))
|
||
|
||
for spec in selected_specs:
|
||
target_path = _target_path(context.root, spec.target)
|
||
source_content = spec.source.read_bytes()
|
||
source_mode = spec.source.stat().st_mode & 0o777
|
||
if not target_path.exists():
|
||
plan.actions.append(
|
||
Action(
|
||
"create",
|
||
spec.target,
|
||
f"物化 {selected.id} bundle",
|
||
source_content,
|
||
source_mode,
|
||
)
|
||
)
|
||
elif not target_path.is_file():
|
||
plan.blockers.append(f"目标不是文件：{spec.target}")
|
||
elif target_path.read_bytes() != source_content:
|
||
if _managed_file_is_unchanged(
|
||
target_path,
|
||
spec.target,
|
||
spec,
|
||
state_hashes,
|
||
):
|
||
plan.actions.append(
|
||
Action(
|
||
"update",
|
||
spec.target,
|
||
f"更新 {selected.id} bundle",
|
||
source_content,
|
||
source_mode,
|
||
)
|
||
)
|
||
else:
|
||
plan.blockers.append(f"managed 文件已被用户修改，拒绝覆盖：{spec.target}")
|
||
|
||
for target in sorted(all_managed_targets - selected_targets):
|
||
path = _target_path(context.root, target)
|
||
if path.is_file():
|
||
spec = specs_by_target.get(target)
|
||
if _managed_file_is_unchanged(path, target, spec, state_hashes):
|
||
plan.actions.append(
|
||
Action("delete", target, "移除非所选项目类型的 managed 文件")
|
||
)
|
||
else:
|
||
plan.blockers.append(f"旧 bundle 文件已被用户修改，拒绝删除：{target}")
|
||
|
||
fields = _all_env_fields(context, selected)
|
||
example_content = _render_example(fields)
|
||
local_env_content, required_inputs = _render_local_env(
|
||
context.root,
|
||
fields,
|
||
values,
|
||
)
|
||
plan.required_inputs.extend(required_inputs)
|
||
if required_inputs:
|
||
keys = ", ".join(item["key"] for item in required_inputs)
|
||
plan.blockers.append(f"缺少必填本机配置：{keys}")
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
LOCAL_ENV_EXAMPLE_PATH,
|
||
example_content,
|
||
"按项目类型生成本机配置模板",
|
||
)
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
LOCAL_ENV_PATH,
|
||
local_env_content,
|
||
"生成或合并本机配置",
|
||
redact_digest=True,
|
||
)
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
SCAN_CONFIG_PATH,
|
||
_render_scan(context, selected),
|
||
"生成 Hook 活动扫描配置",
|
||
)
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
REGISTRY_PATH,
|
||
_render_registry(context, selected),
|
||
"按活动 Skill 集合生成 Registry",
|
||
)
|
||
|
||
active_skill_ids = set(_active_skill_ids(context, selected))
|
||
for skill_id in active_skill_ids:
|
||
skill = context.skills[skill_id]
|
||
target = _target_path(context.root, skill["target"])
|
||
planned_targets = {
|
||
PurePosixPath(spec.target).parts[:3] for spec in selected_specs
|
||
}
|
||
target_parts = PurePosixPath(skill["target"]).parts
|
||
planned = target_parts[:3] in planned_targets
|
||
if not (target / "SKILL.md").is_file() and not planned:
|
||
plan.blockers.append(f"活动 Skill 缺少 SKILL.md：{skill_id}")
|
||
|
||
if not skip_git_files:
|
||
common_gitignore = _read_fragments(
|
||
context.common_path,
|
||
context.common,
|
||
"gitignore_fragments",
|
||
)
|
||
type_gitignore = _read_fragments(
|
||
selected.manifest_path,
|
||
selected.manifest,
|
||
"gitignore_fragments",
|
||
)
|
||
gitignore_body = "\n\n".join([*common_gitignore, *type_gitignore])
|
||
try:
|
||
gitignore_content = _render_managed_block(
|
||
context.root / ".gitignore",
|
||
gitignore_body,
|
||
GITIGNORE_START,
|
||
GITIGNORE_END,
|
||
)
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
".gitignore",
|
||
gitignore_content,
|
||
"更新 cursor-init managed block",
|
||
)
|
||
except (InitError, UnicodeDecodeError) as exc:
|
||
plan.blockers.append(str(exc))
|
||
|
||
common_attributes = _read_fragments(
|
||
context.common_path,
|
||
context.common,
|
||
"gitattributes_fragments",
|
||
)
|
||
type_attributes = _read_fragments(
|
||
selected.manifest_path,
|
||
selected.manifest,
|
||
"gitattributes_fragments",
|
||
)
|
||
attributes_body = "\n\n".join([*common_attributes, *type_attributes])
|
||
try:
|
||
attributes_content = _render_managed_block(
|
||
context.root / ".gitattributes",
|
||
attributes_body,
|
||
GITATTRIBUTES_START,
|
||
GITATTRIBUTES_END,
|
||
)
|
||
_add_content_action(
|
||
plan,
|
||
context.root,
|
||
".gitattributes",
|
||
attributes_content,
|
||
"更新 cursor-init managed block",
|
||
)
|
||
except (InitError, UnicodeDecodeError) as exc:
|
||
plan.blockers.append(str(exc))
|
||
|
||
return plan
|
||
|
||
|
||
def _write_atomic(path: Path, content: bytes) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
|
||
temporary = Path(temporary_name)
|
||
try:
|
||
with os.fdopen(descriptor, "wb") as stream:
|
||
stream.write(content)
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
os.replace(temporary, path)
|
||
finally:
|
||
if temporary.exists():
|
||
temporary.unlink()
|
||
|
||
|
||
def _sentinel_bytes(context: Context, plan: Plan) -> bytes:
|
||
timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
|
||
lines = [
|
||
"# .cursor/.init-done — cursor-init 写入的初始化标记",
|
||
"initialized_by: cursor-init",
|
||
f"initialized_at: {timestamp}",
|
||
f"schema_version: {SCHEMA_VERSION}",
|
||
f"skill_version: {context.catalog['skill_version']}",
|
||
f"project_type: {plan.project_type}",
|
||
f"bundle_version: {plan.bundle_version}",
|
||
f"manifest_digest: {plan.manifest_digest}",
|
||
f"gitignore_generated: {'true' if plan.gitignore_generated else 'false'}",
|
||
"",
|
||
]
|
||
return "\n".join(lines).encode("utf-8")
|
||
|
||
|
||
def _state_bytes(context: Context, plan: Plan) -> bytes:
|
||
selected = context.types[plan.project_type]
|
||
managed_files = {
|
||
spec.target: _sha256_file(spec.source)
|
||
for spec in _expand_manifest_files(selected)
|
||
}
|
||
return _json_bytes(
|
||
{
|
||
"schema_version": SCHEMA_VERSION,
|
||
"project_type": plan.project_type,
|
||
"bundle_version": plan.bundle_version,
|
||
"manifest_digest": plan.manifest_digest,
|
||
"managed_files": managed_files,
|
||
}
|
||
)
|
||
|
||
|
||
def apply_plan(context: Context, plan: Plan, expected_digest: str) -> None:
|
||
if plan.digest != expected_digest:
|
||
raise InitError(
|
||
f"dry-run 摘要已变化：期望 {expected_digest}，当前 {plan.digest}；请重新确认"
|
||
)
|
||
if plan.blockers:
|
||
raise InitError("存在阻塞项，拒绝 apply：\n- " + "\n- ".join(plan.blockers))
|
||
|
||
for action in sorted(plan.actions, key=lambda item: item.operation != "delete"):
|
||
path = _target_path(context.root, action.path)
|
||
if action.operation == "delete":
|
||
if path.is_file():
|
||
path.unlink()
|
||
elif action.operation in {"create", "update"} and action.content is not None:
|
||
_write_atomic(path, action.content)
|
||
if action.mode is not None:
|
||
path.chmod(action.mode)
|
||
else:
|
||
raise InitError(f"未知 action：{action.operation} {action.path}")
|
||
|
||
validation_plan = build_plan(
|
||
context,
|
||
plan.project_type,
|
||
skip_git_files=not plan.gitignore_generated,
|
||
)
|
||
validation_actions = [
|
||
action
|
||
for action in validation_plan.actions
|
||
if action.path != LOCAL_ENV_PATH
|
||
]
|
||
if validation_plan.blockers or validation_actions:
|
||
details = [*validation_plan.blockers]
|
||
details.extend(
|
||
f"{action.operation}: {action.path}" for action in validation_actions
|
||
)
|
||
raise InitError("apply 后验证失败，未写 sentinel：\n- " + "\n- ".join(details))
|
||
|
||
_write_atomic(
|
||
_target_path(context.root, STATE_PATH),
|
||
_state_bytes(context, plan),
|
||
)
|
||
_write_atomic(
|
||
_target_path(context.root, SENTINEL_PATH),
|
||
_sentinel_bytes(context, plan),
|
||
)
|
||
|
||
|
||
def _parse_sentinel(path: Path) -> dict[str, str]:
|
||
result: dict[str, str] = {}
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith("#") or ":" not in stripped:
|
||
continue
|
||
key, value = stripped.split(":", 1)
|
||
result[key.strip()] = value.strip()
|
||
return result
|
||
|
||
|
||
def check_project(context: Context, project_type: str | None = None) -> list[str]:
|
||
errors: list[str] = []
|
||
sentinel_path = _target_path(context.root, SENTINEL_PATH)
|
||
sentinel: dict[str, str] = {}
|
||
if sentinel_path.is_file():
|
||
try:
|
||
sentinel = _parse_sentinel(sentinel_path)
|
||
except UnicodeDecodeError as exc:
|
||
errors.append(f"sentinel 不是 UTF-8：{exc}")
|
||
if project_type is None:
|
||
project_type = sentinel.get("project_type")
|
||
if project_type is None:
|
||
return errors
|
||
|
||
selected = context.types.get(project_type)
|
||
if selected is None or selected.status != "ready":
|
||
errors.append(f"sentinel/参数中的项目类型不可用：{project_type}")
|
||
return errors
|
||
skip_git_files = sentinel.get("gitignore_generated") == "false"
|
||
plan = build_plan(
|
||
context,
|
||
project_type,
|
||
skip_git_files=skip_git_files,
|
||
)
|
||
errors.extend(plan.blockers)
|
||
errors.extend(
|
||
f"活动状态漂移：{action.operation} {action.path}"
|
||
for action in plan.actions
|
||
if action.path != LOCAL_ENV_PATH
|
||
)
|
||
if sentinel:
|
||
expected = {
|
||
"schema_version": str(SCHEMA_VERSION),
|
||
"skill_version": str(context.catalog["skill_version"]),
|
||
"project_type": project_type,
|
||
"bundle_version": str(selected.manifest.get("version", 0)),
|
||
"manifest_digest": plan.manifest_digest,
|
||
}
|
||
for key, value in expected.items():
|
||
if sentinel.get(key) != value:
|
||
errors.append(
|
||
f"sentinel 字段不一致：{key}={sentinel.get(key)!r}，期望 {value!r}"
|
||
)
|
||
state = _load_state(context.root)
|
||
expected_state = json.loads(_state_bytes(context, plan).decode("utf-8"))
|
||
if state != expected_state:
|
||
errors.append(f"{STATE_PATH} 与当前 manifest/活动 bundle 不一致")
|
||
return errors
|
||
|
||
|
||
def plan_as_dict(plan: Plan) -> dict[str, Any]:
|
||
return {
|
||
"project_type": plan.project_type,
|
||
"label": plan.label,
|
||
"manifest_digest": plan.manifest_digest,
|
||
"bundle_version": plan.bundle_version,
|
||
"gitignore_generated": plan.gitignore_generated,
|
||
"plan_digest": plan.digest,
|
||
"actions": [action.digest_record() for action in plan.actions],
|
||
"blockers": plan.blockers,
|
||
"foreign": plan.foreign,
|
||
"required_inputs": plan.required_inputs,
|
||
}
|
||
|
||
|
||
def print_plan(plan: Plan) -> None:
|
||
print(f"## cursor-init dry-run — {plan.label} ({plan.project_type})")
|
||
groups = (
|
||
("create", "创建"),
|
||
("update", "更新"),
|
||
("delete", "删除"),
|
||
)
|
||
for operation, label in groups:
|
||
actions = [item for item in plan.actions if item.operation == operation]
|
||
print(f"\n### {label}")
|
||
if actions:
|
||
for action in actions:
|
||
print(f"- {action.path} — {action.reason}")
|
||
else:
|
||
print("- （无）")
|
||
print("\n### Foreign")
|
||
if plan.foreign:
|
||
for path in plan.foreign:
|
||
print(f"- {path}")
|
||
else:
|
||
print("- （无）")
|
||
print("\n### 阻塞")
|
||
if plan.blockers:
|
||
for blocker in plan.blockers:
|
||
print(f"- {blocker}")
|
||
else:
|
||
print("- （无）")
|
||
print(f"\nPLAN_DIGEST={plan.digest}")
|
||
|
||
|
||
def _parse_values(items: list[str]) -> dict[str, str]:
|
||
result: dict[str, str] = {}
|
||
for item in items:
|
||
if "=" not in item:
|
||
raise InitError(f"--set 必须使用 KEY=VALUE：{item}")
|
||
key, value = item.split("=", 1)
|
||
if not key:
|
||
raise InitError(f"--set 的 KEY 不能为空：{item}")
|
||
result[key] = value
|
||
return result
|
||
|
||
|
||
def _validate_input_values(
|
||
fields: list[dict[str, Any]],
|
||
values: dict[str, str],
|
||
) -> None:
|
||
fields_by_key = {field["key"]: field for field in fields}
|
||
for key in values:
|
||
field = fields_by_key.get(key)
|
||
if field is None:
|
||
raise InitError(f"--set 包含当前项目类型未声明的字段：{key}")
|
||
if field.get("sensitive", False):
|
||
raise InitError(
|
||
f"敏感字段不得通过 --set 传入：{key}；"
|
||
f"请手工填写 {LOCAL_ENV_PATH}"
|
||
)
|
||
if not field.get("collect_at_init", True):
|
||
raise InitError(
|
||
f"字段不允许由初始化流程采集：{key}；"
|
||
f"请手工填写 {LOCAL_ENV_PATH}"
|
||
)
|
||
|
||
|
||
def _default_root() -> Path:
|
||
return Path(__file__).resolve().parents[3]
|
||
|
||
|
||
def _interactive(context: Context, skip_git_files: bool) -> int:
|
||
ready = list_ready_types(context)
|
||
print("支持的项目类型：")
|
||
for index, item in enumerate(ready, 1):
|
||
print(f"{index}. {item['label']} ({item['id']})")
|
||
raw = input("请选择项目类型编号：").strip()
|
||
try:
|
||
selected = ready[int(raw) - 1]
|
||
except (ValueError, IndexError):
|
||
raise InitError("无效的项目类型选择")
|
||
|
||
fields = _all_env_fields(context, context.types[selected["id"]])
|
||
values: dict[str, str] = {}
|
||
for field in fields:
|
||
if field.get("sensitive", False) or not field.get("collect_at_init", True):
|
||
continue
|
||
prompt = field.get("prompt", field["key"])
|
||
default = field.get("default", "")
|
||
suffix = f" [{default}]" if default else ""
|
||
value = input(f"{prompt}{suffix}：").strip()
|
||
values[field["key"]] = value or default
|
||
|
||
plan = build_plan(
|
||
context,
|
||
selected["id"],
|
||
values=values,
|
||
skip_git_files=skip_git_files,
|
||
)
|
||
print_plan(plan)
|
||
if plan.blockers:
|
||
return 2
|
||
if input("确认执行以上 dry-run？[y/N]：").strip().lower() != "y":
|
||
print("已取消。")
|
||
return 1
|
||
apply_plan(context, plan, plan.digest)
|
||
print("初始化完成。")
|
||
return 0
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
modes = parser.add_mutually_exclusive_group()
|
||
modes.add_argument("--list-types", action="store_true", help="列出 ready 项目类型")
|
||
modes.add_argument("--describe-type", action="store_true", help="列出类型所需输入")
|
||
modes.add_argument("--dry-run", action="store_true", help="生成初始化预览")
|
||
modes.add_argument("--apply", action="store_true", help="应用已确认预览")
|
||
modes.add_argument("--check", action="store_true", help="校验 catalog 或活动状态")
|
||
modes.add_argument("--sync-registry", action="store_true", help="仅重建活动 Skill Registry")
|
||
parser.add_argument("--target", type=Path, default=_default_root())
|
||
parser.add_argument("--project-type")
|
||
parser.add_argument("--plan-digest")
|
||
parser.add_argument("--set", action="append", default=[], metavar="KEY=VALUE")
|
||
parser.add_argument("--skip-git-files", action="store_true")
|
||
parser.add_argument("--json", action="store_true", dest="as_json")
|
||
args = parser.parse_args(argv)
|
||
|
||
try:
|
||
context = load_context(args.target)
|
||
if args.list_types:
|
||
ready = list_ready_types(context)
|
||
if args.as_json:
|
||
print(json.dumps(ready, ensure_ascii=False, indent=2))
|
||
else:
|
||
for item in ready:
|
||
print(f"{item['id']}\t{item['label']}")
|
||
return 0
|
||
|
||
if args.describe_type:
|
||
if not args.project_type:
|
||
raise InitError("--describe-type 必须提供 --project-type")
|
||
description = describe_type(context, args.project_type)
|
||
if args.as_json:
|
||
print(json.dumps(description, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(f"{description['label']}（{description['id']}）")
|
||
for field in description["local_env_fields"]:
|
||
required = "必填" if field.get("required") else "可选"
|
||
notes = [required]
|
||
if field.get("sensitive", False):
|
||
notes.append("敏感")
|
||
if not field.get("collect_at_init", True):
|
||
notes.append("初始化不采集")
|
||
print(
|
||
f"- {field['key']}：{field.get('prompt', '')}"
|
||
f"（{'，'.join(notes)}）"
|
||
)
|
||
return 0
|
||
|
||
if args.sync_registry:
|
||
project_type = args.project_type
|
||
if project_type is None:
|
||
sentinel_path = _target_path(context.root, SENTINEL_PATH)
|
||
if sentinel_path.is_file():
|
||
project_type = _parse_sentinel(sentinel_path).get("project_type")
|
||
else:
|
||
project_type = "general"
|
||
selected = context.types.get(project_type)
|
||
if selected is None or selected.status != "ready":
|
||
raise InitError(f"项目类型不可用：{project_type}")
|
||
registry = _render_registry(context, selected)
|
||
_write_atomic(_target_path(context.root, REGISTRY_PATH), registry)
|
||
if args.as_json:
|
||
print(json.dumps({"ok": True, "project_type": project_type}))
|
||
else:
|
||
print(f"Registry 已同步：{project_type}")
|
||
return 0
|
||
|
||
if args.check:
|
||
errors = check_project(context, args.project_type)
|
||
if errors:
|
||
if args.as_json:
|
||
print(json.dumps({"ok": False, "errors": errors}, ensure_ascii=False))
|
||
else:
|
||
print("校验失败：")
|
||
for error in errors:
|
||
print(f"- {error}")
|
||
return 2
|
||
if args.as_json:
|
||
print(json.dumps({"ok": True}, ensure_ascii=False))
|
||
else:
|
||
print("校验通过。")
|
||
return 0
|
||
|
||
if args.dry_run or args.apply:
|
||
if not args.project_type:
|
||
raise InitError("--dry-run/--apply 必须提供 --project-type")
|
||
values = _parse_values(args.set)
|
||
selected = context.types.get(args.project_type)
|
||
if selected is None or selected.status != "ready":
|
||
raise InitError(f"项目类型不可用：{args.project_type}")
|
||
_validate_input_values(
|
||
_all_env_fields(context, selected),
|
||
values,
|
||
)
|
||
plan = build_plan(
|
||
context,
|
||
args.project_type,
|
||
values=values,
|
||
skip_git_files=args.skip_git_files,
|
||
)
|
||
if args.dry_run:
|
||
if args.as_json:
|
||
print(json.dumps(plan_as_dict(plan), ensure_ascii=False, indent=2))
|
||
else:
|
||
print_plan(plan)
|
||
return 2 if plan.blockers else 0
|
||
if not args.plan_digest:
|
||
raise InitError("--apply 必须提供 --plan-digest")
|
||
apply_plan(context, plan, args.plan_digest)
|
||
if args.as_json:
|
||
print(json.dumps({"ok": True, "project_type": plan.project_type}))
|
||
else:
|
||
print(f"初始化完成：{plan.label}（{plan.project_type}）")
|
||
return 0
|
||
|
||
return _interactive(context, args.skip_git_files)
|
||
except InitError as exc:
|
||
if args.as_json:
|
||
print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False))
|
||
else:
|
||
print(f"错误：{exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|