guanlan-src/scripts/pack_module.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""P13-2：模块级打包器 `scripts/pack_module.py` + 四模块 `pack.bat`/`pack.sh` 包装。

设计（与既有整体打包 `scripts/pack_dist.py` 的排除规则一致）：
  · 每个系统模块（app_frontEnd / app_backEnd / app_algorithmModel / app_ETL）可**单独打包**为
    `dist/app_<模块>_v<版本>.zip`，内含：该模块目录（排除缓存/数据/产物/日志/临时件）+ `module_manifest.json`
    （名称、版本、生成时间、文件数、逐文件 sha256、字节数、依赖信息）；
  · `--all` 一次打四模块；`--verify <zip>` 开箱核验（逐文件比对 sha256）；
  · 版本单一真源：`app_common/app_common_guanlan/version.py` 的 VERSION（与根 VERSION 一致）。

用法：
  python scripts/pack_module.py --module app_ETL
  python scripts/pack_module.py --all
  python scripts/pack_module.py --verify dist/app_ETL_v2.50.0.zip
"""
from __future__ import annotations

import argparse
import hashlib
import json
import pathlib
import re
import sys
import time
import zipfile

sys.stdout.reconfigure(encoding='utf-8', errors='replace')
ROOT = pathlib.Path(__file__).resolve().parents[1]
MODULES = ('app_frontEnd', 'app_backEnd', 'app_algorithmModel', 'app_ETL')

# 排除规则（与 pack_dist.py 同口径：不入包的东西）
EXCLUDE_DIRS = {'__pycache__', '.git', '.venv', 'node_modules', 'logs', 'run', 'outputs'}
# ★P13：不排除 target —— Java 后端/网关的编译产物与依赖 jar 就在 target/{classes,lib}，
#   排掉会让 app_backEnd 的包跑不起来（实逮）。
EXCLUDE_RE = re.compile(r'(\*\.tmp|\.bak.*|\.old$|\.orig$|\.rej$|\.swp$|~$|^~\$|\.DS_Store$|\.log$|\.jsonl$|_cache$)')


def pkg_version() -> str:
    vf = ROOT / 'app_common/app_common_guanlan/version.py'
    m = re.search(r"^VERSION\s*=\s*'([^']+)'", vf.read_text(encoding='utf-8'), re.M)
    if m:
        return m.group(1)
    rv = (ROOT / 'VERSION').read_text(encoding='utf-8').strip()
    return rv or '0.0.0'


def ext_includes(mod: str) -> list[pathlib.Path]:
    """模块外、但属于该模块运行必需的产物（如前端 Vue 产物 release/web）。由 <module>/pack_include.json 声明。"""
    j = ROOT / mod / 'pack_include.json'
    if not j.exists():
        return []
    try:
        spec = json.loads(j.read_text(encoding='utf-8'))
    except Exception:
        return []
    out = []
    for rel in spec.get('include', []):
        q = ROOT / rel
        if q.is_dir():
            out += [x for x in sorted(q.rglob('*')) if x.is_file()]
        elif q.is_file():
            out.append(q)
    return out


def included_files(mod: str, with_data: bool) -> list[pathlib.Path]:
    base = ROOT / mod
    out = []
    for p in sorted(base.rglob('*')):
        if not p.is_file():
            continue
        rel = p.relative_to(base)
        if any(part in EXCLUDE_DIRS for part in rel.parts):
            continue
        if not with_data and rel.parts and rel.parts[0] in ('data',):
            continue
        if EXCLUDE_RE.search(p.name):
            continue
        out.append(p)
    return out


def sha256(p: pathlib.Path) -> str:
    h = hashlib.sha256()
    with p.open('rb') as f:
        for chunk in iter(lambda: f.read(1 << 20), b''):
            h.update(chunk)
    return h.hexdigest()


def pack(mod: str, outdir: pathlib.Path, with_data: bool = False) -> pathlib.Path:
    assert mod in MODULES, '模块名必须是 %s 之一' % (MODULES,)
    ver = pkg_version()
    outdir.mkdir(parents=True, exist_ok=True)
    zp = outdir / ('app_%s_v%s.zip' % (mod, ver))
    files = included_files(mod, with_data)
    exts = ext_includes(mod)
    files = files + [q for q in exts if q not in files]
    man = {'module': mod, 'version': ver, 'built_at': time.strftime('%Y-%m-%d %H:%M:%S'),
           'with_data': with_data, 'file_count': len(files), 'files': {}, 'root': mod}
    for p in files:
        rel = str(p.relative_to(ROOT)).replace('\\', '/')
        man['files'][rel] = {'sha256': sha256(p), 'size': p.stat().st_size}
    with zipfile.ZipFile(zp, 'w', zipfile.ZIP_DEFLATED) as z:
        for p in files:
            z.write(p, str(p.relative_to(ROOT)).replace('\\', '/'))
        z.writestr('%s/module_manifest.json' % mod, json.dumps(man, ensure_ascii=False, indent=2))
    print('  [OK] %-22s %8d 文件 · %9d B → %s' % (mod, len(files), zp.stat().st_size, zp.name))
    return zp


def verify(zp: pathlib.Path) -> int:
    print('核验 %s' % zp)
    bad = 0
    with zipfile.ZipFile(zp) as z:
        man_name = next((n for n in z.namelist() if n.endswith('module_manifest.json')), None)
        if not man_name:
            print('  [X] 包内没有 module_manifest.json')
            return 1
        man = json.loads(z.read(man_name).decode('utf-8'))
        print('  模块=%s 版本=%s 文件数(声明)=%d' % (man['module'], man['version'], man['file_count']))
        for rel, meta in man['files'].items():
            try:
                data = z.read(rel)
            except KeyError:
                print('  [X] 缺文件 %s' % rel)
                bad += 1
                continue
            h = hashlib.sha256(data).hexdigest()
            if h != meta['sha256']:
                print('  [X] 指纹不符 %s' % rel)
                bad += 1
    print('  %s' % ('全部一致' if not bad else '有 %d 处不一致' % bad))
    return 0 if not bad else 1


def main() -> int:
    ap = argparse.ArgumentParser(description='观澜 · 模块级打包')
    ap.add_argument('--module', choices=MODULES)
    ap.add_argument('--all', action='store_true')
    ap.add_argument('--out', default=str(ROOT / 'dist'))
    ap.add_argument('--with-data', action='store_true', help='连模块 data/ 一起打（默认不含）')
    ap.add_argument('--verify')
    a = ap.parse_args()
    if a.verify:
        return verify(pathlib.Path(a.verify))
    outdir = pathlib.Path(a.out)
    print('观澜 · 模块打包 · 版本 %s · 输出 %s' % (pkg_version(), outdir))
    targets = list(MODULES) if a.all else ([a.module] if a.module else [])
    if not targets:
        ap.print_help()
        return 2
    for m in targets:
        pack(m, outdir, a.with_data)
    return 0


if __name__ == '__main__':
    raise SystemExit(main())