0658d178ad
- Makefile：新增 验收数据库（生成→检查→库层全量，前置校验隔离库连接串）、数据库分片、浏览器测试三个入口； 格式写入 末尾就地重新生成；数据库测试 与 验收数据库 统一排除 浏览器/网络；快检 不再静默跳过类型门； 前端旅程 预检补齐四个必需变量；pytest 目标改用仓内解释器（uv run 在嵌套检出会解析到外层环境）。 - 工具/验证锁.py：验证会话持共享锁，写入口用 --执行 在独占锁内落盘，生成/格式写入/索引生成走同一闸门。 - 工具/并行数据库测试.py：按文件分片并行，缺连接串在采集前拒绝。 - 工具/构建编排.py、环境预检.py、构建资源包.py、维护索引.py 与上述口径对齐。 - .gitignore / CI / README / AGENTS：忽略构建产物、CI 与 Makefile 单一口径、README 按实际实现陈述、AGENTS 补日常入口与验证纪律。
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""生成运行资源与发布清单。
|
||
|
||
工程阶段的生成或预检入口：
|
||
- 从唯一登记输入（工具/资源登记.json）打包 src/muse/资源/运行资源.zip；
|
||
- 写出 src/muse/资源/清单.json（发布身份 + 每个资源的相对路径与 sha256）；
|
||
- 投影 web/dist → src/muse/资源/工作台/（web 构建存在时）。
|
||
|
||
边界：只打包已登记生产资源；不读取未审工作树内容、不打包 tests/.git/当前目录；
|
||
生成物幂等可重现（固定排序、固定压缩时间戳）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import shutil
|
||
import tempfile
|
||
import tomllib
|
||
import zipfile
|
||
from pathlib import Path, PurePosixPath
|
||
from typing import Any
|
||
|
||
from muse.资源加载 import 解析能力声明
|
||
from 效果依赖 import 构建效果依赖
|
||
|
||
仓库根 = Path(__file__).resolve().parent.parent
|
||
登记文件 = 仓库根 / "工具" / "资源登记.json"
|
||
资源目录 = 仓库根 / "src" / "muse" / "资源"
|
||
压缩包名 = "运行资源.zip"
|
||
清单名 = "清单.json"
|
||
_固定时间戳 = (2026, 1, 1, 0, 0, 0)
|
||
|
||
|
||
def 读取登记() -> dict[str, Any]:
|
||
if not 登记文件.is_file():
|
||
raise SystemExit(f"资源登记不存在：{登记文件}（打包前必须先登记生产资源）")
|
||
登记: Any = json.loads(登记文件.read_text(encoding="utf-8"))
|
||
if not isinstance(登记, dict) or "资源" not in 登记:
|
||
raise SystemExit("资源登记不合法：缺少 资源 表")
|
||
return 登记
|
||
|
||
|
||
def _sha256(文件: Path) -> str:
|
||
return hashlib.sha256(文件.read_bytes()).hexdigest()
|
||
|
||
|
||
def 编译能力资源(
|
||
根: Path, 文件清单: list[str], 已实现命令: set[str]
|
||
) -> tuple[dict, dict[str, bytes]]:
|
||
目录: dict[str, dict] = {"role": {}, "method": {}, "operation": {}}
|
||
资源: dict[str, bytes] = {}
|
||
for 相对 in sorted(文件清单):
|
||
文件 = (根 / 相对).resolve()
|
||
if not 文件.is_relative_to(根.resolve()) or not 文件.is_file():
|
||
raise ValueError("能力资源必须是登记根内的实际文件")
|
||
数据 = 文件.read_bytes()
|
||
声明, _ = 解析能力声明(数据)
|
||
分类, 身份 = 声明["category"], 声明["id"]
|
||
if 身份 in 目录[分类]:
|
||
raise ValueError("能力稳定身份重复")
|
||
命令 = 声明.get("commands", [])
|
||
if not isinstance(命令, list) or any(
|
||
not isinstance(v, str) or v not in 已实现命令 for v in 命令
|
||
):
|
||
raise ValueError("能力声明包含未实现命令")
|
||
if 分类 == "operation" and not 命令:
|
||
raise ValueError("操作技能必须声明已实现命令")
|
||
路径 = f"能力/{分类}/{身份}.md"
|
||
资源[路径] = 数据
|
||
目录[分类][身份] = {
|
||
"id": 身份,
|
||
"name": 声明["name"],
|
||
"description": 声明["description"],
|
||
"contract_version": 声明["contract_version"],
|
||
"commands": 命令,
|
||
"resource_path": 路径,
|
||
"sha256": hashlib.sha256(数据).hexdigest(),
|
||
}
|
||
return 目录, 资源
|
||
|
||
|
||
def _仓内源(相对: str, *, 目录: bool = False) -> Path:
|
||
路径 = 仓库根 / 相对
|
||
if Path(相对).is_absolute() or not 路径.resolve().is_relative_to(仓库根.resolve()):
|
||
raise ValueError(f"登记源必须位于仓库内：{相对}")
|
||
if not (路径.is_dir() if 目录 else 路径.is_file()):
|
||
raise ValueError(f"登记源不存在或类型不符：{相对}")
|
||
return 路径
|
||
|
||
|
||
def _预检资源(登记: dict, 能力资源: dict[str, bytes]) -> tuple[list[Path], list[Path]]:
|
||
"""先统一核对源边界和所有目标身份；此阶段不创建 ZIP 或读取 web 文件字节。"""
|
||
目标们: set[str] = set()
|
||
|
||
def 登记目标(路径: str) -> None:
|
||
规范 = PurePosixPath(路径)
|
||
if (
|
||
not 路径
|
||
or "\\" in 路径
|
||
or "\x00" in 路径
|
||
or 规范.is_absolute()
|
||
or (规范.parts and ":" in 规范.parts[0])
|
||
or ".." in 规范.parts
|
||
or 规范.as_posix() != 路径
|
||
or 路径 == "."
|
||
):
|
||
raise ValueError(f"资源目标必须是规范化相对路径：{路径}")
|
||
if 路径 in 目标们:
|
||
raise ValueError(f"资源目标重复：{路径}")
|
||
目标们.add(路径)
|
||
|
||
for 条 in 登记["资源"]:
|
||
_仓内源(str(条["源路径"]))
|
||
登记目标(str(条["打包路径"]))
|
||
for 路径 in 能力资源:
|
||
登记目标(路径)
|
||
迁移文件 = []
|
||
if 登记.get("迁移目录"):
|
||
源目录 = _仓内源(str(登记["迁移目录"]), 目录=True)
|
||
for 文件 in sorted(源目录.glob("V*.sql")):
|
||
_仓内源(文件.relative_to(仓库根).as_posix())
|
||
登记目标(f"迁移/{文件.name}")
|
||
迁移文件.append(文件)
|
||
web_dist = 仓库根 / "web/dist"
|
||
if web_dist.is_symlink():
|
||
raise ValueError("web/dist 不得是软链")
|
||
web文件 = []
|
||
if web_dist.exists():
|
||
if not web_dist.is_dir():
|
||
raise ValueError("web/dist 必须是目录")
|
||
后代 = sorted(web_dist.rglob("*"))
|
||
if any(p.is_symlink() for p in 后代):
|
||
raise ValueError("web/dist 后代不得含软链")
|
||
for 文件 in 后代:
|
||
if 文件.is_file():
|
||
_仓内源(文件.relative_to(仓库根).as_posix())
|
||
登记目标("工作台/" + 文件.relative_to(web_dist).as_posix())
|
||
web文件.append(文件)
|
||
return 迁移文件, web文件
|
||
|
||
|
||
def _写入资源(资源目录: Path) -> dict[str, Any]:
|
||
登记 = 读取登记()
|
||
发布身份 = str(登记.get("发布身份", ""))
|
||
if not 发布身份.strip():
|
||
raise SystemExit("资源登记缺少 发布身份")
|
||
条目: list[dict[str, str]] = 登记["资源"]
|
||
能力资源: dict[str, bytes] = {}
|
||
if 登记.get("能力文件"):
|
||
from muse.接入.http.应用 import 导出合同
|
||
|
||
合同 = 导出合同()
|
||
命令 = {
|
||
值["operationId"]
|
||
for 路径 in 合同["paths"].values()
|
||
for 值 in 路径.values()
|
||
if "operationId" in 值
|
||
}
|
||
目录, 能力资源 = 编译能力资源(仓库根, 登记["能力文件"], 命令)
|
||
能力资源["能力/目录.json"] = json.dumps(
|
||
目录, ensure_ascii=False, sort_keys=True, indent=2
|
||
).encode()
|
||
迁移文件, web文件 = _预检资源(登记, 能力资源)
|
||
资源目录.mkdir(parents=True, exist_ok=True)
|
||
压缩包路径 = 资源目录 / 压缩包名
|
||
|
||
已打包: dict[str, str] = {}
|
||
with zipfile.ZipFile(压缩包路径, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||
for 条 in sorted(条目, key=lambda x: str(x.get("打包路径", ""))):
|
||
源 = 仓库根 / str(条["源路径"])
|
||
打包路径 = str(条["打包路径"])
|
||
if not 源.is_file():
|
||
raise SystemExit(f"登记资源不存在：{源}")
|
||
信息 = zipfile.ZipInfo(打包路径, date_time=_固定时间戳)
|
||
信息.compress_type = zipfile.ZIP_DEFLATED
|
||
信息.external_attr = 0o644 << 16
|
||
数据 = 源.read_bytes()
|
||
zf.writestr(信息, 数据)
|
||
已打包[打包路径] = hashlib.sha256(数据).hexdigest()
|
||
for 路径, 数据 in sorted(能力资源.items()):
|
||
if 路径 in 已打包:
|
||
raise ValueError("生成能力资源与直接登记路径重复")
|
||
信息 = zipfile.ZipInfo(路径, date_time=_固定时间戳)
|
||
信息.compress_type = zipfile.ZIP_DEFLATED
|
||
信息.external_attr = 0o644 << 16
|
||
zf.writestr(信息, 数据)
|
||
已打包[路径] = hashlib.sha256(数据).hexdigest()
|
||
|
||
# 可编辑安装和正式 wheel 使用相同的包内迁移入口。
|
||
迁移源 = 登记.get("迁移目录")
|
||
if 迁移源:
|
||
源目录 = 仓库根 / 迁移源
|
||
if not 源目录.is_dir():
|
||
raise SystemExit("登记的迁移目录不存在")
|
||
迁移资源 = 资源目录 / "迁移"
|
||
if 迁移资源.exists():
|
||
shutil.rmtree(迁移资源)
|
||
迁移资源.mkdir()
|
||
for 文件 in 迁移文件:
|
||
数据 = 文件.read_bytes()
|
||
(迁移资源 / 文件.name).write_bytes(数据)
|
||
已打包[f"迁移/{文件.name}"] = hashlib.sha256(数据).hexdigest()
|
||
|
||
# 效果依赖按登记/编译/迁移生成阶段取全量保守闭包，不纳入工作台产物。
|
||
效果依赖 = 构建效果依赖(
|
||
仓库根,
|
||
dict(已打包),
|
||
{
|
||
"registered": sorted(str(条["打包路径"]) for 条 in 条目),
|
||
"capabilities": sorted(能力资源),
|
||
"migrations": sorted(
|
||
set(已打包) - {str(条["打包路径"]) for 条 in 条目} - set(能力资源)
|
||
),
|
||
},
|
||
)
|
||
|
||
# 工作台投影：web 构建产物存在时带上；不存在时如实留空。
|
||
工作台目录 = 资源目录 / "工作台"
|
||
if 工作台目录.exists():
|
||
shutil.rmtree(工作台目录)
|
||
web_dist = 仓库根 / "web" / "dist"
|
||
if web_dist.is_dir():
|
||
工作台目录.mkdir()
|
||
for 文件 in web文件:
|
||
if 文件.is_symlink() or any(p.is_symlink() for p in 文件.parents if p != 仓库根):
|
||
raise ValueError("web/dist 在预检后变为软链")
|
||
相对 = 文件.relative_to(web_dist)
|
||
目标 = 工作台目录 / 相对
|
||
目标.parent.mkdir(parents=True, exist_ok=True)
|
||
数据 = 文件.read_bytes()
|
||
目标.write_bytes(数据)
|
||
已打包["工作台/" + 相对.as_posix()] = hashlib.sha256(数据).hexdigest()
|
||
|
||
清单 = {
|
||
"schema_version": 1,
|
||
"效果依赖": 效果依赖,
|
||
"发布身份": 发布身份,
|
||
"资源": {路径: {"sha256": 哈希} for 路径, 哈希 in sorted(已打包.items())},
|
||
}
|
||
代码文件 = {
|
||
文件.relative_to(仓库根).as_posix(): _sha256(文件)
|
||
for 文件 in (仓库根 / "src/muse").rglob("*")
|
||
if 文件.suffix in {".py", ".ts"}
|
||
and "资源" not in 文件.relative_to(仓库根 / "src/muse").parts
|
||
}
|
||
代码哈希 = hashlib.sha256(json.dumps(代码文件, sort_keys=True).encode()).hexdigest()
|
||
with (仓库根 / "pyproject.toml").open("rb") as 文件:
|
||
包版本 = tomllib.load(文件)["project"]["version"]
|
||
清单.update({"代码版本": 包版本, "代码哈希": 代码哈希, "合同版本": 1})
|
||
清单["构建身份"] = hashlib.sha256(
|
||
json.dumps(清单, ensure_ascii=False, sort_keys=True).encode()
|
||
).hexdigest()
|
||
(资源目录 / 清单名).write_text(
|
||
json.dumps(清单, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||
)
|
||
return 清单
|
||
|
||
|
||
def 打包() -> dict[str, Any]:
|
||
"""在独立目录完成全部输入校验与输出，再替换旧资源；失败保留旧产物。"""
|
||
资源目录.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.TemporaryDirectory(prefix=".资源构建-", dir=资源目录.parent) as 临时:
|
||
暂存 = Path(临时) / "新资源"
|
||
清单 = _写入资源(暂存)
|
||
旧资源 = Path(临时) / "旧资源"
|
||
if 资源目录.exists():
|
||
资源目录.rename(旧资源)
|
||
try:
|
||
暂存.rename(资源目录)
|
||
except BaseException:
|
||
if 旧资源.exists():
|
||
旧资源.rename(资源目录)
|
||
raise
|
||
return 清单
|
||
|
||
|
||
def 主() -> int:
|
||
解析器 = argparse.ArgumentParser(description="生成运行资源与发布清单")
|
||
解析器.parse_args()
|
||
清单 = 打包()
|
||
print(f"构建资源包：发布身份 {清单['发布身份']}，资源 {len(清单['资源'])} 项")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(主())
|