feat: 增加伴生包加载器、manifest配置,Excel处理优化

This commit is contained in:
2026-07-21 16:26:16 +08:00
parent 45e4f1ecf2
commit 91bb61444e
19 changed files with 1364 additions and 82 deletions
+283
View File
@@ -0,0 +1,283 @@
"""导出插件分发包到 plugins/ 目录。
把以下内容打包到 plugins/
1. build/ 下的 .pydmil + openpyxl + et_xmlfile
2. 从当前 Python 环境拷贝 stdlib 子包(.py 文件)
3. 从 Python 安装根目录拷贝 C 扩展 .pyd(如 _elementtree.pyd
4. 生成 manifest.json 供 runtime_hook.py 读取
产物 plugins/ 目录就是完整的"插件分发包"——宿主直接放到 exe 旁边即可。
用法:
# 前置:先编译 .pyd
python build_pyd.py --clean
# 导出插件分发包
python tools/export_runtime.py
# 产物:plugins/ 目录
"""
from __future__ import annotations
import json
import platform
import shutil
import sys
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
BUILD_DIR = ROOT / "build"
PLUGINS_DIR = ROOT / "plugins"
# 用 importlib 加载 manifest(绕过 mil/__init__.py 的 PySide6 依赖)
import importlib.util
_spec = importlib.util.spec_from_file_location(
"mil_manifest", ROOT / "mil" / "core" / "manifest.py"
)
_manifest = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_manifest)
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 _copy_pyd_files() -> list[str]:
"""把 build/ 下的 .pyd 拷到 plugins/,返回已拷贝的包名列表。"""
copied: list[str] = []
for pkg in _manifest.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 _find_stdlib_zip() -> Path | None:
"""定位 Python stdlib 的 zip 文件(如 python310.zip)。"""
for p in sys.path:
path = Path(p)
if path.is_file() and path.suffix == ".zip" and "python" in path.name.lower():
return path
return None
def _extract_from_zip(zip_path: Path, pkg_name: str, dst: Path) -> bool:
"""从 zip 文件提取 stdlib 包目录到 plugins/。
Args:
zip_path: python310.zip 路径
pkg_name: 顶层包名(如 "xml"
dst: 目标目录 plugins/xml
Returns:
是否成功提取
"""
try:
with zipfile.ZipFile(zip_path, "r") as zf:
members = [m for m in zf.namelist() if m.startswith(f"{pkg_name}/")]
if not members:
return False
zf.extractall(PLUGINS_DIR, members=members)
print(f"[export] stdlib {pkg_name}: {zip_path} (zip) -> {dst}")
return True
except Exception as e:
print(f"[export] ERROR: 从 zip 提取 {pkg_name} 失败: {e}")
return False
def _copy_stdlib_from_fs(pkg_name: str, src: Path) -> bool:
"""从文件系统拷贝 stdlib 包目录到 plugins/。"""
dst = PLUGINS_DIR / pkg_name
if dst.exists():
shutil.rmtree(dst)
try:
shutil.copytree(src, dst, dirs_exist_ok=False)
print(f"[export] stdlib {pkg_name}: {src} (fs) -> {dst}")
return True
except Exception as e:
print(f"[export] ERROR: 拷贝 {pkg_name} 失败: {e}")
return False
def _copy_stdlib_packages() -> list[str]:
"""把 manifest 声明的 stdlib 子包从当前 Python 环境拷到 plugins/。
支持两种 stdlib 存储形态:
1. 文件系统目录(<python>/Lib/xml/)→ shutil.copytree
2. zip 压缩包(python310.zip/xml/)→ zipfile 提取
"""
copied: list[str] = []
seen: set[str] = set()
stdlib_zip = _find_stdlib_zip()
for module_name in _manifest.all_stdlib_modules():
top = module_name.split(".")[0]
if top in seen:
continue
seen.add(top)
dst = PLUGINS_DIR / top
if dst.exists():
shutil.rmtree(dst)
# 优先从文件系统找
try:
import importlib.util
spec = importlib.util.find_spec(top)
if spec and spec.submodule_search_locations:
src = Path(spec.submodule_search_locations[0])
if src.is_dir() and ".zip" not in str(src):
if _copy_stdlib_from_fs(top, src):
copied.append(top)
continue
except (ImportError, ValueError):
pass
# 从 zip 提取
if stdlib_zip:
if _extract_from_zip(stdlib_zip, top, dst):
copied.append(top)
continue
print(f"[export] WARN: 找不到 stdlib 包 {top}fs 和 zip 均无)")
return copied
def _copy_py_packages() -> list[str]:
"""把 COMPANION_PACKAGES_PY 声明的包从 site-packages 拷 .py 源码到 plugins/。
这些包不编译 .pyd,保留 .py 源码形式,import 走标准 importlib 机制,
能正确从 sys.path 找到插件自带的 stdlib xml。
"""
copied: list[str] = []
for pkg in _manifest.COMPANION_PACKAGES_PY:
try:
import importlib.util
spec = importlib.util.find_spec(pkg)
if spec is None or not spec.submodule_search_locations:
print(f"[export] WARN: 找不到包 {pkg} 的源码目录")
continue
src = Path(spec.submodule_search_locations[0])
if not src.is_dir():
print(f"[export] WARN: {pkg} 源码路径不是目录: {src}")
continue
dst = PLUGINS_DIR / pkg
# 排除 __pycache__,只拷 .py 文件
shutil.copytree(src, dst,
ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
copied.append(pkg)
print(f"[export] {pkg}: {src} (.py) -> {dst}")
except Exception as e:
print(f"[export] ERROR: 拷贝 {pkg} 源码失败: {e}")
return copied
def _copy_c_extensions() -> list[str]:
"""从 Python 安装目录拷贝 C 扩展 .pyd 到 plugins/。
C 扩展(如 _elementtree.pyd)是顶层模块,不在 stdlib 包目录里。
Windows 下 .pyd 可能在 py_root/ 或 py_root/DLLs/,都查一遍。
"""
copied: list[str] = []
py_root = Path(sys.base_prefix)
search_dirs = [py_root, py_root / "DLLs"]
for stdlib_pkg, ext_names in _manifest.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 _read_plugin_meta() -> tuple[str, str, str]:
"""从 mil/__init__.py 源码提取插件名/版本/描述(绕过 PySide6 import)。
Returns:
(plugin_name, plugin_version, plugin_description)
"""
import re
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
def _write_manifest(copied_packages: list[str], stdlib_packages: list[str],
c_extensions: list[str]) -> None:
"""生成 plugins/manifest.json,含插件元数据供远程 zip 扫描读取。"""
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,
# 主业务包名固定为 mil,用作 zip 文件名前缀与本地子目录名
"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 main() -> None:
if not BUILD_DIR.exists():
print("[export] ERROR: build/ 不存在,请先运行 python build_pyd.py", file=sys.stderr)
sys.exit(1)
if PLUGINS_DIR.exists():
shutil.rmtree(PLUGINS_DIR)
PLUGINS_DIR.mkdir(parents=True)
copied_pyd = _copy_pyd_files()
# openpyxl/et_xmlfile/xml.etree 已编进 mil.pyd,不需要单独拷。
# 只需拷 C 扩展(_elementtree.pyd, pyexpat.pyd),运行时走标准 import 加载。
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)} 个(含已编入的 openpyxl/et_xmlfile/xml.etree")
print(f"[export] C 扩展: {len(copied_cext)} 个(运行时从插件目录加载)")
print(f"[export] Python 版本: {platform.python_version()}")
print(f"[export] 下一步: 把 plugins/ 目录放到宿主 exe 旁边")
if __name__ == "__main__":
main()
+71
View File
@@ -0,0 +1,71 @@
"""把 export_runtime.py 产物打包成带版本号的 zip 分发包。
产物:dist/<tool_name>-<version>-<python_tag>.zip
zip 内部扁平结构:解压到 plugins/<tool_name>/ 即得完整目录。
用法:
python build_pyd.py --clean
python tools/export_runtime.py
python tools/pack_zip.py
"""
from __future__ import annotations
import json
import platform
import re
import sys
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
PLUGINS_DIR = ROOT / "plugins"
DIST_DIR = ROOT / "dist"
def _read_plugin_meta() -> tuple[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)
return name, f"{major}.{minor}.{micro}"
def _python_tag() -> str:
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 main() -> None:
if not PLUGINS_DIR.exists() or not any(PLUGINS_DIR.iterdir()):
print("[pack] ERROR: plugins/ 不存在或为空,请先运行 export_runtime.py",
file=sys.stderr)
sys.exit(1)
name, version = _read_plugin_meta()
manifest = json.loads((PLUGINS_DIR / "manifest.json").read_text(encoding="utf-8"))
# 主业务包名由 export_runtime 写入 manifest,固定为 mil
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 相对 plugins/,解压到子目录即得完整布局
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 复制到远程 Y:/SE/xufeifei/plugins/")
if __name__ == "__main__":
main()
+118
View File
@@ -0,0 +1,118 @@
"""宿主启动时的 runtime hook。
宿主打包时通过 --runtime-hook 指定本文件,宿主 exe 启动时会在 main 之前执行本脚本。
作用:把 exe 旁边的 plugins/ 目录加进 sys.path
让插件自带的 .pyd + stdlib 子包(含 C 扩展)可以被 Python import 机制找到。
宿主打包命令示例(PyInstaller):
pyinstaller --onefile --runtime-hook tools/runtime_hook.py host_main.py
宿主打包命令示例(Nuitka):
nuitka --standalone --onefile --include-module=runtime_hook host_main.py
宿主部署目录布局:
host.exe
plugins/ ← 由 mil_sdk/tools/export_runtime.py 生成
└── mil/ ← 每插件一个子目录(zip 解压产物)
├── mil.cp311-*.pyd
├── openpyxl.cp311-*.pyd
├── et_xmlfile.cp311-*.pyd
├── xml/ ← 插件自带的 stdlib 子包(含 _elementtree.pyd
│ └── etree/
└── manifest.json
兼容旧扁平布局:plugins/ 直接含 manifest.json(过渡期)。
注意:本文件不依赖任何第三方包,只用 stdlibos / sys / json / pathlib)。
因为它在宿主 sys.path 配置好之前执行,不能 import 任何外部包。
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
def _resolve_plugins_dir() -> Path | None:
"""定位 plugins/ 目录。
优先级:
1. 环境变量 MIL_PLUGIN_DIR
2. sys.executable 旁边的 plugins/onedir 模式)
3. sys._MEIPASS 旁边的 plugins/onefile 模式解压目录)
"""
# 1) 环境变量
env = os.environ.get("MIL_PLUGIN_DIR")
if env:
p = Path(env)
if p.is_dir():
return p.resolve()
# 2) onedirexe 旁边的 plugins/
exe_dir = Path(sys.executable).resolve().parent
plugins = exe_dir / "plugins"
if plugins.is_dir():
return plugins.resolve()
# 3) onefilePyInstaller 解压目录旁边的 plugins/
meipass = getattr(sys, "_MEIPASS", None)
if meipass:
plugins = Path(meipass).parent / "plugins"
if plugins.is_dir():
return plugins.resolve()
return None
def _check_python_version(manifest_path: Path) -> None:
"""校验插件的 Python 版本与当前解释器兼容(minor 版本必须一致)。"""
import json
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except Exception:
return # manifest 损坏不阻止启动,让 import 错误自己暴露
plugin_py = manifest.get("python_version", "")
host_py = f"{sys.version_info.major}.{sys.version_info.minor}"
if not plugin_py.startswith(host_py):
sys.stderr.write(
f"[mil] WARNING: 插件 Python 版本({plugin_py}) 与宿主({host_py}) 可能不兼容\n"
)
def _insert_path(path: Path) -> None:
"""把 path 加到 sys.path[0],已存在则跳过。"""
p = str(path)
if p not in sys.path:
sys.path.insert(0, p)
def setup() -> None:
"""把 plugins/ 或 plugins/*/ 子目录加进 sys.path 最前面。
兼容两种布局:
- 旧扁平:plugins/ 直接含 manifest.json(过渡期)
- 新子目录:plugins/<插件名>/ 各自含 manifest.json
"""
plugins_dir = _resolve_plugins_dir()
if plugins_dir is None:
return
# 旧扁平布局:plugins/ 直接有 manifest.json
if (plugins_dir / "manifest.json").exists():
_check_python_version(plugins_dir / "manifest.json")
_insert_path(plugins_dir)
return
# 新子目录布局:逐个扫描 plugins/*/
for sub in plugins_dir.iterdir():
if not sub.is_dir():
continue
if (sub / "manifest.json").exists():
_check_python_version(sub / "manifest.json")
_insert_path(sub)
# 模块加载时立即执行(PyInstaller runtime hook 的约定)
setup()
+93
View File
@@ -0,0 +1,93 @@
"""编译完跑一遍:把 build/ 下的伴生包 .pyd 挂到 sys.modules
再用真实接口(openpyxl.load_workbook)写读一个最小 Excel,确认链路可用。
用法:
python tools/verify_companions.py
注意:本脚本直接 import mil.core.companion_loader**不**经过 mil/__init__.py
避免在无 PySide6 的开发机上因 UI 桥接而炸(生产宿主环境必然有 PySide6)。
"""
from __future__ import annotations
import argparse
import importlib.util
import sys
import tempfile
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
# 不走 mil/__init__.py,单独加载 companion_loader
spec = importlib.util.spec_from_file_location(
"mil_companion_loader", ROOT / "mil" / "core" / "companion_loader.py"
)
companion_loader = importlib.util.module_from_spec(spec)
spec.loader.exec_module(companion_loader)
load_companions = companion_loader.load_companions
# 伴生包清单:与 mil/__init__.py 中的 COMPANION_PACKAGES 保持一致
COMPANION_PACKAGES: list[str] = [
"openpyxl",
"et_xmlfile",
]
def main() -> int:
parser = argparse.ArgumentParser(description="验证伴生包 .pyd 加载链路")
parser.add_argument(
"--fallback",
action="store_true",
help="未发现产物时回退到 site-packages 已装版本(开发自测用)",
)
args = parser.parse_args()
loaded = load_companions(COMPANION_PACKAGES)
print(f"[verify] 加载器返回: {loaded}")
if not loaded and not args.fallback:
print(
"[verify] FAIL: 没有挂载到任何伴生包(是否忘了 python build_pyd.py?)\n"
" 调试期可加 --fallback 走 site-packages 自测加载器逻辑。",
file=sys.stderr,
)
return 1
if not loaded and args.fallback:
# 开发自测:build/mil_runtime 都没有时,从 sys.path 兜底
import importlib
for pkg in COMPANION_PACKAGES:
importlib.import_module(pkg)
loaded.append(pkg)
print(f"[verify] fallback 后挂载: {loaded}")
# 用 openpyxl 真接口验证:写一个最小 xlsx,立刻读回来
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
if __name__ == "__main__":
sys.exit(main())