guanlan-src/guanlan.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""观澜 · 如东样板 v2 启动器 (跨平台, 只用标准库). 用法:
  python guanlan.py check    自检: Python/依赖/端口/数据产物/发布件/Ollama
  python guanlan.py serve    启动全部服务 (网关 + 工作台 + CMS + 仿真×2 + 三维), 写 run/pids.json, 等 healthz
  python guanlan.py status   各服务端口/进程状态
  python guanlan.py stop     停止本启动器起的全部服务
  python guanlan.py open     打印入口地址 (并尝试用系统浏览器打开)
  python guanlan.py qa       离线模型已知答案回放 (需服务已起; 参数见 scripts/llm_known_answer.py)
端口与路径在 configs/serve.json (没有则用内置默认). 组件默认只绑 127.0.0.1;
门户网关可用 serve.json 的 public_host 放到所有网卡 (用户令 2026-09-19「监听所有 IP」, 见 §7)。"""
from __future__ import annotations
import json, os, re, socket, subprocess, sys, time, urllib.request, warnings, webbrowser, signal
from pathlib import Path
ROOT = Path(__file__).resolve().parent; RUN = ROOT / "run"; LOGS = ROOT / "logs"; PIDS = RUN / "pids.json"
sys.path.insert(0, str(ROOT))
from src import paths as P          # 配置/路径唯一取用口 (P.SERVE_JSON / P.config()…, 见 docs §11)
WIN = os.name == "nt"


def _hermetic():
    """摘掉机器全局的 PYTHONPATH: 本项目依赖都装在 .venv 里, 而 PYTHONPATH 会被插到 sys.path 前面,
    顶掉 venv 里 pin 住的版本 (实测 requests 2.33.0 顶掉 2.34.2), 还会让 pip 少装传递依赖。
    子进程从 env() 继承的是摘干净之后的环境。"""
    for d in [x for x in os.environ.pop("PYTHONPATH", "").split(os.pathsep) if x]:
        while d in sys.path: sys.path.remove(d)


_hermetic()
DEFAULT = {
    "host": "127.0.0.1", "public_host": "0.0.0.0", "gateway": 28084, "detail": 18033, "cms": 18020, "sim": 18791, "sim_sys": 18792, "viewer": 64292, "ollama": 11434,
    "release_dir": "release", "viewer_dir": "release/viewer", "sim_dir": "resources/oem_envision_sc1_rudong2014",
    "python": "",   # 空 = 当前解释器 (安装脚本建的 .venv)
}
# ★`host` = 组件(内部 API: detail/cms/viewer/sim)监听地址, **127.0.0.1**（只本机, 不联网）。
#   `public_host` = **门户网关**监听地址（用户令 2026-09-19「观澜改为监听所有 IP」）:
#     出厂/内置默认 **"0.0.0.0"** = 门户在任何网卡上都能访问（别的机器开 http://<本机IP>:28084/）;
#     空串 ⇒ 跟 host（只本机）; 填某个内网 IP ⇒ 只放那一个地址。configs/serve.json 是运行期真源,
#     这里只是"配置读不到时的兜底", 两边口径保持一致。
#   ★只放网关一个口子出去，**组件仍绑本机** —— 对外只有一个 28084 入口（网关自己做路由与改写），
#   内部 API（/api/*、CMS 18020、viewer 64292、sim 18791）不直接暴露。
#   ★页面**没有鉴权**：对外监听等于把全场的运行数据摊开。请同时做一层限制（防火墙白名单/反代加
#   认证/只绑内网 IP），别把它直接挂公网。改法与回退口径见 README 第二节与 docs/系统设计说明.md §7。


def jload(p: Path):
    """读 JSON 配置。必须用 utf-8-sig: 记事本和 PowerShell 5.1 的 `Set-Content -Encoding UTF8`
    写出的都是 **带 BOM** 的 UTF-8, 裸 utf-8 读会 JSONDecodeError (Unexpected UTF-8 BOM)。"""
    return json.loads(Path(p).read_text(encoding="utf-8-sig"))


def cfg():
    c = dict(DEFAULT); p = P.SERVE_JSON          # 端口/路径真源 (配置取用口 = src/paths.py)
    if p.exists():
        try: c.update(jload(p))
        except Exception as ex:  # 配置写坏不该让整个产品起不来: 退回内置默认端口, 但要说清楚
            print(f"  [X] {p} 读不了 ({ex.__class__.__name__}: {str(ex)[:100]}); 本次用内置默认端口。"
                  f" 重跑 install 脚本可重写该文件")
    return c


def py(c):
    """解释器: serve.json 的 python (相对路径按**安装根**解析) → 本目录 .venv (按平台探测) → 当前解释器。
    2026-09-11: 原来只认绝对路径 —— 安装目录整个拷到别的电脑/别的盘后, 那个绝对路径就失效了。"""
    v = (c.get("python") or "").strip()
    if v:
        p = Path(v)
        p = p if p.is_absolute() else (ROOT / p)
        if p.exists(): return str(p)
    for rel in (".venv/Scripts/python.exe", ".venv/bin/python", ".venv/bin/python3"):
        p = ROOT / rel
        if p.exists(): return str(p)
    return sys.executable
def up(host, port, t=0.4):
    try:
        with socket.create_connection((host, port), timeout=t): return True
    except OSError: return False


def env(c):
    e = dict(os.environ); rel = str(ROOT / c["release_dir"])
    raw = Path(c.get("raw_dir", "data/raw")); raw = raw if raw.is_absolute() else (ROOT / raw)
    e.update(GUANLAN_RELEASE_DIR=rel, GUANLAN_PORTAL=str(ROOT / c["release_dir"] / "portal.html"), GUANLAN_STATIC=rel, WINDSCADA_ROOT=str(ROOT), WINDCMS_ROOT=str(ROOT),
             WINDSCADA_RUDONG_SRC=str(raw), GUANLAN_OLLAMA_PORT=str(c["ollama"]), PYTHONUTF8="1", PYTHONIOENCODING="utf-8"); return e


def public_host(c) -> str:
    """门户网关对外监听地址: `public_host` 空则跟 `host`（出厂=127.0.0.1 只本机）。"""
    return (str(c.get("public_host") or "").strip() or str(c["host"]))


def lan_ips():
    """本机可用的 IPv4（给"该用什么地址打开"那句话用; 取不到就空）。"""
    out = []
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            ip = info[4][0]
            if ip and not ip.startswith("127.") and ip not in out:
                out.append(ip)
    except OSError:
        pass
    return out


def services(c):
    P = py(c); h = c["host"]; hp = public_host(c)
    return [
        ("detail", c["detail"], [P, "scripts/windscada_serve.py", "--host", h, "--port", str(c["detail"])]),
        ("cms", c["cms"], [P, "scripts/windcms.py", "serve", "--farm", "rudong", "--port", str(c["cms"])]),
        ("viewer", c["viewer"], [P, "scripts/static_server.py", "--port", str(c["viewer"]), "--host", h,
                                 "--dir", str(ROOT / c["viewer_dir"]), "--comp", "viewer"]),
        ("sim_sys", c["sim_sys"], [P, "-u", str(ROOT / c["release_dir"] / "sim_sys_server.py")]),
        ("sim", c["sim"], [P, "scripts/static_server.py", "--port", str(c["sim"]), "--host", h,
                               "--dir", str(ROOT / c["sim_dir"]), "--comp", "sim"]),
        # ★只有网关用 public_host（用户令 2026-09-19「监听所有 IP」）: 组件一律留在 host(本机)
        ("gateway", c["gateway"], [P, "scripts/guanlan_gateway.py", "--host", hp, "--port", str(c["gateway"])]),
    ]


def spawn(cmd, log: Path, e):
    f = open(log, "ab")
    kw = dict(cwd=str(ROOT), env=e, stdout=f, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL)
    if WIN:
        # 2026-09-16: 统一到 CREATE_NO_WINDOW (src/proc.py) —— 原先用 DETACHED_PROCESS, 子进程没有控制台,
        # 而 .venv 的 python.exe 是**转发器**, 它会再拉一个真解释器; 落地实测那一步会冒出可见黑窗。
        # CREATE_NO_WINDOW 给的是"没有窗口的控制台", stdio 照旧重定向到日志, 且与 DETACHED 互斥故只用前者。
        from src import proc as _proc
        kw["creationflags"] = _proc.flags()
    else: kw["start_new_session"] = True
    return subprocess.Popen(cmd, **kw).pid


def alive(pid):
    if not pid:
        return False
    if WIN:
        # errors="replace" 必须有: tasklist 的输出是**控制台代码页**(中文 Windows = GBK), 而本进程
        # 可能是 PYTHONUTF8=1 起的(默认文本编码变 UTF-8) → 解码失败会让 r.stdout 变成 None,
        # 下一行 `str(pid) in None` 直接 TypeError。2026-09-12 由运维控制台的动作进程实测逮到。
        # ★ 走 src.proc.run_text: 本函数也会被**无控制台**的进程调用 (网关/运维任务), 裸 spawn tasklist
        #   会给它新建可见控制台 → 反复闪窗 (2026-09-16)。
        from src import proc as _proc
        r = _proc.run_text(["tasklist", "/FI", f"PID eq {pid}"])
        return r.stdout is not None and str(pid) in r.stdout
    try: os.kill(pid, 0); return True
    except OSError: return False


def kill(pid):
    from src import proc as _proc
    if WIN: _proc.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True)
    else:
        try: os.killpg(os.getpgid(pid), signal.SIGTERM)
        except OSError:
            try: os.kill(pid, signal.SIGTERM)
            except OSError: pass


def cmd_check(c):
    ok = True
    def row(name, good, note=""): 
        nonlocal ok; ok &= bool(good); print(f"  [{'OK' if good else 'FAIL'}] {name}{(' — ' + note) if note else ''}")
    print(f"观澜·如东样板 v2 自检 @ {ROOT}"); v = sys.version_info; row("Python ≥ 3.11", v >= (3, 11), f"{v.major}.{v.minor}.{v.micro} ({sys.executable})")
    # 源码可编译性 —— 重点是**非法转义序列**: 普通字符串里写 `\d` `\*` 这类, Python 3.12 只给
    # SyntaxWarning, 未来版本会变成 SyntaxError。这类问题在开发机上永远看不见, 换台机器/换个解释器才炸
    # (2026-09-11 逮到 src\windcms\serve.py 里 JS 正则的两处, 已按"补成合法转义"修好)。
    srcs = sorted(list((ROOT / "src").rglob("*.py")) + list((ROOT / "scripts").rglob("*.py")))
    bad = []
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        for f in srcs:
            try: compile(f.read_text(encoding="utf-8"), str(f), "exec")
            except SyntaxError as e: bad.append(f"{f.relative_to(ROOT)}:{e.lineno} {e.msg}")
        for it in w:
            if issubclass(it.category, SyntaxWarning): bad.append(f"{Path(str(it.filename)).name}:{it.lineno} {it.message}")
    row(f"源码可编译 ({len(srcs)} 个文件, 无非法转义/语法错)", not bad, "; ".join(bad[:3]))
    # 语言包闸前移 (2026-09-19 实逮): `src/windscada/ui/build.py` 在**出页之前**会跑 lang.check(),
    # 不过就抛 RuntimeError —— 设计是对的(不会把中文残留发给用户), 但代价是**页面 500**:
    # 实测我加了一条英文措辞里含 'spectra'(36 条内部审核黑话之一), `/v2` 与 `/detail/` 当场全 500。
    # 这类错误在部署前就该报出来, 所以在这里也跑一遍(与出页同一个 check(), 不是另写一套)。
    try:
        from src.windscada import lang as _lang
        _lb = _lang.check()
        row(f"语言包 (前后端界面文案 zh/en 成对, {len(_lang.L)} 条)", not _lb, "; ".join(_lb[:3]))
    except Exception as e:                       # 语言包读不出来同样要报, 不静默
        row("语言包", False, f"{type(e).__name__}: {e}")
    # 子进程口径自检 (2026-09-16 加): 这三条是**回归守卫** —— 它们各自都真炸过一次, 而炸法都是"静默":
    #   ① capture_output=True 与显式 stdout 互斥 → 异常被 except 吞成 alive=False ⇒ 在跑的任务被
    #      误判成"被强杀", 页面显示完成、重算按钮可再点(可能并发重算);
    #   ② CREATE_NO_WINDOW 新建的控制台会把子进程标准句柄吸走 ⇒ 重算日志一个字都没有、任务挂住;
    #   ③ 不弹窗 (NO_WINDOW 位必须真的设上, 否则"点按钮闪黑窗"回来)。
    try:
        from src import proc as _proc
        _r = _proc.run_text([sys.executable, "-c", "print('proc-ok')"])
        row("子进程口径: run_text(capture_output) 可用", _r.returncode == 0 and "proc-ok" in (_r.stdout or ""),
            "src/proc.py (capture_output 与显式 stdout 互斥, 冲突会让 job_running 误判任务已死)")
        import tempfile as _tf
        _lg = Path(_tf.gettempdir()) / "_guanlan_proc_check.log"
        _p = _proc.spawn([sys.executable, "-c", "print('spawn-ok')"], log=_lg); _p.wait(timeout=60)
        _txt = _lg.read_text(encoding="utf-8", errors="replace") if _lg.exists() else ""
        try: _lg.unlink()
        except OSError: pass
        row("子进程口径: spawn(log=) 输出进日志", "spawn-ok" in _txt,
            "CREATE_NO_WINDOW 会吸走句柄, 不显式继承则日志空白 (重算看起来卡住)")
        row("子进程口径: 无窗口位已设 (CREATE_NO_WINDOW)", bool(_proc.flags() & 0x08000000) or os.name != "nt",
            f"flags={hex(_proc.flags())}")
    except Exception as _e:
        row("子进程口径自检", False, f"{type(_e).__name__}: {_e}")
    # 入口引用闭合 (2026-09-16 现场实炸后加): 目标机 (D:\产品\app) 双击 start.bat 报
    # "无法找到脚本文件 start_hidden.vbs" —— 包少带了一个被引用的文件。这条守卫在**装机后**复查一遍:
    # start.bat/check.bat/stop.bat/install.* 引用的每个文件都必须在位, 缺一个就 FAIL。
    # (那个 .vbs 已按用户令删除, 无窗口启动改用 pythonw.exe + scripts\guanlan_start_hidden.py;
    #  守卫与实现语言无关, 照样生效 —— 现在它盯的是 start.bat → scripts\guanlan_start_hidden.py 这类引用。)
    try:
        from src.entry_refs import encoding_problems, missing_refs, referenced_tree_files
        _miss = missing_refs(ROOT)
        _nref = sum(len(v) for v in referenced_tree_files(ROOT).values())
        row(f"入口脚本引用闭合 ({_nref} 条: start.bat→guanlan_start_hidden.py 等)", not _miss,
            "; ".join(f"{e} 缺 {r}" for e, r in _miss[:3]) or "每个入口引用的文件都在位")
        _enc = encoding_problems(ROOT)
        row("入口脚本编码守则 (.ps1 = UTF-8 BOM + CRLF; .bat = 纯 ASCII/CRLF/无 BOM/无 `<<`; .sh = LF)", not _enc,
            "; ".join(_enc[:2]) or "编码与字符都对 (PS 5.1 才不会把中文按 GBK 解错; 批处理里不放中文/重定向符)")
    except Exception as _e:
        row("入口脚本引用闭合", False, f"{type(_e).__name__}: {_e}")
    # 版本管理 (2026-09-17 用户令): ① 版本记录 md 必须与 src/version.py 一致（否则包里会出现
    # "包说 v2.5.0、记录表最后一行还是 v2.4.0"）；② 卸载入口必须在位（装得上卸不掉 = 交付缺陷）。
    try:
        from src import version as _V
        import importlib.util as _ilu2
        _sp = _ilu2.spec_from_file_location('_version_log', ROOT / "scripts" / "version_log.py")
        _vm = _ilu2.module_from_spec(_sp)
        _sp.loader.exec_module(_vm)
        _doc = ROOT / "docs" / "版本记录.md"
        _ok = _doc.is_file() and _vm.block_of(_doc.read_text(encoding="utf-8")) == _vm.body().strip()
        row(f"版本管理: {_V.NAME} v{_V.VERSION}（记录表与代码一致, 包名 {_V.package_name()}）", _ok,
            "" if _ok else "docs/版本记录.md 与 src/version.py 不一致 —— 跑 python scripts/version_log.py --write")
        _un = [f for f in ("uninstall.bat", "uninstall.sh") if (ROOT / f).is_file()]
        _unpy = ROOT / "scripts" / "guanlan_uninstall.py"
        _uok = len(_un) == 2 and _unpy.is_file()
        row("卸载入口在位 (uninstall.bat / uninstall.sh → scripts/guanlan_uninstall.py)", _uok,
            f"{len(_un)}/2 个入口" + ("" if _uok else " —— 缺卸载入口: 装得上卸不掉"))
    except Exception as _e:
        row("版本管理与卸载入口", False, f"{type(_e).__name__}: {_e}")
    # 模块化目录与边界（用户令 2026-09-22「按算法、数据接入管理、前端、后端等系统模块重构，每模块一个目录」）:
    #   组件化/高内聚低耦合/可插拔这条**必须机器可查** —— 边界审计查五件事: 目录结构齐、模块间只经 api 调用、
    #   依赖方向合规（按 configs/modules.yaml 的 allow）、公共层不依赖业务模块、api 的转发目标真实存在。
    try:
        import importlib.util as _ilu5
        _sp5 = _ilu5.spec_from_file_location('_modb', ROOT / "scripts" / "module_boundary_audit.py")
        _mb = _ilu5.module_from_spec(_sp5)
        _sp5.loader.exec_module(_mb)
        _rc5, _probs, _notes, _mods = _mb.audit(verbose=False)
        row(f"模块化目录与边界（{len(_mods)} 模块：算法/数据接入/前端/后端/本体/公共/质量门）", _rc5 == 0,
            ("登记 %d 模块 · 模块间只经 api 调用 · 依赖方向合规 · 接口目标全在位" % len(_mods)
             if _rc5 == 0 else
             "违规 %d 条 ⇒ " % len(_probs) + "; ".join(f'{r}:{m[:60]}' for r, m in _probs[:2])))
    except Exception as _e:
        row("模块化目录与边界", False, f"{type(_e).__name__}: {_e}")
    # 交付文档三件在位（2026-09-22 用户令「需求分析/系统设计说明/数据要求说明」）: 文档名里带**当前版本号**,
    # 所以升一次版本号, 旧文档名就过期了 —— 不发 FAIL 的话, 包会悄悄带着 v2.9.2 名字的文档进 v2.10.0 交付。
    # ★2026-09-22 用户令:「三份 docx 不同步到远端服务器」⇒ 交付部署可以不携带这批文档。
    #   判据先看 `docs/src` 在不在: 不在 = 本机是**不随文档的部署**(远端演示机就是这种), 只报提示行、不 FAIL,
    #   且在 import python-docx **之前**分流 —— 那台机没装 docx 依赖, 直接进渲染器会炸成 ModuleNotFoundError。
    try:
        if not (ROOT / "docs" / "src").is_dir():
            row("交付文档三件 (需求分析/系统设计说明/数据要求说明)", True,
                "本机未部署交付文档（docs/src 不在位, 按部署口径跳过；需要时跑 "
                "python scripts/delivery_docs_build.py 重建, 依赖 python-docx）")
        else:
            import importlib.util as _ilu4
            _sp4 = _ilu4.spec_from_file_location('_ddb', ROOT / "scripts" / "delivery_docs_build.py")
            _ddb = _ilu4.module_from_spec(_sp4)
            _sp4.loader.exec_module(_ddb)
            _V2 = _ddb.VERSION
            _miss, _stale, _figmiss, _leak = [], [], [], {}
            for _k, _spec in _ddb.DOCS.items():
                _out = ROOT / _spec['out'].format(v=_V2)
                _src = ROOT / _spec['src'].format(v=_V2)
                if not _out.is_file():
                    _miss.append(_out.name)
                    _old = sorted((ROOT / "docs").glob(_spec['out'].format(v='*').split('/')[-1]))
                    _stale += [p.name for p in _old if p.name != _out.name]
                if not _src.is_file():
                    _miss.append(_src.name)
                    continue
                _mdtxt = _src.read_text(encoding='utf-8')
                for _w, _n in _ddb.forbidden_hits(_mdtxt).items():      # ★对外件去标识化硬门
                    _leak[_w] = _leak.get(_w, 0) + _n
                if _spec.get('strict_terms'):                            # ★数据要求说明: 去实现细节/去现状
                    for _w, _n in _ddb.strict_hits(_mdtxt).items():
                        _leak['严格:' + _w] = _leak.get('严格:' + _w, 0) + _n
                for _rel in re.findall(r'!\[[^\]]*\]\((figures/[^)]+)\)', _mdtxt):
                    if not (ROOT / "docs" / _rel).is_file():
                        _figmiss.append(_rel)
            _miss += _figmiss
            # ★图表器源码也要扫：图题是代码里的字符串，源码里带样本场字样就一定会画进图（对外件不许有）
            _figsrc = ROOT / "scripts" / "delivery_docs_figures.py"
            if _figsrc.is_file():
                for _w, _n in _ddb.forbidden_hits(_figsrc.read_text(encoding='utf-8')).items():
                    _leak['图表器:' + _w] = _leak.get('图表器:' + _w, 0) + _n
            _ok4 = not _miss and not _leak
            row(f"交付文档三件 (需求分析/系统设计说明/数据要求说明 · v{_V2})", _ok4,
                ("3/3 份在位, 文档名含当前版本号, 插图全在位, 去标识化 clean" if _ok4 else
                 (f"缺 {len(_miss)} 件: {', '.join(_miss[:3])}" if _miss else '')
                 + (f"；同名旧版残留: {', '.join(_stale[:3])}" if _stale else '')
                 + (f"；去标识化泄漏 {_leak}（对外件不得出现样本场标识）" if _leak else '')
                 + " —— 跑 python scripts/delivery_docs_build.py 重建"))
    except Exception as _e:
        row("交付文档三件", False, f"{type(_e).__name__}: {_e}")
    # 反向呼应审计 (用户令 2026-09-17): 自**输出**回溯 功能与算法 → **输入**, 逐件问"这件产物能不能
    # 由输入推导出来"。正向那几条跨度判据查不出"盘上这件产物到底有没有来路"(缺件时一条都不报),
    # 反向一对账就清楚了: 哪些有来路(生成端+输入+判据)、哪些没有(无生成端, 只能从交付包补)。
    try:
        import importlib.util as _ilu3
        _spec3 = _ilu3.spec_from_file_location('_rev_audit', ROOT / "scripts" / "products_reverse_audit.py")
        _ra = _ilu3.module_from_spec(_spec3)
        _spec3.loader.exec_module(_ra)
        _rows, _v, _cnt, _fails, _uncl, _rc = _ra.audit(None, verbose=False)
        row(f"反向呼应 (输出→功能/算法→输入): 成立 {_cnt.get('✓', 0)} 件 · 不成立 {_cnt.get('✗', 0)} 件 "
            f"· 人工件 {_cnt.get('◆', 0)} 件 · 无法验证 {_cnt.get('~', 0)} 件 · 未归类 {_cnt.get('?', 0)} 件", _rc == 0,
            "；".join(_fails[:2]) or "每件产物都能指到生成端与输入, 或如实标了 shipped/人工件(docs §13.6)")
    except Exception as _e:
        row("反向呼应审计", False, f"{type(_e).__name__}: {_e}")
    # 页面归口 (2026-09-17 用户令 1): 门户里哪些页面是"数据派生快照"(属于产物)、它们陈旧没陈旧。
    # 用户问过"#findings/#sim/#documents 该不该随输入数据变、算不算产物" —— 答案落在
    # configs/portal_pages.yaml 登记表 + scripts/pages_audit.py 的检查里, 这里只报结论。
    # 退出码: 0 一致 / 5 缺文件或未登记 / 6 溯源缺失 / 7 **页面陈旧**(数据变了页面没变) / 8 交付件缺版本 / 9 分类错。
    try:
        import importlib.util as _ilu
        _spec = _ilu.spec_from_file_location('_pages_audit', ROOT / "scripts" / "pages_audit.py")
        _m = _ilu.module_from_spec(_spec)
        _spec.loader.exec_module(_m)
        _rc, _res, _gaps = _m.audit()
        _bad = [r for r in _res if r[0] not in ("OK", "i")]
        _known = [r for r in _res if r[0] == "i"]
        _note = (f"检查 {len(_res)} 项: 不一致 {len(_bad)}"
                 + (f" ({'; '.join(f'{r[1]}: {r[2][:60]}' for r in _bad[:2])})" if _bad else "")
                 + f" · 已知缺口 {len(_known)} (引用悬空/无溯源, 登记表明写理由)"
                 + (f" · 未登记页 {len(_gaps)}" if _gaps else ""))
        row("页面归口: 页面是否随数据变 / 是否陈旧 (configs/portal_pages.yaml)", _rc == 0, _note)
    except Exception as _e:
        row("页面归口审计", False, f"{type(_e).__name__}: {_e}")
    # 配置与日志的统一口径 (2026-09-17 用户令 2): 配置目录/文件、日志目录/命名/格式。
    # 两个检查器都是"看一眼实物 + 代码"的静态检查, 很便宜; 有问题就 FAIL 并指到具体文件。
    for _script, _label in (("config_audit.py", "配置统一: 目录约定/命名/内容/引用闭合/不手拼路径"),
                            ("log_audit.py", "日志统一: logs/ 唯一落点/命名/行格式/保留策略")):
        try:
            import importlib.util as _ilu2
            _sp = _ilu2.spec_from_file_location(f'_{_script[:-3]}', ROOT / "scripts" / _script)
            _mod = _ilu2.module_from_spec(_sp)
            _sp.loader.exec_module(_mod)
            _rc2, _res2, _info2 = _mod.audit()
            _lvl = {}
            for _l in _res2:
                _lvl[_l[0]] = _lvl.get(_l[0], 0) + 1
            _bad2 = [r for r in _res2 if r[0] in ("X", "!")]
            row(_label, _rc2 == 0,
                (f"rc={_rc2}: " + "; ".join(f"{r[1]}: {r[2][:70]}" for r in _bad2[:2])) if _bad2
                else f"无不一致 · 已知缺口/白名单 {_lvl.get('i', 0)} · 提示 {_lvl.get('?', 0)}")
        except Exception as _e:
            row(_label, False, f"{type(_e).__name__}: {_e}")
    # 输入数据放置体检 (2026-09-17 用户令 3): data/raw 的结构/命名是否合约定、源类齐不齐。
    # 判 FAIL 的只有**结构类**问题 (放错层/名字不对/压缩包没解压) —— 那类问题会让摄入静默读不到;
    # "缺某类源件"只作为提示 (新装的机器本来就没数据, 不该因此判失败)。
    try:
        import importlib.util as _ilu3
        _sp3 = _ilu3.spec_from_file_location('_raw_check', ROOT / "scripts" / "raw_data_check.py")
        _rc3 = _ilu3.module_from_spec(_sp3)
        _sp3.loader.exec_module(_rc3)
        _r3, _n3, _s3 = _rc3.check_tree(Path(_rc3.station_dir()), deep=False)
        _struct = [x for x in _r3 if x[3] in (_rc3.RC_STRUCT, _rc3.RC_CLASH)]
        _gaps = [x for x in _r3 if x[3] == _rc3.RC_GAP]
        if not _s3.get('exists'):
            row("输入数据放置体检 (data/raw)", True,
                "尚未放置输入数据 —— 属预期; 放置规范见 docs/输入数据放置指导_v0.1.md")
        else:
            row("输入数据放置体检 (data/raw 结构/命名/源类)", not _struct,
                (f"{len(_struct)} 项结构问题: " + "; ".join(f"{x[1]}: {x[2][:60]}" for x in _struct[:2])) if _struct
                else f"{_s3.get('files', 0):,} 件源件 · 结构合规"
                     + (f" · 缺 {len(_gaps)} 类/项: " + "; ".join(f"{x[1]}: {x[2][:50]}" for x in _gaps[:2]) if _gaps else "")
                     + (f" · {len(_n3)} 条提示" if _n3 else ""))
    except Exception as _e:
        row("输入数据放置体检", False, f"{type(_e).__name__}: {_e}")
    for m in ("numpy", "pandas", "pyarrow", "polars", "yaml", "matplotlib", "plotly", "jinja2", "docx"):
        try: __import__(m); row(f"依赖 {m}", True)
        except Exception as ex: row(f"依赖 {m}", False, f"未安装: {ex.__class__.__name__} (运行 install 脚本)")
    # ★2026-09-16 用户令: 打包"不含 输入数据/产物/日志"。产物缺失时, 若包内 dist-manifest.json 就声明了
    #   no_products/no_data, 这里显示 `--`(待重算) 而**不是 FAIL** —— 否则开箱自检会把"本来就该重算"的
    #   状态报成失败, install 脚本收尾的那次 check 也会失败 (把正常状态说成问题)。
    _man = {}
    try:
        _mp = ROOT / "dist-manifest.json"
        _man = jload(_mp) if _mp.is_file() else {}
    except Exception:
        _man = {}
    _no_products = bool(_man.get("no_products"))
    for rel, n in (("outputs/rudong/windscada", "L0/L1 产物 (parquet)"), ("outputs/rudong/ontology/objects.json", "本体对象库"), ("outputs/rudong/sop/findings.json", "findings"), ("outputs/rudong/guanlan/facts_contract_v0.json", "事实契约"),
                   (c["release_dir"] + "/portal.html", "门户"), (c["release_dir"] + "/sim_sys_server.py", "仿真合页服务"), (c["release_dir"] + "/如东SWT40_控制律仿真台_20260906.zip", "仿真合页资料包"), (c["viewer_dir"], "三维 viewer 资产"), (c["sim_dir"], "仿真回放资产"), (c["release_dir"] + "/如东", "治理清单交付件 (可选)")):
        p = ROOT / rel; exists = p.exists() and (any(p.iterdir()) if p.is_dir() else p.stat().st_size > 0)
        if "可选" in n: print(f"  [{'OK' if exists else '--'}] {n}: {rel}")
        elif _no_products and rel.startswith("outputs/"):
            print(f"  [--] {n}: {rel}  (本包按用户令**未随产物** —— 把现场包放好后跑 "
                  f"scripts/place_raw_data.py --src <现场包目录> --scope full 再 scripts/rebuild_all.py)")
        else: row(n, exists, rel)
    raw = Path(c.get("raw_dir", "data/raw")); raw = raw if raw.is_absolute() else (ROOT / raw)
    print(f"  [{'OK' if raw.exists() else '--'}] 原始件目录 (补数据放这里): {raw}{'' if raw.exists() else ' (不存在; 原始件不随包分发, 只影响摄入命令, 不影响页面)'}")
    for name, port, _ in services(c): print(f"  [{'运行中' if up(c['host'], port) else '空闲'}] 端口 {port} ({name})")
    # 模型闸: 探针 (generate 非 tags) + 档位模型是否都在本机 + 摘要记录; 无模型时问答不可用, 其余页面不受影响
    sys.path.insert(0, str(ROOT))
    try:
        from src.ontology import llm_gate; mcfg = jload(P.MODELS_JSON)
        pr = llm_gate.probe(mcfg["tiers"]["default_qa"]["model"], timeout=90); row("本机模型 Ollama 探针 (默认档)", pr["ok"], f"{pr.get('latency_s')} s, digest {pr.get('digest')}" if pr["ok"] else pr.get("err", "") + " (启动 Ollama 并按 configs/models.json 的 pull_commands 拉模型)")
        try:
            with urllib.request.urlopen(f"http://{c['host']}:{c['ollama']}/api/tags", timeout=10) as r: have = {m["name"] for m in json.load(r).get("models", [])}
            for tier, t in mcfg["tiers"].items():
                name = t["model"]; ok_ = name in have or (name + ":latest") in have; print(f"  [{'OK' if ok_ else '--'}] 模型档 {tier}: {name}{'' if ok_ else ' (未下载: ' + 'ollama pull ' + name + ')'}")
        except Exception as ex: print(f"  [--] 模型清单不可读: {ex.__class__.__name__}")
        st = llm_gate.state(); print(f"  [{'FAIL' if st['open'] else 'OK'}] 模型可用性检查: {'已熔断 (连续失败后自动停用, 60 秒后重试)' if st['open'] else '正常'}; 审计日志 {st['audit']}")
    except Exception as ex: row("模型可用性检查", False, f"{ex.__class__.__name__}: {str(ex)[:100]}")
    print("结论:", "全绿, 可 serve" if ok else "有 FAIL 项, 先按提示处理"); return 0 if ok else 2


def cmd_serve(c):
    from src import logfile as _lf; _lf.prefix_stdout('serve')      # 统一日志格式 (用户令 2)
    RUN.mkdir(exist_ok=True); LOGS.mkdir(exist_ok=True); pids = jload(PIDS) if PIDS.exists() else {}; e = env(c)
    for name, port, cmd in services(c):
        if up(c["host"], port): print(f"  {name} {port} 已在运行 (复用)"); continue
        if name == "gateway": time.sleep(1.5)
        pid = spawn(cmd, LOGS / f"{name}.log", e); pids[name] = {"pid": pid, "port": port}; print(f"  {name} {port} 启动 pid {pid}")
    PIDS.write_text(json.dumps(pids, ensure_ascii=False, indent=1), encoding="utf-8")
    url = f"http://{c['host']}:{c['gateway']}/healthz"
    for _ in range(40):
        try:
            with urllib.request.urlopen(url, timeout=20) as r: d = json.load(r); break
        except Exception: time.sleep(1)
    else: print("网关 healthz 60 s 内未就绪, 看 logs/gateway.log"); return 2
    bad = [m for m in d["modules"] if not m["ok"]]; print(f"入口 http://{c['host']}:{c['gateway']}/  状态 {d['status']} {d['mode']}  模块 {len(d['modules']) - len(bad)}/{len(d['modules'])} ok")
    hp = public_host(c)
    if hp not in ("127.0.0.1", "localhost"):
        ips = lan_ips()
        print(f"  对外监听: 网关绑在 {hp}:{c['gateway']}（组件仍在 {c['host']}）"
              + (f" ⇒ 别的机器可开 http://{ips[0]}:{c['gateway']}/" if ips else ""))
        print("  ★页面没有鉴权: 对外暴露 = 全场运行数据对同网段/公网可见。请在防火墙侧限来源 IP, "
              "或改 public_host 回 127.0.0.1（只本机 + SSH 隧道）")
    for m in bad: print(f"  [DOWN] {m['path']} {m['name']} (看 logs/)")
    return 0 if not bad else 1


def cmd_status(c):
    pids = jload(PIDS) if PIDS.exists() else {}
    for name, port, _ in services(c):
        p = pids.get(name, {}).get("pid"); print(f"  {name:8s} 端口 {port} {'监听中' if up(c['host'], port) else '未监听'}  pid {p if p else '-'} {'存活' if p and alive(p) else ''}")
    return 0


def cmd_stop(c):
    if not PIDS.exists(): print("没有 pids.json (不是本启动器起的, 或已停)"); return 0
    pids = jload(PIDS)
    for name, v in pids.items(): kill(v["pid"]); print(f"  停 {name} pid {v['pid']}")
    time.sleep(1.5); left = [n for n, p, _ in services(c) if up(c["host"], p)]; PIDS.unlink()
    print("全部已停" if not left else f"仍在监听 (可能是别的实例): {left}"); return 0


def cmd_version(c):
    """版本号 / 规则 / 版本记录 (用户令 2026-09-17: 打包版本管理)。"""
    from src import version as V
    print(f"{V.NAME}  v{V.VERSION}  ({V.EDITION})")
    for ln in V.rule_lines():          # 规则 + 例 + 包名（措辞与 README/版本记录同一份）
        print("  " + ln)
    print("  版本表 (完整见 docs/版本记录.md):")
    for h in V.HISTORY:
        print(f"    v{h.get('version', '?'):<8} {h.get('date', '?')}  [{h.get('level', '?')}]  {h.get('title', '')[:60]}")
    return 0


def main():
    a = sys.argv[1] if len(sys.argv) > 1 else "check"; c = cfg()
    if a == "open": url = f"http://{c['host']}:{c['gateway']}/"; print(url); webbrowser.open(url); return 0
    if a == "qa": return subprocess.call([py(c), str(ROOT / "scripts/llm_known_answer.py"), "--port", str(c["detail"])] + sys.argv[2:], cwd=str(ROOT))
    return {"check": cmd_check, "serve": cmd_serve, "status": cmd_status, "stop": cmd_stop,
            "version": cmd_version}.get(a, lambda c: (print(__doc__), 2)[1])(c)


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