vent_agent_2.0/release/build.py

#!/usr/bin/env python3
"""构建版本化 Linux 部署包；默认打包当前工作区并运行离线检查。"""
from __future__ import annotations

import argparse
from datetime import datetime, timezone
import gzip
from importlib import metadata
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tarfile
import tempfile

sys.dont_write_bytecode = True
from common import sha256, validate_package, validation_env, version

ROOT = Path(__file__).resolve().parents[1]
ROOT_FILES = {"main.py", "config.py", "README.md", "docs/DEVELOPMENT.md", "docs/BUSINESS_AGENTS.md", "docs/MAIN_FAN.md", "docs/TEST_LAB.md", "requirements.txt", "pytest.ini", "VERSION", ".env.example"}
TREE_SUFFIXES = {
    "algorithms": {".py", ".md", ".json", ".txt"},
    "runtime": {".py"},
    "extensions": {".py", ".md", ".txt", ".json", ".yaml", ".yml"},
    "agents": {".py"}, "api": {".py"}, "billing": {".py"}, "db": {".py", ".sql"},
    "obs": {".py"}, "scheduler": {".py"}, "tools": {".py"},
    "skills": {".md", ".txt", ".json", ".yaml", ".yml"},
    "static": {".html", ".css", ".js", ".md", ".LICENSE", ".svg", ".png", ".ico", ".woff", ".woff2"},
    "tests": {".py", ".html"},
    "mock_gateway": {".py"},
    "mock_result_platform": {".py"},
    "examples": {".json", ".md"},
}
RELEASE_FILES = {"build.py", "common.py", "deploy.py", "verify.py", "install.sh", "service.sh",
                 "lock.py", "requirements.lock", "lock.meta.json", "baseline.constraints", "README.md",
                 "vent-agent.service.example"}
EXCLUDED_PARTS = {"__pycache__", "node_modules", "data", "workspace", "logs", "dist", "venv"}


def source_files(root: Path) -> list[Path]:
    """白名单收集源文件；运行目录、隐藏文件和符号链接不会被打包。"""
    selected = [root / name for name in sorted(ROOT_FILES)]
    selected += [root / "release" / name for name in sorted(RELEASE_FILES)]
    selected += [root / "scripts" / name for name in ("selfcheck.py", "demo_main_fan.py", "mock_shaanxi_gateway.py", "mock_result_platform.py", "import_main_fan_facts.py", "run_main_fan_document.py", "refresh_main_fan_mines.py", "run_main_fan_batch.py", "mock_main_fan_reports.py", "verify_main_fan_batch.py", "compact_main_fan_history.py")]
    for directory, suffixes in TREE_SUFFIXES.items():
        base = root / directory
        if base.is_symlink():
            raise ValueError(f"不接受源目录符号链接：{directory}")
        for current, dirs, names in os.walk(base, followlinks=False):
            dirs[:] = sorted(d for d in dirs if not d.startswith(".") and d not in EXCLUDED_PARTS)
            for directory_name in dirs:
                if (Path(current) / directory_name).is_symlink():
                    raise ValueError(f"不接受源目录符号链接：{Path(current) / directory_name}")
            for name in sorted(names):
                path = Path(current) / name
                if not name.startswith(".") and path.suffix in suffixes:
                    selected.append(path)
    for path in selected:
        if not path.is_file() or any(p.is_symlink() for p in (path, *path.parents) if p != root and root in p.parents):
            raise ValueError(f"源文件缺失或是符号链接：{path.relative_to(root)}")
    return sorted(selected, key=lambda p: p.relative_to(root).as_posix())


def git_info(root: Path) -> dict:
    def git(*args):
        try:
            result = subprocess.run(["git", "-C", str(root), *args], text=True, capture_output=True)
        except FileNotFoundError:
            return None
        return result.stdout.strip() if result.returncode == 0 else None
    status = git("status", "--porcelain")
    return {"commit": git("rev-parse", "HEAD"), "dirty": None if status is None else bool(status)}


def verify_lock(root: Path, check_environment: bool) -> None:
    meta = json.loads((root / "release/lock.meta.json").read_text())
    for name in ("requirements.txt", "release/baseline.constraints", "release/requirements.lock"):
        if meta["sha256"].get(name) != sha256(root / name):
            raise ValueError(f"依赖输入或锁文件已变化，请先运行 release/lock.py：{name}")
    if not check_environment:
        return
    from packaging.requirements import Requirement
    # uv 的 requirements 输出：每项以 name==version 开始，后接哈希续行。
    for line in (root / "release/requirements.lock").read_text().splitlines():
        if not line or line[0].isspace() or line.startswith("#"):
            continue
        req = Requirement(line.removesuffix("\\").strip())
        if req.marker and not req.marker.evaluate():
            continue
        if metadata.version(req.name) not in req.specifier:
            raise ValueError(f"构建环境与锁文件不一致：{req.name}；请安装 release/requirements.lock")


def build(root: Path, output: Path, release_version: str, *, checks: bool = True) -> Path:
    if sys.version_info[:2] != (3, 12):
        raise ValueError("本发布流程使用 Python 3.12")
    release_version = version(release_version)
    verify_lock(root, checks)
    output.mkdir(parents=True, exist_ok=True)
    name = f"vent-agent-{release_version}-linux"
    archive = output / f"{name}.tar.gz"
    checksum = output / f"{name}.tar.gz.sha256"
    if archive.exists() or checksum.exists():
        raise ValueError(f"版本产物已存在，不覆盖：{archive}")
    with tempfile.TemporaryDirectory(prefix="vent-release-") as temporary:
        temp = Path(temporary)
        staged = temp / name
        staged.mkdir()
        sources = source_files(root)
        for source in sources:
            target = staged / source.relative_to(root)
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(source, target)
            target.chmod(0o755 if target.suffix == ".sh" else 0o644)
        (staged / "VERSION").write_text(release_version + "\n")
        if checks:
            env = validation_env(temp)
            subprocess.run([sys.executable, "-m", "pytest", "-p", "no:cacheprovider"], cwd=staged, env=env, check=True)
            subprocess.run([sys.executable, "scripts/selfcheck.py"], cwd=staged, env=env, check=True)
            # 自检可能建出空运行目录。它们不会进入发布包。
        files = {}
        for source in sources:
            relative = source.relative_to(root).as_posix()
            path = staged / relative
            files[relative] = {"sha256": sha256(path), "size": path.stat().st_size}
        manifest = {"format": 1, "version": release_version, "python": "3.12", "target": "linux",
                    "built_at": datetime.now(timezone.utc).isoformat(), "source": git_info(root),
                    "checks": "passed" if checks else "skipped", "files": files}
        (staged / "release-manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
        temporary_archive = temp / archive.name
        with temporary_archive.open("wb") as raw, gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz:
            with tarfile.open(fileobj=gz, mode="w") as tar:
                for relative in sorted([*files, "release-manifest.json"]):
                    path = staged / relative
                    info = tar.gettarinfo(str(path), arcname=f"{name}/{relative}")
                    info.uid = info.gid = 0
                    info.uname = info.gname = ""
                    info.mtime = 0
                    with path.open("rb") as content:
                        tar.addfile(info, content)
        # 验证真正归档后的内容，避免只检查 staging 而遗漏打包错误。
        extracted = temp / "verify"
        with tarfile.open(temporary_archive) as tar:
            tar.extractall(extracted, filter="data")
        validate_package(extracted / name)
        # 使用排他创建，重复构建同一版本不会覆盖已有发布文件。
        with archive.open("xb") as target, temporary_archive.open("rb") as source:
            shutil.copyfileobj(source, target)
        with checksum.open("x") as target:
            target.write(f"{sha256(archive)}  {archive.name}\n")
    return archive


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--version", default=(ROOT / "VERSION").read_text().strip())
    parser.add_argument("--output", type=Path, default=ROOT / "release/dist")
    parser.add_argument("--skip-checks", action="store_true", help="仅调试：跳过检查，清单将标记 skipped")
    args = parser.parse_args()
    try:
        path = build(ROOT, args.output.resolve(), args.version, checks=not args.skip_checks)
    except (ValueError, subprocess.CalledProcessError, metadata.PackageNotFoundError) as exc:
        parser.exit(1, f"构建失败：{exc}\n")
    print(f"发布包：{path}\n校验文件：{path}.sha256")


if __name__ == "__main__":
    main()