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."""
from __future__ import annotations
import json, os, 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", "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)
}


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 services(c):
    P = py(c); h = c["host"]
    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"]),
        ("gateway", c["gateway"], [P, "scripts/guanlan_gateway.py", "--host", h, "--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-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("入口脚本编码守则 (install.ps1 = UTF-8 BOM + CRLF; .bat = CRLF 无 BOM)", not _enc,
            "; ".join(_enc[:2]) or "编码都对 (PS 5.1 才不会把中文按 GBK 解错)")
    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}")
    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")
    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 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}.get(a, lambda c: (print(__doc__), 2)[1])(c)


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