guanlan-src/app_common/app_common_guanlan/bootstrap.py

# -*- coding: utf-8 -*-
r"""观澜 · **模块包**独立安装/自检/运行/卸载的引导器（P9-B 起，配合"分别打包"）。

背景（用户令 2026-09-28）：前端、后端、算法、数据接入及管理要能**分别打包**。每个模块包 = 公共层 +
该模块 + 它依赖的模块（按 `configs/modules.yaml` 的 `allow` 递归）。单包装进总包时，模块目录里的
`install/run/uninstall` 直接调用安装根统一入口；**单独解压一个模块包时**（包根没有安装根那套 `install.ps1`
或本模块不是集成层），走这里：自建 `.venv`、离线装依赖、自检、按模块起服务或跑链、卸载。

用法（在**模块包根目录**下执行；也可用模块目录里的 install/run/uninstall 脚本转发过来）：

    python app_common/app_common_guanlan/bootstrap.py install   --module app_frontEnd
    python app_common/app_common_guanlan/bootstrap.py check     --module app_frontEnd
    python app_common/app_common_guanlan/bootstrap.py run       --module app_frontEnd [--port 28100]
    python app_common/app_common_guanlan/bootstrap.py run       --module app_ETL --cmd rebuild_all -- --dry-run
    python app_common/app_common_guanlan/bootstrap.py list      --module app_algorithmModel
    python app_common/app_common_guanlan/bootstrap.py uninstall --module app_frontEnd [--purge]

退出码：0 成功 · 2 用法/环境不满足 · 3 自检不过 · 4 运行失败 · 5 安装失败。
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import subprocess
import sys
import time

WIN = os.name == 'nt'

# 每个模块的"可运行入口"（模块包里 run 的候选；`--cmd` 取这些名字）
RUNNABLES = {
    'app_backEnd': {'serve': '启动全部服务（网关 + 工作台 + CMS + 仿真×2 + 三维）',
                    'check': '系统自检',
                    'status': '各服务端口/进程状态',
                    'stop': '停止本启动器起的服务'},
    'app_frontEnd': {'portal': '起一个静态服务把门户与工作台页面放出来（默认端口 28100）'},
    'app_ETL': {'raw_scan': '输入数据扫描（发现新增/变化 → 该重跑哪几步）',
                'raw_data_check': '输入数据放置体检',
                'rebuild_all': '整链重算（默认加 --dry-run，真要跑请显式给 -- --apply）',
                'scada_slim_build': '10 分钟标准仓重建'},
    'app_algorithmModel': {'rudong_model_run': '判级与曲线主链（L0/谱闸/定级）',
                           'rudong_fusion_run': '融合面与四轴',
                           'rudong_line_energy_share': '发电量线与能量占比',
                           'tcm_mask_thresholds_build': 'TCM 掩码阈值反解'},
    'app_ontology': {'ontology_release_build': '本体发布层 r1/r2 生成',
                     'sop_findings_from_ledger': '由重算台账生成 claim'},
    'app_common': {},
    'app_qualityGate': {'module_boundary_audit': '模块边界审计',
                        'config_audit': '配置统一审计',
                        'pages_audit': '页面归口审计',
                        'chain_gap_check': '重算链缺口检查',
                        'delivery_docs_build': '三件交付文档构建与自检'},
}
MODULES = tuple(RUNNABLES)
# 支持"分别打包"的四个模块（用户令点名的）+ 其余三个（同一机制，顺带可用）
FOUR = ('app_frontEnd', 'app_backEnd', 'app_algorithmModel', 'app_ETL')


def root_of(f: str | pathlib.Path) -> pathlib.Path:
    """包根 = 同时含 `VERSION`（或 `configs/`）与 `app_common/` 的最近上层目录。"""
    p = pathlib.Path(f).resolve()
    for d in [p.parent] + list(p.parents):
        if (d / 'app_common').is_dir() and ((d / 'VERSION').is_file() or (d / 'configs').is_dir()):
            return d
    return p.parents[1]


ROOT = root_of(__file__)
INFO = ROOT / 'install-info.json'


def py_in_venv() -> str:
    return str(ROOT / '.venv' / ('Scripts/python.exe' if WIN else 'bin/python'))


def say(msg: str = '') -> None:
    print(msg, flush=True)


def load_modules() -> dict:
    """读安装根单件配置 `modules.yaml`（走配置取用口；包结构异常时才兜底手拼）。"""
    try:
        from app_common.app_common_guanlan.api import paths as _P
        p = _P.config('modules.yaml')
    except Exception:  # noqa: BLE001
        p = ROOT / 'configs' / 'modules.yaml'      # config-path-allow: 取用口不可用时的兜底
    if not p.is_file():
        return {}
    try:
        import yaml
        reg = yaml.safe_load(p.read_text(encoding='utf-8')) or {}
        return {m['name']: m for m in (reg.get('modules') or [])}
    except Exception:  # noqa: BLE001  （无 yaml 时退回最小解析：只认 name/allow 行）
        out, cur = {}, None
        for ln in p.read_text(encoding='utf-8').splitlines():
            s = ln.strip()
            if s.startswith('- name:'):
                cur = s.split(':', 1)[1].strip()
                out[cur] = {'allow': []}
            elif cur and s.startswith('allow:'):
                body = s.split('[', 1)[-1].rstrip(']')
                out[cur]['allow'] = [x.strip() for x in body.split(',') if x.strip()]
        return out


def closure(name: str, reg: dict | None = None) -> list[str]:
    """依赖闭包（依赖在前，自身在后）；app_common 永远在最前。"""
    reg = reg if reg is not None else load_modules()
    seen: list[str] = []

    def walk(n: str) -> None:
        if n in seen:
            return
        for d in (reg.get(n, {}).get('allow') or []):
            walk(d)
        seen.append(n)

    walk(name)
    if 'app_common' in reg and 'app_common' not in seen:
        seen.insert(0, 'app_common')
    return seen


def module_api_files(m: str) -> list[pathlib.Path]:
    d = ROOT / m / f'{m}_guanlan'
    return sorted(d.glob('*.py')) + sorted(d.rglob('*/api.py'))


def cmd_check(a) -> int:
    say(f'== 自检（模块包 · {a.module}）==')
    say(f'   包根: {ROOT}')
    bad: list[str] = []
    reg = load_modules()
    if not reg:
        bad.append('读不到 configs/modules.yaml（模块包必须带安装根单件配置）')
    cl = closure(a.module, reg)
    say(f'   依赖闭包: {" → ".join(cl)}')
    # ① 目录与入口
    for m in cl:
        d = ROOT / m
        if not (d / f'{m}_guanlan' / 'api.py').is_file():
            bad.append(f'{m}/{m}_guanlan/api.py 不在位')
        if not ((d / 'README.MD').is_file() or (d / 'README.md').is_file()):
            bad.append(f'{m}/README.MD 不在位')
        for s in ('install.bat', 'install.sh', 'uninstall.bat', 'uninstall.sh', 'run.bat', 'run.sh'):
            if not (d / s).is_file():
                bad.append(f'{m}/{s} 不在位')
    # ② 版本一致（真源是 app_common 的 version.py；根 VERSION 是给它做镜像的标记文件）
    import re as _re
    vp = ROOT / 'app_common' / 'app_common_guanlan' / 'version.py'
    v2 = ''
    if vp.is_file():
        m = _re.search(r"^VERSION\s*=\s*'([^']+)'", vp.read_text(encoding='utf-8'), _re.M)
        v2 = m.group(1) if m else ''
    vf = ROOT / 'VERSION'
    v1 = ''
    if vf.is_file():
        m1 = _re.search(r"([0-9]+\.[0-9]+\.[0-9]+)", vf.read_text(encoding='utf-8'))
        v1 = m1.group(1) if m1 else vf.read_text(encoding='utf-8').strip()
    if v1 and v2 and v1 != v2:
        bad.append(f'VERSION({v1}) 与版本真源 app_common/.../version.py({v2}) 不一致')
    say(f'   版本: VERSION={v1 or "—"} · 版本真源={v2 or "—"}')
    # ③ 配置单源（安装根单件 vs 模块 configs/）
    try:
        from app_common.app_common_guanlan.api import paths as _P2
        root_cfg_dir = _P2.CONFIGS
    except Exception:  # noqa: BLE001
        root_cfg_dir = ROOT / 'configs'            # config-path-allow: 取用口不可用时的兜底
    root_cfg = {p.relative_to(root_cfg_dir).as_posix() for p in root_cfg_dir.rglob('*') if p.is_file()}
    for m in cl:
        mc = ROOT / m / 'configs'
        if not mc.is_dir():
            continue
        for p in mc.rglob('*'):
            if p.is_file() and p.relative_to(mc).as_posix() in root_cfg:
                bad.append(f'配置双份: configs/{p.relative_to(mc).as_posix()} 与 {m}/configs/ 同名（必须单源）')
    # ④ 导入冒烟（每个模块的公开面）
    sys.path.insert(0, str(ROOT))
    for m in cl:
        mod = f'{m}.{m}_guanlan.api'
        try:
            __import__(mod)
        except Exception as e:  # noqa: BLE001
            bad.append(f'导入 {mod} 失败: {type(e).__name__}: {str(e)[:90]}')
    if bad:
        for b in bad:
            say(f'   [X] {b}')
        say(f'结论: 自检不过（{len(bad)} 项）')
        return 3
    say('   [OK] 目录与入口齐全 · 版本一致 · 配置单源 · 公开面可导入')
    say('结论: 自检通过')
    return 0


def cmd_install(a) -> int:
    say(f'== 安装（模块包 · {a.module}）==')
    vpy = py_in_venv()
    if not pathlib.Path(vpy).is_file():
        say(f'   建 .venv: {ROOT / ".venv"}')
        r = subprocess.run([sys.executable, '-m', 'venv', str(ROOT / '.venv')])
        if r.returncode:
            say('   [X] 建虚拟环境失败')
            return 5
    req = ROOT / 'requirements.txt'
    if req.is_file():
        wheels = ROOT / 'wheels'
        args = [vpy, '-m', 'pip', 'install', '-q', '-r', str(req)]
        offline = wheels.is_dir() and any(wheels.rglob('*.whl'))
        if offline:
            args += ['--no-index', '--find-links', str(wheels)]
        say(f'   装依赖（{"离线 wheels" if offline else "在线 pip"}）…')
        r = subprocess.run(args)
        if r.returncode:
            say('   [X] 依赖安装失败')
            return 5
    info = {}
    if INFO.is_file():
        try:
            info = json.loads(INFO.read_text(encoding='utf-8'))
        except Exception:  # noqa: BLE001
            info = {}
    info.update(dict(name='观澜·模块包', version=(ROOT / 'VERSION').read_text(encoding='utf-8').strip()
                     if (ROOT / 'VERSION').is_file() else '',
                     module=a.module, mode='module-package', python=vpy,
                     installed_at=time.strftime('%Y-%m-%d %H:%M:%S')))
    INFO.write_text(json.dumps(info, ensure_ascii=False, indent=1), encoding='utf-8')
    say(f'   已写安装记录 {INFO.name}')
    return cmd_check(a)


def _run_script(script: str, extra: list[str]) -> int:
    p = ROOT / 'scripts' / f'{script}.py'
    if not p.is_file():
        say(f'   [X] 找不到 scripts/{script}.py')
        return 4
    cmd = [py_in_venv() if pathlib.Path(py_in_venv()).is_file() else sys.executable, '-X', 'utf8', str(p)] + extra
    say('   $ ' + ' '.join(cmd))
    return subprocess.run(cmd, cwd=str(ROOT)).returncode


def cmd_list(a) -> int:
    say(f'== {a.module} 可运行入口 ==')
    for k, v in RUNNABLES.get(a.module, {}).items():
        say(f'   {k:26s} {v}')
    say('\n用法: run --module %s --cmd <名> [-- 透传给该脚本的参数]' % a.module)
    return 0


def cmd_run(a) -> int:
    m = a.module
    say(f'== 运行（模块包 · {m}）==')
    if m == 'app_backEnd':
        if a.cmd in (None, 'serve') and (ROOT / 'guanlan.py').is_file():
            return subprocess.run([py_in_venv() if pathlib.Path(py_in_venv()).is_file() else sys.executable,
                                   '-X', 'utf8', str(ROOT / 'guanlan.py'), 'serve'], cwd=str(ROOT)).returncode
        if a.cmd in ('check', 'status', 'stop'):
            return subprocess.run([py_in_venv() if pathlib.Path(py_in_venv()).is_file() else sys.executable,
                                   '-X', 'utf8', str(ROOT / 'guanlan.py'), a.cmd], cwd=str(ROOT)).returncode
    if m == 'app_frontEnd' and a.cmd in (None, 'portal'):
        port = str(a.port or 28100)
        return _run_script('static_server', ['--port', port, '--host', '127.0.0.1', '--dir', str(ROOT / 'release')])
    cmd = a.cmd or (next(iter(RUNNABLES.get(m, {})), None))
    if not cmd:
        say('   本模块没有可直接运行的入口；用 list 看可选名')
        return 2
    extra = list(a.rest or [])
    if m == 'app_ETL' and cmd == 'rebuild_all' and not extra:
        extra = ['--dry-run']          # 默认只看计划，真跑要显式给参数
    return _run_script(cmd, extra)


def cmd_uninstall(a) -> int:
    say(f'== 卸载（模块包 · {a.module}）==')
    import shutil
    keep_root = ROOT
    # 有质量门（含卸载器）时走统一卸载器，行为与总包一致
    un = ROOT / 'scripts' / 'guanlan_uninstall.py'
    if un.is_file():
        say('   走统一卸载器 scripts/guanlan_uninstall.py（与总包同一套行为）')
        cmd = [py_in_venv() if pathlib.Path(py_in_venv()).is_file() else sys.executable, '-X', 'utf8', str(un)]
        if a.purge:
            cmd.append('--purge')
        return subprocess.run(cmd, cwd=str(keep_root)).returncode
    for rel in ('.venv', 'run'):
        p = keep_root / rel
        if p.exists():
            shutil.rmtree(p, ignore_errors=True)
            say(f'   已删 {rel}/')
    if INFO.is_file():
        INFO.unlink()
        say('   已删安装记录 install-info.json')
    if a.purge:
        for rel in ('data', 'outputs', 'logs'):
            p = keep_root / rel
            if p.exists():
                shutil.rmtree(p, ignore_errors=True)
                say(f'   已删 {rel}/（--purge）')
    say('结论: 卸载完成（保留程序目录；要连目录一起删就手工删包根）')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description='观澜 · 模块包引导器（安装/自检/运行/卸载）')
    ap.add_argument('action', choices=('install', 'check', 'run', 'list', 'uninstall'))
    ap.add_argument('--module', required=True, choices=MODULES)
    ap.add_argument('--cmd', default=None)
    ap.add_argument('--port', type=int, default=None)
    ap.add_argument('--purge', action='store_true')
    ap.add_argument('rest', nargs='*', help='透传给被调脚本的参数（放在 -- 之后）')
    a = ap.parse_args()
    if (ROOT / 'app_common' / 'app_common_guanlan' / 'api.py').is_file() is False:
        say('[X] 包根看起来不是模块包（缺 app_common/app_common_guanlan/api.py）')
        return 2
    return {'install': cmd_install, 'check': cmd_check, 'run': cmd_run, 'list': cmd_list,
            'uninstall': cmd_uninstall}[a.action](a)


if __name__ == '__main__':
    sys.exit(main())