686aaaa421
冻结地图1二十分钟纵切版的平衡、证据与金标登记。 新增 ReferenceAsset/2 清单、策略、release、受信快照及 CLI/Service/acceptance provenance /4 消费链；保持 survivor live、R1 签名与部署关闭。
295 lines
12 KiB
Python
295 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""生成《山海行纪》参照资产消费清单的确定性命令。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
import unicodedata
|
||
from pathlib import Path
|
||
|
||
|
||
DEFAULT_REGISTRY = Path("contracts/play-loop/reference-asset-registry.initial.json")
|
||
DEFAULT_RECORD_ID = "gac-shanhai-xingji"
|
||
DEFAULT_MANIFEST_ID = "survivor-gold-v1-manifest"
|
||
MANIFEST_SCHEMA_VERSION = "ReferenceAssetConsumptionManifest/1"
|
||
CANONICALIZATION = "reference-asset-consumption-manifest/1"
|
||
DEFAULT_SOURCE_MODE = "worktree"
|
||
SOURCE_MODES = ("worktree", "clean-archive", "clean_archive")
|
||
TEMPORARY_OUTPUT_RE = re.compile(r"(^|[._-])(cache|debug|generated|output|tmp|temp)([._-]|$)", re.IGNORECASE)
|
||
|
||
|
||
class ManifestGenerationError(Exception):
|
||
"""表示清单输入不可信或不完整的稳定生成错误。"""
|
||
|
||
def __init__(self, code: str, logical_path: str, detail: str) -> None:
|
||
super().__init__(detail)
|
||
self.code = code
|
||
self.logical_path = logical_path
|
||
self.detail = detail
|
||
|
||
|
||
def canonical_json_bytes(value: object) -> bytes:
|
||
"""按项目契约生成无 BOM、无尾随换行的 canonical JSON 字节。"""
|
||
# sort_keys 使用 Unicode code point 排序，ensure_ascii=False 保留非 ASCII 原字节。
|
||
return json.dumps(
|
||
value,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
allow_nan=False,
|
||
).encode("utf-8")
|
||
|
||
|
||
def logical_path(repo_root: Path, path: Path) -> str:
|
||
"""把路径转换为仓根相对 POSIX NFC 路径，拒绝越出仓根的引用。"""
|
||
try:
|
||
relative = path.resolve().relative_to(repo_root.resolve()).as_posix()
|
||
except ValueError as exc:
|
||
raise ManifestGenerationError(
|
||
"reference_path_escape", "<input>", "输入路径必须位于可信仓根内",
|
||
) from exc
|
||
normalized = unicodedata.normalize("NFC", relative)
|
||
if normalized != relative or not relative or relative.startswith("/"):
|
||
raise ManifestGenerationError("reference_path_invalid", relative or "<empty>", "路径不是 NFC 相对 POSIX 路径")
|
||
return normalized
|
||
|
||
|
||
def read_regular_file(repo_root: Path, path: Path) -> tuple[str, bytes]:
|
||
"""以仓内逻辑路径读取普通文件，避免把链接或目录当作消费输入。"""
|
||
relative = logical_path(repo_root, path)
|
||
try:
|
||
file_stat = path.lstat()
|
||
except FileNotFoundError as exc:
|
||
raise ManifestGenerationError("reference_missing", relative, "纳入文件不存在") from exc
|
||
except OSError as exc:
|
||
raise ManifestGenerationError("reference_unreadable", relative, "纳入文件不可读取") from exc
|
||
if stat.S_ISLNK(file_stat.st_mode):
|
||
raise ManifestGenerationError("reference_symlink", relative, "纳入文件不得是 symlink")
|
||
if not stat.S_ISREG(file_stat.st_mode):
|
||
raise ManifestGenerationError("reference_not_regular", relative, "纳入对象必须是普通文件")
|
||
try:
|
||
raw = path.read_bytes()
|
||
except OSError as exc:
|
||
raise ManifestGenerationError("reference_unreadable", relative, "纳入文件不可读取") from exc
|
||
return relative, raw
|
||
|
||
|
||
def load_registry(repo_root: Path, registry_ref: str, record_id: str) -> dict:
|
||
"""读取当前 /1 registry，只从 active 记录机械取得已批准 designRef。"""
|
||
registry_path = repo_root / registry_ref
|
||
relative, raw = read_regular_file(repo_root, registry_path)
|
||
try:
|
||
registry = json.loads(raw.decode("utf-8"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||
raise ManifestGenerationError("reference_registry_invalid", relative, "registry 不是 UTF-8 JSON") from exc
|
||
if not isinstance(registry, dict) or registry.get("schemaVersion") != "ReferenceAssetRegistry/1":
|
||
raise ManifestGenerationError("reference_registry_invalid", relative, "生成器只接受冻结的 Registry/1 输入")
|
||
matches = [
|
||
record for record in registry.get("records", [])
|
||
if isinstance(record, dict) and record.get("recordId") == record_id
|
||
]
|
||
if len(matches) != 1 or matches[0].get("lifecycleStatus") != "active":
|
||
raise ManifestGenerationError("reference_registry_invalid", relative, "目标记录必须唯一且为 active")
|
||
design_refs = matches[0].get("designRef")
|
||
if not isinstance(design_refs, list) or not design_refs or not all(isinstance(ref, str) for ref in design_refs):
|
||
raise ManifestGenerationError("reference_registry_invalid", relative, "active 记录缺少已批准 designRef")
|
||
return {"record": matches[0], "designRefs": design_refs}
|
||
|
||
|
||
def normalize_source_mode(mode: str) -> str:
|
||
"""规范化输入来源模式，默认只允许真实 Git worktree。"""
|
||
normalized = mode.replace("_", "-")
|
||
if normalized not in ("worktree", "clean-archive"):
|
||
raise ManifestGenerationError("reference_source_mode_invalid", "<input>", "输入来源模式不受支持")
|
||
return normalized
|
||
|
||
|
||
def require_git_worktree(repo_root: Path) -> None:
|
||
"""确认仓根自身是 Git worktree，禁止把父仓或无元数据归档当作真实仓。"""
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "-C", str(repo_root), "rev-parse", "--show-toplevel"],
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
except OSError as exc:
|
||
raise ManifestGenerationError(
|
||
"reference_git_metadata_unavailable", "<input>", "真实 worktree 模式需要可用的 Git 元数据",
|
||
) from exc
|
||
if result.returncode != 0:
|
||
raise ManifestGenerationError(
|
||
"reference_git_metadata_unavailable", "<input>", "真实 worktree 模式需要可用的 Git 元数据",
|
||
)
|
||
try:
|
||
git_root = Path(result.stdout.strip()).resolve()
|
||
except (OSError, ValueError) as exc:
|
||
raise ManifestGenerationError(
|
||
"reference_git_metadata_unavailable", "<input>", "真实 worktree 模式需要可用的 Git 元数据",
|
||
) from exc
|
||
if not result.stdout.strip() or git_root != repo_root.resolve():
|
||
raise ManifestGenerationError(
|
||
"reference_git_metadata_unavailable", "<input>", "真实 worktree 模式需要可用的 Git 元数据",
|
||
)
|
||
|
||
|
||
def is_git_ignored(repo_root: Path, path: Path) -> bool:
|
||
"""使用 Git 原生语义判断候选是否 ignored；未知状态一律拒绝继续。"""
|
||
relative = logical_path(repo_root, path)
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "-C", str(repo_root), "check-ignore", "--quiet", "--", relative],
|
||
capture_output=True,
|
||
check=False,
|
||
)
|
||
except OSError as exc:
|
||
raise ManifestGenerationError(
|
||
"reference_git_check_failed", relative, "Git ignore 检查失败",
|
||
) from exc
|
||
if result.returncode == 0:
|
||
return True
|
||
if result.returncode == 1:
|
||
return False
|
||
raise ManifestGenerationError("reference_git_check_failed", relative, "Git ignore 检查失败")
|
||
|
||
|
||
def is_temporary_output(path: Path) -> bool:
|
||
"""排除固定白名单下可识别的调试、缓存、生成和临时输出命名。"""
|
||
return TEMPORARY_OUTPUT_RE.search(path.name) is not None
|
||
|
||
|
||
def selected_paths(
|
||
repo_root: Path,
|
||
registry_ref: str,
|
||
record_id: str,
|
||
mode: str = DEFAULT_SOURCE_MODE,
|
||
) -> list[Path]:
|
||
"""按固定白名单收集输入，再应用 Git 与临时输出排除规则。"""
|
||
source_mode = normalize_source_mode(mode)
|
||
if source_mode == "worktree":
|
||
require_git_worktree(repo_root)
|
||
game_root = repo_root / "game-runtime/games/shanhai-xingji"
|
||
fixed = [game_root / name for name in ("README.md", "index.html", "entry.js")]
|
||
fixed.append(game_root / "assets/manifest.json")
|
||
fixed.extend(sorted((game_root / "src").glob("*.js"), key=lambda path: logical_path(repo_root, path).encode("utf-8")))
|
||
fixed.extend(sorted((game_root / "assets/atlas").glob("*.json"), key=lambda path: logical_path(repo_root, path).encode("utf-8")))
|
||
design_refs = load_registry(repo_root, registry_ref, record_id)["designRefs"]
|
||
fixed.extend(repo_root / ref for ref in design_refs)
|
||
unique: dict[str, Path] = {}
|
||
for path in fixed:
|
||
relative = logical_path(repo_root, path)
|
||
if relative in unique:
|
||
raise ManifestGenerationError("reference_path_invalid", relative, "消费路径不得重复")
|
||
unique[relative] = path
|
||
selected = []
|
||
for relative in sorted(unique, key=lambda value: value.encode("utf-8")):
|
||
path = unique[relative]
|
||
if is_temporary_output(path):
|
||
continue
|
||
if source_mode == "worktree" and is_git_ignored(repo_root, path):
|
||
continue
|
||
selected.append(path)
|
||
return selected
|
||
|
||
|
||
def build_manifest(
|
||
repo_root: Path,
|
||
registry_ref: str = DEFAULT_REGISTRY.as_posix(),
|
||
record_id: str = DEFAULT_RECORD_ID,
|
||
manifest_id: str = DEFAULT_MANIFEST_ID,
|
||
mode: str = DEFAULT_SOURCE_MODE,
|
||
) -> dict:
|
||
"""读取批准范围并生成只含原始 size/hash 的 manifest 对象。"""
|
||
entries = []
|
||
for path in selected_paths(repo_root, registry_ref, record_id, mode):
|
||
relative, raw = read_regular_file(repo_root, path)
|
||
entries.append({
|
||
"path": relative,
|
||
"size": len(raw),
|
||
"sha256": hashlib.sha256(raw).hexdigest(),
|
||
})
|
||
# 路径排序使用 UTF-8 字节，确保不同语言实现得到同一数组顺序。
|
||
entries.sort(key=lambda entry: entry["path"].encode("utf-8"))
|
||
return {
|
||
"schemaVersion": MANIFEST_SCHEMA_VERSION,
|
||
"manifestId": manifest_id,
|
||
"canonicalization": CANONICALIZATION,
|
||
"entries": entries,
|
||
}
|
||
|
||
|
||
def parse_args(argv: list[str]) -> argparse.Namespace:
|
||
"""解析生成命令参数；默认值固定到仓内首个 active 金标。"""
|
||
parser = argparse.ArgumentParser(description="生成参照资产消费 manifest")
|
||
parser.add_argument("--repo-root", type=Path, required=True, help="可信仓根")
|
||
parser.add_argument("--output", type=Path, required=True, help="manifest 输出文件")
|
||
parser.add_argument("--registry", default=DEFAULT_REGISTRY.as_posix(), help="仓根相对 Registry/1 路径")
|
||
parser.add_argument("--record-id", default=DEFAULT_RECORD_ID, help="active 记录 ID")
|
||
parser.add_argument("--manifest-id", default=DEFAULT_MANIFEST_ID, help="manifest 逻辑 ID")
|
||
parser.add_argument(
|
||
"--mode",
|
||
"--source-mode",
|
||
dest="mode",
|
||
choices=SOURCE_MODES,
|
||
default=DEFAULT_SOURCE_MODE,
|
||
help="输入来源模式；默认 worktree，归档必须显式选择 clean-archive",
|
||
)
|
||
parser.add_argument(
|
||
"--clean-archive",
|
||
dest="mode",
|
||
action="store_const",
|
||
const="clean-archive",
|
||
help="显式选择无 Git 元数据的 clean-archive 输入",
|
||
)
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
"""执行生成并输出可审计的稳定摘要，不输出文件内容或机器绝对路径。"""
|
||
args = parse_args(argv or sys.argv[1:])
|
||
repo_root = args.repo_root.resolve()
|
||
try:
|
||
manifest = build_manifest(
|
||
repo_root,
|
||
registry_ref=args.registry,
|
||
record_id=args.record_id,
|
||
manifest_id=args.manifest_id,
|
||
mode=args.mode,
|
||
)
|
||
raw = canonical_json_bytes(manifest)
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_bytes(raw)
|
||
output_ref = args.output.resolve().relative_to(repo_root).as_posix() if args.output.resolve().is_relative_to(repo_root) else "<output>"
|
||
print(
|
||
f"generated reference manifest recordId={args.record_id} "
|
||
f"entries={len(manifest['entries'])} sha256={hashlib.sha256(raw).hexdigest()} "
|
||
f"path={output_ref}",
|
||
)
|
||
return 0
|
||
except ManifestGenerationError as exc:
|
||
print(
|
||
f"reference_manifest_generation_failed code={exc.code} "
|
||
f"recordId={args.record_id} path={exc.logical_path}",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
except (OSError, ValueError, TypeError) as exc:
|
||
# 未预期的本地 I/O/参数错误仍只报告稳定类别，不泄露绝对路径或输入内容。
|
||
print(
|
||
f"reference_manifest_generation_failed code=reference_generation_error "
|
||
f"recordId={args.record_id} path=<input> detail={type(exc).__name__}",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|