mil_sdk/build_pyd.py

"""通过 Nuitka 把 mil 包及其纯 Python 伴生包编译为 .pyd，并完成导出、打包、部署、验证全流程。

策略说明：
- **mil 业务包**（mil/ 整个包：core + ui）→ 编 pyd；
- **Qt 桥接代码**（mil/ui/*_ui.py）→ 保留为源码；
- **纯 Python 伴生包**（COMPANIONS 列表）→ 逐个编 pyd，产物随 mil 一起分发；
- **Qt .ui 资源文件** → 运行时由 mil 包内的 uic.loadUiType 直接读源码路径；
- **PySide6** → --nofollow-import-to，运行时走宿主 pip 安装的二进制。

前置条件：
- openpyxl 3.x 的 C 扩展依赖 et_xmlfile 必须升到 >= 2.0（纯 Python 实现），
  否则 Nuitka --module 模式会丢弃 _xmlfile.pyd 导致运行期 ImportError。
    pip install --upgrade --force-reinstall "et_xmlfile>=2.0"
    pip install --upgrade --force-reinstall openpyxl

用法（在工程根 F:\MyProject\mil_sdk 下）:
    python build_pyd.py                 # 编译 + 导出 + 打包 + 部署（默认全流程）
    python build_pyd.py --clean         # 清理 build/ 后走全流程
    python build_pyd.py --no-progress   # 关闭进度条（CI 友好）
    python build_pyd.py --only mil      # 只编指定包（逗号分隔）
    python build_pyd.py --no-deploy     # 不复制 zip 到网络盘
    python build_pyd.py --verify        # 单独验证伴生包加载链路
    python build_pyd.py --verify --fallback  # 验证时允许回退到 site-packages

全流程四步：
    1. Nuitka 编译 .pyd → build/
    2. export 聚合 → plugins/
    3. pack 打包 → dist/<tool>-<ver>-<tag>.zip
    4. 复制 zip → Y:/SE/xufeifei/plugins/（--no-deploy 跳过）

产物布局:
    build/
      mil.cp311-win_amd64.pyd          ← 业务包
    plugins/                           ← 中间产物
    dist/
      mil-<version>-cp311-win_amd64.zip ← 最终交付物
"""
from __future__ import annotations

import argparse
import importlib.util
import json
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent
BUILD_DIR = ROOT / "build"
PLUGINS_DIR = ROOT / "plugins"
DIST_DIR = ROOT / "dist"
REMOTE_DEPLOY_DIR = Path("Y:/SE/xufeifei/plugins")


# ============================================================
# 共享工具函数
# ============================================================

def _python_tag() -> str:
    """返回 Python ABI tag，如 cp311-win_amd64。"""
    impl = "cp" + str(sys.version_info.major) + str(sys.version_info.minor)
    plat = "win_amd64" if platform.system() == "Windows" else platform.machine()
    return f"{impl}-{plat}"


def _read_plugin_meta() -> tuple[str, str, str]:
    """从 mil/__init__.py 源码提取插件名/版本/描述（绕过 PySide6 import）。"""
    src = (ROOT / "mil" / "__init__.py").read_text(encoding="utf-8")
    name = re.search(r'NMAE\s*=\s*"([^"]+)"', src).group(1)
    major = re.search(r"__MAJOR_VER:\s*int\s*=\s*(\d+)", src).group(1)
    minor = re.search(r"__MINOR_VER:\s*int\s*=\s*(\d+)", src).group(1)
    micro = re.search(r"__MICRO_VER:\s*int\s*=\s*(\d+)", src).group(1)
    desc_match = re.search(r'DESCRIPITION\s*=\s*"""([\s\S]*?)"""', src)
    desc = desc_match.group(1).strip() if desc_match else ""
    return name, f"{major}.{minor}.{micro}", desc


# ============================================================
# 伴生包清单（从 manifest.py 加载，绕过 mil/__init__.py 的 PySide6 依赖）
# ============================================================
_spec = importlib.util.spec_from_file_location(
    "mil_manifest", ROOT / "mil" / "core" / "manifest.py"
)
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
COMPANIONS: list[str] = _mod.COMPANION_PACKAGES_PYD


# ============================================================
# Nuitka 编译
# ============================================================

def _resolve_entry(pkg: str) -> Path:
    """定位包的入口 __init__.py。

    业务包（mil）从工程根找；第三方包从当前 Python 的 site-packages 找。
    """
    local = ROOT / pkg / "__init__.py"
    if local.is_file():
        return local

    try:
        spec = importlib.util.find_spec(pkg)
        if spec is not None and spec.origin:
            origin = Path(spec.origin)
            if origin.is_file():
                return origin
    except (ImportError, ValueError):
        pass

    raise FileNotFoundError(f"找不到包 {pkg} 的入口 __init__.py")


def build_one(pkg: str, no_progress: bool) -> Path:
    """对单个包跑 Nuitka --module，产出 <pkg>.<python-tag>.pyd。

    关键约束：--module 模式下要编"包"必须传"包目录路径"（末尾带 /），
    不能传 __init__.py。Nuitka 看到目录就会自动按包模式处理。
    """
    entry = _resolve_entry(pkg)
    pkg_dir = entry.parent
    pkg_name = pkg_dir.name
    is_local = pkg_dir.parent == ROOT

    module_arg = str(pkg_dir) + "/"

    cmd: list[str] = [
        sys.executable, "-m", "nuitka",
        "--module",
        module_arg,
        f"--include-package={pkg_name}",
        f"--output-dir={BUILD_DIR}",
    ]

    if not is_local:
        cmd += [
            "--nofollow-import-to=xml",
            "--nofollow-import-to=xml.etree",
            "--nofollow-import-to=xml.etree.ElementTree",
            "--nofollow-import-to=zipfile",
            "--nofollow-import-to=tempfile",
            "--nofollow-import-to=datetime",
            "--nofollow-import-to=decimal",
            "--nofollow-import-to=re",
            "--nofollow-import-to=json",
            "--nofollow-import-to=argparse",
            "--nofollow-import-to=logging",
        ]

    if is_local:
        for pkg in _mod.COMPANION_PACKAGES_INLINE:
            cmd.append(f"--include-package={pkg}")
        for mod in _mod.COMPANION_STDLIB_INLINE:
            cmd.append(f"--include-module={mod}")
        for ext_pkgs in _mod.COMPANION_C_EXTENSIONS.values():
            for ext in ext_pkgs:
                cmd.append(f"--nofollow-import-to={ext}")
        cmd.append("--nofollow-import-to=pyexpat")

    cmd += ["--no-deployment-flag=frame-useless-set-trace"]
    if no_progress:
        cmd.append("--no-progress")

    subprocess.run(cmd, cwd=str(pkg_dir.parent), check=True)

    candidates = list(BUILD_DIR.glob(f"{pkg_name}.*.pyd"))
    if not candidates:
        candidates = list((BUILD_DIR / pkg_name).glob(f"{pkg_name}.*.pyd"))
    if not candidates:
        raise FileNotFoundError(
            f"Nuitka 未产出预期的 {pkg_name}.*.pyd（已扫遍 build/）"
        )
    return candidates[0]


# ============================================================
# 导出 plugins/ 分发包（原 tools/export_runtime.py）
# ============================================================

def _copy_pyd_files() -> list[str]:
    """把 build/ 下的 .pyd 拷到 plugins/，返回已拷贝的包名列表。"""
    copied: list[str] = []
    for pkg in _mod.COMPANION_PACKAGES + ["mil"]:
        matches = list(BUILD_DIR.glob(f"{pkg}.*.pyd"))
        if not matches:
            matches = list((BUILD_DIR / pkg).glob(f"{pkg}.*.pyd"))
        if not matches:
            print(f"[export] WARN: build/ 下未找到 {pkg}.*.pyd，跳过")
            continue

        dst = PLUGINS_DIR / matches[0].name
        shutil.copy2(matches[0], dst)
        copied.append(pkg)
        print(f"[export] {pkg}: {matches[0].name} -> {dst}")

    return copied


def _copy_c_extensions() -> list[str]:
    """从 Python 安装目录拷贝 C 扩展 .pyd 到 plugins/。"""
    copied: list[str] = []
    py_root = Path(sys.base_prefix)
    search_dirs = [py_root, py_root / "DLLs"]

    for stdlib_pkg, ext_names in _mod.COMPANION_C_EXTENSIONS.items():
        for ext_name in ext_names:
            candidates: list[Path] = []
            for d in search_dirs:
                candidates = list(d.glob(f"{ext_name}.*.pyd"))
                if candidates:
                    break
                candidates = list(d.glob(f"{ext_name}*.pyd"))
                if candidates:
                    break
            if not candidates:
                print(f"[export] WARN: 找不到 C 扩展 {ext_name}.pyd"
                      f"（已查 {', '.join(str(d) for d in search_dirs)}）")
                continue

            dst = PLUGINS_DIR / candidates[0].name
            shutil.copy2(candidates[0], dst)
            copied.append(ext_name)
            print(f"[export] C 扩展 {ext_name}: {candidates[0]} -> {dst}")

    return copied


def _write_manifest(copied_packages: list[str], stdlib_packages: list[str],
                    c_extensions: list[str]) -> None:
    """生成 plugins/manifest.json，供宿主 runtime hook 读取并配置 sys.path。"""
    name, version, desc = _read_plugin_meta()
    manifest = {
        "python_tag": _python_tag(),
        "python_version": platform.python_version(),
        "packages": copied_packages,
        "stdlib": stdlib_packages,
        "c_extensions": c_extensions,
        "plugin_name": name,
        "plugin_version": version,
        "plugin_description": desc,
        "tool_name": "mil",
    }
    dst = PLUGINS_DIR / "manifest.json"
    dst.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"[export] manifest -> {dst}")


def _export_runtime() -> None:
    """导出插件分发包到 plugins/ 目录。"""
    if not BUILD_DIR.exists():
        print("[export] ERROR: build/ 不存在，请先运行 python build_pyd.py", file=sys.stderr)
        raise SystemExit(1)

    if PLUGINS_DIR.exists():
        shutil.rmtree(PLUGINS_DIR)
    PLUGINS_DIR.mkdir(parents=True)

    copied_pyd = _copy_pyd_files()
    copied_cext = _copy_c_extensions()
    _write_manifest(copied_pyd, [], copied_cext)

    print(f"\n[export] 完成！plugins/ 目录已就绪：{PLUGINS_DIR}")
    print(f"[export]   .pyd 文件: {len(copied_pyd)} 个（含已编入 mil.pyd 的 openpyxl/et_xmlfile）")
    print(f"[export]   C 扩展: {len(copied_cext)} 个（运行时从插件目录加载）")
    print(f"[export] Python 版本: {platform.python_version()}")
    print(f"[export] 下一步: 把 plugins/ 目录放到宿主 exe 旁边")


# ============================================================
# 打包 zip 分发包（原 tools/pack_zip.py）
# ============================================================

def _pack_zip() -> None:
    """把 plugins/ 目录打包成带版本号的 zip 分发包。"""
    if not PLUGINS_DIR.exists() or not any(PLUGINS_DIR.iterdir()):
        print("[pack] ERROR: plugins/ 不存在或为空，请先执行导出步骤", file=sys.stderr)
        raise SystemExit(1)

    name, version, _ = _read_plugin_meta()
    manifest = json.loads((PLUGINS_DIR / "manifest.json").read_text(encoding="utf-8"))
    tool_name = manifest["tool_name"]

    zip_name = f"{tool_name}-{version}-{_python_tag()}.zip"
    DIST_DIR.mkdir(parents=True, exist_ok=True)
    zip_path = DIST_DIR / zip_name

    with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for f in PLUGINS_DIR.rglob("*"):
            if f.is_file():
                arcname = f.relative_to(PLUGINS_DIR).as_posix()
                zf.write(f, arcname)

    file_count = len(zipfile.ZipFile(zip_path).namelist())
    print(f"[pack] 打包完成: {zip_path}")
    print(f"[pack]   插件: {name} v{version}")
    print(f"[pack]   文件数: {file_count}")
    print(f"[pack] 下一步: 把 zip 复制到远程 {REMOTE_DEPLOY_DIR}")


# ============================================================
# 部署
# ============================================================

def _deploy_zip() -> None:
    """把 dist/ 下最新生成的 zip 复制到网络盘远程分发目录。"""
    zips = sorted(DIST_DIR.glob("*.zip"), key=lambda f: f.stat().st_mtime, reverse=True)
    if not zips:
        print("[build_pyd] WARN: dist/ 下无 zip 文件，跳过部署", file=sys.stderr)
        return
    src = zips[0]
    if not REMOTE_DEPLOY_DIR.exists():
        print(f"[build_pyd] WARN: 远程目录不可达: {REMOTE_DEPLOY_DIR}，跳过部署",
              file=sys.stderr)
        return
    dst = REMOTE_DEPLOY_DIR / src.name
    shutil.copy2(src, dst)
    print(f"[build_pyd] 步骤 4/4: 已部署 {src.name} -> {dst}")


# ============================================================
# 验证伴生包加载链路（原 tools/verify_companions.py）
# ============================================================

def verify_companions(fallback: bool = False) -> int:
    """验证 build/ 下伴生包 .pyd 的 openpyxl 读写闭环是否可用。

    直接 import mil.core.companion_loader，不经过 mil/__init__.py，
    避免在无 PySide6 的开发机上因 UI 桥接而炸。
    """
    spec = importlib.util.spec_from_file_location(
        "mil_companion_loader", ROOT / "mil" / "core" / "companion_loader.py"
    )
    loader_mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(loader_mod)
    load_companions = loader_mod.load_companions

    packages = list(_mod.COMPANION_PACKAGES)
    loaded = load_companions(packages)
    print(f"[verify] 加载器返回: {loaded}")

    if not loaded and not fallback:
        print(
            "[verify] FAIL: 没有挂载到任何伴生包（是否忘了 python build_pyd.py？）\n"
            "        调试期可加 --fallback 走 site-packages 自测加载器逻辑。",
            file=sys.stderr,
        )
        return 1

    if not loaded and fallback:
        import importlib as _il
        for pkg in packages:
            _il.import_module(pkg)
            loaded.append(pkg)
        print(f"[verify] fallback 后挂载: {loaded}")

    import openpyxl  # noqa: WPS433
    print(f"[verify] openpyxl 来自: {openpyxl.__file__}")

    with tempfile.TemporaryDirectory() as td:
        xlsx_path = Path(td) / "smoke.xlsx"

        wb = openpyxl.Workbook()
        ws = wb.active
        ws["A1"] = "time"
        ws["B1"] = "sig1"
        ws["A2"] = 0.0
        ws["B2"] = 1
        ws["A3"] = 1.0
        ws["B3"] = 2
        wb.save(xlsx_path)

        wb2 = openpyxl.load_workbook(xlsx_path)
        ws2 = wb2.active
        assert ws2["A1"].value == "time", ws2["A1"].value
        assert ws2["B1"].value == "sig1", ws2["B1"].value
        assert float(ws2["A3"].value) == 1.0, ws2["A3"].value
        assert int(ws2["B3"].value) == 2, ws2["B3"].value

    print("[verify] PASS: openpyxl 写读闭环 ✓")
    return 0


# ============================================================
# 主入口
# ============================================================

def main() -> None:
    """编译 mil 包及其伴生依赖为 .pyd，导出为 plugins/ 目录，打包为 zip 并部署至网络盘。

    通过 --verify 可跳过编译，仅验证已编译产物的 openpyxl 读写闭环。
    """
    parser = argparse.ArgumentParser(description="编译 mil + 伴生包为 .pyd，并打包成 zip")
    parser.add_argument("--clean", action="store_true", help="清理 build/ 与上次产物")
    parser.add_argument("--no-progress", action="store_true", help="关闭进度条（CI 友好）")
    parser.add_argument("--no-deploy", action="store_true", help="不复制 zip 到网络盘")
    parser.add_argument("--only", type=str, default="",
                        help="只编指定包（逗号分隔），如 --only mil,openpyxl；默认全编")
    parser.add_argument("--verify", action="store_true", help="验证伴生包 .pyd 加载链路（不编译）")
    parser.add_argument("--fallback", action="store_true",
                        help="验证时，未发现产物则回退到 site-packages（开发自测用）")
    args = parser.parse_args()

    if args.verify:
        sys.exit(verify_companions(args.fallback))

    if args.clean and BUILD_DIR.exists():
        shutil.rmtree(BUILD_DIR)

    targets = ["mil"]
    if args.only:
        targets = [p.strip() for p in args.only.split(",") if p.strip()]
    else:
        targets += COMPANIONS

    print("[build_pyd] 步骤 1/4: Nuitka 编译 .pyd")
    for pkg in targets:
        print(f"[build_pyd] 正在编译: {pkg}")
        out = build_one(pkg, args.no_progress)
        print(f"[build_pyd] {pkg} -> {out}")

    try:
        print("[build_pyd] 步骤 2/4: 导出 plugins/ 分发包")
        _export_runtime()

        print("[build_pyd] 步骤 3/4: 打包 zip")
        _pack_zip()
    except SystemExit as e:
        if e.code != 0:
            print(f"[build_pyd] 打包流程中止（退出码 {e.code}）", file=sys.stderr)
            sys.exit(e.code)

    if not args.no_deploy:
        _deploy_zip()


if __name__ == "__main__":
    main()