muse-agent-example/工具/验收安装包.py

"""验收指定实际 wheel：隔离解压，仓外运行，缺失及损坏资源必须拒绝。

指定 wheel 时不重新构建；省略参数则增量构建并消费本批 wheel。
不安装依赖、不启动服务；PG/浏览器/恢复验收仍由各自环境承担。
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import sysconfig
import tempfile
import zipfile
from pathlib import Path

_探针 = r"""
import hashlib, json, os, pathlib, socket, sys
def deny_connection(self, address):
    raise AssertionError("包入口解析期间不得连接外部服务")
socket.socket.connect = deny_connection
prefix = pathlib.Path(sys.argv[1]).resolve()
source = pathlib.Path(sys.argv[3]).resolve()
def reject_source_reads(event, args):
    if event in {"open", "os.listdir", "os.scandir"} and args and isinstance(args[0], (str, bytes)):
        target = pathlib.Path(os.fsdecode(args[0])).resolve()
        if target.is_relative_to(source) or ".git" in target.parts:
            raise AssertionError("installed package attempted to read source or Git")
sys.addaudithook(reject_source_reads)
sys.path.insert(0, str(prefix))
sys.path.append(sys.argv[2])
import muse, muse.启动, muse.配置
from muse.启动 import 应用装配, 构建
assert callable(应用装配) and callable(构建)
assert pathlib.Path(muse.__file__).resolve().is_relative_to(prefix), muse.__file__
from muse.资源加载 import 核对资源
from muse.共享.错误 import 资源缺失错误
try:
    manifest = 核对资源()
except 资源缺失错误 as error:
    print(json.dumps({"拒绝": type(error).__name__, "原因": str(error)}, ensure_ascii=False))
    sys.exit(23)
files = {"src/" + f.relative_to(prefix).as_posix(): hashlib.sha256(f.read_bytes()).hexdigest()
         for f in (prefix / "muse").rglob("*")
         if f.suffix in {".py", ".ts"} and "资源" not in f.relative_to(prefix / "muse").parts}
code_hash = hashlib.sha256(json.dumps(files, sort_keys=True).encode()).hexdigest()
assert code_hash == manifest["代码哈希"]
from muse.编排.生成规划 import 规划模板
from muse.审校修订.完整审校 import 文学模板
from muse.资源加载 import 读取能力
for kind, resource in [("outline", "规划/生成作品规划.md"),
                       ("fine_outline", "规划/细化章节计划.md")]:
    text, digest = 规划模板(kind)
    assert text and digest == manifest["资源"][resource]["sha256"]
text, digest = 文学模板()
assert text and digest == manifest["资源"]["审校/创作质量审阅.md"]["sha256"]
for capability in ("write-next-chapter", "check-consistency"):
    assert 读取能力("operation", capability)["正文"]
print(json.dumps({"构建身份": manifest["构建身份"], "资源数": len(manifest["资源"]),
                  "加载位置": str(muse.__file__)}, ensure_ascii=False))
"""


def 验收(轮子: Path) -> dict:
    if not 轮子.is_file() or 轮子.suffix != ".whl":
        raise ValueError("必须提供本批实际 wheel 文件")
    包摘要 = hashlib.sha256(轮子.read_bytes()).hexdigest()
    环境 = {k: v for k, v in os.environ.items() if k != "PYTHONPATH" and not k.startswith("MUSE_")}
    环境["PYTHONDONTWRITEBYTECODE"] = "1"
    with tempfile.TemporaryDirectory(prefix="muse-包验收-") as 临时:
        根 = Path(临时)
        前缀 = 根 / "installed"
        with zipfile.ZipFile(轮子) as zf:
            名称们 = zf.namelist()
            if len(名称们) != len(set(名称们)):
                raise ValueError("wheel 存在重复路径")
            for 名 in 名称们:
                if Path(名).is_absolute() or {
                    "..",
                    ".git",
                    "tests",
                    ".agents.local",
                    ".env",
                }.intersection(Path(名).parts):
                    raise ValueError(f"wheel 包含不允许的路径：{名}")
            zf.extractall(前缀)
        结果 = []

        def 运行(场景: str, 期望: int) -> None:
            进程 = subprocess.run(
                [
                    sys.executable,
                    "-I",
                    "-S",
                    "-B",
                    "-c",
                    _探针,
                    str(前缀),
                    sysconfig.get_path("purelib"),
                    str(Path(__file__).resolve().parents[1]),
                ],
                cwd=根,
                env=环境,
                capture_output=True,
                text=True,
                timeout=120,
            )
            if 进程.returncode != 期望:
                raise RuntimeError(
                    f"{场景} exit={进程.returncode}，期望={期望}\n{进程.stdout}\n{进程.stderr}"
                )
            结果.append({"场景": 场景, "exit": 进程.returncode, "输出": json.loads(进程.stdout)})

        运行("完整实际wheel", 0)
        清单文件 = 前缀 / "muse/资源/清单.json"
        原清单 = 清单文件.read_bytes()
        清单文件.unlink()
        运行("缺清单必须拒绝", 23)
        清单文件.write_bytes(原清单)
        资源 = 前缀 / "muse/资源/运行资源.zip"
        原包 = 资源.read_bytes()
        资源.unlink()
        运行("缺资源必须拒绝", 23)
        资源.write_bytes(原包)
        with zipfile.ZipFile(资源) as zf:
            条目 = [(info, zf.read(info.filename)) for info in zf.infolist()]
        with zipfile.ZipFile(资源, "w") as zf:
            for 序号, (信息, 数据) in enumerate(条目):
                zf.writestr(信息, 数据 + b"tamper" if 序号 == 0 else 数据)
        运行("资源字节损坏必须拒绝", 23)
    if hashlib.sha256(轮子.read_bytes()).hexdigest() != 包摘要:
        raise RuntimeError("验收期间原 wheel 发生变化")
    return {"wheel": str(轮子), "sha256": 包摘要, "场景": 结果}


def 主() -> int:
    解析器 = argparse.ArgumentParser(description=__doc__)
    解析器.add_argument("wheel", type=Path, nargs="?")
    解析器.add_argument("--报告", type=Path)
    参数 = 解析器.parse_args()
    构建报告 = None
    if 参数.wheel is None:
        from 构建编排 import 执行构建

        构建报告 = 执行构建(轮子=True)
        轮子 = Path(构建报告["wheel"])
    else:
        轮子 = 参数.wheel.resolve()
    报告 = 验收(轮子)
    if 构建报告 is not None:
        报告["同批构建"] = 构建报告
    文本 = json.dumps(报告, ensure_ascii=False, indent=2) + "\n"
    if 参数.报告:
        参数.报告.parent.mkdir(parents=True, exist_ok=True)
        参数.报告.write_text(文本)
    print(文本)
    return 0


if __name__ == "__main__":
    raise SystemExit(主())