#!/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" 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 = ROOT / "configs/serve.json" 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, "-m", "http.server", str(c["viewer"]), "--bind", h, "--directory", str(ROOT / c["viewer_dir"])]), ("sim_sys", c["sim_sys"], [P, "-u", str(ROOT / c["release_dir"] / "sim_sys_server.py")]), ("sim", c["sim"], [P, "-m", "http.server", str(c["sim"]), "--bind", h, "--directory", str(ROOT / c["sim_dir"])]), ("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/start_hidden.vbs/install.* 引用的每个文件都必须在位, 缺一个就 FAIL。 try: from src.entry_refs import 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→start_hidden.vbs 等)", not _miss, "; ".join(f"{e} 缺 {r}" for e, r in _miss[:3]) or "每个入口引用的文件都在位") 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(ROOT / "configs/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): 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())