vent_agent_2.0/release/common.py

"""发布脚本共用的版本、清单和隔离验证环境。仅依赖 Python 标准库。"""
from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re

VERSION_RE = re.compile(r"(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*)?\Z")


def version(value: str) -> str:
    value = value.strip()
    if not VERSION_RE.fullmatch(value):
        raise ValueError("版本格式应为 2.0.0 或 2.0.0-rc.1")
    return value


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def validate_package(root: Path, *, installed: bool = False) -> dict:
    root = root.resolve()
    manifest = json.loads((root / "release-manifest.json").read_text())
    if manifest.get("format") != 1 or manifest.get("python") != "3.12":
        raise ValueError("不支持的发布清单或 Python 版本")
    if version((root / "VERSION").read_text()) != version(manifest["version"]):
        raise ValueError("VERSION 与发布清单不一致")
    files = manifest.get("files", {})
    for required in ("VERSION", "main.py", "config.py", ".env.example", "release/deploy.py", "release/requirements.lock"):
        if required not in files:
            raise ValueError(f"发布包缺少必需文件：{required}")
    for name, expected in files.items():
        relative = PurePosixPath(name)
        if relative.is_absolute() or ".." in relative.parts or "\\" in name:
            raise ValueError(f"非法清单路径：{name}")
        path = root / name
        if any(part.is_symlink() for part in (path, *path.parents) if part != root and root in part.parents):
            raise ValueError(f"发布文件不允许是符号链接：{name}")
        if not path.is_file() or sha256(path) != expected["sha256"]:
            raise ValueError(f"文件缺失或校验失败：{name}")
    if not installed:
        actual = set()
        for path in root.rglob("*"):
            if path.is_symlink():
                raise ValueError(f"发布包包含符号链接：{path.relative_to(root)}")
            if path.is_file():
                actual.add(path.relative_to(root).as_posix())
        if actual != set(files) | {"release-manifest.json"}:
            raise ValueError("发布包包含清单外文件")
    return manifest


def validation_env(temporary: Path) -> dict[str, str]:
    """测试和自检只使用临时 SQLite 与假密钥，不读取服务器配置。"""
    env = dict(os.environ)
    env.update(DEEP_AGENT_ENV_PRIORITY="none", DB_DIALECT="sqlite",
               DB_SQLITE_PATH=str(temporary / "validation.sqlite"), CHECKPOINTER="memory",
               OPENAI_API_KEY="sk-release-offline", OPENAI_BASE_URL="https://example.invalid",
               DEEPAGENT_MODEL="deepseek-flash", BILLING_ALLOW_UNPRICED="0",
               PYTHONDONTWRITEBYTECODE="1", PYTHONPATH="", LOG_LEVEL="WARNING")
    env["ALGORITHM_JOB_DIR"] = str(temporary / "algorithm_jobs")
    return env