118 lines
5.4 KiB
Python
118 lines
5.4 KiB
Python
"""为无损底座迁移保存并校验工作区的只读 Git 状态证据。"""
|
|
|
|
# 导入命令行参数解析器以支持生成和校验报告。
|
|
import argparse
|
|
# 导入 JSON 序列化工具以写入 UTF-8 状态证据。
|
|
import json
|
|
# 导入子进程工具以直接调用 Git 而不经过 shell。
|
|
import subprocess
|
|
# 导入可调用协议和类型别名支持。
|
|
from collections.abc import Callable
|
|
# 导入路径类型以约束工作区和报告位置。
|
|
from pathlib import Path
|
|
# 导入任意 JSON 对象的静态类型。
|
|
from typing import Any
|
|
|
|
# 定义只允许执行的 Git 只读子命令首参数集合。
|
|
READ_ONLY_GIT_COMMANDS = frozenset({"branch", "rev-parse", "status"})
|
|
# 定义便于测试注入的 Git 调用函数类型。
|
|
GitRunner = Callable[..., str]
|
|
|
|
|
|
# 执行受限的 Git 只读命令并返回标准输出文本。
|
|
def run_git(worktree: Path, *args: str) -> str:
|
|
# 解析目标工作区以确保 Git 和安全目录使用同一个绝对路径。
|
|
resolved = worktree.resolve()
|
|
# 拒绝空命令，避免形成未约束的 Git 调用。
|
|
if not args:
|
|
raise ValueError("git command is required")
|
|
# 拒绝任何不在白名单中的 Git 子命令。
|
|
if args[0] not in READ_ONLY_GIT_COMMANDS:
|
|
raise ValueError("git command is not read-only")
|
|
# 将安全目录限定为当前查询工作区，避免写入全局 Git 配置。
|
|
safe_directory = resolved.as_posix()
|
|
# 以参数数组执行 Git，禁止 shell 解释路径或输入内容。
|
|
result = subprocess.run(
|
|
["git", "-c", f"safe.directory={safe_directory}", "-C", str(resolved), *args],
|
|
check=True,
|
|
capture_output=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
# 返回 Git 的标准输出，供调用方以确定性方式解析。
|
|
return result.stdout
|
|
|
|
|
|
# 收集一个工作区的分支、提交与简短状态，不读取环境变量或业务数据。
|
|
def collect_workspace_state(worktree: Path, runner: GitRunner = run_git) -> dict[str, object]:
|
|
# 解析绝对路径，防止报告中出现随当前目录变化的相对路径。
|
|
resolved = worktree.resolve()
|
|
# 依次执行已白名单化的 Git 查询并构造安全状态字典。
|
|
return {
|
|
"path": str(resolved),
|
|
"branch": runner(resolved, "branch", "--show-current").strip(),
|
|
"head": runner(resolved, "rev-parse", "HEAD").strip(),
|
|
"status": runner(resolved, "status", "--short").splitlines(),
|
|
}
|
|
|
|
|
|
# 将已收集的状态以 UTF-8 JSON 写入调用者指定的报告文件。
|
|
def write_preflight_report(target: Path, states: list[dict[str, object]]) -> None:
|
|
# 确保报告父目录存在，但不创建或改动任何工作区内容。
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
# 使用稳定缩进和 UTF-8 编码写入仅由调用方提供的状态数据。
|
|
target.write_text(json.dumps(states, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
# 从磁盘读取先前报告并验证指定工作区状态完全一致。
|
|
def verify_preflight_report(target: Path, worktrees: list[Path]) -> list[str]:
|
|
# 以 UTF-8 读取 JSON 报告，避免系统默认编码影响比较。
|
|
expected_value: Any = json.loads(target.read_text(encoding="utf-8"))
|
|
# 拒绝非列表报告，防止错误文件被误当作迁移证据。
|
|
if not isinstance(expected_value, list):
|
|
raise ValueError("preflight report must be a list")
|
|
# 为每个当前工作区重新采集只读 Git 状态。
|
|
current = [collect_workspace_state(worktree) for worktree in worktrees]
|
|
# 返回 JSON 表示不同的工作区路径，空列表代表完全一致。
|
|
return [
|
|
str(item["path"])
|
|
for item, expected in zip(current, expected_value, strict=True)
|
|
if item != expected
|
|
]
|
|
|
|
|
|
# 解析 CLI 工作区参数并执行报告生成或校验。
|
|
def main() -> int:
|
|
# 创建命令行解析器并限定所有输入为显式路径。
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
# 要求调用者指定报告文件路径。
|
|
parser.add_argument("--report", type=Path, required=True)
|
|
# 允许多次提供要采集或校验的工作区路径。
|
|
parser.add_argument("--worktree", type=Path, action="append", required=True)
|
|
# 启用校验模式时不覆盖报告。
|
|
parser.add_argument("--verify", action="store_true")
|
|
# 解析用户传入的参数。
|
|
arguments = parser.parse_args()
|
|
# 在校验模式下输出差异并返回非零状态。
|
|
if arguments.verify:
|
|
# 比较当前状态和报告状态。
|
|
differences = verify_preflight_report(arguments.report, arguments.worktree)
|
|
# 输出机器和人工都可识别的校验结果。
|
|
print("UNCHANGED" if not differences else f"CHANGED: {', '.join(differences)}")
|
|
# 有任何差异时返回失败状态。
|
|
return 0 if not differences else 1
|
|
# 收集调用方明确列出的工作区状态。
|
|
states = [collect_workspace_state(worktree) for worktree in arguments.worktree]
|
|
# 写入新的迁移前证据报告。
|
|
write_preflight_report(arguments.report, states)
|
|
# 输出报告保存位置，避免输出任何敏感运行配置。
|
|
print(f"WROTE: {arguments.report}")
|
|
# 报告创建成功时返回零状态。
|
|
return 0
|
|
|
|
|
|
# 仅在脚本直接执行时运行命令行入口。
|
|
if __name__ == "__main__":
|
|
# 用 main 的返回值作为进程退出码。
|
|
raise SystemExit(main())
|